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;
53/**
54 * A recording can span processes: a `runtime_exit` mid-transcript marks where
55 * one agent process ended, and a respawned replay agent continues from the
56 * cursor the previous one left in the status file.
57 */
58function resumedCursor(): number {
59 try {
60 const previous = JSON.parse(NodeFS.readFileSync(statusPath ?? "", "utf8")) as {
61 readonly cursor?: unknown;
62 readonly failure?: unknown;
63 };
64 const resumeAt = typeof previous.cursor === "number" ? previous.cursor : 0;
65 return previous.failure === undefined &&
66 transcript.entries[resumeAt - 1]?.type === "runtime_exit"
67 ? resumeAt
68 : 0;
69 } catch {
70 return 0;
71 }
72}
73let cursor = resumedCursor();
74let stopped = false;
75let nextAgentRequestId = 1;
76const pendingClientRequestIds = new Map<string, string | number>();
77const pendingAgentRequestMethods = new Map<string, string>();
78
79// The test reads status while this agent may still be running; rename makes each
80// write atomic so a reader never sees a truncated file.
81function writeStatus(failure?: unknown): void {
82 const temporaryStatusPath = `${replayStatusPath}.${process.pid}.tmp`;
83 NodeFS.writeFileSync(
84 temporaryStatusPath,
85 JSON.stringify({
86 scenario: transcript.scenario,
87 cursor,
88 total: transcript.entries.length,
89 ...(failure === undefined ? {} : { failure }),
90 }),
91 "utf8",
92 );
93 NodeFS.renameSync(temporaryStatusPath, replayStatusPath);
94}
95
96function stableStringify(value: unknown): string {
97 if (Array.isArray(value)) {
98 return `[${value.map(stableStringify).join(",")}]`;
99 }
100 if (typeof value === "object" && value !== null) {
101 const record = value as Record<string, unknown>;
102 return `{${Object.keys(record)
103 .toSorted()
104 .map((key) => `${JSON.stringify(key)}:${stableStringify(record[key])}`)
105 .join(",")}}`;
106 }
107 return JSON.stringify(value);
108}
109
110function expandExpectedString(value: string): string {
111 return value.replaceAll("<workspace>", replayWorkspace);
112}
113
114function matchesExpected(expected: unknown, actual: unknown): boolean {
115 if (expected === "<any>") return true;
116 if (typeof expected === "string" && typeof actual === "string") {
117 const expanded = expandExpectedString(expected);
118 if (!expanded.includes("<any>")) return expanded === actual;
119 const parts = expanded.split("<any>");
120 let offset = 0;
121 for (const part of parts) {
122 const index = actual.indexOf(part, offset);
123 if (index === -1) return false;
124 offset = index + part.length;
125 }
126 return true;
127 }
128 if (Array.isArray(expected)) {
129 return (
130 Array.isArray(actual) &&
131 expected.length === actual.length &&
132 expected.every((entry, index) => matchesExpected(entry, actual[index]))
133 );
134 }
135 if (typeof expected === "object" && expected !== null) {
136 if (typeof actual !== "object" || actual === null || Array.isArray(actual)) return false;
137 const expectedRecord = expected as Record<string, unknown>;
138 const actualRecord = actual as Record<string, unknown>;
139 const expectedKeys = Object.keys(expectedRecord).toSorted();
140 const actualKeys = Object.keys(actualRecord).toSorted();
141 return (
142 stableStringify(expectedKeys) === stableStringify(actualKeys) &&
143 expectedKeys.every((key) => matchesExpected(expectedRecord[key], actualRecord[key]))
144 );
145 }
146 return Object.is(expected, actual);
147}
148
149function stopWithFailure(detail: string, actual?: unknown): void {
150 if (stopped) return;
151 stopped = true;
152 const entry = transcript.entries[cursor];
153 const failure = {
154 detail,
155 cursor,
156 expected: entry,
157 ...(actual === undefined ? {} : { actual }),
158 };
159 writeStatus(failure);
160 process.stderr.write(`ACP replay mismatch: ${JSON.stringify(failure)}\n`);
161 process.exitCode = 1;
162 process.stdin.pause();
163}
164
165function send(message: JsonRpcMessage): void {
166 process.stdout.write(`${JSON.stringify(message)}\n`);
167}
168
169function pendingClientRequestId(method: string): string | number | undefined {
170 return pendingClientRequestIds.get(method);
171}
172
173function logicalIncoming(message: JsonRpcMessage): LogicalFrame | undefined {
174 if (typeof message.method === "string") {
175 return {
176 kind:
177 message.id === undefined || message.id === null || message.id === ""
178 ? "notification"
179 : "request",
180 method: message.method,
181 ...(message.params === undefined ? {} : { params: message.params }),
182 };
183 }
184 if (message.id === undefined || message.id === null) return undefined;
185 const method = pendingAgentRequestMethods.get(String(message.id));
186 if (method === undefined) return undefined;
187 return {
188 kind: "response",
189 method,
190 ...(message.result === undefined ? {} : { result: message.result }),
191 ...(message.error === undefined ? {} : { error: message.error }),
192 };
193}
194
195function materializeInbound(value: unknown): unknown {
196 if (typeof value === "string") return expandExpectedString(value);
197 if (Array.isArray(value)) return value.map(materializeInbound);
198 if (typeof value !== "object" || value === null) return value;
199 return Object.fromEntries(
200 Object.entries(value).map(([key, entry]) => [key, materializeInbound(entry)]),
201 );
202}
203
204function inboundMessage(recorded: LogicalFrame): JsonRpcMessage | undefined {
205 const frame = materializeInbound(recorded) as LogicalFrame;
206 switch (frame.kind) {
207 case "notification":
208 return {
209 jsonrpc: "2.0",
210 method: frame.method,
211 ...(frame.params === undefined ? {} : { params: frame.params }),
212 };
213 case "request": {
214 const id = nextAgentRequestId;
215 nextAgentRequestId += 1;
216 pendingAgentRequestMethods.set(String(id), frame.method);
217 return {
218 jsonrpc: "2.0",
219 id,
220 method: frame.method,
221 ...(frame.params === undefined ? {} : { params: frame.params }),
222 headers: [],
223 };
224 }
225 case "response": {
226 const id = pendingClientRequestId(frame.method);
227 if (id === undefined) {
228 stopWithFailure(`No pending client request for ${frame.method}`, frame);
229 return undefined;
230 }
231 pendingClientRequestIds.delete(frame.method);
232 return {
233 jsonrpc: "2.0",
234 id,
235 ...(frame.result === undefined ? {} : { result: frame.result }),
236 ...(frame.error === undefined ? {} : { error: frame.error }),
237 };
238 }
239 }
240}
241
242// T3 may kill the agent after its last reply. Record the batch, including trailing
243// runtime exits, before publishing any frames.
244function flushInbound(): void {
245 const messages: Array<JsonRpcMessage> = [];
246 while (!stopped) {
247 const entry = transcript.entries[cursor];
248 if (entry === undefined || entry.type === "expect_outbound") break;
249 if (entry.type === "runtime_exit") {
250 if (entry.status !== "success" && entry.status !== "cancelled") {
251 stopWithFailure(`Recorded runtime exit was ${entry.status ?? "unknown"}`, entry.error);
252 break;
253 }
254 cursor += 1;
255 // Frames after a mid-transcript exit belong to the next process.
256 if (cursor < transcript.entries.length) break;
257 continue;
258 }
259 const frame = entry.frame as LogicalFrame;
260 if (
261 typeof frame !== "object" ||
262 frame === null ||
263 !["notification", "request", "response"].includes(frame.kind) ||
264 typeof frame.method !== "string"
265 ) {
266 stopWithFailure("Invalid emit_inbound logical ACP frame", entry.frame);
267 break;
268 }
269 const message = inboundMessage(frame);
270 if (message === undefined) break;
271 messages.push(message);
272 cursor += 1;
273 }
274 if (!stopped) writeStatus();
275 messages.forEach(send);
276}
277
278function handleMessage(message: JsonRpcMessage): void {
279 if (stopped) return;
280 const actual = logicalIncoming(message);
281 if (actual === undefined) {
282 stopWithFailure("Could not identify outbound ACP frame", message);
283 return;
284 }
285 const entry = transcript.entries[cursor];
286 if (entry?.type !== "expect_outbound" || !matchesExpected(entry.frame, actual)) {
287 stopWithFailure("Unexpected outbound ACP frame", actual);
288 if (actual.kind === "request" && message.id !== undefined && message.id !== null) {
289 send({
290 jsonrpc: "2.0",
291 id: message.id,
292 error: { code: -32603, message: "ACP replay frame mismatch" },
293 });
294 }
295 return;
296 }
297 if (actual.kind === "request" && message.id !== undefined && message.id !== null) {
298 pendingClientRequestIds.set(actual.method, message.id);
299 } else if (actual.kind === "response" && message.id !== undefined && message.id !== null) {
300 pendingAgentRequestMethods.delete(String(message.id));
301 }
302 cursor += 1;
303 flushInbound();
304}
305
306flushInbound();
307
308const input = NodeReadline.createInterface({ input: process.stdin, crlfDelay: Infinity });
309input.on("line", (line) => {
310 if (stopped || line.trim().length === 0) return;
311 try {
312 handleMessage(JSON.parse(line) as JsonRpcMessage);
313 } catch (cause) {
314 stopWithFailure("Failed to decode outbound ACP JSON-RPC", String(cause));
315 }
316});
317
318input.on("close", () => {
319 const processEnded = transcript.entries[cursor - 1]?.type === "runtime_exit";
320 if (!stopped && cursor !== transcript.entries.length && !processEnded) {
321 stopWithFailure("ACP replay input closed before transcript completion");
322 }
323});