1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, describe, it } from "@effect/vitest";
3import {
4 ProviderDriverKind,
5 ProviderInstanceId,
6 type ProviderReplayTranscript,
7} from "@t3tools/contracts";
8import * as DateTime from "effect/DateTime";
9import * as Effect from "effect/Effect";
10import * as Layer from "effect/Layer";
11import * as Option from "effect/Option";
12import * as Stream from "effect/Stream";
13import { TestClock } from "effect/testing";
14
15import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
16import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
17import { AcpRegistryOrchestratorReplayHarness } from "../Adapters/AcpRegistryAdapterV2.testkit.ts";
18import * as EffectWorker from "../EffectWorker.ts";
19import * as Orchestrator from "../Orchestrator.ts";
20import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts";
21import { materializeFixtureInput, SIMPLE_PROMPT } from "./fixtures/shared.ts";
22import {
23 itRunsReplayFixtures,
24 readReplayFixtureTranscript,
25 runReplayFixture,
26} from "./ProviderReplayFixtures.ts";
27import { layerProviderReplay } from "./ProviderReplayHarness.ts";
28import { materializeReplayTranscriptRuntimeInstructions } from "./ReplayRuntimeInstructions.ts";
29
30describe("ACP registry replay fixtures", () => {
31 it.effect.each([
32 { name: "open", firstTurn: false, code: -32602 },
33 { name: "switch", firstTurn: true, code: -32602 },
34 { name: "authentication", firstTurn: false, code: -32000 },
35 ])("preserves the ACP model-write $name failure through orchestration", (testCase) =>
36 Effect.gen(function* () {
37 const name = `acp_model_write_${testCase.name}`;
38 const sessionId = "grok-replay-session-1";
39 const rejectedModel = "replay-model-b";
40 const modelSelection = {
41 instanceId: ProviderInstanceId.make("acpRegistry"),
42 model: rejectedModel,
43 };
44 const recorded = yield* readReplayFixtureTranscript(
45 new URL("./fixtures/simple/registry_transcript.ndjson", import.meta.url),
46 );
47 const sessionIndex = recorded.entries.findIndex(
48 (entry) => "label" in entry && entry.label === "session.created",
49 );
50 assert.isAtLeast(sessionIndex, 0);
51 const transcript = {
52 ...recorded,
53 scenario: name,
54 entries: [
55 ...recorded.entries.slice(0, sessionIndex),
56 {
57 type: "emit_inbound",
58 label: "session.created",
59 frame: {
60 kind: "response",
61 method: "session/new",
62 result: {
63 sessionId,
64 configOptions: [
65 {
66 id: "model",
67 name: "Model",
68 category: "model",
69 type: "select",
70 currentValue: "replay-model-a",
71 options: [
72 { value: "replay-model-a", name: "Model A" },
73 { value: rejectedModel, name: "Model B" },
74 ],
75 },
76 ],
77 },
78 },
79 },
80 ...(testCase.firstTurn
81 ? recorded.entries
82 .slice(sessionIndex + 1)
83 .filter((entry) => entry.type !== "runtime_exit")
84 : []),
85 {
86 type: "expect_outbound",
87 label: "session.set_model",
88 frame: {
89 kind: "request",
90 method: "session/set_config_option",
91 params: { sessionId, configId: "model", value: rejectedModel },
92 },
93 },
94 {
95 type: "emit_inbound",
96 label: "session.set_model.rejected",
97 frame: {
98 kind: "response",
99 method: "session/set_config_option",
100 error: {
101 code: testCase.code,
102 message: testCase.code === -32602 ? "Invalid provider" : "Authentication required",
103 },
104 },
105 },
106 ],
107 } satisfies ProviderReplayTranscript;
108 const materialized = yield* materializeFixtureInput({
109 scenario: name,
110 fixtureInput: {
111 steps: Array.from({ length: testCase.firstTurn ? 2 : 1 }, () => ({
112 type: "message" as const,
113 text: SIMPLE_PROMPT,
114 })),
115 },
116 driver: ProviderDriverKind.make("acpRegistry"),
117 modelSelection: {
118 ...modelSelection,
119 model: testCase.firstTurn ? "replay-model-a" : rejectedModel,
120 },
121 });
122 const scenario = {
123 ...materialized,
124 name,
125 transcript: yield* AcpRegistryOrchestratorReplayHarness.decodeTranscript(
126 materializeReplayTranscriptRuntimeInstructions(transcript, {
127 driver: ProviderDriverKind.make("acpRegistry"),
128 model: "replay-model-a",
129 }),
130 ),
131 runtimePolicyOverride: { cwd: yield* checkpointWorkspace(name) },
132 };
133 yield* Effect.gen(function* () {
134 const orchestrator = yield* Orchestrator.OrchestratorV2;
135 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
136 if (testCase.firstTurn) yield* EffectWorker.runDaemon.pipe(Effect.forkScoped);
137 for (const [index, command] of materialized.commands.entries()) {
138 const isFirstTurn =
139 testCase.firstTurn && command.type === "message.dispatch" && index === 1;
140 yield* orchestrator.dispatch(
141 command.type === "message.dispatch" && !isFirstTurn
142 ? { ...command, modelSelection }
143 : command,
144 );
145 if (isFirstTurn) {
146 // Wait for the committed terminal event before switching models.
147 yield* orchestrator.streamStoredEvents.pipe(
148 Stream.filter(
149 ({ event }) => event.type === "run.updated" && event.payload.status === "completed",
150 ),
151 Stream.runHead,
152 );
153 }
154 }
155 if (testCase.firstTurn) {
156 yield* orchestrator.streamStoredEvents.pipe(
157 Stream.filter(
158 ({ event }) => event.type === "run.updated" && event.payload.status === "failed",
159 ),
160 Stream.runHead,
161 );
162 } else {
163 // Failed opens retry on the outbox schedule. Advance to each recorded
164 // deadline, draining effects instead of polling the projection.
165 for (let attempt = 0; attempt < 10; attempt++) {
166 yield* worker.drain();
167 const next = yield* worker.nextClaimableAt();
168 if (Option.isNone(next)) break;
169 const now = yield* DateTime.now;
170 yield* TestClock.adjust(
171 Math.max(0, DateTime.toEpochMillis(next.value) - DateTime.toEpochMillis(now)),
172 );
173 }
174 }
175 const projection = yield* orchestrator.getThreadProjection(
176 materialized.projectionThreadIds[0]!,
177 );
178 assert.deepEqual(
179 projection.runs.map((run) => run.status),
180 testCase.firstTurn ? ["completed", "failed"] : ["failed"],
181 );
182 const errors = projection.turnItems.filter((item) => item.type === "error");
183 assert.lengthOf(errors, 1);
184 assert.equal(errors[0]?.failure.class, "provider_error");
185 if (testCase.code === -32602) {
186 assert.equal(
187 errors[0]?.failure.message,
188 'The agent did not accept the model "replay-model-b". Choose another model for this provider, then retry.',
189 );
190 } else {
191 assert.match(
192 errors[0]?.failure.message ?? "",
193 /^Failed to open acpRegistry provider session \S+\.$/u,
194 );
195 }
196 }).pipe(
197 Effect.provide(
198 layerProviderReplay(scenario, AcpRegistryOrchestratorReplayHarness, {
199 runEffectWorker: false,
200 }),
201 ),
202 );
203 }).pipe(
204 Effect.scoped,
205 provideDeterministicTestRuntime,
206 Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer)),
207 ),
208 );
209
210 itRunsReplayFixtures("acpRegistry", (input) =>
211 runReplayFixture({ ...input, harness: AcpRegistryOrchestratorReplayHarness }),
212 );
213});