1import { assert, describe, it } from "@effect/vitest";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { OrchestrationV2Command, ProviderInstanceId } from "@t3tools/contracts";
4import * as Effect from "effect/Effect";
5import * as Schema from "effect/Schema";
6
7import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
8import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts";
9import { ORCHESTRATOR_REPLAY_FIXTURES } from "./fixtures/index.ts";
10import { materializeFixtureInput } from "./fixtures/shared.ts";
11import { readProviderReplayTranscript } from "@t3tools/provider-testing/replayTranscript";
12
13const decodeCommand = Schema.decodeUnknownEffect(OrchestrationV2Command);
14const readTranscript = Effect.fn("readOrchestratorReplayContractTranscript")(function* (file: URL) {
15 return yield* readProviderReplayTranscript(file);
16}, Effect.provide(NodeServices.layer));
17
18function assertUnique(values: ReadonlyArray<string>, label: string) {
19 assert.deepEqual(new Set(values).size, values.length, `${label} must be unique`);
20}
21
22describe("orchestrator replay fixture contract", () => {
23 it.effect(
24 "defines one stable input and provider-specific replay/output contracts per scenario",
25 () =>
26 Effect.gen(function* () {
27 assertUnique(
28 ORCHESTRATOR_REPLAY_FIXTURES.map((fixture) => fixture.name),
29 "fixture names",
30 );
31
32 for (const fixture of ORCHESTRATOR_REPLAY_FIXTURES) {
33 assert.isAtLeast(fixture.providers.length, 1, `${fixture.name} must have providers`);
34 assertUnique(
35 fixture.providers.map((provider) => provider.driver),
36 `${fixture.name} provider variants`,
37 );
38
39 for (const provider of fixture.providers) {
40 const transcript = yield* readTranscript(provider.transcriptFile);
41 const replayGateLabels = fixture
42 .buildInput()
43 .steps.flatMap((step) =>
44 step.type === "release_replay_gate" ||
45 step.type === "release_replay_gate_after_waiting"
46 ? [step.label]
47 : [],
48 );
49 const materialized = yield* materializeFixtureInput({
50 scenario: fixture.name,
51 fixtureInput: fixture.buildInput(),
52 driver: provider.driver,
53 modelSelection: provider.modelSelection,
54 }).pipe(Effect.provide(IdAllocator.layer), provideDeterministicTestRuntime);
55 const firstCommand = materialized.commands[0];
56
57 assert.equal(transcript.scenario, provider.recordedScenario ?? fixture.name);
58 if (provider.driver === "acpRegistry") {
59 assert.include(
60 ["acpRegistry", "grok"],
61 transcript.provider,
62 "ACP Registry may retarget protocol-standard Grok ACP evidence",
63 );
64 } else {
65 assert.equal(transcript.provider, provider.driver);
66 }
67 assert.equal(
68 provider.modelSelection.instanceId,
69 ProviderInstanceId.make(provider.driver),
70 );
71 assert.isDefined(materialized.projectionThreadIds[0]);
72 assert.equal(firstCommand?.type, "thread.create");
73 if (firstCommand?.type !== "thread.create") {
74 throw new Error(`${fixture.name}/${provider.driver} must start with thread.create`);
75 }
76 assert.equal(firstCommand.threadId, materialized.projectionThreadIds[0]);
77 // These fixture steps coordinate the test runtime; every other
78 // input step dispatches a command.
79 const commandProducingSteps = fixture
80 .buildInput()
81 .steps.filter(
82 (step) =>
83 step.type !== "advance_clock" &&
84 step.type !== "await_run_status" &&
85 step.type !== "finish_held_run" &&
86 step.type !== "capture_shell_snapshot" &&
87 step.type !== "release_replay_gate" &&
88 step.type !== "release_replay_gate_after_waiting",
89 );
90 assert.equal(materialized.commands.length, commandProducingSteps.length + 1);
91 assert.isAtLeast(materialized.steps.length, materialized.commands.length);
92 assert.equal(typeof provider.assertOutput, "function");
93 for (const label of replayGateLabels) {
94 assert.isTrue(
95 transcript.entries.some(
96 (entry) => entry.type === "emit_inbound" && entry.label === label,
97 ),
98 `${fixture.name}/${provider.driver} replay gate ${label} must label an inbound frame`,
99 );
100 }
101 if (provider.transcriptEntriesThroughLabel !== undefined) {
102 assert.isTrue(
103 transcript.entries.some(
104 (entry) =>
105 entry.type === "emit_inbound" &&
106 entry.label === provider.transcriptEntriesThroughLabel,
107 ),
108 `${fixture.name}/${provider.driver} transcript slice must name an inbound frame`,
109 );
110 }
111
112 assertUnique(
113 materialized.commands.map((command) => command.commandId),
114 `${fixture.name}/${provider.driver} command IDs`,
115 );
116
117 for (const command of materialized.commands) {
118 yield* decodeCommand(command);
119 }
120
121 for (const command of materialized.commands) {
122 assert.isTrue(
123 materialized.steps.some(
124 (step) =>
125 (step.type === "dispatch" && step.command === command) ||
126 (step.type === "respond_to_next_runtime_request" &&
127 step.commandId === command.commandId),
128 ),
129 `${fixture.name}/${provider.driver} command ${command.commandId} must appear in the timeline`,
130 );
131 }
132 }
133 }
134 }),
135 );
136
137 it.effect("keeps Codex fixture transcripts at the codex app-server boundary", () =>
138 Effect.gen(function* () {
139 for (const fixture of ORCHESTRATOR_REPLAY_FIXTURES) {
140 for (const provider of fixture.providers.filter((entry) => entry.driver === "codex")) {
141 const transcript = yield* readTranscript(provider.transcriptFile);
142 const first = transcript.entries[0];
143 const last = transcript.entries.at(-1);
144
145 assert.equal(transcript.protocol, "codex.app-server");
146 assert.equal(first?.type, "expect_outbound");
147 if (first?.type === "expect_outbound") {
148 assert.equal(first.label, "initialize");
149 }
150 assert.deepEqual(last, {
151 type: "runtime_exit",
152 status: "success",
153 });
154 }
155 }
156 }),
157 );
158});