1#!/usr/bin/env node
2// @effect-diagnostics nodeBuiltinImport:off
3import * as NodeFS from "node:fs";
4import * as NodeReadline from "node:readline";
5
6interface ReplayEntry {
7 readonly type: "emit_inbound" | "expect_outbound" | "runtime_exit";
8 readonly label?: string;
9 readonly frame?: unknown;
10 readonly status?: "success" | "error" | "cancelled";
11 readonly error?: unknown;
12}
13
14interface ReplayTranscript {
15 readonly scenario: string;
16 readonly entries: ReadonlyArray<ReplayEntry>;
17}
18
19interface LogicalFrame {
20 readonly kind: "notification" | "request" | "response";
21 readonly method: string;
22 readonly params?: unknown;
23 readonly result?: unknown;
24 readonly error?: unknown;
25}
26
27interface JsonRpcMessage {
28 readonly jsonrpc?: string;
29 readonly id?: string | number | null;
30 readonly method?: string;
31 readonly params?: unknown;
32 readonly result?: unknown;
33 readonly error?: unknown;
34 readonly headers?: ReadonlyArray<unknown>;
35}
36
37const encodedTranscript = process.env.T3_ACP_REPLAY_TRANSCRIPT;
38const transcriptPath = process.env.T3_ACP_REPLAY_TRANSCRIPT_PATH;
39const statusPath = process.env.T3_ACP_REPLAY_STATUS_PATH;
40const replayWorkspace = process.env.T3_ACP_REPLAY_WORKSPACE ?? process.cwd();
41
42if ((encodedTranscript === undefined && transcriptPath === undefined) || statusPath === undefined) {
43 process.stderr.write("ACP replay requires transcript and status environment variables.\n");
44 process.exit(2);
45}
46
47const replayStatusPath = statusPath;
48const transcript = JSON.parse(
49 transcriptPath === undefined
50 ? Buffer.from(encodedTranscript ?? "", "base64").toString("utf8")
51 : NodeFS.readFileSync(transcriptPath, "utf8"),
52) as ReplayTranscript;
53let cursor = 0;
54let stopped = false;
55let nextAgentRequestId = 1;
56const pendingClientRequestIds = new Map<string, string | number>();
57const pendingAgentRequestMethods = new Map<string, string>();
58
59function writeStatus(failure?: unknown): void {
60 NodeFS.writeFileSync(
61 replayStatusPath,
62 JSON.stringify({
63 scenario: transcript.scenario,
64 cursor,
65 total: transcript.entries.length,
66 ...(failure === undefined ? {} : { failure }),
67 }),
68 "utf8",
69 );
70}
71
72function stableStringify(value: unknown): string {
73 if (Array.isArray(value)) {
74 return `[${value.map(stableStringify).join(",")}]`;
75 }
76 if (typeof value === "object" && value !== null) {
77 const record = value as Record<string, unknown>;
78 return `{${Object.keys(record)
79 .toSorted()
80 .map((key) => `${JSON.stringify(key)}:${stableStringify(record[key])}`)
81 .join(",")}}`;
82 }
83 return JSON.stringify(value);
84}
85
86function expandExpectedString(value: string): string {
87 return value.replaceAll("<workspace>", replayWorkspace);
88}
89
90function matchesExpected(expected: unknown, actual: unknown): boolean {
91 if (expected === "<any>") return true;
92 if (typeof expected === "string" && typeof actual === "string") {
93 const expanded = expandExpectedString(expected);
94 if (!expanded.includes("<any>")) return expanded === actual;
95 const parts = expanded.split("<any>");
96 let offset = 0;
97 for (const part of parts) {
98 const index = actual.indexOf(part, offset);
99 if (index === -1) return false;
100 offset = index + part.length;
101 }
102 return true;
103 }
104 if (Array.isArray(expected)) {
105 return (
106 Array.isArray(actual) &&
107 expected.length === actual.length &&
108 expected.every((entry, index) => matchesExpected(entry, actual[index]))
109 );
110 }
111 if (typeof expected === "object" && expected !== null) {
112 if (typeof actual !== "object" || actual === null || Array.isArray(actual)) return false;
113 const expectedRecord = expected as Record<string, unknown>;
114 const actualRecord = actual as Record<string, unknown>;
115 const expectedKeys = Object.keys(expectedRecord).toSorted();
116 const actualKeys = Object.keys(actualRecord).toSorted();
117 return (
118 stableStringify(expectedKeys) === stableStringify(actualKeys) &&
119 expectedKeys.every((key) => matchesExpected(expectedRecord[key], actualRecord[key]))
120 );
121 }
122 return Object.is(expected, actual);
123}
124
125function stopWithFailure(detail: string, actual?: unknown): void {
126 if (stopped) return;
127 stopped = true;
128 const entry = transcript.entries[cursor];
129 const failure = {
130 detail,
131 cursor,
132 expected: entry,
133 ...(actual === undefined ? {} : { actual }),
134 };
135 writeStatus(failure);
136 process.stderr.write(`ACP replay mismatch: ${JSON.stringify(failure)}\n`);
137 process.exitCode = 1;
138 process.stdin.pause();
139}
140
141function advance(): void {
142 cursor += 1;
143 writeStatus();
144}
145
146function send(message: JsonRpcMessage): void {
147 process.stdout.write(`${JSON.stringify(message)}\n`);
148}
149
150function pendingClientRequestId(method: string): string | number | undefined {
151 return pendingClientRequestIds.get(method);
152}
153
154function logicalIncoming(message: JsonRpcMessage): LogicalFrame | undefined {
155 if (typeof message.method === "string") {
156 return {
157 kind:
158 message.id === undefined || message.id === null || message.id === ""
159 ? "notification"
160 : "request",
161 method: message.method,
162 ...(message.params === undefined ? {} : { params: message.params }),
163 };
164 }
165 if (message.id === undefined || message.id === null) return undefined;
166 const method = pendingAgentRequestMethods.get(String(message.id));
167 if (method === undefined) return undefined;
168 return {
169 kind: "response",
170 method,
171 ...(message.result === undefined ? {} : { result: message.result }),
172 ...(message.error === undefined ? {} : { error: message.error }),
173 };
174}
175
176// Recorded agent frames name workspace paths (tool inputs, fs requests) as
177// <workspace>; point them at this replay's workspace like expectations do.
178function materializeInbound(value: unknown): unknown {
179 if (typeof value === "string") return expandExpectedString(value);
180 if (Array.isArray(value)) return value.map(materializeInbound);
181 if (typeof value !== "object" || value === null) return value;
182 return Object.fromEntries(
183 Object.entries(value).map(([key, entry]) => [key, materializeInbound(entry)]),
184 );
185}
186
187function emitInbound(recorded: LogicalFrame): void {
188 const frame = materializeInbound(recorded) as LogicalFrame;
189 switch (frame.kind) {
190 case "notification":
191 send({
192 jsonrpc: "2.0",
193 method: frame.method,
194 ...(frame.params === undefined ? {} : { params: frame.params }),
195 });
196 return;
197 case "request": {
198 const id = nextAgentRequestId;
199 nextAgentRequestId += 1;
200 pendingAgentRequestMethods.set(String(id), frame.method);
201 send({
202 jsonrpc: "2.0",
203 id,
204 method: frame.method,
205 ...(frame.params === undefined ? {} : { params: frame.params }),
206 headers: [],
207 });
208 return;
209 }
210 case "response": {
211 const id = pendingClientRequestId(frame.method);
212 if (id === undefined) {
213 stopWithFailure(`No pending client request for ${frame.method}`, frame);
214 return;
215 }
216 pendingClientRequestIds.delete(frame.method);
217 send({
218 jsonrpc: "2.0",
219 id,
220 ...(frame.result === undefined ? {} : { result: frame.result }),
221 ...(frame.error === undefined ? {} : { error: frame.error }),
222 });
223 }
224 }
225}
226
227function flushInbound(): void {
228 while (!stopped) {
229 const entry = transcript.entries[cursor];
230 if (entry === undefined || entry.type === "expect_outbound") return;
231 if (entry.type === "runtime_exit") {
232 if (entry.status !== "success" && entry.status !== "cancelled") {
233 stopWithFailure(`Recorded runtime exit was ${entry.status ?? "unknown"}`, entry.error);
234 return;
235 }
236 advance();
237 continue;
238 }
239 const frame = entry.frame as LogicalFrame;
240 if (
241 typeof frame !== "object" ||
242 frame === null ||
243 !["notification", "request", "response"].includes(frame.kind) ||
244 typeof frame.method !== "string"
245 ) {
246 stopWithFailure("Invalid emit_inbound logical ACP frame", entry.frame);
247 return;
248 }
249 emitInbound(frame);
250 if (stopped) return;
251 advance();
252 }
253}
254
255function handleMessage(message: JsonRpcMessage): void {
256 if (stopped) return;
257 const actual = logicalIncoming(message);
258 if (actual === undefined) {
259 stopWithFailure("Could not identify outbound ACP frame", message);
260 return;
261 }
262 const entry = transcript.entries[cursor];
263 if (entry?.type !== "expect_outbound" || !matchesExpected(entry.frame, actual)) {
264 if (actual.kind === "request" && message.id !== undefined && message.id !== null) {
265 send({
266 jsonrpc: "2.0",
267 id: message.id,
268 error: { code: -32603, message: "ACP replay frame mismatch" },
269 });
270 }
271 stopWithFailure("Unexpected outbound ACP frame", actual);
272 return;
273 }
274 if (actual.kind === "request" && message.id !== undefined && message.id !== null) {
275 pendingClientRequestIds.set(actual.method, message.id);
276 } else if (actual.kind === "response" && message.id !== undefined && message.id !== null) {
277 pendingAgentRequestMethods.delete(String(message.id));
278 }
279 advance();
280 flushInbound();
281}
282
283writeStatus();
284flushInbound();
285
286const input = NodeReadline.createInterface({ input: process.stdin, crlfDelay: Infinity });
287input.on("line", (line) => {
288 if (stopped || line.trim().length === 0) return;
289 try {
290 handleMessage(JSON.parse(line) as JsonRpcMessage);
291 } catch (cause) {
292 stopWithFailure("Failed to decode outbound ACP JSON-RPC", String(cause));
293 }
294});
295
296input.on("close", () => {
297 if (!stopped && cursor !== transcript.entries.length) {
298 stopWithFailure("ACP replay input closed before transcript completion");
299 }
300});