1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, describe, it } from "@effect/vitest";
3import { ProviderDriverKind, type ProviderReplayEntry } from "@t3tools/contracts";
4import * as Effect from "effect/Effect";
5import * as FileSystem from "effect/FileSystem";
6import * as Layer from "effect/Layer";
7import * as Path from "effect/Path";
8
9import {
10 ClaudeOrchestratorReplayHarness,
11 makeClaudeRestartReplayHarness,
12} from "../Adapters/ClaudeAdapterV2.testkit.ts";
13import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
14import * as EffectWorker from "../EffectWorker.ts";
15import * as SqlitePersistence from "../../persistence/Sqlite.ts";
16import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts";
17import { CLAUDE_BACKGROUND_SUBAGENT_AFTER_ROOT_PROMPT } from "./fixtures/claude_background_subagent_after_root/input.ts";
18import {
19 CLAUDE_MODEL_SELECTION,
20 materializeFixtureInput,
21 projectionFor,
22} from "./fixtures/shared.ts";
23import { runOrchestratorV2ProviderReplayScenario } from "./ProviderReplayHarness.ts";
24import * as ProviderReplayHarness from "./ProviderReplayHarness.ts";
25import { runOrchestratorV2Scenario } from "./OrchestratorScenario.ts";
26import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
27import { readProviderReplayTranscript } from "@t3tools/provider-testing/replayTranscript";
28
29const SCENARIO = "claude_background_subagent_after_root";
30const SESSION_ID = "cca274e4-25ae-4171-b972-bbb31118517e";
31const FIRST_AFTER_RESTART = "Is the background subagent done yet?";
32const SECOND_AFTER_RESTART = "Thanks. Anything else?";
33const NOTE = [
34 "Note: the T3 server restarted, and this background work was cancelled before it finished. It will not report back:",
35 "- subagent: Background subagent test",
36].join("\n");
37
38/**
39 * The recorded background-subagent session cut by a restart right after the
40 * root turn settled: the subagent's frames never arrive. A fresh runtime then
41 * resumes the native session; each prompt frame it sends is pinned, so these
42 * resumed turns are the provider's view of what T3 told it.
43 */
44const readRestartTranscript = Effect.fn("readRestartTranscript")(function* (
45 resumedPrompts: ReadonlyArray<string>,
46) {
47 const recorded = yield* readProviderReplayTranscript(
48 new URL(`./fixtures/${SCENARIO}/claude_transcript.ndjson`, import.meta.url),
49 );
50 const rootResult = recorded.entries.findIndex(
51 (entry) => entry.type === "emit_inbound" && entry.label === "result",
52 );
53 // The restarted server resumes the recorded session instead of creating it.
54 const queryOpen = recorded.entries[0];
55 const openFrame = queryOpen?.type === "expect_outbound" ? queryOpen.frame : undefined;
56 const queryOptions: unknown =
57 typeof openFrame === "object" && openFrame !== null
58 ? Reflect.get(openFrame, "options")
59 : undefined;
60 if (typeof queryOptions !== "object" || queryOptions === null) {
61 throw new Error(`${SCENARIO} must open with a recorded query.open frame.`);
62 }
63 const { sessionId: _sessionId, ...resumeOptions } = queryOptions as Record<string, unknown>;
64 const resumedTurn = (label: string, text: string): ReadonlyArray<ProviderReplayEntry> => [
65 {
66 type: "expect_outbound",
67 label: `prompt.offer:${label}`,
68 frame: {
69 type: "prompt.offer",
70 message: {
71 type: "user",
72 message: { role: "user", content: text },
73 parent_tool_use_id: null,
74 },
75 },
76 },
77 {
78 type: "emit_inbound",
79 label: `assistant:${label}`,
80 frame: {
81 type: "assistant",
82 message: {
83 model: "claude-sonnet-4-6",
84 id: `msg_restart_${label}`,
85 type: "message",
86 role: "assistant",
87 content: [{ type: "text", text: `REPLY_${label}` }],
88 stop_reason: null,
89 stop_sequence: null,
90 usage: { input_tokens: 1, output_tokens: 1 },
91 },
92 parent_tool_use_id: null,
93 session_id: SESSION_ID,
94 uuid: `restart-assistant-${label}`,
95 },
96 },
97 {
98 type: "emit_inbound",
99 label: `result:${label}`,
100 frame: {
101 type: "result",
102 subtype: "success",
103 is_error: false,
104 num_turns: 1,
105 result: `REPLY_${label}`,
106 stop_reason: "end_turn",
107 duration_ms: 1,
108 duration_api_ms: 1,
109 total_cost_usd: 0,
110 usage: { input_tokens: 1, output_tokens: 1 },
111 modelUsage: {},
112 permission_denials: [],
113 session_id: SESSION_ID,
114 uuid: `restart-result-${label}`,
115 },
116 },
117 ];
118 return yield* ClaudeOrchestratorReplayHarness.decodeTranscript({
119 ...recorded,
120 entries: [
121 ...recorded.entries.slice(0, rootResult + 1),
122 // The server dies here, with the background subagent still running.
123 { type: "runtime_exit", status: "success" },
124 {
125 type: "expect_outbound",
126 label: "query.open:resume",
127 frame: { type: "query.open", options: { ...resumeOptions, resume: SESSION_ID } },
128 },
129 ...resumedPrompts.flatMap((prompt, index) => resumedTurn(String(index + 1), prompt)),
130 ],
131 });
132});
133
134const runRestart = Effect.fn("runRestart")(function* (input: {
135 readonly resumedPrompts: ReadonlyArray<string>;
136 readonly userMessagesAfterRestart: ReadonlyArray<string>;
137 readonly continueThreadsAfterServerUpdate: boolean;
138}) {
139 const transcript = yield* readRestartTranscript(input.resumedPrompts);
140 const workspace = yield* checkpointWorkspace(SCENARIO);
141 const fs = yield* FileSystem.FileSystem;
142 const path = yield* Path.Path;
143 const tempDir = yield* Effect.acquireRelease(
144 fs.makeTempDirectory({ prefix: "t3-orchestration-v2-restart-note-" }),
145 (directory) => fs.remove(directory, { recursive: true, force: true }).pipe(Effect.orDie),
146 );
147 const materialized = yield* materializeFixtureInput({
148 scenario: SCENARIO,
149 fixtureInput: {
150 steps: [
151 { type: "message", text: CLAUDE_BACKGROUND_SUBAGENT_AFTER_ROOT_PROMPT },
152 ...input.userMessagesAfterRestart.map((text) => ({ type: "message" as const, text })),
153 ],
154 },
155 driver: ProviderDriverKind.make("claudeAgent"),
156 modelSelection: CLAUDE_MODEL_SELECTION,
157 });
158 // Phase 1 ends once the root run settles; the subagent is still open.
159 const firstIdle = materialized.steps.findIndex((step) => step.type === "await_thread_idle");
160 const phase1Steps = materialized.steps.slice(0, firstIdle + 1);
161 const phase2Steps = materialized.steps.slice(firstIdle + 1);
162 const { harness, assertComplete } = makeClaudeRestartReplayHarness(transcript);
163 const layerDatabase = SqlitePersistence.layerFromPath(path.join(tempDir, "state.sqlite")).pipe(
164 Layer.provide(NodeServices.layer),
165 );
166 const scenario = (name: string, steps: typeof materialized.steps) => ({
167 name: `${SCENARIO}:${name}`,
168 transcript,
169 commands: steps.flatMap((step) => (step.type === "dispatch" ? [step.command] : [])),
170 steps,
171 projectionThreadIds: materialized.projectionThreadIds,
172 runtimePolicyOverride: { cwd: workspace },
173 });
174
175 const before = yield* Effect.scoped(
176 runOrchestratorV2ProviderReplayScenario(scenario("before-restart", phase1Steps), harness, {
177 databaseLayer: layerDatabase,
178 }),
179 );
180 const settled = projectionFor(before, SCENARIO);
181 assert.equal(settled.runs[0]?.status, "completed");
182 assert.equal(settled.subagents[0]?.status, "running");
183
184 // Drain recovery before submitting any user work, which would otherwise
185 // take precedence over an incorrectly queued automatic continuation.
186 const restartScenario = scenario("restarted", []);
187 const restarted = yield* Effect.scoped(
188 Effect.gen(function* () {
189 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
190 yield* worker.drain();
191 return yield* runOrchestratorV2Scenario(restartScenario);
192 }).pipe(
193 Effect.provide(
194 ProviderReplayHarness.layerProviderReplay(restartScenario, harness, {
195 databaseLayer: layerDatabase,
196 recoverOnStartup: true,
197 continueThreadsAfterServerUpdate: input.continueThreadsAfterServerUpdate,
198 }),
199 ),
200 ),
201 );
202 const recovered = projectionFor(restarted, SCENARIO);
203 assert.deepEqual(
204 recovered.runs.map((run) => run.status),
205 ["completed"],
206 );
207 assert.equal(recovered.subagents[0]?.status, "cancelled");
208 assert.equal(recovered.runs[0]?.restartCancelledBackgroundWork?.[0]?.kind, "subagent");
209
210 const after = yield* Effect.scoped(
211 runOrchestratorV2ProviderReplayScenario(scenario("after-restart", phase2Steps), harness, {
212 databaseLayer: layerDatabase,
213 continueThreadsAfterServerUpdate: input.continueThreadsAfterServerUpdate,
214 }),
215 );
216 // The replay runner rejects any prompt frame that differs from the transcript.
217 yield* assertComplete;
218 const projection = projectionFor(after, SCENARIO);
219 assert.equal(projection.subagents[0]?.status, "cancelled");
220 return projection;
221});
222
223const userTexts = (projection: ReturnType<typeof projectionFor>) =>
224 projection.turnItems.flatMap((item) => (item.type === "user_message" ? [item.text] : []));
225
226describe("restart-cancelled background work", () => {
227 it.effect.each([false, true])(
228 "keeps the thread settled and tells the next user turn once when continuation is %s",
229 (continueThreadsAfterServerUpdate) =>
230 Effect.scoped(
231 Effect.gen(function* () {
232 const projection = yield* runRestart({
233 // The first turn after the restart carries the note; the second does not.
234 resumedPrompts: [
235 `${NOTE}\n\nUser message:\n${FIRST_AFTER_RESTART}`,
236 SECOND_AFTER_RESTART,
237 ],
238 userMessagesAfterRestart: [FIRST_AFTER_RESTART, SECOND_AFTER_RESTART],
239 continueThreadsAfterServerUpdate,
240 });
241 assert.deepEqual(
242 projection.runs.map((run) => run.status),
243 ["completed", "completed", "completed"],
244 );
245 assert.isFalse(
246 projection.runs.some((run) => run.restartContinuationOfRunId !== undefined),
247 );
248 // The note reaches the provider only; the timeline keeps what the user sent.
249 assert.deepEqual(userTexts(projection), [
250 CLAUDE_BACKGROUND_SUBAGENT_AFTER_ROOT_PROMPT,
251 FIRST_AFTER_RESTART,
252 SECOND_AFTER_RESTART,
253 ]);
254 }).pipe(
255 provideDeterministicTestRuntime,
256 Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer)),
257 ),
258 ),
259 );
260});