Open History to see the last commit
1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { assert, describe, it } from "@effect/vitest";
4import * as Effect from "effect/Effect";
5import * as FileSystem from "effect/FileSystem";
6import * as Layer from "effect/Layer";
7import * as Path from "effect/Path";
8import * as Schema from "effect/Schema";
9import * as SqlClient from "effect/sql/SqlClient";
10import * as CodexReplay from "effect-codex-app-server/replay";
11import {
12 type OrchestrationV2ThreadProjection,
13 ProviderDriverKind,
14 type ThreadId,
15} from "@t3tools/contracts";
16
17import {
18 ClaudeOrchestratorReplayHarness,
19 makeClaudeRestartReplayHarness,
20} from "../Adapters/ClaudeAdapterV2.testkit.ts";
21import { CodexOrchestratorReplayHarness } from "../Adapters/CodexAdapterV2.testkit.ts";
22import * as CodexAdapterV2Testkit from "../Adapters/CodexAdapterV2.testkit.ts";
23import {
24 type CursorAgentSdkReplayTranscript,
25 CursorOrchestratorReplayHarness,
26 makeCursorAgentSdkReplayRunner,
27} from "../Adapters/CursorAdapterV2.testkit.ts";
28import * as CursorAdapterV2Testkit from "../Adapters/CursorAdapterV2.testkit.ts";
29import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
30import * as EffectWorker from "../EffectWorker.ts";
31import * as SqlitePersistence from "../../persistence/Sqlite.ts";
32import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts";
33import {
34 CLAUDE_MODEL_SELECTION,
35 CODEX_MODEL_SELECTION,
36 CURSOR_MODEL_SELECTION,
37 materializeFixtureInput,
38 type MaterializedOrchestratorFixtureInput,
39 PROVIDER_THREAD_RESUME_FIRST_PROMPT,
40 PROVIDER_THREAD_RESUME_SECOND_PROMPT,
41} from "./fixtures/shared.ts";
42import {
43 assertAssistantTextIncludes,
44 assertBaseProjection,
45 assertConversationMessageRoles,
46 assertRunOrdinals,
47 assertSemanticProjectionIntegrity,
48 assertTurnItemTypes,
49 assertUserMessagesInclude,
50 projectionFor,
51} from "./fixtures/shared.ts";
52import {
53 runOrchestratorV2Scenario,
54 type OrchestratorV2ScenarioResult,
55} from "./OrchestratorScenario.ts";
56import {
57 layerProviderReplay,
58 runOrchestratorV2ProviderReplayScenario,
59} from "./ProviderReplayHarness.ts";
60import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
61import { materializeReplayTranscriptRuntimeInstructions } from "./ReplayRuntimeInstructions.ts";
62import {
63 materializeReplayTranscriptWorkspace,
64 readProviderReplayTranscript,
65} from "@t3tools/provider-testing/replayTranscript";
66
67const FIRST_FINAL = "provider thread resume fixture first turn complete";
68const SECOND_FINAL = "provider thread resume fixture second turn complete";
69
70const decodeCodexTranscript = Schema.decodeUnknownEffect(
71 CodexReplay.CodexAppServerReplayTranscript,
72);
73const readRawTranscript = Effect.fn("readRecoveryTranscript")(function* (file: URL) {
74 return yield* readProviderReplayTranscript(file);
75});
76const readCodexTranscript = Effect.fn("readCodexRecoveryTranscript")(function* (workspace: string) {
77 const transcript = yield* readRawTranscript(
78 new URL("./fixtures/provider_thread_resume/codex_transcript.ndjson", import.meta.url),
79 );
80 return yield* decodeCodexTranscript(materializeReplayTranscriptWorkspace(transcript, workspace));
81});
82const decodePromptPair = Schema.decodeUnknownEffect(Schema.Tuple([Schema.String, Schema.String]));
83const readClaudeSubagentResumeTranscript = Effect.fn("readClaudeSubagentResumeTranscript")(
84 function* () {
85 return yield* ClaudeOrchestratorReplayHarness.decodeTranscript(
86 yield* readRawTranscript(
87 new URL(
88 "./fixtures/claude_subagent_resume_after_restart/claude_transcript.ndjson",
89 import.meta.url,
90 ),
91 ),
92 );
93 },
94);
95/** Remove the subagent's first run so restart cancels it after root settlement; resume carries the restart note. */
96const readClaudeCancelledSubagentResumeTranscript = Effect.fn(
97 "readClaudeCancelledSubagentResumeTranscript",
98)(function* () {
99 const recorded = yield* readRawTranscript(
100 new URL(
101 "./fixtures/claude_subagent_resume_after_restart/claude_transcript.ndjson",
102 import.meta.url,
103 ),
104 );
105 const frameField = (index: number, key: string): unknown => {
106 const entry = recorded.entries[index];
107 return entry?.type === "emit_inbound" && typeof entry.frame === "object" && entry.frame !== null
108 ? Reflect.get(entry.frame, key)
109 : undefined;
110 };
111 const indices = recorded.entries.map((_, index) => index);
112 const firstSubagentFrame = indices.find(
113 (index) => typeof frameField(index, "parent_tool_use_id") === "string",
114 );
115 const firstOutcome = indices.find(
116 (index) => frameField(index, "subtype") === "task_notification",
117 );
118 const firstExit = recorded.entries.findIndex((entry) => entry.type === "runtime_exit");
119 if (firstSubagentFrame === undefined || firstOutcome === undefined || firstExit < 0) {
120 throw new Error("Expected the recorded subagent's first run.");
121 }
122 const entries = recorded.entries
123 // A CLI exit settles the task; hold at query.open:2 until the crash snapshot instead.
124 .filter(
125 (_, index) => (index < firstSubagentFrame || index > firstOutcome) && index !== firstExit,
126 )
127 .map((entry) => {
128 if (entry.type !== "expect_outbound" || entry.label !== "prompt.offer:2") return entry;
129 const frame = entry.frame as {
130 readonly message: { readonly message: { readonly content: string } };
131 };
132 return {
133 ...entry,
134 frame: {
135 ...frame,
136 message: {
137 ...frame.message,
138 message: {
139 ...frame.message.message,
140 content: [
141 "Note: the T3 server restarted, and this background work was cancelled before it finished. It will not report back on its own, so resume or restart what you still need and wait for its result:",
142 "- subagent: Audit src/math.ts (id a44c0ed810da9e4f3)",
143 "",
144 "User message:",
145 frame.message.message.content,
146 ].join("\n"),
147 },
148 },
149 },
150 };
151 });
152 return yield* ClaudeOrchestratorReplayHarness.decodeTranscript({ ...recorded, entries });
153});
154const readCursorTranscript = Effect.fn("readCursorRecoveryTranscript")(function* () {
155 const transcript = yield* readRawTranscript(
156 new URL("./fixtures/provider_thread_resume/cursor_transcript.ndjson", import.meta.url),
157 );
158 return yield* CursorOrchestratorReplayHarness.decodeTranscript(
159 materializeReplayTranscriptRuntimeInstructions(transcript, {
160 driver: ProviderDriverKind.make("cursor"),
161 model: CURSOR_MODEL_SELECTION.model,
162 }),
163 );
164});
165
166function splitAfterFirstIdle(materialized: MaterializedOrchestratorFixtureInput) {
167 const splitIndex = materialized.steps.findIndex((step) => step.type === "await_thread_idle");
168 if (splitIndex < 0) {
169 throw new Error("Expected fixture to contain await_thread_idle after the first turn.");
170 }
171
172 const phase1Steps = materialized.steps.slice(0, splitIndex + 1);
173 const phase2Steps = materialized.steps.slice(splitIndex + 1);
174 return {
175 phase1Steps,
176 phase2Steps,
177 phase1Commands: phase1Steps.flatMap((step) => (step.type === "dispatch" ? [step.command] : [])),
178 phase2Commands: phase2Steps.flatMap((step) => (step.type === "dispatch" ? [step.command] : [])),
179 };
180}
181
182const runCursorRecovery = Effect.fn("runCursorRecovery")(function* (input: {
183 readonly transcript: CursorAgentSdkReplayTranscript;
184 readonly runner: ReturnType<typeof makeCursorAgentSdkReplayRunner>;
185}) {
186 const fs = yield* FileSystem.FileSystem;
187 const path = yield* Path.Path;
188 const tempDir = yield* Effect.acquireRelease(
189 fs.makeTempDirectory({
190 prefix: "t3-orchestration-v2-cursor-recovery-",
191 }),
192 (directory) => fs.remove(directory, { recursive: true, force: true }).pipe(Effect.orDie),
193 );
194 yield* fs.makeDirectory(tempDir, { recursive: true });
195 const dbPath = path.join(tempDir, "state.sqlite");
196 const materialized = yield* materializeFixtureInput({
197 scenario: "provider_thread_resume",
198 fixtureInput: {
199 steps: [
200 { type: "message", text: PROVIDER_THREAD_RESUME_FIRST_PROMPT },
201 { type: "message", text: PROVIDER_THREAD_RESUME_SECOND_PROMPT },
202 ],
203 },
204 driver: ProviderDriverKind.make("cursor"),
205 modelSelection: CURSOR_MODEL_SELECTION,
206 });
207 const { phase1Commands, phase1Steps, phase2Commands, phase2Steps } =
208 splitAfterFirstIdle(materialized);
209 const options = {
210 databaseLayer: SqlitePersistence.layerFromPath(dbPath).pipe(
211 Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
212 ),
213 };
214 const harness = {
215 ...CursorOrchestratorReplayHarness,
216 makeProviderAdapterRegistryLayer: () =>
217 CursorAdapterV2Testkit.layer(input.transcript, {
218 runner: input.runner,
219 assertCompleteOnFinalize: false,
220 }),
221 };
222
223 yield* Effect.scoped(
224 runOrchestratorV2ProviderReplayScenario(
225 {
226 name: "provider_thread_resume/cursor:first-runtime",
227 transcript: input.transcript,
228 commands: phase1Commands,
229 steps: phase1Steps,
230 projectionThreadIds: materialized.projectionThreadIds,
231 runtimePolicyOverride: { cwd: tempDir },
232 },
233 harness,
234 options,
235 ),
236 );
237
238 const result = yield* Effect.scoped(
239 runOrchestratorV2ProviderReplayScenario(
240 {
241 name: "provider_thread_resume/cursor:second-runtime",
242 transcript: input.transcript,
243 commands: phase2Commands,
244 steps: phase2Steps,
245 projectionThreadIds: materialized.projectionThreadIds,
246 runtimePolicyOverride: { cwd: tempDir },
247 },
248 harness,
249 // A real restart requeues effects, like checkpoint capture, the first runtime left running.
250 { ...options, recoverOnStartup: true },
251 ),
252 );
253
254 assertBaseProjection({
255 result,
256 transcript: input.transcript,
257 runCount: 2,
258 runStatuses: ["completed", "completed"],
259 });
260 const projection = projectionFor(result, input.transcript.scenario);
261 assertSemanticProjectionIntegrity(projection);
262 assertRunOrdinals(projection, [1, 2]);
263 assertConversationMessageRoles(projection, ["user", "assistant", "user", "assistant"]);
264 assertTurnItemTypes(projection, ["user_message", "assistant_message"]);
265 assertUserMessagesInclude(projection, [
266 PROVIDER_THREAD_RESUME_FIRST_PROMPT,
267 PROVIDER_THREAD_RESUME_SECOND_PROMPT,
268 ]);
269 assertAssistantTextIncludes(projection, FIRST_FINAL);
270 assertAssistantTextIncludes(projection, SECOND_FINAL);
271 assert.lengthOf(projection.providerThreads, 1);
272});
273
274describe("orchestrator replay recovery", () => {
275 it.effect(
276 "resumes a provider-native Codex thread after recreating the orchestrator runtime",
277 () =>
278 Effect.scoped(
279 Effect.gen(function* () {
280 // Use a throwaway checkpoint workspace; cold full-repo capture exceeds CI wait budgets.
281 const workspace = yield* checkpointWorkspace("provider_thread_resume");
282 const transcript = yield* readCodexTranscript(workspace);
283 const fs = yield* FileSystem.FileSystem;
284 const path = yield* Path.Path;
285 const tempDir = yield* Effect.acquireRelease(
286 fs.makeTempDirectory({
287 prefix: "t3-orchestration-v2-recovery-",
288 }),
289 (directory) =>
290 fs.remove(directory, { recursive: true, force: true }).pipe(Effect.orDie),
291 );
292 yield* fs.makeDirectory(tempDir, { recursive: true });
293 const dbPath = path.join(tempDir, "state.sqlite");
294 const driver = yield* CodexReplay.makeReplayDriver(transcript);
295 const materialized = yield* materializeFixtureInput({
296 scenario: "provider_thread_resume",
297 fixtureInput: {
298 steps: [
299 { type: "message", text: PROVIDER_THREAD_RESUME_FIRST_PROMPT },
300 { type: "message", text: PROVIDER_THREAD_RESUME_SECOND_PROMPT },
301 ],
302 },
303 driver: ProviderDriverKind.make("codex"),
304 modelSelection: CODEX_MODEL_SELECTION,
305 });
306 const { phase1Commands, phase1Steps, phase2Commands, phase2Steps } =
307 splitAfterFirstIdle(materialized);
308
309 const harness = {
310 ...CodexOrchestratorReplayHarness,
311 makeProviderAdapterRegistryLayer: () =>
312 CodexAdapterV2Testkit.layer({ transcript, driver }),
313 };
314 const options = {
315 databaseLayer: SqlitePersistence.layerFromPath(dbPath).pipe(
316 Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
317 ),
318 };
319
320 yield* runOrchestratorV2ProviderReplayScenario(
321 {
322 name: "provider_thread_resume/codex:first-runtime",
323 transcript,
324 commands: phase1Commands,
325 steps: phase1Steps,
326 projectionThreadIds: materialized.projectionThreadIds,
327 runtimePolicyOverride: { cwd: workspace },
328 },
329 harness,
330 options,
331 );
332
333 const result = yield* runOrchestratorV2ProviderReplayScenario(
334 {
335 name: "provider_thread_resume/codex:second-runtime",
336 transcript,
337 commands: phase2Commands,
338 steps: phase2Steps,
339 projectionThreadIds: materialized.projectionThreadIds,
340 runtimePolicyOverride: { cwd: workspace },
341 },
342 harness,
343 { ...options, recoverOnStartup: true },
344 );
345
346 assertBaseProjection({
347 result,
348 transcript,
349 runCount: 2,
350 runStatuses: ["completed", "completed"],
351 });
352 const projection = projectionFor(result, transcript.scenario);
353 assertSemanticProjectionIntegrity(projection);
354 assertRunOrdinals(projection, [1, 2]);
355 assertConversationMessageRoles(projection, ["user", "assistant", "user", "assistant"]);
356 assertTurnItemTypes(projection, ["user_message", "assistant_message"]);
357 assertUserMessagesInclude(projection, [
358 PROVIDER_THREAD_RESUME_FIRST_PROMPT,
359 PROVIDER_THREAD_RESUME_SECOND_PROMPT,
360 ]);
361 assertAssistantTextIncludes(projection, FIRST_FINAL);
362 assertAssistantTextIncludes(projection, SECOND_FINAL);
363 assert.lengthOf(projection.providerThreads, 1);
364 }).pipe(
365 provideDeterministicTestRuntime,
366 Effect.provide(
367 Layer.merge(IdAllocator.layer, Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
368 ),
369 ),
370 ),
371 );
372
373 it.effect(
374 "resumes a provider-native Cursor thread after recreating the orchestrator runtime",
375 () =>
376 Effect.scoped(
377 Effect.gen(function* () {
378 const transcript = yield* readCursorTranscript();
379 const runner = makeCursorAgentSdkReplayRunner(transcript);
380 yield* runCursorRecovery({ transcript, runner });
381 yield* runner.assertComplete;
382 }).pipe(
383 provideDeterministicTestRuntime,
384 Effect.provide(
385 Layer.merge(IdAllocator.layer, Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
386 ),
387 ),
388 ),
389 );
390 it.effect("keeps a Claude subagent resumed after a server restart in its own thread", () =>
391 Effect.scoped(
392 Effect.gen(function* () {
393 const transcript = yield* readClaudeSubagentResumeTranscript();
394 const [launchPrompt, resumePrompt] = yield* decodePromptPair(transcript.metadata?.prompts);
395 const workspace = yield* checkpointWorkspace(transcript.scenario);
396 const fs = yield* FileSystem.FileSystem;
397 const path = yield* Path.Path;
398 const tempDir = yield* Effect.acquireRelease(
399 fs.makeTempDirectory({ prefix: "t3-orchestration-v2-claude-recovery-" }),
400 (directory) => fs.remove(directory, { recursive: true, force: true }).pipe(Effect.orDie),
401 );
402 const materialized = yield* materializeFixtureInput({
403 scenario: transcript.scenario,
404 fixtureInput: {
405 steps: [
406 { type: "message", text: launchPrompt },
407 { type: "message", text: resumePrompt },
408 ],
409 },
410 driver: ProviderDriverKind.make("claudeAgent"),
411 modelSelection: CLAUDE_MODEL_SELECTION,
412 });
413 const { phase1Commands, phase1Steps, phase2Commands, phase2Steps } =
414 splitAfterFirstIdle(materialized);
415 const { harness, assertComplete } = makeClaudeRestartReplayHarness(transcript);
416 const options = {
417 databaseLayer: SqlitePersistence.layerFromPath(path.join(tempDir, "state.sqlite")).pipe(
418 Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
419 ),
420 };
421
422 yield* Effect.scoped(
423 runOrchestratorV2ProviderReplayScenario(
424 {
425 name: `${transcript.scenario}:first-runtime`,
426 transcript,
427 commands: phase1Commands,
428 steps: phase1Steps,
429 projectionThreadIds: materialized.projectionThreadIds,
430 runtimePolicyOverride: { cwd: workspace },
431 },
432 harness,
433 options,
434 ),
435 );
436 const result = yield* Effect.scoped(
437 runOrchestratorV2ProviderReplayScenario(
438 {
439 name: `${transcript.scenario}:second-runtime`,
440 transcript,
441 commands: phase2Commands,
442 steps: phase2Steps,
443 projectionThreadIds: materialized.projectionThreadIds,
444 runtimePolicyOverride: { cwd: workspace },
445 },
446 harness,
447 { ...options, recoverOnStartup: true },
448 ),
449 );
450 yield* assertComplete;
451
452 const projection = projectionFor(result, transcript.scenario);
453 assertSemanticProjectionIntegrity(projection);
454 assert.lengthOf(projection.subagents, 1);
455 const subagent = projection.subagents[0];
456 assert.equal(subagent?.status, "completed");
457 const childThreads = [...result.projections.values()].filter(
458 (candidate) => candidate.thread.lineage.parentThreadId === projection.thread.id,
459 );
460 assert.lengthOf(childThreads, 1);
461 const child = childThreads[0];
462 assert.equal(child?.thread.id, subagent?.childThreadId);
463 assert.deepEqual(child?.thread.modelSelection, {
464 instanceId: CLAUDE_MODEL_SELECTION.instanceId,
465 model: "claude-sonnet-5",
466 });
467
468 const conversation = (child?.turnItems ?? [])
469 .toSorted((left, right) => left.ordinal - right.ordinal)
470 .flatMap((item) =>
471 item.type === "user_message" || item.type === "assistant_message"
472 ? [`${item.type === "user_message" ? "user" : "assistant"}:${item.text}`]
473 : [],
474 );
475 const launchTask = conversation[0];
476 assert.deepEqual(
477 conversation.map((entry) => entry.slice(0, entry.indexOf(":"))),
478 ["user", "assistant", "user", "assistant"],
479 );
480 assert.include(launchTask, "Your job is just the file");
481 assert.include(conversation[1], "is 1 line long");
482 assert.include(conversation[2], "Look again at");
483 assert.include(conversation[3], "Bug/edge case found and fixed");
484 const childCommands = (child?.turnItems ?? []).filter(
485 (item) => item.type === "command_execution",
486 );
487 assert.lengthOf(childCommands, 3);
488 assert.isFalse(
489 projection.turnItems.some(
490 (item) =>
491 item.type === "assistant_message" && item.text.includes("Bug/edge case found"),
492 ),
493 "the resumed subagent's reply leaked into the parent thread",
494 );
495 }).pipe(
496 provideDeterministicTestRuntime,
497 Effect.provide(
498 Layer.merge(IdAllocator.layer, Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
499 ),
500 ),
501 ),
502 );
503 it.effect("reopens the thread of a Claude subagent the restart cancelled when resumed", () =>
504 Effect.scoped(
505 Effect.gen(function* () {
506 const transcript = yield* readClaudeCancelledSubagentResumeTranscript();
507 const [launchPrompt, resumePrompt] = yield* decodePromptPair(transcript.metadata?.prompts);
508 const workspace = yield* checkpointWorkspace(transcript.scenario);
509 const fs = yield* FileSystem.FileSystem;
510 const path = yield* Path.Path;
511 const tempDir = yield* Effect.acquireRelease(
512 fs.makeTempDirectory({ prefix: "t3-orchestration-v2-claude-recovery-" }),
513 (directory) => fs.remove(directory, { recursive: true, force: true }).pipe(Effect.orDie),
514 );
515 const materialized = yield* materializeFixtureInput({
516 scenario: transcript.scenario,
517 fixtureInput: {
518 steps: [
519 { type: "message", text: launchPrompt },
520 { type: "message", text: resumePrompt },
521 ],
522 },
523 driver: ProviderDriverKind.make("claudeAgent"),
524 modelSelection: CLAUDE_MODEL_SELECTION,
525 });
526 const { phase1Commands, phase1Steps, phase2Commands, phase2Steps } =
527 splitAfterFirstIdle(materialized);
528 const { harness, assertComplete } = makeClaudeRestartReplayHarness(transcript);
529 const options = {
530 databaseLayer: SqlitePersistence.layerFromPath(path.join(tempDir, "state.sqlite")).pipe(
531 Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
532 ),
533 };
534 const childThreadOf = (result: OrchestratorV2ScenarioResult, parentThreadId: ThreadId) =>
535 [...result.projections.values()].find(
536 (candidate) => candidate.thread.lineage.parentThreadId === parentThreadId,
537 );
538 const childRootStatus = (child: OrchestrationV2ThreadProjection | undefined) =>
539 child?.nodes.find((node) => node.kind === "root_turn" && node.runId === null)?.status;
540
541 const snapshotPath = path.join(tempDir, "restart-state.sqlite");
542 const beforeScenario = {
543 name: `${transcript.scenario}:cancelled-first-runtime`,
544 transcript,
545 commands: phase1Commands,
546 steps: phase1Steps,
547 projectionThreadIds: materialized.projectionThreadIds,
548 runtimePolicyOverride: { cwd: workspace },
549 };
550 const before = yield* Effect.scoped(
551 Effect.gen(function* () {
552 const result = yield* runOrchestratorV2Scenario(beforeScenario);
553 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
554 yield* worker.drain();
555 // Preserve the orphan before scoped teardown settles its provider work.
556 const sql = yield* SqlClient.SqlClient;
557 yield* sql`VACUUM INTO ${snapshotPath}`;
558 return result;
559 }).pipe(
560 Effect.provide(
561 Layer.merge(
562 layerProviderReplay(beforeScenario, harness, options),
563 options.databaseLayer,
564 ),
565 ),
566 ),
567 );
568 const settled = projectionFor(before, transcript.scenario);
569 assert.equal(settled.subagents[0]?.status, "running");
570 assert.equal(childRootStatus(childThreadOf(before, settled.thread.id)), "running");
571
572 const result = yield* Effect.scoped(
573 runOrchestratorV2ProviderReplayScenario(
574 {
575 name: `${transcript.scenario}:cancelled-second-runtime`,
576 transcript,
577 commands: phase2Commands,
578 steps: phase2Steps,
579 projectionThreadIds: materialized.projectionThreadIds,
580 runtimePolicyOverride: { cwd: workspace },
581 },
582 harness,
583 {
584 databaseLayer: SqlitePersistence.layerFromPath(snapshotPath).pipe(
585 Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
586 ),
587 recoverOnStartup: true,
588 continueThreadsAfterServerUpdate: false,
589 },
590 ),
591 );
592 yield* assertComplete;
593
594 const projection = projectionFor(result, transcript.scenario);
595 assertSemanticProjectionIntegrity(projection);
596 assert.equal(projection.subagents[0]?.status, "completed");
597 const child = childThreadOf(result, projection.thread.id);
598 assert.equal(child?.thread.id, projection.subagents[0]?.childThreadId);
599 assert.equal(childRootStatus(child), "completed");
600 const conversation = (child?.turnItems ?? [])
601 .toSorted((left, right) => left.ordinal - right.ordinal)
602 .flatMap((item) =>
603 item.type === "user_message" || item.type === "assistant_message"
604 ? [`${item.type === "user_message" ? "user" : "assistant"}:${item.text}`]
605 : [],
606 );
607 assert.deepEqual(
608 conversation.map((entry) => entry.slice(0, entry.indexOf(":"))),
609 ["user", "user", "assistant"],
610 );
611 assert.include(conversation[0], "Your job is just the file");
612 assert.include(conversation[1], "Look again at");
613 assert.include(conversation[2], "Bug/edge case found and fixed");
614 assert.lengthOf(
615 (child?.turnItems ?? []).filter((item) => item.type === "command_execution"),
616 2,
617 );
618 }).pipe(
619 provideDeterministicTestRuntime,
620 Effect.provide(
621 Layer.merge(IdAllocator.layer, Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
622 ),
623 ),
624 ),
625 );
626});