1import {
2 ProviderDriverKind,
3 ProviderReplayEntry,
4 type ProviderReplayTranscript,
5} from "@t3tools/contracts";
6import * as Effect from "effect/Effect";
7import * as Crypto from "effect/Crypto";
8import * as FileSystem from "effect/FileSystem";
9import * as Layer from "effect/Layer";
10import * as Schema from "effect/Schema";
11import type * as Scope from "effect/Scope";
12import * as Stream from "effect/Stream";
13import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
14import * as EffectAcpErrors from "effect-acp/errors";
15
16import * as AcpSessionRuntime from "@t3tools/provider-acp/server/AcpSessionRuntime";
17import type { ProviderReplayGate } from "@t3tools/provider-testing/replayGate";
18import { ACP_PROTOCOL, type AcpAdapterV2RuntimeInput } from "@t3tools/provider-acp/server/adapter";
19
20export const AcpReplayTranscript = Schema.Struct({
21 provider: ProviderDriverKind,
22 protocol: Schema.Literal(ACP_PROTOCOL),
23 version: Schema.String,
24 scenario: Schema.String,
25 metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
26 entries: Schema.Array(ProviderReplayEntry),
27});
28export type AcpReplayTranscript = typeof AcpReplayTranscript.Type;
29
30const decodeAcpReplayTranscriptSchema = Schema.decodeUnknownEffect(AcpReplayTranscript);
31const encodeReplayTranscriptJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
32
33export class AcpReplayTranscriptDecodeError extends Schema.TaggedError<AcpReplayTranscriptDecodeError>()(
34 "AcpReplayTranscriptDecodeError",
35 {
36 expectedProvider: ProviderDriverKind,
37 driver: Schema.optional(Schema.String),
38 protocol: Schema.optional(Schema.String),
39 scenario: Schema.optional(Schema.String),
40 cause: Schema.Defect(),
41 },
42) {
43 override get message(): string {
44 return `Failed to decode ${this.expectedProvider} ACP replay transcript for scenario ${this.scenario ?? "<unknown>"}.`;
45 }
46}
47
48const isAcpReplayTranscriptDecodeError = Schema.is(AcpReplayTranscriptDecodeError);
49
50function metadataFromTranscript(transcript: ProviderReplayTranscript): {
51 readonly provider?: string;
52 readonly protocol?: string;
53 readonly scenario?: string;
54} {
55 return {
56 provider: transcript.provider,
57 protocol: transcript.protocol,
58 scenario: transcript.scenario,
59 };
60}
61
62export function decodeAcpReplayTranscript(
63 transcript: ProviderReplayTranscript,
64 expectedProvider: ProviderDriverKind,
65 options: { readonly retargetProvider?: boolean } = {},
66): Effect.Effect<AcpReplayTranscript, AcpReplayTranscriptDecodeError> {
67 const candidate = options.retargetProvider
68 ? { ...transcript, provider: expectedProvider }
69 : transcript;
70 return decodeAcpReplayTranscriptSchema(candidate).pipe(
71 Effect.filterOrFail(
72 (decoded) => decoded.provider === expectedProvider,
73 () =>
74 new AcpReplayTranscriptDecodeError({
75 expectedProvider,
76 ...metadataFromTranscript(transcript),
77 cause: `Expected provider ${expectedProvider}, received ${transcript.provider}`,
78 }),
79 ),
80 Effect.mapError((cause) =>
81 isAcpReplayTranscriptDecodeError(cause)
82 ? cause
83 : new AcpReplayTranscriptDecodeError({
84 expectedProvider,
85 ...metadataFromTranscript(transcript),
86 cause,
87 }),
88 ),
89 );
90}
91
92interface ReplayStatus {
93 readonly cursor: number;
94 readonly total: number;
95 readonly failure?: unknown;
96}
97
98function decodeReplayStatus(raw: string): ReplayStatus {
99 const value: unknown = JSON.parse(raw);
100 if (
101 typeof value !== "object" ||
102 value === null ||
103 typeof Reflect.get(value, "cursor") !== "number" ||
104 typeof Reflect.get(value, "total") !== "number"
105 ) {
106 throw new Error("ACP replay status is malformed.");
107 }
108 return value as ReplayStatus;
109}
110
111export function makeAcpReplayCompletenessAssertion(
112 fileSystem: FileSystem.FileSystem,
113 statusPath: string,
114 transcript: AcpReplayTranscript,
115): Effect.Effect<void, EffectAcpErrors.AcpError> {
116 return fileSystem.readFileString(statusPath).pipe(
117 Effect.mapError(
118 (cause) =>
119 new EffectAcpErrors.AcpTransportError({
120 detail: `Failed to read ACP replay status for ${transcript.scenario}`,
121 cause,
122 }),
123 ),
124 Effect.flatMap((raw) =>
125 Effect.try({
126 try: () => decodeReplayStatus(raw),
127 catch: (cause) =>
128 new EffectAcpErrors.AcpTransportError({
129 detail: `Failed to decode ACP replay status for ${transcript.scenario}`,
130 cause,
131 }),
132 }),
133 ),
134 Effect.flatMap((status) => {
135 if (
136 status.failure === undefined &&
137 status.cursor === transcript.entries.length &&
138 status.total === transcript.entries.length
139 ) {
140 return Effect.void;
141 }
142 return Effect.fail(
143 new EffectAcpErrors.AcpTransportError({
144 detail: `ACP replay did not consume all frames for ${transcript.scenario}`,
145 cause: status,
146 }),
147 );
148 }),
149 );
150}
151
152function acpReplayAgentArgs(scriptPath: string): ReadonlyArray<string> {
153 return ["--experimental-strip-types", scriptPath];
154}
155
156/**
157 * Holds the replay agent's inbound lines at a gate the scenario releases.
158 * The agent writes one line per `emit_inbound` entry in transcript order, so
159 * the Nth line it writes belongs to the Nth inbound entry and carries its
160 * label. Lines after a held one wait with it, which keeps wire order.
161 */
162function holdGatedReplayLines(
163 transcript: AcpReplayTranscript,
164 replayGate: ProviderReplayGate,
165): NonNullable<AcpSessionRuntime.AcpSessionRuntimeOptions["transformStdout"]> {
166 const inboundLabels = transcript.entries.flatMap((entry) =>
167 entry.type === "emit_inbound" ? [entry.label] : [],
168 );
169 let inboundIndex = 0;
170 const encoder = new TextEncoder();
171 return (stdout) =>
172 stdout.pipe(
173 Stream.decodeText,
174 Stream.splitLines,
175 Stream.mapEffect((line) => {
176 const label = line.trim().length === 0 ? undefined : inboundLabels[inboundIndex++];
177 return Effect.promise((signal) => replayGate.beforeEmit(label, signal)).pipe(
178 Effect.as(encoder.encode(`${line}\n`)),
179 );
180 }),
181 );
182}
183
184export function makeAcpReplayRuntime(input: {
185 readonly transcript: AcpReplayTranscript;
186 readonly statusPath: string;
187 readonly scriptPath: string;
188 readonly childProcessSpawner: ChildProcessSpawner.ChildProcessSpawner["Service"];
189 readonly fileSystem: FileSystem.FileSystem;
190 readonly cancelMeta?: AcpSessionRuntime.AcpSessionRuntimeOptions["cancelMeta"];
191 readonly initializeMeta?: AcpSessionRuntime.AcpSessionRuntimeOptions["initializeMeta"];
192 readonly replayGate?: ProviderReplayGate;
193}): (
194 runtimeInput: AcpAdapterV2RuntimeInput,
195) => Effect.Effect<
196 AcpSessionRuntime.AcpSessionRuntime["Service"],
197 EffectAcpErrors.AcpError,
198 Crypto.Crypto | Scope.Scope
199> {
200 // Recorded transcripts outgrow the kernel's 128 KiB limit on one environment
201 // variable, so the agent reads the transcript from a file beside its status.
202 const transcriptPath = `${input.statusPath}.transcript.json`;
203 return (runtimeInput) =>
204 Effect.gen(function* () {
205 yield* input.fileSystem
206 .writeFileString(transcriptPath, encodeReplayTranscriptJson(input.transcript))
207 .pipe(
208 Effect.mapError(
209 (cause) =>
210 new EffectAcpErrors.AcpTransportError({
211 detail: `Failed to write ACP replay transcript for ${input.transcript.scenario}`,
212 cause,
213 }),
214 ),
215 );
216 const context = yield* Layer.build(
217 AcpSessionRuntime.layer({
218 ...runtimeInput,
219 spawn: {
220 command: process.execPath,
221 args: acpReplayAgentArgs(input.scriptPath),
222 cwd: runtimeInput.cwd,
223 env: {
224 ...process.env,
225 T3_ACP_REPLAY_TRANSCRIPT_PATH: transcriptPath,
226 T3_ACP_REPLAY_STATUS_PATH: input.statusPath,
227 T3_ACP_REPLAY_WORKSPACE: runtimeInput.cwd,
228 },
229 },
230 authMethodId: "replay",
231 ...(input.cancelMeta === undefined ? {} : { cancelMeta: input.cancelMeta }),
232 ...(input.initializeMeta === undefined ? {} : { initializeMeta: input.initializeMeta }),
233 ...(input.replayGate === undefined
234 ? {}
235 : { transformStdout: holdGatedReplayLines(input.transcript, input.replayGate) }),
236 }).pipe(
237 Layer.provide(
238 Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, input.childProcessSpawner),
239 ),
240 ),
241 );
242 return yield* Effect.service(AcpSessionRuntime.AcpSessionRuntime).pipe(
243 Effect.provide(context),
244 );
245 });
246}