authorgravatar for maria@kuuro.netmaria <maria@kuuro.net> 2026-10-04 08:49:14-03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-04 08:49:14-03:00
log253dea360ef0c743c5de8ba04f8412a7eb4694ac
tree99ff5d3f40f6ce77d123b25e3269df2788a6f781
parentb3b6ae2cebbae39a706038911dbb8c85c7120be7
signature Signed by PGP key B5690EEEBB952194

fix(server): keep settled threads asleep after restarts (#15604)

6 files changed, 305 insertions(+), 271 deletions(-)

apps/server/src/orchestration-v2/Orchestrator.ts+2-2
......@@ -4382,8 +4382,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
43824382 const source = projection.runs.find((run) => run.id === command.restartContinuationOfRunId);
43834383 if (
43844384 !source ||
4385 (source.status !== "cancelled" &&
4386 !isRestartNoteSource(source, projection.providerTurns)) ||
4385 source.status !== "cancelled" ||
4386 isRestartNoteSource(source, projection.providerTurns) ||
43874387 projection.thread.archivedAt !== null ||
43884388 projection.thread.deletedAt !== null ||
43894389 projection.thread.providerInstanceId !== source.providerInstanceId ||
apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts+2-34
......@@ -149,33 +149,6 @@ function resolveStaleBackgroundItemProviderInstanceId(
149149 return projection.providerThreads[0]?.providerInstanceId ?? projection.thread.providerInstanceId;
150150}
151151
152/**
153 * Provider threads with background work reconciliation would cancel and
154 * record for their next turn.
155 */
156function providerThreadsWithOpenBackgroundWork(
157 projection: ProjectionStore.ProjectionRuntimeRecoveryState,
158): ReadonlySet<ProviderThreadId> {
159 const ids = new Set<ProviderThreadId>();
160 for (const item of projection.turnItems ?? []) {
161 if (
162 !isBackgroundCapableTurnItemType(item.type) ||
163 !isNonterminalTurnItemStatus(item.status) ||
164 isAppOwnedDelegationItem(item)
165 )
166 continue;
167 const providerThreadId =
168 item.providerThreadId ??
169 projection.runs.find((run) => run.id === item.runId)?.providerThreadId;
170 if (providerThreadId != null) ids.add(providerThreadId);
171 }
172 for (const thread of projection.providerThreads ?? []) {
173 if (thread.ownerNodeId === null && providerThreadHasPendingBackgroundTasks(thread))
174 ids.add(thread.id);
175 }
176 return ids;
177}
178
179152/**
180153 * A provider thread's latest started run: the last turn that provider saw.
181154 * Restart recovery records the thread's cancelled background work on it, and
......@@ -668,7 +641,7 @@ export const make = Effect.gen(function* () {
668641 }
669642 const continuationRun =
670643 continueAfterRestart && trigger === "startup"
671 ? restartContinuationRun(projection, new Set(cancelledBackgroundWork.keys()))
644 ? restartContinuationRun(projection)
672645 : undefined;
673646 const effects: Array<EffectOutbox.PendingOrchestrationEffectV2> = continuationRun
674647 ? [
......@@ -804,12 +777,7 @@ export const make = Effect.gen(function* () {
804777 .continueThreadsAfterServerUpdate
805778 )
806779 return;
807 // Shutdown reconciliation cancels the background work below, so a
808 // settled thread's continuation must be captured while it is still open.
809 const run = restartContinuationRun(
810 projection,
811 providerThreadsWithOpenBackgroundWork(projection),
812 );
780 const run = restartContinuationRun(projection);
813781 if (!run) return;
814782 const commandId = CommandId.make(`command:restart-prepare:${run.id}`);
815783 yield* eventSink.writeWithEffects({
apps/server/src/orchestration-v2/RestartContinuation.test.ts+144-65
......@@ -32,6 +32,8 @@ const driver = ProviderDriverKind.make("codex");
3232const providerThreadId = ProviderThreadId.make("provider-thread:restart");
3333const sessionId = ProviderSessionId.make("session:restart");
3434const attemptId = RunAttemptId.make("attempt:restart");
35// "No project" threads belong to the environment's Scratch project.
36const scratchProjectId = ProjectId.make("project:scratch");
3537
3638function makeProjection() {
3739 return {
......@@ -155,76 +157,63 @@ it("recovers an admitted continuation after another crash before provider start"
155157 assert.equal(restartContinuationRun(starting)?.id, runId);
156158});
157159
158it("continues a settled root run only when the restart cancelled its background work", () => {
160it("does not continue settled root runs with restart-cancelled background work", () => {
159161 const projection = makeProjection();
160162 const settled = {
161163 ...projection,
162 runs: [{ ...projection.runs[0]!, status: "completed" as const }],
164 runs: [
165 {
166 ...projection.runs[0]!,
167 status: "completed" as const,
168 restartCancelledBackgroundWork: [{ kind: "shell" as const, label: "sleep 25" }],
169 },
170 ],
163171 providerThreads: [{ ...projection.providerThreads[0]!, status: "idle" as const }],
164172 providerSessions: [{ ...projection.providerSessions[0]!, status: "stopped" as const }],
165173 providerTurns: [],
166174 };
167 const lostWork = new Set([providerThreadId]);
168 assert.isUndefined(restartContinuationRun(settled));
169 assert.equal(restartContinuationRun(settled, lostWork)?.id, runId);
170 // Work an older provider thread launched (before a provider switch) is not
171 // this run's: its provider was never told about it and cannot continue it.
172 assert.isUndefined(
173 restartContinuationRun(settled, new Set([ProviderThreadId.make("provider-thread:claude")])),
174 );
175 for (const invalid of [
176 { ...settled, thread: { ...settled.thread, archivedAt: {} } },
177 { ...settled, thread: { ...settled.thread, deletedAt: {} } },
178 { ...settled, runs: [{ ...settled.runs[0]!, status: "failed" as const }] },
179 {
180 ...settled,
181 providerThreads: [{ ...settled.providerThreads[0]!, nativeThreadRef: null }],
182 },
183 ])
184 assert.isUndefined(
185 restartContinuationRun(invalid as OrchestrationV2ThreadProjection, lostWork),
186 );
175 for (const projectId of [scratchProjectId, projection.thread.projectId])
176 for (const status of ["completed", "waiting"] as const)
177 assert.isUndefined(
178 restartContinuationRun({
179 ...settled,
180 thread: { ...settled.thread, projectId },
181 runs: [{ ...settled.runs[0]!, status }],
182 }),
183 );
187184});
188185
189it.effect("prompts a settled thread's continuation with the note of its lost work", () =>
190 Effect.gen(function* () {
191 const base = makeProjection();
192 const work = [{ kind: "shell" as const, label: "sleep 25 && echo DONE" }];
193 const projection = {
194 ...base,
195 runs: [{ ...base.runs[0]!, status: "completed", restartCancelledBackgroundWork: work }],
196 providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }],
197 } as unknown as OrchestrationV2ThreadProjection;
198 const commands: Parameters<
199 ThreadManagementService.ThreadManagementService["Service"]["dispatch"]
200 >[0][] = [];
201 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
202 Effect.provide(
203 Layer.merge(
204 Layer.mock(ThreadManagementService.ThreadManagementService)({
205 getThreadRecords: () => Effect.succeed(projection),
206 recoverDelegatedTask: () => Effect.void,
207 dispatch: (command) => {
208 commands.push(command);
209 return Effect.succeed({} as never);
210 },
211 }),
212 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
186it.effect.each(["completed", "waiting", "cancelled"] as const)(
187 "ignores a pending restart continuation for a %s run whose provider turn settled",
188 (status) =>
189 Effect.gen(function* () {
190 const base = makeProjection();
191 const work = [{ kind: "shell" as const, label: "sleep 25 && echo DONE" }];
192 const projection = {
193 ...base,
194 runs: [{ ...base.runs[0]!, status, restartCancelledBackgroundWork: work }],
195 providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }],
196 } as unknown as OrchestrationV2ThreadProjection;
197 const commands: Parameters<
198 ThreadManagementService.ThreadManagementService["Service"]["dispatch"]
199 >[0][] = [];
200 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
201 Effect.provide(
202 Layer.merge(
203 Layer.mock(ThreadManagementService.ThreadManagementService)({
204 getThreadRecords: () => Effect.succeed(projection),
205 recoverDelegatedTask: () => Effect.void,
206 dispatch: (command) => {
207 commands.push(command);
208 return Effect.succeed({} as never);
209 },
210 }),
211 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
212 ),
213213 ),
214 ),
215 );
216 assert.lengthOf(commands, 1);
217 const command = commands[0]!;
218 assert.equal(
219 command.type === "message.dispatch" ? command.restartContinuationOfRunId : null,
220 runId,
221 );
222 assert.include(
223 command.type === "message.dispatch" ? command.text : "",
224 "sleep 25 && echo DONE",
225 );
226 assert.notInclude(command.type === "message.dispatch" ? command.text : "", "Continue where");
227 }),
214 );
215 assert.lengthOf(commands, 0);
216 }),
228217);
229218
230219it.effect("does not continue a failed run that lost background work", () =>
......@@ -419,7 +408,7 @@ it.effect("prepares no continuation for background work another provider thread
419408 );
420409 yield* recovery.prepareForShutdown;
421410 assert.lengthOf(writes, 0);
422 // The same work on the Codex run's own provider thread is continued.
411 // Work on the settled run's own provider thread must not wake it either.
423412 const ownWork = {
424413 ...projection,
425414 turnItems: [{ ...projection.turnItems[0]!, providerThreadId }],
......@@ -446,13 +435,103 @@ it.effect("prepares no continuation for background work another provider thread
446435 ),
447436 );
448437 yield* ownRecovery.prepareForShutdown;
449 assert.deepEqual(
450 writes.map((write) => write.effects[0]?.request),
451 [{ type: "provider-runtime.continue", sourceRunId: runId }],
452 );
438 assert.lengthOf(writes, 0);
453439 }),
454440);
455441
442it.effect.each([
443 ["completed", scratchProjectId],
444 ["completed", ProjectId.make("restart-project")],
445 ["waiting", scratchProjectId],
446 ["waiting", ProjectId.make("restart-project")],
447] as const)(
448 "cleans up background work without waking a %s run in project %s",
449 ([status, projectId]) =>
450 Effect.gen(function* () {
451 const base = makeProjection();
452 const projection = {
453 ...base,
454 thread: { ...base.thread, projectId },
455 runs: [{ ...base.runs[0]!, status }],
456 providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }],
457 turnItems: [
458 {
459 id: "turn-item:background-subagent",
460 runId,
461 nodeId: null,
462 providerThreadId,
463 providerInstanceId: instanceId,
464 type: "subagent",
465 subagentId: "subagent:background",
466 title: "Background reviewer",
467 status: "running",
468 },
469 ],
470 subagents: [
471 {
472 id: "subagent:background",
473 runId,
474 driver,
475 providerInstanceId: instanceId,
476 status: "running",
477 },
478 ],
479 } as unknown as OrchestrationV2ThreadProjection;
480 for (const trigger of ["startup", "shutdown"] as const) {
481 const commits: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0][] = [];
482 const writes: Parameters<EventSink.EventSinkV2["Service"]["writeWithEffects"]>[0][] = [];
483 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
484 Effect.provide(
485 Layer.mergeAll(
486 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
487 Layer.mock(ProjectionStore.ProjectionStoreV2)({
488 getRecoveryThreadIds: () => Effect.succeed([threadId]),
489 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
490 }),
491 Layer.mock(EventSink.EventSinkV2)({
492 writeWithEffects: (input) =>
493 Effect.sync(() => {
494 writes.push(input);
495 return [];
496 }),
497 commitCommand: (input) =>
498 Effect.sync(() => {
499 commits.push(input);
500 return { committed: true, cancelledEffectCount: 0 } as never;
501 }),
502 }),
503 IdAllocator.layer,
504 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
505 runRecoveryOnce: Effect.succeed(false),
506 }),
507 Layer.mock(EffectOutbox.EffectOutboxV2)({
508 listByCommandId: () => Effect.succeed([]),
509 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
510 }),
511 ),
512 ),
513 );
514 if (trigger === "shutdown") yield* recovery.prepareForShutdown;
515 yield* recovery.reconcile(trigger);
516 assert.lengthOf(writes, 0);
517 assert.lengthOf(commits, 1);
518 assert.lengthOf(commits[0]!.effects, 0);
519 const events = commits[0]!.events;
520 for (const type of ["turn-item.updated", "subagent.updated"] as const)
521 assert.isTrue(
522 events.some((event) => event.type === type && event.payload.status === "cancelled"),
523 );
524 const note = events.find((event) => event.type === "run.background-work-cancelled");
525 assert.equal(note?.runId, runId);
526 assert.deepEqual(note?.payload.restartCancelledBackgroundWork, [
527 { kind: "subagent", label: "Background reviewer", id: "turn-item:background-subagent" },
528 ]);
529 if (status === "completed")
530 assert.isFalse(events.some((event) => event.type === "run.updated"));
531 }
532 }),
533);
534
456535it.effect("does not cancel or resume a run that completes while shutdown intent commits", () =>
457536 Effect.gen(function* () {
458537 let projection = makeProjection();
apps/server/src/orchestration-v2/RestartContinuation.ts+16-23
......@@ -4,7 +4,6 @@ import {
44 CommandId,
55 MessageId,
66 type OrchestrationV2Run,
7 type ProviderThreadId,
87 type RunId,
98 type ThreadId,
109} from "@t3tools/contracts";
......@@ -23,16 +22,14 @@ import {
2322const CONTINUE_PROMPT = "Continue where you left off.";
2423
2524/**
26 * The run a restart continuation resumes, if any: an unfinished root run, or a
27 * settled one whose own provider thread lost background work in the restart
28 * (`cancelledWorkProviderThreadIds`, which recovery records on that thread).
25 * Resume only an unfinished root turn. Leftover background work is cleaned up
26 * separately and reported on the next user turn; it must not wake a settled run.
2927 */
3028export function restartContinuationRun(
3129 projection: Pick<
3230 ProjectionRuntimeRecoveryState,
3331 "thread" | "runs" | "providerThreads" | "providerSessions" | "providerTurns"
3432 >,
35 cancelledWorkProviderThreadIds: ReadonlySet<ProviderThreadId> = new Set(),
3633): OrchestrationV2Run | undefined {
3734 if (projection.thread.archivedAt !== null || projection.thread.deletedAt !== null) return;
3835 // Queued runs never started; recovery holds them behind the cut run.
......@@ -46,13 +43,8 @@ export function restartContinuationRun(
4643 if (!run) return;
4744 const preparedContinuation =
4845 run.status === "starting" && run.restartContinuationOfRunId !== undefined;
49 // Background work outlived this settled turn; the provider has no live turn.
50 const settledWithCancelledWork =
51 (run.status === "completed" || run.status === "waiting") &&
52 run.providerThreadId !== null &&
53 cancelledWorkProviderThreadIds.has(run.providerThreadId);
54 if (run.status !== "running" && !preparedContinuation && !settledWithCancelledWork) return;
55 const liveTurnRequired = !preparedContinuation && !settledWithCancelledWork;
46 if (run.status !== "running" && !preparedContinuation) return;
47 const liveTurnRequired = !preparedContinuation;
5648 if (projection.thread.providerInstanceId !== run.providerInstanceId) return;
5749 const providerThread = projection.providerThreads.find(
5850 (thread) => thread.id === run.providerThreadId,
......@@ -73,16 +65,13 @@ export function restartContinuationRun(
7365 const session = projection.providerSessions.find(
7466 (candidate) => candidate.id === providerThread.providerSessionId,
7567 );
76 // A settled thread's session may already be stopped and out of the recovery
77 // read; the continuation reopens it from the provider thread's native ref.
7868 // Most adapters keep a live session "ready" through its turns, so only a
7969 // stopped or failed session rules out a live turn.
8070 if (
81 session === undefined
82 ? !settledWithCancelledWork
83 : session.providerInstanceId !== run.providerInstanceId ||
84 session.driver !== providerThread.driver ||
85 (liveTurnRequired && (session.status === "stopped" || session.status === "error"))
71 session === undefined ||
72 session.providerInstanceId !== run.providerInstanceId ||
73 session.driver !== providerThread.driver ||
74 (liveTurnRequired && (session.status === "stopped" || session.status === "error"))
8675 )
8776 return;
8877 if (
......@@ -119,10 +108,14 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta
119108
120109 if (projection.messages.some((message) => message.id === messageId)) return;
121110 const source = projection.runs.find((run) => run.id === input.sourceRunId);
122 // A settled source prompts with the note of the background work it lost.
123 const noteSource =
124 source !== undefined && isRestartNoteSource(source, projection.providerTurns);
125 if (!source || (source.status !== "cancelled" && !noteSource)) return;
111 // Pending effects from older versions may target settled background work,
112 // including waiting runs that reconciliation subsequently cancelled.
113 if (
114 !source ||
115 source.status !== "cancelled" ||
116 isRestartNoteSource(source, projection.providerTurns)
117 )
118 return;
126119 // A user submission after reconciliation takes precedence over an automatic
127120 // prompt. Queued runs never started and stay held behind this one.
128121 if (
apps/server/src/orchestration-v2/runtimeLayer.test.ts+78-76
......@@ -2645,84 +2645,86 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => {
26452645 }),
26462646 );
26472647
2648 it.effect("does not admit a restart continuation of a failed run that lost background work", () =>
2649 Effect.gen(function* () {
2650 const orchestrator = yield* Orchestrator.OrchestratorV2;
2651 const eventSink = yield* EventSink.EventSinkV2;
2652 const threadId = ThreadId.make("runtime-layer-restart-failed-source");
2653 yield* orchestrator.dispatch({
2654 type: "thread.create",
2655 createdBy: "user",
2656 creationSource: "web",
2657 commandId: CommandId.make("restart-failed-create"),
2658 threadId,
2659 projectId: ProjectId.make("restart-project"),
2660 title: "Restart",
2661 modelSelection,
2662 runtimeMode: "full-access",
2663 interactionMode: "default",
2664 branch: null,
2665 worktreePath: "/tmp/runtime-layer-restart-failed",
2666 });
2667 yield* orchestrator.dispatch({
2668 type: "message.dispatch",
2669 createdBy: "user",
2670 creationSource: "web",
2671 commandId: CommandId.make("restart-failed-user-message"),
2672 threadId,
2673 messageId: MessageId.make("restart-failed-user-message"),
2674 text: "Original work",
2675 attachments: [],
2676 modelSelection,
2677 dispatchMode: { type: "start_immediately" },
2678 });
2679 const original = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
2680 const now = yield* DateTime.now;
2681 yield* eventSink.commitCommand({
2682 commandId: CommandId.make("restart-failed-reconcile"),
2683 threadId,
2684 commandType: "provider-runtime.reconcile",
2685 acceptedAt: now,
2686 events: [
2687 {
2688 id: EventId.make("restart-failed-run"),
2689 type: "run.updated",
2690 threadId,
2691 runId: original.id,
2692 occurredAt: now,
2693 payload: { ...original, status: "failed", completedAt: now },
2694 },
2695 {
2696 id: EventId.make("restart-failed-work"),
2697 type: "run.background-work-cancelled",
2698 threadId,
2699 runId: original.id,
2700 occurredAt: now,
2701 payload: {
2648 it.effect.each(["failed", "completed", "waiting", "cancelled"] as const)(
2649 "does not admit a restart continuation of a %s run with only lost background work",
2650 (status) =>
2651 Effect.gen(function* () {
2652 const orchestrator = yield* Orchestrator.OrchestratorV2;
2653 const eventSink = yield* EventSink.EventSinkV2;
2654 const threadId = ThreadId.make(`runtime-layer-restart-${status}-source`);
2655 yield* orchestrator.dispatch({
2656 type: "thread.create",
2657 createdBy: "user",
2658 creationSource: "web",
2659 commandId: CommandId.make(`restart-${status}-create`),
2660 threadId,
2661 projectId: ProjectId.make("restart-project"),
2662 title: "Restart",
2663 modelSelection,
2664 runtimeMode: "full-access",
2665 interactionMode: "default",
2666 branch: null,
2667 worktreePath: "/tmp/runtime-layer-restart-failed",
2668 });
2669 yield* orchestrator.dispatch({
2670 type: "message.dispatch",
2671 createdBy: "user",
2672 creationSource: "web",
2673 commandId: CommandId.make(`restart-${status}-user-message`),
2674 threadId,
2675 messageId: MessageId.make(`restart-${status}-user-message`),
2676 text: "Original work",
2677 attachments: [],
2678 modelSelection,
2679 dispatchMode: { type: "start_immediately" },
2680 });
2681 const original = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
2682 const now = yield* DateTime.now;
2683 yield* eventSink.commitCommand({
2684 commandId: CommandId.make(`restart-${status}-reconcile`),
2685 threadId,
2686 commandType: "provider-runtime.reconcile",
2687 acceptedAt: now,
2688 events: [
2689 {
2690 id: EventId.make(`restart-${status}-run`),
2691 type: "run.updated",
2692 threadId,
27022693 runId: original.id,
2703 restartCancelledBackgroundWork: [{ kind: "shell", label: "sleep 25" }],
2694 occurredAt: now,
2695 payload: { ...original, status, completedAt: now },
27042696 },
2705 },
2706 ],
2707 effects: [],
2708 });
2709 yield* orchestrator.dispatch({
2710 type: "message.dispatch",
2711 createdBy: "agent",
2712 creationSource: "server",
2713 commandId: CommandId.make("restart-failed-continuation"),
2714 threadId,
2715 messageId: MessageId.make("restart-failed-continuation"),
2716 text: "Note: the T3 server restarted.",
2717 attachments: [],
2718 modelSelection,
2719 dispatchMode: { type: "start_immediately" },
2720 restartContinuationOfRunId: original.id,
2721 });
2722 const projection = yield* orchestrator.getThreadProjection(threadId);
2723 assert.lengthOf(projection.runs, 1);
2724 assert.equal(projection.runs[0]?.status, "failed");
2725 }),
2697 {
2698 id: EventId.make(`restart-${status}-work`),
2699 type: "run.background-work-cancelled",
2700 threadId,
2701 runId: original.id,
2702 occurredAt: now,
2703 payload: {
2704 runId: original.id,
2705 restartCancelledBackgroundWork: [{ kind: "shell", label: "sleep 25" }],
2706 },
2707 },
2708 ],
2709 effects: [],
2710 });
2711 yield* orchestrator.dispatch({
2712 type: "message.dispatch",
2713 createdBy: "agent",
2714 creationSource: "server",
2715 commandId: CommandId.make(`restart-${status}-continuation`),
2716 threadId,
2717 messageId: MessageId.make(`restart-${status}-continuation`),
2718 text: "Note: the T3 server restarted.",
2719 attachments: [],
2720 modelSelection,
2721 dispatchMode: { type: "start_immediately" },
2722 restartContinuationOfRunId: original.id,
2723 });
2724 const projection = yield* orchestrator.getThreadProjection(threadId);
2725 assert.lengthOf(projection.runs, 1);
2726 assert.equal(projection.runs[0]?.status, status);
2727 }),
27262728 );
27272729
27282730 it.effect("rejects settling a thread while a run is active", () =>
apps/server/src/orchestration-v2/testkit/OrchestratorReplayRestartBackgroundNote.integration.test.ts+63-71
......@@ -11,6 +11,7 @@ import {
1111 makeClaudeRestartReplayHarness,
1212} from "../Adapters/ClaudeAdapterV2.testkit.ts";
1313import * as IdAllocator from "../IdAllocator.ts";
14import * as EffectWorker from "../EffectWorker.ts";
1415import { makeSqlitePersistenceLive } from "../../persistence/Layers/Sqlite.ts";
1516import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts";
1617import { CLAUDE_BACKGROUND_SUBAGENT_AFTER_ROOT_PROMPT } from "./fixtures/claude_background_subagent_after_root/input.ts";
......@@ -19,7 +20,11 @@ import {
1920 materializeFixtureInput,
2021 projectionFor,
2122} from "./fixtures/shared.ts";
22import { runOrchestratorV2ProviderReplayScenario } from "./ProviderReplayHarness.ts";
23import {
24 makeOrchestratorV2ProviderReplayLayer,
25 runOrchestratorV2ProviderReplayScenario,
26} from "./ProviderReplayHarness.ts";
27import { runOrchestratorV2Scenario } from "./OrchestratorScenario.ts";
2328import { checkpointWorkspace } from "./ReplayFixtureWorkspace.ts";
2429import { readProviderReplayTranscript } from "./ReplayTranscriptNdjson.ts";
2530
......@@ -155,21 +160,7 @@ const runRestart = Effect.fn("runRestart")(function* (input: {
155160 // Phase 1 ends once the root run settles; the subagent is still open.
156161 const firstIdle = materialized.steps.findIndex((step) => step.type === "await_thread_idle");
157162 const phase1Steps = materialized.steps.slice(0, firstIdle + 1);
158 const threadId = materialized.projectionThreadIds[0]!;
159 const phase2Steps = [
160 // Let recovery's continuation (run 2) finish before the next user message.
161 ...(input.continueThreadsAfterServerUpdate
162 ? [
163 {
164 type: "await_run_status" as const,
165 threadId,
166 runId: (yield* IdAllocator.IdAllocatorV2).derive.run({ threadId, ordinal: 2 }),
167 status: "completed" as const,
168 },
169 ]
170 : []),
171 ...materialized.steps.slice(firstIdle + 1),
172 ];
163 const phase2Steps = materialized.steps.slice(firstIdle + 1);
173164 const { harness, assertComplete } = makeClaudeRestartReplayHarness(transcript);
174165 const databaseLayer = makeSqlitePersistenceLive(path.join(tempDir, "state.sqlite")).pipe(
175166 Layer.provide(NodeServices.layer),
......@@ -192,10 +183,35 @@ const runRestart = Effect.fn("runRestart")(function* (input: {
192183 assert.equal(settled.runs[0]?.status, "completed");
193184 assert.equal(settled.subagents[0]?.status, "running");
194185
186 // Drain recovery before submitting any user work, which would otherwise
187 // take precedence over an incorrectly queued automatic continuation.
188 const restartScenario = scenario("restarted", []);
189 const restarted = yield* Effect.scoped(
190 Effect.gen(function* () {
191 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
192 yield* worker.drain();
193 return yield* runOrchestratorV2Scenario(restartScenario);
194 }).pipe(
195 Effect.provide(
196 makeOrchestratorV2ProviderReplayLayer(restartScenario, harness, {
197 databaseLayer,
198 recoverOnStartup: true,
199 continueThreadsAfterServerUpdate: input.continueThreadsAfterServerUpdate,
200 }),
201 ),
202 ),
203 );
204 const recovered = projectionFor(restarted, SCENARIO);
205 assert.deepEqual(
206 recovered.runs.map((run) => run.status),
207 ["completed"],
208 );
209 assert.equal(recovered.subagents[0]?.status, "cancelled");
210 assert.equal(recovered.runs[0]?.restartCancelledBackgroundWork?.[0]?.kind, "subagent");
211
195212 const after = yield* Effect.scoped(
196213 runOrchestratorV2ProviderReplayScenario(scenario("after-restart", phase2Steps), harness, {
197214 databaseLayer,
198 recoverOnStartup: true,
199215 continueThreadsAfterServerUpdate: input.continueThreadsAfterServerUpdate,
200216 }),
201217 );
......@@ -210,61 +226,37 @@ const userTexts = (projection: ReturnType<typeof projectionFor>) =>
210226 projection.turnItems.flatMap((item) => (item.type === "user_message" ? [item.text] : []));
211227
212228describe("restart-cancelled background work", () => {
213 it.effect("tells the next provider turn once that its background subagent died", () =>
214 Effect.scoped(
215 Effect.gen(function* () {
216 const projection = yield* runRestart({
217 // The first turn after the restart carries the note; the second does not.
218 resumedPrompts: [
219 `${NOTE}\n\nUser message:\n${FIRST_AFTER_RESTART}`,
229 it.effect.each([false, true])(
230 "keeps the thread settled and tells the next user turn once when continuation is %s",
231 (continueThreadsAfterServerUpdate) =>
232 Effect.scoped(
233 Effect.gen(function* () {
234 const projection = yield* runRestart({
235 // The first turn after the restart carries the note; the second does not.
236 resumedPrompts: [
237 `${NOTE}\n\nUser message:\n${FIRST_AFTER_RESTART}`,
238 SECOND_AFTER_RESTART,
239 ],
240 userMessagesAfterRestart: [FIRST_AFTER_RESTART, SECOND_AFTER_RESTART],
241 continueThreadsAfterServerUpdate,
242 });
243 assert.deepEqual(
244 projection.runs.map((run) => run.status),
245 ["completed", "completed", "completed"],
246 );
247 assert.isFalse(
248 projection.runs.some((run) => run.restartContinuationOfRunId !== undefined),
249 );
250 // The note reaches the provider only; the timeline keeps what the user sent.
251 assert.deepEqual(userTexts(projection), [
252 CLAUDE_BACKGROUND_SUBAGENT_AFTER_ROOT_PROMPT,
253 FIRST_AFTER_RESTART,
220254 SECOND_AFTER_RESTART,
221 ],
222 userMessagesAfterRestart: [FIRST_AFTER_RESTART, SECOND_AFTER_RESTART],
223 continueThreadsAfterServerUpdate: false,
224 });
225 assert.deepEqual(
226 projection.runs.map((run) => run.status),
227 ["completed", "completed", "completed"],
228 );
229 assert.isFalse(projection.runs.some((run) => run.restartContinuationOfRunId !== undefined));
230 // The note reaches the provider only; the timeline keeps what the user sent.
231 assert.deepEqual(userTexts(projection), [
232 CLAUDE_BACKGROUND_SUBAGENT_AFTER_ROOT_PROMPT,
233 FIRST_AFTER_RESTART,
234 SECOND_AFTER_RESTART,
235 ]);
236 }).pipe(
237 provideDeterministicTestRuntime,
238 Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer)),
239 ),
240 ),
241 );
242
243 it.effect("continues a settled thread with the note when restart continuation is on", () =>
244 Effect.scoped(
245 Effect.gen(function* () {
246 const projection = yield* runRestart({
247 // Recovery's continuation prompts with the note; the user turn after it does not.
248 resumedPrompts: [NOTE, FIRST_AFTER_RESTART],
249 userMessagesAfterRestart: [FIRST_AFTER_RESTART],
250 continueThreadsAfterServerUpdate: true,
251 });
252 const [root, continuation, user] = projection.runs;
253 assert.deepEqual(
254 projection.runs.map((run) => run.status),
255 ["completed", "completed", "completed"],
256 );
257 assert.equal(continuation?.restartContinuationOfRunId, root?.id);
258 assert.isUndefined(user?.restartContinuationOfRunId);
259 assert.deepEqual(userTexts(projection), [
260 CLAUDE_BACKGROUND_SUBAGENT_AFTER_ROOT_PROMPT,
261 NOTE,
262 FIRST_AFTER_RESTART,
263 ]);
264 }).pipe(
265 provideDeterministicTestRuntime,
266 Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer)),
255 ]);
256 }).pipe(
257 provideDeterministicTestRuntime,
258 Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer)),
259 ),
267260 ),
268 ),
269261 );
270262});