1// @effect-diagnostics nodeBuiltinImport:off
2import * as NodeChildProcess from "node:child_process";
3import * as NodeFS from "node:fs";
4import * as NodeReadline from "node:readline";
5import * as NodeURL from "node:url";
6
7import { describe, expect, it } from "@effect/vitest";
8
9import { parseReplayTurns, SCENARIOS, streamChunks } from "./dev-replay-agent.ts";
10
11interface AgentMessage {
12 readonly id?: number;
13 readonly method?: string;
14 readonly params?: {
15 readonly update?: {
16 readonly sessionUpdate?: string;
17 readonly toolCallId?: string;
18 readonly status?: string;
19 };
20 };
21 readonly result?: { readonly sessionId?: string; readonly stopReason?: string };
22 readonly error?: { readonly code?: number };
23}
24
25/**
26 * Runs the agent over stdio and returns every message it sent until each request is answered.
27 * Permission requests get `permissionResult`.
28 */
29async function talkToAgent(
30 requests: ReadonlyArray<(sessionId: string) => object>,
31 permissionResult: object = { outcome: { outcome: "selected", optionId: "allow-once" } },
32) {
33 const agent = NodeChildProcess.spawn(
34 process.execPath,
35 [NodeURL.fileURLToPath(new URL("./dev-replay-agent.ts", import.meta.url))],
36 { stdio: ["pipe", "pipe", "inherit"] },
37 );
38 const lines = NodeReadline.createInterface({ input: agent.stdout })[Symbol.asyncIterator]();
39 const received: Array<AgentMessage> = [];
40 let sessionId = "";
41 try {
42 for (const [index, request] of requests.entries()) {
43 agent.stdin.write(
44 `${JSON.stringify({ jsonrpc: "2.0", id: index + 1, ...request(sessionId) })}\n`,
45 );
46 for (;;) {
47 const message = JSON.parse((await lines.next()).value as string) as AgentMessage;
48 received.push(message);
49 if (message.method === "session/request_permission") {
50 const answer = { jsonrpc: "2.0", id: message.id, result: permissionResult };
51 agent.stdin.write(`${JSON.stringify(answer)}\n`);
52 }
53 if (message.id !== index + 1 || message.method !== undefined) continue;
54 sessionId = message.result?.sessionId ?? sessionId;
55 break;
56 }
57 }
58 } finally {
59 agent.kill();
60 }
61 return { sessionId, received };
62}
63
64describe("dev replay agent", () => {
65 it("keeps sessions and tool ids unique across processes and reloads", async () => {
66 const prompt = (sessionId: string) => ({ method: "session/prompt", params: { sessionId } });
67 const selectTodoList = (sessionId: string) => ({
68 method: "session/set_model",
69 params: { sessionId, modelId: "todo-list" },
70 });
71 const run = () =>
72 talkToAgent([
73 () => ({ method: "session/new", params: { cwd: "/tmp" } }),
74 selectTodoList,
75 prompt,
76 prompt,
77 ]);
78 const [first, second] = await Promise.all([run(), run()]);
79 expect(first.sessionId).not.toBe(second.sessionId);
80 // A server restart loads the same session in a new agent process.
81 const reloaded = await talkToAgent([
82 () => ({ method: "session/load", params: { sessionId: first.sessionId, cwd: "/tmp" } }),
83 () => selectTodoList(first.sessionId),
84 () => prompt(first.sessionId),
85 ]);
86 const toolCallIds = [...first.received, ...reloaded.received].flatMap((message) =>
87 message.params?.update?.toolCallId === undefined ? [] : [message.params.update.toolCallId],
88 );
89 expect(new Set(toolCallIds).size).toBe(6);
90 });
91
92 it.each(SCENARIOS)("$id plays a turn without session setup frames", ({ file }) => {
93 const turns = parseReplayTurns(NodeFS.readFileSync(new URL(file, import.meta.url), "utf8"));
94 expect(turns.length).toBeGreaterThan(0);
95 const updates = turns.flatMap((turn) =>
96 turn.frames.flatMap(({ frame }) =>
97 frame.method === "session/update"
98 ? [(frame.params as { update: { sessionUpdate: string } }).update.sessionUpdate]
99 : [],
100 ),
101 );
102 expect(updates).toContain("agent_message_chunk");
103 expect(updates).not.toContain("available_commands_update");
104 });
105
106 it("keeps permission requests and the recorded stop reason in order", () => {
107 const turns = parseReplayTurns(
108 NodeFS.readFileSync(new URL(SCENARIOS[0].file, import.meta.url), "utf8"),
109 );
110 const methods = turns[0]!.frames.map(({ frame }) => frame.method);
111 expect(methods).toContain("session/request_permission");
112 expect(turns[0]!.stopReason).toBe("end_turn");
113 });
114
115 it("rejects an unknown scenario instead of playing another one", async () => {
116 const { received } = await talkToAgent([
117 () => ({ method: "session/new", params: { cwd: "/tmp" } }),
118 (sessionId) => ({ method: "session/set_model", params: { sessionId, modelId: "typo" } }),
119 ]);
120 expect(received.at(-1)?.error?.code).toBe(-32602);
121 });
122
123 it("ends the turn with a failed tool call when the user rejects an approval", async () => {
124 const { received } = await talkToAgent(
125 [
126 () => ({ method: "session/new", params: { cwd: "/tmp" } }),
127 (sessionId) => ({ method: "session/prompt", params: { sessionId } }),
128 ],
129 { outcome: { outcome: "selected", optionId: "reject-once" } },
130 );
131 const updates = received.flatMap((message) =>
132 message.params?.update === undefined ? [] : [message.params.update],
133 );
134 expect(updates.at(-1)).toMatchObject({ sessionUpdate: "tool_call_update", status: "failed" });
135 expect(updates.some((update) => update.toolCallId?.endsWith("showcase-test"))).toBe(false);
136 expect(received.at(-1)?.result?.stopReason).toBe("end_turn");
137 });
138
139 it("ends the turn as cancelled when the client cancels an open approval", async () => {
140 const { received } = await talkToAgent(
141 [
142 () => ({ method: "session/new", params: { cwd: "/tmp" } }),
143 (sessionId) => ({ method: "session/prompt", params: { sessionId } }),
144 ],
145 { outcome: { outcome: "cancelled" } },
146 );
147 const requestIndex = received.findIndex(
148 (message) => message.method === "session/request_permission",
149 );
150 expect(received.slice(requestIndex + 1).some((message) => message.method !== undefined)).toBe(
151 false,
152 );
153 expect(received.at(-1)?.result?.stopReason).toBe("cancelled");
154 });
155
156 it("leaves frames after a prompt's response out of that turn", () => {
157 const update = (text: string) => ({
158 type: "emit_inbound",
159 frame: {
160 kind: "notification",
161 method: "session/update",
162 params: {
163 update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text } },
164 },
165 },
166 });
167 const transcript = [
168 { type: "expect_outbound", frame: { kind: "request", method: "session/prompt" } },
169 update("root"),
170 { type: "emit_inbound", frame: { kind: "response", method: "session/prompt", result: {} } },
171 update("background"),
172 ]
173 .map((entry) => JSON.stringify(entry))
174 .join("\n");
175 const turns = parseReplayTurns(transcript);
176 expect(turns).toHaveLength(1);
177 expect(turns[0]!.frames).toHaveLength(1);
178 });
179
180 it("streams text in chunks that rebuild the original", () => {
181 const text = "One two three four five\n\n```ts\nx\n```";
182 const chunks = streamChunks(text);
183 expect(chunks.length).toBeGreaterThan(1);
184 expect(chunks.join("")).toBe(text);
185 });
186});