Open History to see the last commit
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, it } from "@effect/vitest";
3import type { OrchestrationV2DomainEvent, ProviderReplayTranscript } from "@t3tools/contracts";
4import * as Effect from "effect/Effect";
5import * as FileSystem from "effect/FileSystem";
6import * as Path from "effect/Path";
7
8import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
9import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
10import { readProviderReplayTranscript } from "@t3tools/provider-testing/replayTranscript";
11import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts";
12import { ORCHESTRATOR_REPLAY_FIXTURES } from "./fixtures/index.ts";
13import {
14 assertProviderNativeSubagentRootTurns,
15 materializeFixtureInput,
16 type OrchestratorFixtureInput,
17 type ProviderOrchestratorReplayVariant,
18} from "./fixtures/shared.ts";
19import {
20 runOrchestratorV2ProviderReplayScenario,
21 type OrchestratorV2ProviderReplayHarness,
22} from "./ProviderReplayHarness.ts";
23import { materializeReplayTranscriptRuntimeInstructions } from "./ReplayRuntimeInstructions.ts";
24
25// Each driver's fixtures run from its own <Provider>ReplayFixtures.integration.test.ts,
26// so a focused run loads one adapter. A fixture for a driver missing here would
27// silently never run, so loading this module fails instead.
28const DRIVERS_WITH_REPLAY_FIXTURE_FILES = new Set([
29 "acpRegistry",
30 "claudeAgent",
31 "codex",
32 "cursor",
33 "fx",
34 "grok",
35 "kiro",
36 "muse",
37 "opencode",
38 "pi",
39]);
40for (const fixture of ORCHESTRATOR_REPLAY_FIXTURES) {
41 for (const provider of fixture.providers) {
42 if (!DRIVERS_WITH_REPLAY_FIXTURE_FILES.has(provider.driver)) {
43 throw new Error(
44 `${fixture.name}/${provider.driver} has no ReplayFixtures test file to run it.`,
45 );
46 }
47 }
48}
49
50export const readReplayFixtureTranscript = Effect.fn("readOrchestratorReplayTranscript")(function* (
51 file: URL,
52) {
53 return yield* readProviderReplayTranscript(file);
54}, Effect.provide(NodeServices.layer));
55
56function normalizeTestError(cause: unknown): Error {
57 return cause instanceof Error ? cause : new Error(String(cause));
58}
59
60function transcriptEntriesThroughLabel(
61 transcript: ProviderReplayTranscript,
62 label: string | undefined,
63): ProviderReplayTranscript {
64 if (label === undefined) {
65 return transcript;
66 }
67 const entryIndex = transcript.entries.findIndex(
68 (entry) => entry.type === "emit_inbound" && entry.label === label,
69 );
70 if (entryIndex === -1) {
71 throw new Error(`${transcript.scenario} is missing inbound transcript label ${label}.`);
72 }
73 return {
74 ...transcript,
75 entries: transcript.entries.slice(0, entryIndex + 1),
76 };
77}
78
79function isStreamingAssistantEvent(event: OrchestrationV2DomainEvent): boolean {
80 switch (event.type) {
81 case "node.updated":
82 return event.payload.kind === "assistant_message" && event.payload.status === "running";
83 case "message.updated":
84 return event.payload.role === "assistant" && event.payload.streaming;
85 case "turn-item.updated":
86 return event.payload.type === "assistant_message" && event.payload.streaming;
87 default:
88 return false;
89 }
90}
91
92export interface ReplayFixtureCase {
93 readonly fixtureName: string;
94 readonly buildInput: () => OrchestratorFixtureInput;
95 readonly driver: ProviderOrchestratorReplayVariant;
96 readonly transformTranscript?: (transcript: ProviderReplayTranscript) => ProviderReplayTranscript;
97}
98
99export interface ReplayFixtureRun<
100 Transcript extends ProviderReplayTranscript,
101 Error,
102 WorkspaceError = never,
103> extends ReplayFixtureCase {
104 readonly harness: OrchestratorV2ProviderReplayHarness<Transcript, Error>;
105 /** Rewrites recorded workspace paths to the replay workspace, for drivers that record them. */
106 readonly materializeWorkspace?: (
107 transcript: ProviderReplayTranscript,
108 workspace: string,
109 ) => Effect.Effect<ProviderReplayTranscript, WorkspaceError>;
110}
111
112const runFixtureProvider = Effect.fn("runOrchestratorReplayFixture")(function* <
113 Transcript extends ProviderReplayTranscript,
114 Error,
115 WorkspaceError,
116>(input: ReplayFixtureRun<Transcript, Error, WorkspaceError>) {
117 const recordedTranscript = yield* readReplayFixtureTranscript(input.driver.transcriptFile);
118 const rawTranscript = input.transformTranscript?.(recordedTranscript) ?? recordedTranscript;
119 const replayTranscript = materializeReplayTranscriptRuntimeInstructions(
120 transcriptEntriesThroughLabel(rawTranscript, input.driver.transcriptEntriesThroughLabel),
121 { driver: input.driver.driver, model: input.driver.modelSelection.model },
122 );
123 const fixtureInput = input.buildInput();
124 const workspace = yield* checkpointWorkspace(input.fixtureName, fixtureInput.workspaceFiles);
125 const transcript = yield* input.harness.decodeTranscript(
126 input.materializeWorkspace === undefined
127 ? replayTranscript
128 : yield* input.materializeWorkspace(replayTranscript, workspace),
129 );
130 const materialized = yield* materializeFixtureInput({
131 scenario: input.fixtureName,
132 fixtureInput,
133 driver: input.driver.driver,
134 modelSelection: input.driver.modelSelection,
135 }).pipe(Effect.provide(IdAllocator.layer), provideDeterministicTestRuntime);
136 const scenario = {
137 name: `${input.fixtureName}/${input.driver.driver}`,
138 transcript,
139 commands: materialized.commands,
140 steps: materialized.steps,
141 projectionThreadIds: materialized.projectionThreadIds,
142 runtimePolicyOverride: {
143 ...input.driver.runtimePolicyOverride,
144 cwd: workspace,
145 },
146 };
147
148 const result = yield* runOrchestratorV2ProviderReplayScenario(
149 scenario,
150 input.harness,
151 input.driver.runContinuationWorker === true ? { runContinuationWorker: true } : {},
152 ).pipe(provideDeterministicTestRuntime);
153 input.driver.assertOutput(result, transcript);
154 assertProviderNativeSubagentRootTurns(result);
155 const expectedAbsentWorkspacePaths = input.driver.expectedAbsentWorkspacePaths;
156 if (expectedAbsentWorkspacePaths !== undefined) {
157 yield* Effect.gen(function* () {
158 const fs = yield* FileSystem.FileSystem;
159 const path = yield* Path.Path;
160 for (const relativePath of expectedAbsentWorkspacePaths) {
161 assert.isFalse(
162 yield* fs.exists(path.join(workspace, relativePath)),
163 `${input.fixtureName}/${input.driver.driver} must not create ${relativePath} in the replay workspace`,
164 );
165 }
166 }).pipe(Effect.provide(NodeServices.layer));
167 }
168 assert.isFalse(
169 result.domainEvents.some(isStreamingAssistantEvent),
170 "buffered delivery must not persist streaming assistant artifacts",
171 );
172 const projectionThreadId = materialized.projectionThreadIds[0];
173 assert.isDefined(projectionThreadId);
174 const projection = result.projections.get(projectionThreadId);
175 assert.isDefined(projection);
176 const latestRun = projection.runs.at(-1);
177 // A `set_model` step moves later runs to another model of the same instance.
178 const finalModel = fixtureInput.steps.findLast((step) => step.type === "set_model");
179 assert.deepEqual(
180 latestRun?.modelSelection,
181 finalModel?.type === "set_model"
182 ? { ...input.driver.modelSelection, model: finalModel.model }
183 : input.driver.modelSelection,
184 );
185 if (projection.runs.some((run) => run.status === "completed")) {
186 const threadStartCheckpoint = projection.checkpoints.find(
187 (checkpoint) => checkpoint.ordinalWithinScope === 0 && checkpoint.appRunOrdinal === null,
188 );
189 assert.isDefined(
190 threadStartCheckpoint,
191 "completed threads must retain an addressable thread-start checkpoint",
192 );
193 assert.equal(threadStartCheckpoint.status, "ready");
194 }
195 return result;
196});
197
198/** Replays one fixture variant through OrchestratorV2 and runs the shared output checks. */
199export function runReplayFixture<
200 Transcript extends ProviderReplayTranscript,
201 Error,
202 WorkspaceError = never,
203>(input: ReplayFixtureRun<Transcript, Error, WorkspaceError>) {
204 return runFixtureProvider(input).pipe(Effect.mapError(normalizeTestError), Effect.scoped);
205}
206
207export function findReplayFixture(name: string) {
208 return ORCHESTRATOR_REPLAY_FIXTURES.find((fixture) => fixture.name === name);
209}
210
211/**
212 * Registers one test per recorded fixture variant of `driver`. Call it inside the
213 * provider file's `describe`. `run` picks the harness for each variant.
214 */
215export function itRunsReplayFixtures<E>(
216 driver: string,
217 run: (input: ReplayFixtureCase) => Effect.Effect<unknown, E>,
218) {
219 it.effect.each(
220 ORCHESTRATOR_REPLAY_FIXTURES.flatMap((fixture) =>
221 fixture.providers
222 .filter((provider) => provider.driver === driver)
223 .map((provider) => [fixture.name, provider.driver, fixture, provider] as const),
224 ),
225 )("runs %s/%s through OrchestratorV2 using deterministic replay", ([, , fixture, provider]) =>
226 run({ fixtureName: fixture.name, buildInput: fixture.buildInput, driver: provider }),
227 );
228}