1import { assert, it } from "@effect/vitest";
2import {
3 MessageId,
4 NodeId,
5 ProjectId,
6 ProviderDriverKind,
7 ProviderInstanceId,
8 ProviderSessionId,
9 ProviderThreadId,
10 ProviderTurnId,
11 RunAttemptId,
12 RunId,
13 RuntimeRequestId,
14 ThreadId,
15 type OrchestrationV2Command,
16 type OrchestrationV2ThreadProjection,
17} from "@t3tools/contracts";
18import * as DateTime from "effect/DateTime";
19import * as Effect from "effect/Effect";
20import * as Layer from "effect/Layer";
21import * as ServerSettings from "../serverSettings.ts";
22import { restartContinuationRun, continueRestartedRun } from "./RestartContinuation.ts";
23import * as ThreadManagementService from "./ThreadManagementService.ts";
24import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts";
25import * as ProjectionStore from "./ProjectionStore.ts";
26import * as EventSink from "./EventSink.ts";
27import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
28import * as EffectWorker from "./EffectWorker.ts";
29import * as EffectOutbox from "./EffectOutbox.ts";
30
31const threadId = ThreadId.make("thread:restart");
32const runId = RunId.make("run:restart");
33const instanceId = ProviderInstanceId.make("codex");
34const driver = ProviderDriverKind.make("codex");
35const providerThreadId = ProviderThreadId.make("provider-thread:restart");
36const sessionId = ProviderSessionId.make("session:restart");
37const attemptId = RunAttemptId.make("attempt:restart");
38// "No project" threads belong to the environment's Scratch project.
39const scratchProjectId = ProjectId.make("project:scratch");
40const now = DateTime.makeUnsafe("2026-10-03T12:00:00.000Z");
41
42function makeProjection() {
43 return {
44 thread: {
45 id: threadId,
46 projectId: ProjectId.make("restart-project"),
47 providerInstanceId: instanceId,
48 archivedAt: null,
49 deletedAt: null,
50 },
51 runs: [
52 {
53 id: runId,
54 ordinal: 1,
55 providerInstanceId: instanceId,
56 modelSelection: { instanceId, model: "gpt-6" },
57 providerThreadId,
58 activeAttemptId: attemptId,
59 status: "running",
60 },
61 ],
62 providerThreads: [
63 {
64 id: providerThreadId,
65 appThreadId: threadId,
66 ownerNodeId: null,
67 driver,
68 providerInstanceId: instanceId,
69 providerSessionId: sessionId,
70 nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" },
71 status: "active",
72 },
73 ],
74 providerSessions: [
75 { id: sessionId, driver, providerInstanceId: instanceId, status: "running" },
76 ],
77 providerTurns: [
78 {
79 id: ProviderTurnId.make("turn:restart"),
80 providerThreadId,
81 runAttemptId: attemptId,
82 status: "running",
83 },
84 ],
85 runtimeRequests: [],
86 attempts: [],
87 nodes: [],
88 subagents: [],
89 messages: [],
90 turnItems: [],
91 } as unknown as OrchestrationV2ThreadProjection;
92}
93
94it("requires matching saved native state for an unfinished root run", () => {
95 const projection = makeProjection();
96 assert.equal(restartContinuationRun(projection, now)?.id, runId);
97 for (const invalid of [
98 { ...projection, thread: { ...projection.thread, archivedAt: {} } },
99 { ...projection, thread: { ...projection.thread, deletedAt: {} } },
100 {
101 ...projection,
102 thread: { ...projection.thread, providerInstanceId: ProviderInstanceId.make("other") },
103 },
104 {
105 ...projection,
106 providerThreads: [{ ...projection.providerThreads[0]!, nativeThreadRef: null }],
107 },
108 {
109 ...projection,
110 providerSessions: [
111 {
112 ...projection.providerSessions[0]!,
113 providerInstanceId: ProviderInstanceId.make("other"),
114 },
115 ],
116 },
117 { ...projection, providerTurns: [] },
118 ...[
119 "queued",
120 "preparing",
121 "starting",
122 "waiting",
123 "completed",
124 "cancelled",
125 "failed",
126 "interrupted",
127 ].map((status) => ({ ...projection, runs: [{ ...projection.runs[0]!, status }] })),
128 ])
129 assert.isUndefined(restartContinuationRun(invalid as OrchestrationV2ThreadProjection, now));
130});
131
132it("continues a live turn whose session the adapter never marked running", () => {
133 const projection = makeProjection();
134 // Codex, Claude, Cursor and ACP sessions stay "ready" for their whole life.
135 const withSessionStatus = (status: string) =>
136 ({
137 ...projection,
138 providerSessions: [{ ...projection.providerSessions[0]!, status }],
139 }) as OrchestrationV2ThreadProjection;
140 for (const status of ["starting", "ready", "running", "waiting"])
141 assert.equal(restartContinuationRun(withSessionStatus(status), now)?.id, runId, status);
142 for (const status of ["stopped", "error"])
143 assert.isUndefined(restartContinuationRun(withSessionStatus(status), now), status);
144});
145
146it("recovers an admitted continuation after another crash before provider start", () => {
147 const projection = makeProjection();
148 const starting = {
149 ...projection,
150 runs: [
151 {
152 ...projection.runs[0]!,
153 status: "starting" as const,
154 restartContinuationOfRunId: RunId.make("run:previous-crash"),
155 },
156 ],
157 providerThreads: [{ ...projection.providerThreads[0]!, status: "idle" as const }],
158 providerSessions: [{ ...projection.providerSessions[0]!, status: "stopped" as const }],
159 providerTurns: [],
160 };
161 assert.equal(restartContinuationRun(starting, now)?.id, runId);
162});
163
164it("does not continue settled root runs with restart-cancelled background work", () => {
165 const projection = makeProjection();
166 const settled = {
167 ...projection,
168 runs: [
169 {
170 ...projection.runs[0]!,
171 status: "completed" as const,
172 restartCancelledBackgroundWork: [{ kind: "shell" as const, label: "sleep 25" }],
173 },
174 ],
175 providerThreads: [{ ...projection.providerThreads[0]!, status: "idle" as const }],
176 providerSessions: [{ ...projection.providerSessions[0]!, status: "stopped" as const }],
177 providerTurns: [],
178 };
179 for (const projectId of [scratchProjectId, projection.thread.projectId])
180 for (const status of ["completed", "waiting"] as const)
181 assert.isUndefined(
182 restartContinuationRun(
183 {
184 ...settled,
185 thread: { ...settled.thread, projectId },
186 runs: [{ ...settled.runs[0]!, status }],
187 },
188 now,
189 ),
190 );
191});
192
193it.effect.each(["completed", "waiting", "cancelled"] as const)(
194 "ignores a pending restart continuation for a %s run whose provider turn settled",
195 (status) =>
196 Effect.gen(function* () {
197 const base = makeProjection();
198 const work = [{ kind: "shell" as const, label: "sleep 25 && echo DONE" }];
199 const projection = {
200 ...base,
201 runs: [{ ...base.runs[0]!, status, restartCancelledBackgroundWork: work }],
202 providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }],
203 } as unknown as OrchestrationV2ThreadProjection;
204 const commands: Parameters<
205 ThreadManagementService.ThreadManagementService["Service"]["dispatch"]
206 >[0][] = [];
207 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
208 Effect.provide(
209 Layer.merge(
210 Layer.mock(ThreadManagementService.ThreadManagementService)({
211 getThreadRecords: () => Effect.succeed(projection),
212 recoverDelegatedTask: () => Effect.void,
213 dispatch: (command) => {
214 commands.push(command);
215 return Effect.succeed({} as never);
216 },
217 }),
218 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
219 ),
220 ),
221 );
222 assert.lengthOf(commands, 0);
223 }),
224);
225
226it.effect.each(["completed", "waiting", "cancelled"] as const)(
227 "wakes a %s run whose monitor the restart cancelled, with the note as its prompt",
228 (status) =>
229 Effect.gen(function* () {
230 const base = makeProjection();
231 const work = [
232 { kind: "monitor" as const, label: "Watch CI run 42" },
233 { kind: "shell" as const, label: "vp run dev" },
234 ];
235 const projection = {
236 ...base,
237 runs: [{ ...base.runs[0]!, status, restartCancelledBackgroundWork: work }],
238 providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }],
239 } as unknown as OrchestrationV2ThreadProjection;
240 const commands: Parameters<
241 ThreadManagementService.ThreadManagementService["Service"]["dispatch"]
242 >[0][] = [];
243 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
244 Effect.provide(
245 Layer.merge(
246 Layer.mock(ThreadManagementService.ThreadManagementService)({
247 getThreadRecords: () => Effect.succeed(projection),
248 recoverDelegatedTask: () => Effect.void,
249 dispatch: (command) => {
250 commands.push(command);
251 return Effect.succeed({} as never);
252 },
253 }),
254 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
255 ),
256 ),
257 );
258 assert.lengthOf(commands, 1);
259 const command = commands[0]!;
260 assert.equal(command.type, "message.dispatch");
261 if (command.type !== "message.dispatch") return;
262 assert.equal(command.restartContinuationOfRunId, runId);
263 assert.include(command.text, "- monitor: Watch CI run 42");
264 assert.include(command.text, "- shell: vp run dev");
265 }),
266);
267
268it.effect("does not continue a failed run that lost background work", () =>
269 Effect.gen(function* () {
270 const base = makeProjection();
271 const projection = {
272 ...base,
273 runs: [
274 {
275 ...base.runs[0]!,
276 status: "failed",
277 restartCancelledBackgroundWork: [{ kind: "shell" as const, label: "sleep 25" }],
278 },
279 ],
280 providerTurns: [{ ...base.providerTurns[0]!, status: "failed" }],
281 } as unknown as OrchestrationV2ThreadProjection;
282 const commands: Parameters<
283 ThreadManagementService.ThreadManagementService["Service"]["dispatch"]
284 >[0][] = [];
285 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
286 Effect.provide(
287 Layer.merge(
288 Layer.mock(ThreadManagementService.ThreadManagementService)({
289 getThreadRecords: () => Effect.succeed(projection),
290 recoverDelegatedTask: () => Effect.void,
291 dispatch: (command) => {
292 commands.push(command);
293 return Effect.succeed({} as never);
294 },
295 }),
296 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
297 ),
298 ),
299 );
300 assert.lengthOf(commands, 0);
301 }),
302);
303
304it.effect.each([
305 [false, undefined],
306 [true, undefined],
307 [false, true],
308 [true, false],
309] as const)(
310 "atomically records restart intent with cancellation when opt-in is %s and project override is %s",
311 ([enabled, projectOverride]) =>
312 Effect.gen(function* () {
313 let committed: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | undefined;
314 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
315 Effect.provide(
316 Layer.mergeAll(
317 ServerSettings.layerTest({
318 continueThreadsAfterServerUpdate: enabled,
319 projectSettingsOverrides:
320 projectOverride === undefined
321 ? {}
322 : {
323 [ProjectId.make("restart-project")]: {
324 continueThreadsAfterServerUpdate: projectOverride,
325 },
326 },
327 }),
328 Layer.mock(ProjectionStore.ProjectionStoreV2)({
329 getRecoveryThreadIds: () => Effect.succeed([threadId]),
330 getRuntimeRecoveryProjection: () => Effect.succeed(makeProjection()),
331 }),
332 Layer.mock(EventSink.EventSinkV2)({
333 commitCommand: (input) => {
334 committed = input;
335 return Effect.succeed({ committed: true, cancelledEffectCount: 1 } as never);
336 },
337 }),
338 IdAllocator.layer,
339 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
340 runRecoveryOnce: Effect.succeed(false),
341 }),
342 Layer.mock(EffectOutbox.EffectOutboxV2)({
343 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
344 }),
345 ),
346 ),
347 );
348 yield* recovery.reconcile("startup");
349 assert.isDefined(committed);
350 assert.isTrue(
351 committed!.events.some(
352 (event) => event.type === "run.updated" && event.payload.status === "cancelled",
353 ),
354 );
355 assert.lengthOf(committed!.effects, (projectOverride ?? enabled) ? 1 : 0);
356 if (projectOverride ?? enabled)
357 assert.deepEqual(committed!.effects[0]?.request, {
358 type: "provider-runtime.continue",
359 sourceRunId: runId,
360 });
361 }),
362);
363
364it.effect("lets a cut blocking delegate_task wake its parent like an async task", () =>
365 Effect.gen(function* () {
366 const base = makeProjection();
367 const task = (id: string, completionWake: "settled_only" | "always", status: string) => ({
368 id: NodeId.make(id),
369 threadId,
370 runId,
371 origin: "app_owned",
372 childThreadId: ThreadId.make(`thread:${id}`),
373 driver,
374 providerInstanceId: instanceId,
375 completionWake,
376 status,
377 });
378 const projection = {
379 ...base,
380 subagents: [
381 task("task:waited", "settled_only", "running"),
382 task("task:async", "always", "running"),
383 task("task:finished", "settled_only", "completed"),
384 ],
385 } as unknown as OrchestrationV2ThreadProjection;
386 let committed: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | undefined;
387 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
388 Effect.provide(
389 Layer.mergeAll(
390 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
391 Layer.mock(ProjectionStore.ProjectionStoreV2)({
392 getRecoveryThreadIds: () => Effect.succeed([threadId]),
393 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
394 }),
395 Layer.mock(EventSink.EventSinkV2)({
396 commitCommand: (input) => {
397 committed = input;
398 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
399 },
400 }),
401 IdAllocator.layer,
402 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}),
403 Layer.mock(EffectOutbox.EffectOutboxV2)({
404 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
405 }),
406 ),
407 ),
408 );
409 yield* recovery.reconcile("startup");
410 const upgraded = (committed?.events ?? []).flatMap((event) =>
411 event.type === "subagent.updated" ? [event.payload] : [],
412 );
413 assert.deepEqual(
414 upgraded.map((row) => ({ id: row.id, completionWake: row.completionWake })),
415 [{ id: NodeId.make("task:waited"), completionWake: "always" }],
416 );
417 }),
418);
419
420it.effect("does not duplicate delivery and yields to newer user work or opt-out", () =>
421 Effect.gen(function* () {
422 let projection = makeProjection();
423 projection = { ...projection, runs: [{ ...projection.runs[0]!, status: "cancelled" }] };
424 const commands: Parameters<
425 ThreadManagementService.ThreadManagementService["Service"]["dispatch"]
426 >[0][] = [];
427 const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({
428 getThreadRecords: () => Effect.succeed(projection),
429 recoverDelegatedTask: () => Effect.void,
430 dispatch: (command) => {
431 commands.push(command);
432 if (command.type === "message.dispatch")
433 projection = { ...projection, messages: [{ id: command.messageId } as never] };
434 return Effect.succeed({} as never);
435 },
436 });
437 const layerEnabled = Layer.merge(
438 layerThreads,
439 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
440 );
441 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
442 Effect.provide(layerEnabled),
443 );
444 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
445 Effect.provide(layerEnabled),
446 );
447 assert.lengthOf(commands, 1);
448 assert.match(String(commands[0]!.commandId), /run:restart$/);
449 if (commands[0]!.type === "message.dispatch")
450 assert.equal(commands[0]!.restartContinuationOfRunId, runId);
451 projection = {
452 ...projection,
453 messages: [],
454 runs: [
455 ...projection.runs,
456 {
457 ...projection.runs[0]!,
458 id: RunId.make("run:user-newer"),
459 ordinal: 2,
460 status: "completed",
461 },
462 ],
463 };
464 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
465 Effect.provide(layerEnabled),
466 );
467 assert.lengthOf(commands, 1);
468 projection = { ...projection, runs: [projection.runs[0]!] };
469 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
470 Effect.provide(
471 Layer.merge(
472 layerThreads,
473 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: false }),
474 ),
475 ),
476 );
477 assert.lengthOf(commands, 1);
478 }),
479);
480
481it.effect("prepares no continuation for background work another provider thread launched", () =>
482 Effect.gen(function* () {
483 const base = makeProjection();
484 const claudeThreadId = ProviderThreadId.make("provider-thread:claude");
485 // A Claude turn left a background subagent open, then the thread switched
486 // to Codex and its turn settled. Recovery would cancel Claude's work.
487 const projection = {
488 ...base,
489 runs: [{ ...base.runs[0]!, status: "completed" }],
490 providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }],
491 turnItems: [
492 {
493 id: "turn-item:claude-subagent",
494 runId: RunId.make("run:claude"),
495 providerThreadId: claudeThreadId,
496 type: "subagent",
497 status: "running",
498 },
499 ],
500 } as unknown as OrchestrationV2ThreadProjection;
501 const writes: Parameters<EventSink.EventSinkV2["Service"]["writeWithEffects"]>[0][] = [];
502 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
503 Effect.provide(
504 Layer.mergeAll(
505 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
506 Layer.mock(ProjectionStore.ProjectionStoreV2)({
507 getRecoveryThreadIds: () => Effect.succeed([threadId]),
508 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
509 }),
510 Layer.mock(EventSink.EventSinkV2)({
511 writeWithEffects: (input) =>
512 Effect.sync(() => {
513 writes.push(input);
514 return [];
515 }),
516 }),
517 IdAllocator.layer,
518 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}),
519 Layer.mock(EffectOutbox.EffectOutboxV2)({}),
520 ),
521 ),
522 );
523 yield* recovery.prepareForShutdown;
524 assert.lengthOf(writes, 0);
525 // The same work on the settled run's own provider thread does wake it.
526 const ownWork = {
527 ...projection,
528 turnItems: [{ ...projection.turnItems[0]!, providerThreadId }],
529 } as OrchestrationV2ThreadProjection;
530 const ownRecovery = yield* ProviderRuntimeRecovery.make.pipe(
531 Effect.provide(
532 Layer.mergeAll(
533 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
534 Layer.mock(ProjectionStore.ProjectionStoreV2)({
535 getRecoveryThreadIds: () => Effect.succeed([threadId]),
536 getRuntimeRecoveryProjection: () => Effect.succeed(ownWork),
537 }),
538 Layer.mock(EventSink.EventSinkV2)({
539 writeWithEffects: (input) =>
540 Effect.sync(() => {
541 writes.push(input);
542 return [];
543 }),
544 }),
545 IdAllocator.layer,
546 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}),
547 Layer.mock(EffectOutbox.EffectOutboxV2)({}),
548 ),
549 ),
550 );
551 yield* ownRecovery.prepareForShutdown;
552 assert.lengthOf(writes, 1);
553 assert.deepEqual(writes[0]!.effects[0]?.request, {
554 type: "provider-runtime.continue",
555 sourceRunId: runId,
556 });
557 }),
558);
559
560const backgroundSubagent = {
561 turnItem: {
562 id: "turn-item:background-subagent",
563 runId,
564 nodeId: null,
565 providerThreadId,
566 providerInstanceId: instanceId,
567 type: "subagent",
568 subagentId: "subagent:background",
569 title: "Background reviewer",
570 status: "running",
571 },
572 work: {
573 kind: "subagent" as const,
574 label: "Background reviewer",
575 id: "turn-item:background-subagent",
576 },
577};
578const devServer = {
579 turnItem: {
580 id: "turn-item:dev-server",
581 runId,
582 nodeId: null,
583 providerThreadId,
584 providerInstanceId: instanceId,
585 type: "command_execution",
586 input: "vp run dev",
587 status: "running",
588 },
589 work: { kind: "shell" as const, label: "vp run dev", id: "turn-item:dev-server" },
590};
591
592it.effect.each([
593 ["completed", scratchProjectId, backgroundSubagent, true],
594 ["waiting", ProjectId.make("restart-project"), backgroundSubagent, true],
595 ["completed", ProjectId.make("restart-project"), devServer, false],
596 ["waiting", scratchProjectId, devServer, false],
597] as const)(
598 "cleans up background work of a %s run in project %s and wakes it only for awaited work",
599 ([status, projectId, background, wakes]) =>
600 Effect.gen(function* () {
601 const base = makeProjection();
602 const projection = {
603 ...base,
604 thread: { ...base.thread, projectId },
605 runs: [{ ...base.runs[0]!, status }],
606 providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }],
607 turnItems: [background.turnItem],
608 subagents:
609 background === backgroundSubagent
610 ? [
611 {
612 id: "subagent:background",
613 runId,
614 origin: "provider_native",
615 driver,
616 providerInstanceId: instanceId,
617 status: "running",
618 },
619 ]
620 : [],
621 } as unknown as OrchestrationV2ThreadProjection;
622 for (const trigger of ["startup", "shutdown"] as const) {
623 const commits: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0][] = [];
624 const writes: Parameters<EventSink.EventSinkV2["Service"]["writeWithEffects"]>[0][] = [];
625 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
626 Effect.provide(
627 Layer.mergeAll(
628 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
629 Layer.mock(ProjectionStore.ProjectionStoreV2)({
630 getRecoveryThreadIds: () => Effect.succeed([threadId]),
631 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
632 }),
633 Layer.mock(EventSink.EventSinkV2)({
634 writeWithEffects: (input) =>
635 Effect.sync(() => {
636 writes.push(input);
637 return [];
638 }),
639 commitCommand: (input) =>
640 Effect.sync(() => {
641 commits.push(input);
642 return { committed: true, cancelledEffectCount: 0 } as never;
643 }),
644 }),
645 IdAllocator.layer,
646 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
647 runRecoveryOnce: Effect.succeed(false),
648 }),
649 Layer.mock(EffectOutbox.EffectOutboxV2)({
650 listByCommandId: () => Effect.succeed([]),
651 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
652 }),
653 ),
654 ),
655 );
656 if (trigger === "shutdown") yield* recovery.prepareForShutdown;
657 yield* recovery.reconcile(trigger);
658 // Shutdown records the wake before it cancels the work; a startup
659 // after a kill records it with the cancellation.
660 const recorded = [...writes.flatMap((write) => write.effects), ...commits[0]!.effects];
661 assert.deepEqual(
662 recorded.map((effect) => effect.request),
663 wakes ? [{ type: "provider-runtime.continue", sourceRunId: runId }] : [],
664 );
665 assert.lengthOf(commits, 1);
666 const events = commits[0]!.events;
667 assert.isTrue(
668 events.some(
669 (event) => event.type === "turn-item.updated" && event.payload.status === "cancelled",
670 ),
671 );
672 const note = events.find((event) => event.type === "run.background-work-cancelled");
673 assert.equal(note?.runId, runId);
674 assert.deepEqual(note?.payload.restartCancelledBackgroundWork, [background.work]);
675 if (status === "completed")
676 assert.isFalse(events.some((event) => event.type === "run.updated"));
677 }
678 }),
679);
680
681it("does not wake a settled run for work left by a session that already stopped", () => {
682 const base = makeProjection();
683 const projection = {
684 ...base,
685 runs: [{ ...base.runs[0]!, status: "completed" }],
686 providerThreads: [{ ...base.providerThreads[0]!, status: "idle" }],
687 turnItems: [backgroundSubagent.turnItem],
688 } as unknown as OrchestrationV2ThreadProjection;
689 assert.equal(restartContinuationRun(projection, now)?.id, runId);
690 for (const status of ["stopped", "error"])
691 assert.isUndefined(
692 restartContinuationRun(
693 {
694 ...projection,
695 providerSessions: [{ ...projection.providerSessions[0]!, status }],
696 } as OrchestrationV2ThreadProjection,
697 now,
698 ),
699 status,
700 );
701});
702
703it.effect("does not cancel or resume a run that completes while shutdown intent commits", () =>
704 Effect.gen(function* () {
705 let projection = makeProjection();
706 const commits: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0][] = [];
707 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
708 Effect.provide(
709 Layer.mergeAll(
710 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
711 Layer.mock(ProjectionStore.ProjectionStoreV2)({
712 getRecoveryThreadIds: () => Effect.succeed([threadId]),
713 getRuntimeRecoveryProjection: () => Effect.sync(() => projection),
714 }),
715 Layer.mock(EventSink.EventSinkV2)({
716 writeWithEffects: (input) =>
717 Effect.sync(() => {
718 assert.lengthOf(input.events, 0);
719 assert.equal(input.effects[0]?.request.type, "provider-runtime.continue");
720 projection = {
721 ...projection,
722 runs: [{ ...projection.runs[0]!, status: "completed" }],
723 };
724 return [];
725 }),
726 commitCommand: (input) =>
727 Effect.sync(() => {
728 commits.push(input);
729 return { committed: true, cancelledEffectCount: 0 } as never;
730 }),
731 }),
732 IdAllocator.layer,
733 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
734 runRecoveryOnce: Effect.succeed(false),
735 }),
736 Layer.mock(EffectOutbox.EffectOutboxV2)({
737 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
738 }),
739 ),
740 ),
741 );
742 yield* recovery.prepareForShutdown;
743 yield* recovery.reconcile("shutdown");
744 assert.isFalse(
745 commits.some((commit) => commit.events.some((event) => event.type === "run.updated")),
746 );
747 let dispatched = false;
748 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
749 Effect.provide(
750 Layer.mergeAll(
751 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
752 Layer.mock(ThreadManagementService.ThreadManagementService)({
753 getThreadRecords: () => Effect.succeed(projection),
754 recoverDelegatedTask: () => Effect.void,
755 dispatch: () =>
756 Effect.sync(() => {
757 dispatched = true;
758 return {} as never;
759 }),
760 }),
761 ),
762 ),
763 );
764 assert.isFalse(dispatched);
765 }),
766);
767
768const continuationDispatches = (projection: OrchestrationV2ThreadProjection) =>
769 Effect.gen(function* () {
770 const dispatches: Array<Extract<OrchestrationV2Command, { type: "message.dispatch" }>> = [];
771 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
772 Effect.provide(
773 Layer.merge(
774 Layer.mock(ThreadManagementService.ThreadManagementService)({
775 getThreadRecords: () => Effect.succeed(projection),
776 recoverDelegatedTask: () => Effect.void,
777 dispatch: (command) => {
778 if (command.type === "message.dispatch") dispatches.push(command);
779 return Effect.succeed({} as never);
780 },
781 }),
782 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
783 ),
784 ),
785 );
786 return dispatches;
787 });
788
789const continuationTexts = (projection: OrchestrationV2ThreadProjection) =>
790 continuationDispatches(projection).pipe(
791 Effect.map((dispatches) => dispatches.map((command) => command.text)),
792 );
793
794const cutMidTurn = (extra: Record<string, unknown> = {}) => {
795 const base = makeProjection();
796 return {
797 ...base,
798 runs: [
799 {
800 ...base.runs[0]!,
801 status: "cancelled",
802 userMessageId: MessageId.make("message:user"),
803 ...extra,
804 },
805 ],
806 providerTurns: [{ ...base.providerTurns[0]!, status: "cancelled" }],
807 } as unknown as OrchestrationV2ThreadProjection;
808};
809
810const queuedFollowUp = {
811 id: RunId.make("run:queued"),
812 ordinal: 2,
813 providerInstanceId: instanceId,
814 providerThreadId,
815 status: "queued",
816 queueHeld: true,
817};
818
819it.effect("continues a cut run past queued follow-ups, which stay held", () =>
820 Effect.gen(function* () {
821 const live = makeProjection();
822 assert.equal(
823 restartContinuationRun(
824 {
825 ...live,
826 runs: [...live.runs, queuedFollowUp],
827 } as unknown as OrchestrationV2ThreadProjection,
828 now,
829 )?.id,
830 runId,
831 );
832 const cut = cutMidTurn();
833 const texts = yield* continuationTexts({
834 ...cut,
835 runs: [...cut.runs, queuedFollowUp],
836 } as unknown as OrchestrationV2ThreadProjection);
837 assert.deepEqual(texts, ["Continue where you left off."]);
838 }),
839);
840
841it.effect("continues a resumed queued run that ran after an earlier continuation", () =>
842 Effect.gen(function* () {
843 // The continuation (ordinal 3) ran ahead of the held queue, then the user
844 // resumed the queued run (ordinal 1 here) and the server restarted again.
845 const finishedContinuation = {
846 id: RunId.make("run:earlier-continuation"),
847 ordinal: 3,
848 providerInstanceId: instanceId,
849 providerThreadId,
850 status: "completed",
851 completedAt: DateTime.makeUnsafe("2026-10-03T10:00:00.000Z"),
852 };
853 const live = makeProjection();
854 assert.equal(
855 restartContinuationRun(
856 {
857 ...live,
858 runs: [...live.runs, finishedContinuation],
859 } as unknown as OrchestrationV2ThreadProjection,
860 now,
861 )?.id,
862 runId,
863 );
864 const cut = cutMidTurn({ completedAt: DateTime.makeUnsafe("2026-10-03T10:05:00.000Z") });
865 const texts = yield* continuationTexts({
866 ...cut,
867 runs: [...cut.runs, finishedContinuation],
868 } as unknown as OrchestrationV2ThreadProjection);
869 assert.deepEqual(texts, ["Continue where you left off."]);
870 }),
871);
872
873it.effect("does not continue a run the user asked to stop before the restart", () =>
874 Effect.gen(function* () {
875 const texts = yield* continuationTexts({
876 ...cutMidTurn(),
877 turnItems: [{ id: "turn-item:interrupt", runId, type: "run_interrupt_request" }],
878 } as unknown as OrchestrationV2ThreadProjection);
879 assert.deepEqual(texts, []);
880 }),
881);
882
883it.effect("does not continue a cut /compact or /logout turn", () =>
884 Effect.gen(function* () {
885 for (const text of ["/compact", " /LOGOUT "]) {
886 const texts = yield* continuationTexts({
887 ...cutMidTurn(),
888 messages: [{ id: MessageId.make("message:user"), text, attachments: [] }],
889 } as unknown as OrchestrationV2ThreadProjection);
890 assert.deepEqual(texts, [], text);
891 }
892 const texts = yield* continuationTexts({
893 ...cutMidTurn(),
894 messages: [{ id: MessageId.make("message:user"), text: "/compact later", attachments: [] }],
895 } as unknown as OrchestrationV2ThreadProjection);
896 assert.lengthOf(texts, 1);
897 }),
898);
899
900it.effect("tells a turn cut mid-way about the background work it lost", () =>
901 Effect.gen(function* () {
902 const texts = yield* continuationTexts(
903 cutMidTurn({
904 restartCancelledBackgroundWork: [{ kind: "subagent", label: "Background reviewer" }],
905 }),
906 );
907 assert.lengthOf(texts, 1);
908 assert.include(texts[0]!, "Background reviewer");
909 assert.isTrue(texts[0]!.endsWith("Continue where you left off."));
910 }),
911);
912
913it.effect("shows a continuation as a T3 Code notice whose detail is the prompt", () =>
914 Effect.gen(function* () {
915 const [resumed] = yield* continuationDispatches(cutMidTurn());
916 assert.deepEqual(resumed?.notification, {
917 source: { kind: "system" },
918 outcome: "updated",
919 summary: "T3 Code restarted and resumed this turn",
920 detail: "Continue where you left off.",
921 });
922 }),
923);
924
925it.effect(
926 "carries the note forward when its continuation was cut before reaching the provider",
927 () =>
928 Effect.gen(function* () {
929 const base = makeProjection();
930 const original = {
931 ...base.runs[0]!,
932 id: RunId.make("run:original"),
933 status: "completed",
934 activeAttemptId: RunAttemptId.make("attempt:original"),
935 restartCancelledBackgroundWork: [{ kind: "shell", label: "sleep 25 && echo DONE" }],
936 };
937 const projection = {
938 ...base,
939 runs: [
940 original,
941 {
942 ...base.runs[0]!,
943 ordinal: 2,
944 status: "cancelled",
945 restartContinuationOfRunId: original.id,
946 },
947 ],
948 // The original turn settled; the continuation's attempt never started one.
949 providerTurns: [
950 {
951 ...base.providerTurns[0]!,
952 runAttemptId: original.activeAttemptId,
953 status: "completed",
954 },
955 ],
956 } as unknown as OrchestrationV2ThreadProjection;
957 const texts = yield* continuationTexts(projection);
958 assert.lengthOf(texts, 1);
959 assert.include(texts[0]!, "sleep 25 && echo DONE");
960 assert.notInclude(texts[0]!, "Continue where");
961 }),
962);
963
964it.effect("prepares later threads' continuations when one thread fails", () =>
965 Effect.gen(function* () {
966 const brokenThreadId = ThreadId.make("thread:broken");
967 const writes: Parameters<EventSink.EventSinkV2["Service"]["writeWithEffects"]>[0][] = [];
968 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
969 Effect.provide(
970 Layer.mergeAll(
971 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
972 Layer.mock(ProjectionStore.ProjectionStoreV2)({
973 getRecoveryThreadIds: () => Effect.succeed([brokenThreadId, threadId]),
974 getRuntimeRecoveryProjection: (id) =>
975 id === brokenThreadId
976 ? Effect.fail(
977 new ProjectionStore.ProjectionStoreThreadNotFoundError({ threadId: id }),
978 )
979 : Effect.succeed(makeProjection()),
980 }),
981 Layer.mock(EventSink.EventSinkV2)({
982 writeWithEffects: (input) =>
983 Effect.sync(() => {
984 writes.push(input);
985 return [];
986 }),
987 }),
988 IdAllocator.layer,
989 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}),
990 Layer.mock(EffectOutbox.EffectOutboxV2)({}),
991 ),
992 ),
993 );
994 yield* recovery.prepareForShutdown;
995 assert.deepEqual(
996 writes.map((write) => write.effects[0]?.request),
997 [{ type: "provider-runtime.continue", sourceRunId: runId }],
998 );
999 }),
1000);
1001
1002it.effect("stops continuing after three automatic restart continuations in a row", () =>
1003 Effect.gen(function* () {
1004 const first = RunId.make("run:first-continuation");
1005 const second = RunId.make("run:second-continuation");
1006 const cut = cutMidTurn({ restartContinuationOfRunId: second });
1007 const withTwoLinks = {
1008 ...cut,
1009 runs: [
1010 ...cut.runs,
1011 { ...cut.runs[0]!, id: second, restartContinuationOfRunId: first },
1012 { ...cut.runs[0]!, id: first, restartContinuationOfRunId: undefined },
1013 ],
1014 } as unknown as OrchestrationV2ThreadProjection;
1015 assert.lengthOf(yield* continuationTexts(withTwoLinks), 1);
1016
1017 const withThreeLinks = {
1018 ...withTwoLinks,
1019 runs: withTwoLinks.runs.map((run) =>
1020 run.id === first ? { ...run, restartContinuationOfRunId: RunId.make("run:root") } : run,
1021 ),
1022 } as unknown as OrchestrationV2ThreadProjection;
1023 assert.lengthOf(yield* continuationTexts(withThreeLinks), 0);
1024 }),
1025);
1026
1027it.effect.each([
1028 ["user_input", "pending", "live", false],
1029 ["user_input", "expired", "not_resumable", false],
1030 ["user_input", "cancelled", "not_resumable", false],
1031 ["command", "pending", "live", false],
1032 ["command", "expired", "not_resumable", false],
1033 ["user_input", "resolved", "live", true],
1034 ["command", "cancelled", "live", true],
1035 ["user_input", "pending", "message", true],
1036 ["dynamic_tool_call", "pending", "live", true],
1037 ["auth_refresh", "pending", "live", true],
1038] as const)(
1039 "a %s request that is %s with %s capability allows continuation: %s",
1040 ([kind, status, capability, shouldContinue]) =>
1041 Effect.gen(function* () {
1042 const base = makeProjection();
1043 const runtimeRequests = [
1044 {
1045 id: RuntimeRequestId.make("request:restart"),
1046 nodeId: NodeId.make("node:restart-question"),
1047 providerTurnId: base.providerTurns[0]!.id,
1048 kind,
1049 status,
1050 responseCapability:
1051 capability === "live"
1052 ? { type: capability, providerSessionId: sessionId }
1053 : capability === "not_resumable"
1054 ? { type: capability, reason: "The server restarted." }
1055 : { type: capability },
1056 },
1057 ];
1058 const live = { ...base, runtimeRequests } as unknown as OrchestrationV2ThreadProjection;
1059 assert.equal(restartContinuationRun(live, now) !== undefined, shouldContinue);
1060 const cut = {
1061 ...cutMidTurn(),
1062 runtimeRequests,
1063 } as unknown as OrchestrationV2ThreadProjection;
1064 assert.lengthOf(yield* continuationTexts(cut), shouldContinue ? 1 : 0);
1065 }),
1066);
1067
1068it.effect("does not wake a snoozed or settled thread, but continues once a snooze elapsed", () =>
1069 Effect.gen(function* () {
1070 const live = makeProjection();
1071 for (const held of [
1072 { snoozedUntil: DateTime.add(now, { days: 1 }) },
1073 { settledOverride: "settled" as const },
1074 ]) {
1075 const thread = { ...live.thread, ...held };
1076 assert.isUndefined(restartContinuationRun({ ...live, thread }, now));
1077 assert.deepEqual(yield* continuationTexts({ ...cutMidTurn(), thread }), []);
1078 }
1079 const thread = { ...live.thread, snoozedUntil: DateTime.subtract(now, { minutes: 1 }) };
1080 assert.equal(restartContinuationRun({ ...live, thread }, now)?.id, runId);
1081 }),
1082);
1083
1084it.effect("does not continue a run waiting on a secret the restart closed", () =>
1085 Effect.gen(function* () {
1086 const completedAt = DateTime.makeUnsafe("2026-10-03T10:05:00.000Z");
1087 const secret = (secretStatus: string, at: DateTime.Utc | null) =>
1088 ({ type: "secret_request", runId, secretStatus, completedAt: at }) as never;
1089 assert.isUndefined(
1090 restartContinuationRun({ ...makeProjection(), turnItems: [secret("pending", null)] }, now),
1091 );
1092 const cut = cutMidTurn({ completedAt });
1093 assert.deepEqual(
1094 yield* continuationTexts({ ...cut, turnItems: [secret("cancelled", completedAt)] }),
1095 [],
1096 );
1097 // A secret the user dismissed earlier in the run does not hold it.
1098 assert.lengthOf(
1099 yield* continuationTexts({
1100 ...cut,
1101 turnItems: [secret("cancelled", DateTime.subtract(completedAt, { seconds: 1 }))],
1102 }),
1103 1,
1104 );
1105 }),
1106);
1107
1108it.effect.each(["completed", "failed", "interrupted"] as const)(
1109 "does not continue when the provider turn %s before the app run settled",
1110 (status) =>
1111 Effect.gen(function* () {
1112 const cut = cutMidTurn();
1113 assert.deepEqual(
1114 yield* continuationTexts({
1115 ...cut,
1116 providerTurns: [{ ...cut.providerTurns[0]!, status }],
1117 }),
1118 [],
1119 );
1120 }),
1121);