1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, describe, it } from "@effect/vitest";
3import type {
4 OrchestrationV2DomainEvent,
5 ProviderReplayEntry,
6 ProviderReplayTranscript,
7} from "@t3tools/contracts";
8import * as Effect from "effect/Effect";
9import * as FileSystem from "effect/FileSystem";
10import * as Path from "effect/Path";
11
12import { ClaudeOrchestratorReplayHarness } from "../Adapters/ClaudeAdapterV2.testkit.ts";
13import { CodexOrchestratorReplayHarness } from "../Adapters/CodexAdapterV2.testkit.ts";
14import { CursorOrchestratorReplayHarness } from "../Adapters/CursorAdapterV2.testkit.ts";
15import { AcpRegistryOrchestratorReplayHarness } from "../Adapters/AcpRegistryAdapterV2.testkit.ts";
16import { GrokOrchestratorReplayHarness } from "../Adapters/GrokAdapterV2.testkit.ts";
17import { OpenCodeOrchestratorReplayHarness } from "../Adapters/OpenCodeAdapterV2.testkit.ts";
18import {
19 OPENCODE2_HTTP_PROTOCOL,
20 OpenCode2OrchestratorReplayHarness,
21} from "../Adapters/OpenCode2AdapterV2.testkit.ts";
22import {
23 materializeMuseReplayWorkspace,
24 MuseOrchestratorReplayHarness,
25} from "../Adapters/MuseAdapterV2.testkit.ts";
26import { PiOrchestratorReplayHarness } from "../Adapters/PiAdapterV2.testkit.ts";
27import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
28import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts";
29import { ORCHESTRATOR_REPLAY_FIXTURES } from "./fixtures/index.ts";
30import { messageRestartInput } from "./fixtures/message_steering/input.ts";
31import {
32 assertProviderNativeSubagentRootTurns,
33 materializeFixtureInput,
34 projectionFor,
35 type OrchestratorFixtureInput,
36 type ProviderOrchestratorReplayVariant,
37} from "./fixtures/shared.ts";
38import {
39 runOrchestratorV2ProviderReplayScenario,
40 type OrchestratorV2ProviderReplayHarness,
41} from "./ProviderReplayHarness.ts";
42import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
43import { materializeReplayTranscriptRuntimeInstructions } from "./ReplayRuntimeInstructions.ts";
44import {
45 materializeReplayTranscriptWorkspace,
46 readProviderReplayTranscript,
47} from "@t3tools/provider-testing/replayTranscript";
48
49const readTranscript = Effect.fn("readOrchestratorReplayTranscript")(function* (file: URL) {
50 return yield* readProviderReplayTranscript(file);
51}, Effect.provide(NodeServices.layer));
52
53function normalizeTestError(cause: unknown): Error {
54 return cause instanceof Error ? cause : new Error(String(cause));
55}
56
57function transcriptEntriesThroughLabel(
58 transcript: ProviderReplayTranscript,
59 label: string | undefined,
60): ProviderReplayTranscript {
61 if (label === undefined) {
62 return transcript;
63 }
64 const entryIndex = transcript.entries.findIndex(
65 (entry) => entry.type === "emit_inbound" && entry.label === label,
66 );
67 if (entryIndex === -1) {
68 throw new Error(`${transcript.scenario} is missing inbound transcript label ${label}.`);
69 }
70 return {
71 ...transcript,
72 entries: transcript.entries.slice(0, entryIndex + 1),
73 };
74}
75
76function isStreamingAssistantEvent(event: OrchestrationV2DomainEvent): boolean {
77 switch (event.type) {
78 case "node.updated":
79 return event.payload.kind === "assistant_message" && event.payload.status === "running";
80 case "message.updated":
81 return event.payload.role === "assistant" && event.payload.streaming;
82 case "turn-item.updated":
83 return event.payload.type === "assistant_message" && event.payload.streaming;
84 default:
85 return false;
86 }
87}
88
89const runFixtureProvider = Effect.fn("runOrchestratorReplayFixture")(function* <
90 Transcript extends ProviderReplayTranscript,
91 Error,
92>(input: {
93 readonly fixtureName: string;
94 readonly buildInput: () => OrchestratorFixtureInput;
95 readonly driver: ProviderOrchestratorReplayVariant;
96 readonly harness: OrchestratorV2ProviderReplayHarness<Transcript, Error>;
97 readonly transformTranscript?: (transcript: ProviderReplayTranscript) => ProviderReplayTranscript;
98}) {
99 const recordedTranscript = yield* readTranscript(input.driver.transcriptFile);
100 const rawTranscript = input.transformTranscript?.(recordedTranscript) ?? recordedTranscript;
101 const replayTranscript = materializeReplayTranscriptRuntimeInstructions(
102 transcriptEntriesThroughLabel(rawTranscript, input.driver.transcriptEntriesThroughLabel),
103 { driver: input.driver.driver, model: input.driver.modelSelection.model },
104 );
105 const fixtureInput = input.buildInput();
106 const workspace = yield* checkpointWorkspace(input.fixtureName, fixtureInput.workspaceFiles);
107 const transcript = yield* input.harness.decodeTranscript(
108 input.driver.driver === "codex"
109 ? materializeReplayTranscriptWorkspace(replayTranscript, workspace)
110 : input.driver.driver === "muse"
111 ? yield* materializeMuseReplayWorkspace(replayTranscript, workspace)
112 : replayTranscript,
113 );
114 const materialized = yield* materializeFixtureInput({
115 scenario: input.fixtureName,
116 fixtureInput,
117 driver: input.driver.driver,
118 modelSelection: input.driver.modelSelection,
119 }).pipe(Effect.provide(IdAllocator.layer), provideDeterministicTestRuntime);
120 const scenario = {
121 name: `${input.fixtureName}/${input.driver.driver}`,
122 transcript,
123 commands: materialized.commands,
124 steps: materialized.steps,
125 projectionThreadIds: materialized.projectionThreadIds,
126 runtimePolicyOverride: {
127 ...input.driver.runtimePolicyOverride,
128 cwd: workspace,
129 },
130 };
131
132 const result = yield* runOrchestratorV2ProviderReplayScenario(
133 scenario,
134 input.harness,
135 input.driver.runContinuationWorker === true ? { runContinuationWorker: true } : {},
136 ).pipe(provideDeterministicTestRuntime);
137 input.driver.assertOutput(result, transcript);
138 assertProviderNativeSubagentRootTurns(result);
139 const expectedAbsentWorkspacePaths = input.driver.expectedAbsentWorkspacePaths;
140 if (expectedAbsentWorkspacePaths !== undefined) {
141 yield* Effect.gen(function* () {
142 const fs = yield* FileSystem.FileSystem;
143 const path = yield* Path.Path;
144 for (const relativePath of expectedAbsentWorkspacePaths) {
145 assert.isFalse(
146 yield* fs.exists(path.join(workspace, relativePath)),
147 `${input.fixtureName}/${input.driver.driver} must not create ${relativePath} in the replay workspace`,
148 );
149 }
150 }).pipe(Effect.provide(NodeServices.layer));
151 }
152 assert.isFalse(
153 result.domainEvents.some(isStreamingAssistantEvent),
154 "buffered delivery must not persist streaming assistant artifacts",
155 );
156 const projectionThreadId = materialized.projectionThreadIds[0];
157 assert.isDefined(projectionThreadId);
158 const projection = result.projections.get(projectionThreadId);
159 assert.isDefined(projection);
160 const latestRun = projection.runs.at(-1);
161 assert.deepEqual(latestRun?.modelSelection, input.driver.modelSelection);
162 if (projection.runs.some((run) => run.status === "completed")) {
163 const threadStartCheckpoint = projection.checkpoints.find(
164 (checkpoint) => checkpoint.ordinalWithinScope === 0 && checkpoint.appRunOrdinal === null,
165 );
166 assert.isDefined(
167 threadStartCheckpoint,
168 "completed threads must retain an addressable thread-start checkpoint",
169 );
170 assert.equal(threadStartCheckpoint.status, "ready");
171 }
172 return result;
173});
174
175function runFixtureProviderWithRegisteredHarness(input: {
176 readonly fixtureName: string;
177 readonly buildInput: () => OrchestratorFixtureInput;
178 readonly driver: ProviderOrchestratorReplayVariant;
179 readonly transformTranscript?: (transcript: ProviderReplayTranscript) => ProviderReplayTranscript;
180}) {
181 switch (input.driver.driver) {
182 case "codex":
183 return runFixtureProvider({
184 ...input,
185 harness: CodexOrchestratorReplayHarness,
186 }).pipe(Effect.mapError(normalizeTestError), Effect.scoped);
187 case "claudeAgent":
188 return runFixtureProvider({
189 ...input,
190 harness: ClaudeOrchestratorReplayHarness,
191 }).pipe(Effect.mapError(normalizeTestError), Effect.scoped);
192 case "cursor":
193 return runFixtureProvider({
194 ...input,
195 harness: CursorOrchestratorReplayHarness,
196 }).pipe(Effect.mapError(normalizeTestError), Effect.scoped);
197 case "grok":
198 return runFixtureProvider({
199 ...input,
200 harness: GrokOrchestratorReplayHarness,
201 }).pipe(Effect.mapError(normalizeTestError), Effect.scoped);
202 case "acpRegistry":
203 return runFixtureProvider({
204 ...input,
205 harness: AcpRegistryOrchestratorReplayHarness,
206 }).pipe(Effect.mapError(normalizeTestError), Effect.scoped);
207 case "opencode":
208 // One driver, two runtimes: the transcript's protocol says which one recorded it.
209 return readTranscript(input.driver.transcriptFile).pipe(
210 Effect.flatMap((transcript) =>
211 transcript.protocol === OPENCODE2_HTTP_PROTOCOL
212 ? runFixtureProvider({ ...input, harness: OpenCode2OrchestratorReplayHarness })
213 : runFixtureProvider({ ...input, harness: OpenCodeOrchestratorReplayHarness }),
214 ),
215 Effect.mapError(normalizeTestError),
216 Effect.scoped,
217 );
218 case "pi":
219 return runFixtureProvider({
220 ...input,
221 harness: PiOrchestratorReplayHarness,
222 }).pipe(Effect.mapError(normalizeTestError), Effect.scoped);
223 case "muse":
224 return runFixtureProvider({
225 ...input,
226 harness: MuseOrchestratorReplayHarness,
227 }).pipe(Effect.mapError(normalizeTestError), Effect.scoped);
228 default:
229 return Effect.die(
230 new Error(`No replay harness registered for provider ${input.driver.driver}.`),
231 );
232 }
233}
234
235describe("orchestrator replay fixtures", () => {
236 it.effect.each(
237 ORCHESTRATOR_REPLAY_FIXTURES.flatMap((fixture) =>
238 fixture.providers.map(
239 (provider) => [fixture.name, provider.driver, fixture, provider] as const,
240 ),
241 ),
242 )("runs %s/%s through OrchestratorV2 using deterministic replay", ([, , fixture, provider]) =>
243 runFixtureProviderWithRegisteredHarness({
244 fixtureName: fixture.name,
245 buildInput: fixture.buildInput,
246 driver: provider,
247 }),
248 );
249
250 const steeringFixture = ORCHESTRATOR_REPLAY_FIXTURES.find(
251 (fixture) => fixture.name === "message_steering",
252 );
253 const cursorSteeringProvider = steeringFixture?.providers.find(
254 (provider) => provider.driver === "cursor",
255 );
256 if (cursorSteeringProvider !== undefined) {
257 it.effect("executes explicit Cursor restart_active through the recorded SDK boundary", () =>
258 runFixtureProviderWithRegisteredHarness({
259 fixtureName: "message_steering",
260 buildInput: messageRestartInput,
261 driver: cursorSteeringProvider,
262 }),
263 );
264 }
265
266 // A later OpenCode may change an execution start's shape; the client then
267 // reads it as `unreadable.execution.started`. A subagent's turn and a
268 // background follow-up must still start from it.
269 it.effect.each(
270 (
271 [
272 ["opencode2_subagent", "session.execution.started.2"],
273 ["opencode2_background", "session.execution.started.3"],
274 ] as const
275 ).flatMap(([fixtureName, label]) => {
276 const fixture = ORCHESTRATOR_REPLAY_FIXTURES.find(
277 (candidate) => candidate.name === fixtureName,
278 );
279 const provider = fixture?.providers[0];
280 return fixture === undefined || provider === undefined
281 ? []
282 : [[fixtureName, label, fixture, provider] as const];
283 }),
284 )(
285 "runs %s when %s is an execution start this build cannot decode",
286 ([fixtureName, label, fixture, provider]) =>
287 runFixtureProviderWithRegisteredHarness({
288 fixtureName,
289 buildInput: fixture.buildInput,
290 driver: provider,
291 transformTranscript: (transcript) => ({
292 ...transcript,
293 entries: transcript.entries.map((entry) =>
294 entry.type === "emit_inbound" && entry.label === label
295 ? undecodableEvent(entry)
296 : entry,
297 ),
298 }),
299 }),
300 );
301
302 // A background subagent opens a WebFetch or Write, then stops without
303 // returning its result. The continuation that drains the stop must leave the
304 // child's tool row and node terminal, whether the call opened before the
305 // root settled or while it was idle.
306 const afterRootFixture = ORCHESTRATOR_REPLAY_FIXTURES.find(
307 (candidate) => candidate.name === "claude_background_subagent_after_root",
308 );
309 const afterRootProvider = afterRootFixture?.providers[0];
310 if (afterRootFixture !== undefined && afterRootProvider !== undefined) {
311 const OPEN_TOOL_ID = "toolu_01QDa5jV5g1H9h6QyDfeJohD";
312 const SUBAGENT_TASK_ID = "a2995bfced8019363";
313
314 const moveOpenToolBeforeRootResult = (
315 move: boolean,
316 entries: ReadonlyArray<ProviderReplayEntry>,
317 ): ReadonlyArray<ProviderReplayEntry> => {
318 if (!move) return entries;
319 const toolIndex = entries.findIndex(
320 (entry) =>
321 entry.type === "emit_inbound" &&
322 (entry.frame as Record<string, any>).message?.content?.[0]?.id === OPEN_TOOL_ID,
323 );
324 const resultIndex = entries.findIndex(
325 (entry) => entry.type === "emit_inbound" && entry.label === "result",
326 );
327 if (toolIndex < 0 || resultIndex < 0) throw new Error("transcript shape changed");
328 const toolEntry = entries[toolIndex]!;
329 const rest = entries.filter((_, index) => index !== toolIndex);
330 return [...rest.slice(0, resultIndex), toolEntry, ...rest.slice(resultIndex)];
331 };
332 it.effect.each(
333 (["WebFetch", "Write"] as const).flatMap((tool) =>
334 (["before root settles", "while root is idle"] as const).flatMap((opens) =>
335 (["stopped", "failed"] as const).map((status) => [tool, opens, status] as const),
336 ),
337 ),
338 )(
339 "ends a background subagent's open %s (opened %s) when its notification is %s",
340 ([tool, opens, notificationStatus]) =>
341 runFixtureProviderWithRegisteredHarness({
342 fixtureName: afterRootFixture.name,
343 buildInput: afterRootFixture.buildInput,
344 driver: {
345 ...afterRootProvider,
346 assertOutput: (result) => {
347 const parent = projectionFor(result, afterRootFixture.name);
348 assert.deepEqual(
349 parent.runs.map((run) => run.status),
350 ["completed", "completed"],
351 );
352 const subagent = parent.subagents[0];
353 assert.equal(
354 subagent?.status,
355 notificationStatus === "stopped" ? "cancelled" : "failed",
356 );
357 const childThreadId = subagent?.childThreadId;
358 assert.exists(childThreadId);
359 const child = result.projections.get(childThreadId);
360 assert.exists(child);
361 const fetches = child.turnItems.filter(
362 (item) => item.type === (tool === "WebFetch" ? "web_search" : "file_change"),
363 );
364 assert.lengthOf(fetches, 1, `the ${tool} reached the child thread`);
365 // The stored tool row and every child node end terminal.
366 assert.deepEqual(
367 {
368 openTool: fetches.map((item) => item.status),
369 openChildNodes: child.nodes
370 .filter(
371 (node) =>
372 node.status === "running" ||
373 node.status === "pending" ||
374 node.status === "waiting",
375 )
376 .map((node) => `${node.kind}:${node.status}`),
377 },
378 {
379 openTool: [notificationStatus === "stopped" ? "interrupted" : "failed"],
380 openChildNodes: [],
381 },
382 );
383 },
384 },
385 transformTranscript: (transcript) => ({
386 ...transcript,
387 entries: moveOpenToolBeforeRootResult(
388 opens === "before root settles",
389 transcript.entries.flatMap((entry): ReadonlyArray<ProviderReplayEntry> => {
390 if (entry.type !== "emit_inbound") return [entry];
391 const frame = entry.frame as Record<string, any>;
392 // The subagent's second step becomes a call that never returns.
393 if (
394 frame.type === "assistant" &&
395 frame.message?.content?.[0]?.id === OPEN_TOOL_ID
396 ) {
397 return [
398 {
399 ...entry,
400 frame: {
401 ...frame,
402 message: {
403 ...frame.message,
404 content: [
405 {
406 ...frame.message.content[0],
407 name: tool,
408 input:
409 tool === "WebFetch"
410 ? { url: "https://example.com", prompt: "Summarize" }
411 : { file_path: "notes.txt", content: "SUB_DONE_2" },
412 },
413 ],
414 },
415 },
416 },
417 ];
418 }
419 // Drop the Bash background task, the tool result, and the final report.
420 if (
421 (frame.type === "system" &&
422 (frame.subtype === "task_started" || frame.subtype === "task_notification") &&
423 frame.tool_use_id === OPEN_TOOL_ID) ||
424 (frame.type === "user" &&
425 frame.message?.content?.[0]?.tool_use_id === OPEN_TOOL_ID) ||
426 (frame.type === "assistant" &&
427 frame.parent_tool_use_id !== null &&
428 frame.message?.content?.[0]?.text === "SUB_FINAL_REPORT")
429 ) {
430 return [];
431 }
432 if (frame.type === "system" && frame.subtype === "task_updated") {
433 return [
434 {
435 ...entry,
436 frame: {
437 ...frame,
438 patch: {
439 ...frame.patch,
440 status: notificationStatus === "stopped" ? "killed" : "failed",
441 },
442 },
443 },
444 ];
445 }
446 if (
447 frame.type === "system" &&
448 frame.subtype === "task_notification" &&
449 frame.task_id === SUBAGENT_TASK_ID
450 ) {
451 return [{ ...entry, frame: { ...frame, status: notificationStatus } }];
452 }
453 return [entry];
454 }),
455 ),
456 }),
457 }),
458 );
459 }
460});
461
462/** The same event with an envelope this build cannot decode, as a newer OpenCode may send. */
463function undecodableEvent(entry: ProviderReplayEntry): ProviderReplayEntry {
464 if (entry.type !== "emit_inbound") return entry;
465 const frame = entry.frame as { readonly event: Record<string, unknown> };
466 return { ...entry, frame: { ...frame, event: { ...frame.event, durable: "not-an-envelope" } } };
467}