Open History to see the last commit
1#!/usr/bin/env node
2// @effect-diagnostics nodeBuiltinImport:off globalTimers:off
3/**
4 * A free ACP agent for dev servers. It plays recorded ACP transcripts with simulated
5 * streaming, so the real adapter, orchestration, and UI run without a paid model.
6 * Each scenario is a "model"; each prompt plays the next recorded turn.
7 * Unlike acp-replay-agent.ts, it does not assert the client's frames.
8 */
9import * as NodeCrypto from "node:crypto";
10import * as NodeFS from "node:fs";
11import * as NodeReadline from "node:readline";
12
13interface LogicalFrame {
14 readonly kind: "notification" | "request" | "response";
15 readonly method: string;
16 readonly params?: unknown;
17 readonly result?: unknown;
18}
19
20interface ReplayEntry {
21 readonly type: "emit_inbound" | "expect_outbound" | "runtime_exit" | "transcript_start";
22 readonly frame?: LogicalFrame;
23 readonly afterMs?: number;
24}
25
26interface JsonRpcMessage {
27 readonly jsonrpc?: string;
28 readonly id?: string | number | null;
29 readonly method?: string;
30 readonly params?: Record<string, unknown>;
31 readonly result?: unknown;
32 readonly error?: unknown;
33}
34
35export interface ReplayTurn {
36 /** Frames the agent sends: session updates and agent-to-client requests. */
37 readonly frames: ReadonlyArray<{ readonly frame: LogicalFrame; readonly afterMs?: number }>;
38 readonly stopReason: string;
39}
40
41const FIXTURES = "../src/orchestration-v2/testkit/fixtures";
42
43// Recorded fixtures the replay tests already pin, plus a hand-written UI showcase.
44// Use acpRegistry transcripts: vendor transcripts carry state in frames this agent drops.
45export const SCENARIOS = [
46 { id: "showcase", name: "Showcase", file: "./dev-replay/showcase_transcript.ndjson" },
47 {
48 id: "todo-list",
49 name: "Plan and reads",
50 file: `${FIXTURES}/todo_list/registry_transcript.ndjson`,
51 },
52 {
53 id: "approval",
54 name: "Write approval",
55 file: `${FIXTURES}/tool_call_read_only_on_request/registry_transcript.ndjson`,
56 },
57 {
58 id: "question",
59 name: "Question",
60 file: `${FIXTURES}/acp_elicitation/registry_transcript.ndjson`,
61 },
62 {
63 id: "multi-turn",
64 name: "Multi turn",
65 file: `${FIXTURES}/multi_turn/registry_transcript.ndjson`,
66 },
67] as const;
68
69// Session setup and vendor state stay out: they would replace the replay agent's own models.
70const REPLAYED_UPDATES = new Set([
71 "agent_message_chunk",
72 "agent_thought_chunk",
73 "tool_call",
74 "tool_call_update",
75 "plan",
76 "plan_update",
77 "plan_removed",
78 "session_info_update",
79 "usage_update",
80]);
81
82// Only requests that ask the user. A replayed terminal/create would run a real command.
83const REPLAYED_REQUESTS = new Set(["session/request_permission", "elicitation/create"]);
84
85const CHUNK_DELAY_MS = 30;
86const FRAME_DELAY_MS = 250;
87const MAX_RECORDED_DELAY_MS = 2_000;
88const WORDS_PER_CHUNK = 3;
89
90function isReplayedFrame(frame: LogicalFrame): boolean {
91 if (frame.kind === "request") return REPLAYED_REQUESTS.has(frame.method);
92 if (frame.kind !== "notification" || frame.method !== "session/update") return false;
93 const update = (frame.params as { update?: { sessionUpdate?: unknown } } | undefined)?.update;
94 return typeof update?.sessionUpdate === "string" && REPLAYED_UPDATES.has(update.sessionUpdate);
95}
96
97/** Splits a recorded transcript into the turns that follow each `session/prompt`. */
98export function parseReplayTurns(ndjson: string): ReadonlyArray<ReplayTurn> {
99 const turns: Array<{ frames: Array<ReplayTurn["frames"][number]>; stopReason: string }> = [];
100 // Frames after a prompt's response belong to background work, not to that turn.
101 let turn: (typeof turns)[number] | undefined;
102 for (const line of ndjson.split("\n")) {
103 if (line.trim().length === 0) continue;
104 const entry = JSON.parse(line) as ReplayEntry;
105 const frame = entry.frame;
106 if (frame === undefined) continue;
107 if (
108 entry.type === "expect_outbound" &&
109 frame.kind === "request" &&
110 frame.method === "session/prompt"
111 ) {
112 turn = { frames: [], stopReason: "end_turn" };
113 turns.push(turn);
114 continue;
115 }
116 if (entry.type !== "emit_inbound" || turn === undefined) continue;
117 if (frame.kind === "response" && frame.method === "session/prompt") {
118 const stopReason = (frame.result as { stopReason?: unknown } | undefined)?.stopReason;
119 if (typeof stopReason === "string") turn.stopReason = stopReason;
120 turn = undefined;
121 continue;
122 }
123 if (isReplayedFrame(frame)) {
124 turn.frames.push(entry.afterMs === undefined ? { frame } : { frame, afterMs: entry.afterMs });
125 }
126 }
127 return turns;
128}
129
130/** Splits streamed text into small chunks so the client sees real deltas. */
131export function streamChunks(text: string): ReadonlyArray<string> {
132 const words = text.match(/\S+\s*|\s+/g) ?? [];
133 const chunks: Array<string> = [];
134 for (let index = 0; index < words.length; index += WORDS_PER_CHUNK) {
135 chunks.push(words.slice(index, index + WORDS_PER_CHUNK).join(""));
136 }
137 return chunks.length === 0 ? [text] : chunks;
138}
139
140/** Fills the live session, workspace, and per-prompt tool ids, so repeated turns add rows. */
141function materialize(
142 value: unknown,
143 live: { readonly sessionId: string; readonly workspace: string; readonly turnKey: string },
144): unknown {
145 if (typeof value === "string") return value.replaceAll("<workspace>", live.workspace);
146 if (Array.isArray(value)) return value.map((entry) => materialize(entry, live));
147 if (typeof value !== "object" || value === null) return value;
148 return Object.fromEntries(
149 Object.entries(value).map(([key, entry]) => [
150 key,
151 key === "sessionId"
152 ? live.sessionId
153 : key === "toolCallId"
154 ? `${live.turnKey}-${String(entry)}`
155 : materialize(entry, live),
156 ]),
157 );
158}
159
160/** The tool call a permission answer rejected, if the user declined or cancelled. */
161function permissionRejectedToolCallId(request: LogicalFrame, answer: unknown): string | undefined {
162 if (request.method !== "session/request_permission") return undefined;
163 const params = request.params as {
164 toolCall?: { toolCallId?: string };
165 options?: ReadonlyArray<{ optionId: string; kind: string }>;
166 };
167 const outcome = (answer as { outcome?: { outcome?: string; optionId?: string } } | undefined)
168 ?.outcome;
169 const chosen = params.options?.find((option) => option.optionId === outcome?.optionId);
170 const rejected = outcome?.outcome !== "selected" || chosen?.kind.startsWith("reject") === true;
171 return rejected ? params.toolCall?.toolCallId : undefined;
172}
173
174function runAgent(): void {
175 const turnsByScenario = new Map<string, ReadonlyArray<ReplayTurn>>();
176 const loadTurns = (scenarioId: string) => {
177 const cached = turnsByScenario.get(scenarioId);
178 if (cached !== undefined) return cached;
179 const scenario = SCENARIOS.find((candidate) => candidate.id === scenarioId) ?? SCENARIOS[0];
180 const turns = parseReplayTurns(
181 NodeFS.readFileSync(new URL(scenario.file, import.meta.url), "utf8"),
182 );
183 turnsByScenario.set(scenarioId, turns);
184 return turns;
185 };
186
187 const sessions = new Map<
188 string,
189 { cwd: string; model: string; promptCount: number; cancel: (() => void) | undefined }
190 >();
191 const pendingAgentRequests = new Map<string, (result: unknown) => void>();
192 let nextAgentRequestId = 1;
193
194 const send = (message: JsonRpcMessage) =>
195 process.stdout.write(`${JSON.stringify({ jsonrpc: "2.0", ...message })}\n`);
196 // Resolves when `promise` settles or the turn is cancelled, whichever comes first.
197 const untilAbort = (promise: Promise<unknown>, signal: AbortSignal) =>
198 new Promise<void>((resolve) => {
199 if (signal.aborted) return resolve();
200 const onAbort = () => resolve();
201 signal.addEventListener("abort", onAbort, { once: true });
202 void promise.then(() => {
203 signal.removeEventListener("abort", onAbort);
204 resolve();
205 });
206 });
207 const sleep = (ms: number, signal: AbortSignal) =>
208 untilAbort(
209 new Promise((resolve) => setTimeout(resolve, Math.min(ms, MAX_RECORDED_DELAY_MS))),
210 signal,
211 );
212
213 const playTurn = async (
214 sessionId: string,
215 turn: ReplayTurn,
216 signal: AbortSignal,
217 ): Promise<"cancelled" | undefined> => {
218 const session = sessions.get(sessionId)!;
219 // Random, so a session loaded after a restart cannot reuse earlier tool rows.
220 const live = { sessionId, workspace: session.cwd, turnKey: NodeCrypto.randomUUID() };
221 for (const recorded of turn.frames) {
222 const frame = materialize(recorded.frame, live) as LogicalFrame;
223 if (frame.kind === "request") {
224 await sleep(recorded.afterMs ?? FRAME_DELAY_MS, signal);
225 if (signal.aborted) return;
226 const id = nextAgentRequestId++;
227 let answer: unknown;
228 const response = new Promise((resolve) =>
229 pendingAgentRequests.set(String(id), (result) => resolve((answer = result))),
230 );
231 send({ id, method: frame.method, params: frame.params as Record<string, unknown> });
232 await untilAbort(response, signal);
233 pendingAgentRequests.delete(String(id));
234 // The client answers "cancelled" when the user stops the turn mid-request.
235 const outcome = (answer as { outcome?: { outcome?: string } } | undefined)?.outcome;
236 if (signal.aborted || outcome?.outcome === "cancelled") return "cancelled";
237 const toolCallId = permissionRejectedToolCallId(frame, answer);
238 if (toolCallId === undefined) continue;
239 // The recording only has the approved path, so a rejection ends the turn here.
240 send({
241 method: "session/update",
242 params: {
243 sessionId,
244 update: { sessionUpdate: "tool_call_update", toolCallId, status: "failed" },
245 },
246 });
247 return;
248 }
249 const params = frame.params as { update: { sessionUpdate: string; content?: unknown } };
250 const content = params.update.content as { type?: string; text?: string } | undefined;
251 const text =
252 (params.update.sessionUpdate === "agent_message_chunk" ||
253 params.update.sessionUpdate === "agent_thought_chunk") &&
254 content?.type === "text"
255 ? content.text
256 : undefined;
257 // Live recordings hold one frame per token, so text defaults to the chunk pace.
258 const firstDelay = recorded.afterMs ?? (text === undefined ? FRAME_DELAY_MS : CHUNK_DELAY_MS);
259 const chunks = text === undefined ? [undefined] : streamChunks(text);
260 for (const [index, chunk] of chunks.entries()) {
261 await sleep(index === 0 ? firstDelay : CHUNK_DELAY_MS, signal);
262 if (signal.aborted) return;
263 send({
264 method: frame.method,
265 params:
266 chunk === undefined
267 ? params
268 : { ...params, update: { ...params.update, content: { ...content, text: chunk } } },
269 });
270 }
271 }
272 };
273
274 const handleRequest = async (message: JsonRpcMessage): Promise<unknown> => {
275 const params = message.params ?? {};
276 switch (message.method) {
277 case "initialize":
278 return {
279 protocolVersion: 1,
280 agentCapabilities: {
281 loadSession: true,
282 promptCapabilities: { image: false, audio: false, embeddedContext: true },
283 },
284 authMethods: [],
285 agentInfo: { name: "t3-dev-replay", version: "1" },
286 };
287 case "session/new":
288 case "session/load": {
289 // T3 starts one agent process per thread, so ids must be unique across processes.
290 // Loading after a restart keeps the thread usable; T3 already holds its history.
291 const sessionId =
292 message.method === "session/load" ? String(params.sessionId) : NodeCrypto.randomUUID();
293 const cwd = typeof params.cwd === "string" ? params.cwd : process.cwd();
294 sessions.set(sessionId, { cwd, model: SCENARIOS[0].id, promptCount: 0, cancel: undefined });
295 return {
296 ...(message.method === "session/new" ? { sessionId } : {}),
297 models: {
298 currentModelId: SCENARIOS[0].id,
299 availableModels: SCENARIOS.map((scenario) => ({
300 modelId: scenario.id,
301 name: scenario.name,
302 })),
303 },
304 };
305 }
306 case "session/set_model": {
307 const session = sessions.get(String(params.sessionId));
308 if (!SCENARIOS.some((scenario) => scenario.id === params.modelId)) {
309 throw Object.assign(new Error(`Unknown replay scenario ${String(params.modelId)}`), {
310 code: -32602,
311 });
312 }
313 if (session !== undefined && session.model !== params.modelId) {
314 session.model = String(params.modelId);
315 session.promptCount = 0;
316 }
317 return {};
318 }
319 case "session/prompt": {
320 const sessionId = String(params.sessionId);
321 const session = sessions.get(sessionId);
322 if (session === undefined) throw new Error(`Unknown session ${sessionId}`);
323 const turns = loadTurns(session.model);
324 const turn = turns[session.promptCount % Math.max(turns.length, 1)];
325 session.promptCount += 1;
326 const controller = new AbortController();
327 session.cancel = () => controller.abort();
328 let ended: "cancelled" | undefined;
329 try {
330 if (turn !== undefined) ended = await playTurn(sessionId, turn, controller.signal);
331 } finally {
332 session.cancel = undefined;
333 }
334 return {
335 stopReason:
336 controller.signal.aborted || ended === "cancelled"
337 ? "cancelled"
338 : (turn?.stopReason ?? "end_turn"),
339 };
340 }
341 case "session/close":
342 sessions.get(String(params.sessionId))?.cancel?.();
343 sessions.delete(String(params.sessionId));
344 return {};
345 case "authenticate":
346 case "session/set_mode":
347 return {};
348 default:
349 throw Object.assign(new Error(`Method not found: ${message.method}`), { code: -32601 });
350 }
351 };
352
353 const input = NodeReadline.createInterface({ input: process.stdin, crlfDelay: Infinity });
354 input.on("line", (line) => {
355 if (line.trim().length === 0) return;
356 let message: JsonRpcMessage;
357 try {
358 message = JSON.parse(line) as JsonRpcMessage;
359 } catch {
360 process.stderr.write(`dev replay agent ignored a line that is not JSON\n`);
361 return;
362 }
363 if (message.method === undefined) {
364 const resolve = pendingAgentRequests.get(String(message.id));
365 pendingAgentRequests.delete(String(message.id));
366 resolve?.(message.result);
367 return;
368 }
369 if (message.id === undefined || message.id === null) {
370 if (message.method === "session/cancel") {
371 sessions.get(String(message.params?.sessionId))?.cancel?.();
372 }
373 return;
374 }
375 const id = message.id;
376 handleRequest(message).then(
377 (result) => send({ id, result }),
378 (error: Error & { code?: number }) =>
379 send({ id, error: { code: error.code ?? -32603, message: error.message } }),
380 );
381 });
382}
383
384if (import.meta.main) runAgent();