1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, describe, it } from "@effect/vitest";
3import * as Effect from "effect/Effect";
4import * as FileSystem from "effect/FileSystem";
5import * as Layer from "effect/Layer";
6import * as Path from "effect/Path";
7import * as Schema from "effect/Schema";
8import * as CodexReplay from "effect-codex-app-server/replay";
9import { ProviderDriverKind } from "@t3tools/contracts";
10
11import {
12 ClaudeOrchestratorReplayHarness,
13 makeClaudeRestartReplayHarness,
14} from "../Adapters/ClaudeAdapterV2.testkit.ts";
15import { CodexOrchestratorReplayHarness } from "../Adapters/CodexAdapterV2.testkit.ts";
16import * as CodexAdapterV2Testkit from "../Adapters/CodexAdapterV2.testkit.ts";
17import {
18 type CursorAgentSdkReplayTranscript,
19 CursorOrchestratorReplayHarness,
20 makeCursorAgentSdkReplayRunner,
21} from "../Adapters/CursorAdapterV2.testkit.ts";
22import * as CursorAdapterV2Testkit from "../Adapters/CursorAdapterV2.testkit.ts";
23import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
24import * as SqlitePersistence from "../../persistence/Sqlite.ts";
25import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts";
26import {
27 CLAUDE_MODEL_SELECTION,
28 CODEX_MODEL_SELECTION,
29 CURSOR_MODEL_SELECTION,
30 materializeFixtureInput,
31 type MaterializedOrchestratorFixtureInput,
32 PROVIDER_THREAD_RESUME_FIRST_PROMPT,
33 PROVIDER_THREAD_RESUME_SECOND_PROMPT,
34} from "./fixtures/shared.ts";
35import {
36 assertAssistantTextIncludes,
37 assertBaseProjection,
38 assertConversationMessageRoles,
39 assertRunOrdinals,
40 assertSemanticProjectionIntegrity,
41 assertTurnItemTypes,
42 assertUserMessagesInclude,
43 projectionFor,
44} from "./fixtures/shared.ts";
45import { runOrchestratorV2ProviderReplayScenario } from "./ProviderReplayHarness.ts";
46import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
47import { materializeReplayTranscriptRuntimeInstructions } from "./ReplayRuntimeInstructions.ts";
48import {
49 materializeReplayTranscriptWorkspace,
50 readProviderReplayTranscript,
51} from "@t3tools/provider-testing/replayTranscript";
52
53const FIRST_FINAL = "provider thread resume fixture first turn complete";
54const SECOND_FINAL = "provider thread resume fixture second turn complete";
55
56const decodeCodexTranscript = Schema.decodeUnknownEffect(
57 CodexReplay.CodexAppServerReplayTranscript,
58);
59const readRawTranscript = Effect.fn("readRecoveryTranscript")(function* (file: URL) {
60 return yield* readProviderReplayTranscript(file);
61});
62const readCodexTranscript = Effect.fn("readCodexRecoveryTranscript")(function* (workspace: string) {
63 const transcript = yield* readRawTranscript(
64 new URL("./fixtures/provider_thread_resume/codex_transcript.ndjson", import.meta.url),
65 );
66 return yield* decodeCodexTranscript(materializeReplayTranscriptWorkspace(transcript, workspace));
67});
68const decodePromptPair = Schema.decodeUnknownEffect(Schema.Tuple([Schema.String, Schema.String]));
69const readClaudeSubagentResumeTranscript = Effect.fn("readClaudeSubagentResumeTranscript")(
70 function* () {
71 return yield* ClaudeOrchestratorReplayHarness.decodeTranscript(
72 yield* readRawTranscript(
73 new URL(
74 "./fixtures/claude_subagent_resume_after_restart/claude_transcript.ndjson",
75 import.meta.url,
76 ),
77 ),
78 );
79 },
80);
81const readCursorTranscript = Effect.fn("readCursorRecoveryTranscript")(function* () {
82 const transcript = yield* readRawTranscript(
83 new URL("./fixtures/provider_thread_resume/cursor_transcript.ndjson", import.meta.url),
84 );
85 return yield* CursorOrchestratorReplayHarness.decodeTranscript(
86 materializeReplayTranscriptRuntimeInstructions(transcript, {
87 driver: ProviderDriverKind.make("cursor"),
88 model: CURSOR_MODEL_SELECTION.model,
89 }),
90 );
91});
92
93function splitAfterFirstIdle(materialized: MaterializedOrchestratorFixtureInput) {
94 const splitIndex = materialized.steps.findIndex((step) => step.type === "await_thread_idle");
95 if (splitIndex < 0) {
96 throw new Error("Expected fixture to contain await_thread_idle after the first turn.");
97 }
98
99 const phase1Steps = materialized.steps.slice(0, splitIndex + 1);
100 const phase2Steps = materialized.steps.slice(splitIndex + 1);
101 return {
102 phase1Steps,
103 phase2Steps,
104 phase1Commands: phase1Steps.flatMap((step) => (step.type === "dispatch" ? [step.command] : [])),
105 phase2Commands: phase2Steps.flatMap((step) => (step.type === "dispatch" ? [step.command] : [])),
106 };
107}
108
109const runCursorRecovery = Effect.fn("runCursorRecovery")(function* (input: {
110 readonly transcript: CursorAgentSdkReplayTranscript;
111 readonly runner: ReturnType<typeof makeCursorAgentSdkReplayRunner>;
112}) {
113 const fs = yield* FileSystem.FileSystem;
114 const path = yield* Path.Path;
115 const tempDir = yield* Effect.acquireRelease(
116 fs.makeTempDirectory({
117 prefix: "t3-orchestration-v2-cursor-recovery-",
118 }),
119 (directory) => fs.remove(directory, { recursive: true, force: true }).pipe(Effect.orDie),
120 );
121 yield* fs.makeDirectory(tempDir, { recursive: true });
122 const dbPath = path.join(tempDir, "state.sqlite");
123 const materialized = yield* materializeFixtureInput({
124 scenario: "provider_thread_resume",
125 fixtureInput: {
126 steps: [
127 { type: "message", text: PROVIDER_THREAD_RESUME_FIRST_PROMPT },
128 { type: "message", text: PROVIDER_THREAD_RESUME_SECOND_PROMPT },
129 ],
130 },
131 driver: ProviderDriverKind.make("cursor"),
132 modelSelection: CURSOR_MODEL_SELECTION,
133 });
134 const { phase1Commands, phase1Steps, phase2Commands, phase2Steps } =
135 splitAfterFirstIdle(materialized);
136 const options = {
137 databaseLayer: SqlitePersistence.layerFromPath(dbPath).pipe(Layer.provide(NodeServices.layer)),
138 };
139 const harness = {
140 ...CursorOrchestratorReplayHarness,
141 makeProviderAdapterRegistryLayer: () =>
142 CursorAdapterV2Testkit.layer(input.transcript, {
143 runner: input.runner,
144 assertCompleteOnFinalize: false,
145 }),
146 };
147
148 yield* Effect.scoped(
149 runOrchestratorV2ProviderReplayScenario(
150 {
151 name: "provider_thread_resume/cursor:first-runtime",
152 transcript: input.transcript,
153 commands: phase1Commands,
154 steps: phase1Steps,
155 projectionThreadIds: materialized.projectionThreadIds,
156 runtimePolicyOverride: { cwd: tempDir },
157 },
158 harness,
159 options,
160 ),
161 );
162
163 const result = yield* Effect.scoped(
164 runOrchestratorV2ProviderReplayScenario(
165 {
166 name: "provider_thread_resume/cursor:second-runtime",
167 transcript: input.transcript,
168 commands: phase2Commands,
169 steps: phase2Steps,
170 projectionThreadIds: materialized.projectionThreadIds,
171 runtimePolicyOverride: { cwd: tempDir },
172 },
173 harness,
174 options,
175 ),
176 );
177
178 assertBaseProjection({
179 result,
180 transcript: input.transcript,
181 runCount: 2,
182 runStatuses: ["completed", "completed"],
183 });
184 const projection = projectionFor(result, input.transcript.scenario);
185 assertSemanticProjectionIntegrity(projection);
186 assertRunOrdinals(projection, [1, 2]);
187 assertConversationMessageRoles(projection, ["user", "assistant", "user", "assistant"]);
188 assertTurnItemTypes(projection, ["user_message", "assistant_message"]);
189 assertUserMessagesInclude(projection, [
190 PROVIDER_THREAD_RESUME_FIRST_PROMPT,
191 PROVIDER_THREAD_RESUME_SECOND_PROMPT,
192 ]);
193 assertAssistantTextIncludes(projection, FIRST_FINAL);
194 assertAssistantTextIncludes(projection, SECOND_FINAL);
195 assert.lengthOf(projection.providerThreads, 1);
196});
197
198describe("orchestrator replay recovery", () => {
199 it.effect(
200 "resumes a provider-native Codex thread after recreating the orchestrator runtime",
201 () =>
202 Effect.scoped(
203 Effect.gen(function* () {
204 // Checkpoint against a throwaway git workspace: with no override the
205 // scope cwd falls back to process.cwd(), and a cold full-repo
206 // baseline capture on CI outlives the scenario wait budget.
207 const workspace = yield* checkpointWorkspace("provider_thread_resume");
208 const transcript = yield* readCodexTranscript(workspace);
209 const fs = yield* FileSystem.FileSystem;
210 const path = yield* Path.Path;
211 const tempDir = yield* Effect.acquireRelease(
212 fs.makeTempDirectory({
213 prefix: "t3-orchestration-v2-recovery-",
214 }),
215 (directory) =>
216 fs.remove(directory, { recursive: true, force: true }).pipe(Effect.orDie),
217 );
218 yield* fs.makeDirectory(tempDir, { recursive: true });
219 const dbPath = path.join(tempDir, "state.sqlite");
220 const driver = yield* CodexReplay.makeReplayDriver(transcript);
221 const materialized = yield* materializeFixtureInput({
222 scenario: "provider_thread_resume",
223 fixtureInput: {
224 steps: [
225 { type: "message", text: PROVIDER_THREAD_RESUME_FIRST_PROMPT },
226 { type: "message", text: PROVIDER_THREAD_RESUME_SECOND_PROMPT },
227 ],
228 },
229 driver: ProviderDriverKind.make("codex"),
230 modelSelection: CODEX_MODEL_SELECTION,
231 });
232 const { phase1Commands, phase1Steps, phase2Commands, phase2Steps } =
233 splitAfterFirstIdle(materialized);
234
235 const harness = {
236 ...CodexOrchestratorReplayHarness,
237 makeProviderAdapterRegistryLayer: () =>
238 CodexAdapterV2Testkit.layer({ transcript, driver }),
239 };
240 const options = {
241 databaseLayer: SqlitePersistence.layerFromPath(dbPath).pipe(
242 Layer.provide(NodeServices.layer),
243 ),
244 };
245
246 yield* runOrchestratorV2ProviderReplayScenario(
247 {
248 name: "provider_thread_resume/codex:first-runtime",
249 transcript,
250 commands: phase1Commands,
251 steps: phase1Steps,
252 projectionThreadIds: materialized.projectionThreadIds,
253 runtimePolicyOverride: { cwd: workspace },
254 },
255 harness,
256 options,
257 );
258
259 const result = yield* runOrchestratorV2ProviderReplayScenario(
260 {
261 name: "provider_thread_resume/codex:second-runtime",
262 transcript,
263 commands: phase2Commands,
264 steps: phase2Steps,
265 projectionThreadIds: materialized.projectionThreadIds,
266 runtimePolicyOverride: { cwd: workspace },
267 },
268 harness,
269 options,
270 );
271
272 assertBaseProjection({
273 result,
274 transcript,
275 runCount: 2,
276 runStatuses: ["completed", "completed"],
277 });
278 const projection = projectionFor(result, transcript.scenario);
279 assertSemanticProjectionIntegrity(projection);
280 assertRunOrdinals(projection, [1, 2]);
281 assertConversationMessageRoles(projection, ["user", "assistant", "user", "assistant"]);
282 assertTurnItemTypes(projection, ["user_message", "assistant_message"]);
283 assertUserMessagesInclude(projection, [
284 PROVIDER_THREAD_RESUME_FIRST_PROMPT,
285 PROVIDER_THREAD_RESUME_SECOND_PROMPT,
286 ]);
287 assertAssistantTextIncludes(projection, FIRST_FINAL);
288 assertAssistantTextIncludes(projection, SECOND_FINAL);
289 assert.lengthOf(projection.providerThreads, 1);
290 }).pipe(
291 provideDeterministicTestRuntime,
292 Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer)),
293 ),
294 ),
295 );
296
297 it.effect(
298 "resumes a provider-native Cursor thread after recreating the orchestrator runtime",
299 () =>
300 Effect.scoped(
301 Effect.gen(function* () {
302 const transcript = yield* readCursorTranscript();
303 const runner = makeCursorAgentSdkReplayRunner(transcript);
304 yield* runCursorRecovery({ transcript, runner });
305 yield* runner.assertComplete;
306 }).pipe(
307 provideDeterministicTestRuntime,
308 Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer)),
309 ),
310 ),
311 );
312 it.effect("keeps a Claude subagent resumed after a server restart in its own thread", () =>
313 Effect.scoped(
314 Effect.gen(function* () {
315 // The math.ts subagent from thread 42e302f0 (2026-09-25): launched,
316 // finished, then resumed by SendMessage after the server restarted.
317 const transcript = yield* readClaudeSubagentResumeTranscript();
318 const [launchPrompt, resumePrompt] = yield* decodePromptPair(transcript.metadata?.prompts);
319 const workspace = yield* checkpointWorkspace(transcript.scenario);
320 const fs = yield* FileSystem.FileSystem;
321 const path = yield* Path.Path;
322 const tempDir = yield* Effect.acquireRelease(
323 fs.makeTempDirectory({ prefix: "t3-orchestration-v2-claude-recovery-" }),
324 (directory) => fs.remove(directory, { recursive: true, force: true }).pipe(Effect.orDie),
325 );
326 const materialized = yield* materializeFixtureInput({
327 scenario: transcript.scenario,
328 fixtureInput: {
329 steps: [
330 { type: "message", text: launchPrompt },
331 { type: "message", text: resumePrompt },
332 ],
333 },
334 driver: ProviderDriverKind.make("claudeAgent"),
335 modelSelection: CLAUDE_MODEL_SELECTION,
336 });
337 const { phase1Commands, phase1Steps, phase2Commands, phase2Steps } =
338 splitAfterFirstIdle(materialized);
339 const { harness, assertComplete } = makeClaudeRestartReplayHarness(transcript);
340 const options = {
341 databaseLayer: SqlitePersistence.layerFromPath(path.join(tempDir, "state.sqlite")).pipe(
342 Layer.provide(NodeServices.layer),
343 ),
344 };
345
346 yield* Effect.scoped(
347 runOrchestratorV2ProviderReplayScenario(
348 {
349 name: `${transcript.scenario}:first-runtime`,
350 transcript,
351 commands: phase1Commands,
352 steps: phase1Steps,
353 projectionThreadIds: materialized.projectionThreadIds,
354 runtimePolicyOverride: { cwd: workspace },
355 },
356 harness,
357 options,
358 ),
359 );
360 const result = yield* Effect.scoped(
361 runOrchestratorV2ProviderReplayScenario(
362 {
363 name: `${transcript.scenario}:second-runtime`,
364 transcript,
365 commands: phase2Commands,
366 steps: phase2Steps,
367 projectionThreadIds: materialized.projectionThreadIds,
368 runtimePolicyOverride: { cwd: workspace },
369 },
370 harness,
371 options,
372 ),
373 );
374 yield* assertComplete;
375
376 const projection = projectionFor(result, transcript.scenario);
377 assertSemanticProjectionIntegrity(projection);
378 assert.lengthOf(projection.subagents, 1);
379 const subagent = projection.subagents[0];
380 assert.equal(subagent?.status, "completed");
381 const childThreads = [...result.projections.values()].filter(
382 (candidate) => candidate.thread.lineage.parentThreadId === projection.thread.id,
383 );
384 assert.lengthOf(childThreads, 1);
385 const child = childThreads[0];
386 assert.equal(child?.thread.id, subagent?.childThreadId);
387 // The recovered subagent's replies name the model its thread already
388 // holds, so the effort recorded at launch survives the restart.
389 assert.deepEqual(child?.thread.modelSelection, {
390 instanceId: CLAUDE_MODEL_SELECTION.instanceId,
391 model: "claude-sonnet-5",
392 options: [{ id: "effort", value: "high" }],
393 });
394
395 const conversation = (child?.turnItems ?? [])
396 .toSorted((left, right) => left.ordinal - right.ordinal)
397 .flatMap((item) =>
398 item.type === "user_message" || item.type === "assistant_message"
399 ? [`${item.type === "user_message" ? "user" : "assistant"}:${item.text}`]
400 : [],
401 );
402 const launchTask = conversation[0];
403 assert.deepEqual(
404 conversation.map((entry) => entry.slice(0, entry.indexOf(":"))),
405 ["user", "assistant", "user", "assistant"],
406 );
407 assert.include(launchTask, "Your job is just the file");
408 assert.include(conversation[1], "is 1 line long");
409 assert.include(conversation[2], "Look again at");
410 assert.include(conversation[3], "Bug/edge case found and fixed");
411 // The resumed run's own tool calls land in the child thread too.
412 const childCommands = (child?.turnItems ?? []).filter(
413 (item) => item.type === "command_execution",
414 );
415 assert.lengthOf(childCommands, 3);
416 assert.isFalse(
417 projection.turnItems.some(
418 (item) =>
419 item.type === "assistant_message" && item.text.includes("Bug/edge case found"),
420 ),
421 "the resumed subagent's reply leaked into the parent thread",
422 );
423 }).pipe(
424 provideDeterministicTestRuntime,
425 Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer)),
426 ),
427 ),
428 );
429});