Open History to see the last commit
1// @effect-diagnostics nodeBuiltinImport:off globalTimers:off
2import * as NodeStream from "node:stream";
3
4import { describe, expect, it } from "@effect/vitest";
5import * as Deferred from "effect/Deferred";
6import * as Effect from "effect/Effect";
7import * as Fiber from "effect/Fiber";
8import { vi } from "vite-plus/test";
9
10import { mcpToolSession, runAcpMcpCliFastPath, runAcpMcpStdioBridge } from "./AcpMcpStdioBridge.ts";
11
12function makeHarness(
13 responder: (request: Request) => Promise<Response> | Response,
14 input = new NodeStream.PassThrough(),
15) {
16 const written: Array<string> = [];
17 const requests: Array<{ readonly headers: Headers; readonly body: string }> = [];
18 const bridge = runAcpMcpStdioBridge({
19 endpoint: "http://127.0.0.1:1/mcp",
20 authorization: "Bearer bridge-test",
21 input,
22 output: {
23 write: (chunk: string) => {
24 written.push(chunk);
25 },
26 },
27 fetchImplementation: async (_url, init) => {
28 const request = new Request("http://127.0.0.1:1/mcp", init);
29 requests.push({ headers: request.headers, body: String(init?.body ?? "") });
30 return responder(request);
31 },
32 });
33 return { input, written, requests, bridge };
34}
35
36function pendingFetch(started: Deferred.Deferred<AbortSignal | undefined>) {
37 return (_url: string, init?: RequestInit): Promise<Response> => {
38 Deferred.doneUnsafe(started, Effect.succeed(init?.signal ?? undefined));
39 return new Promise(() => {});
40 };
41}
42
43describe("AcpMcpStdioBridge", () => {
44 it("prints structured results once in the terminal while retaining other content and errors", async () => {
45 const originalEnvironment = process.env;
46 using _environment = {
47 [Symbol.dispose]: () => {
48 process.env = originalEnvironment;
49 },
50 };
51 process.env = {
52 ...originalEnvironment,
53 T3_ACP_MCP_ENDPOINT: "http://127.0.0.1:1/mcp",
54 T3_ACP_MCP_AUTHORIZATION: "Bearer bridge-test",
55 };
56 const structuredContent = { taskId: "task-1", status: "failed" };
57 const explanation = { type: "text", text: "The command failed." };
58 const image = { type: "image", data: "image-data", mimeType: "image/png" };
59 const result = {
60 structuredContent,
61 content: [{ type: "text", text: JSON.stringify(structuredContent) }, explanation, image],
62 isError: true,
63 };
64 using fetch = vi.spyOn(globalThis, "fetch").mockImplementation(async (_url, init) => {
65 if (init?.method === "DELETE") return new Response(null, { status: 204 });
66 const message = JSON.parse(String(init?.body)) as { id?: string; method: string };
67 if (message.method === "notifications/initialized") {
68 return new Response(null, { status: 202 });
69 }
70 return Response.json(
71 {
72 jsonrpc: "2.0",
73 id: message.id,
74 result: message.method === "initialize" ? { protocolVersion: "2025-06-18" } : result,
75 },
76 { headers: { "mcp-session-id": "session-42" } },
77 );
78 });
79 using output = vi.spyOn(process.stdout, "write").mockReturnValue(true);
80
81 await runAcpMcpCliFastPath("acp-mcp-call", ["task_status", '{"taskId":"task-1"}']);
82
83 expect(output).toHaveBeenCalledExactlyOnceWith(
84 `${JSON.stringify({ ...result, content: [explanation, image] })}\n`,
85 );
86 expect(fetch).toHaveBeenLastCalledWith(
87 "http://127.0.0.1:1/mcp",
88 expect.objectContaining({ method: "DELETE" }),
89 );
90 });
91
92 it.effect("forwards requests and replays the session id and protocol version", () =>
93 Effect.gen(function* () {
94 const { input, written, requests, bridge } = makeHarness((request) => {
95 const body = JSON.parse(String(requests.at(-1)?.body ?? "{}")) as { id?: number };
96 void request;
97 return new Response(
98 JSON.stringify({
99 jsonrpc: "2.0",
100 id: body.id ?? null,
101 result: body.id === 1 ? { protocolVersion: "2025-06-18" } : { ok: true },
102 }),
103 {
104 status: 200,
105 headers: { "content-type": "application/json", "mcp-session-id": "session-42" },
106 },
107 );
108 });
109 const fiber = yield* Effect.forkChild(bridge);
110
111 input.write(`${JSON.stringify({ jsonrpc: "2.0", id: 1, method: "initialize" })}\n`);
112 input.write(`${JSON.stringify({ jsonrpc: "2.0", id: 2, method: "tools/list" })}\n`);
113 input.end();
114 yield* Fiber.join(fiber);
115
116 expect(written).toHaveLength(2);
117 expect(JSON.parse(written[0]!)).toMatchObject({ id: 1 });
118 expect(JSON.parse(written[1]!)).toMatchObject({ id: 2, result: { ok: true } });
119 expect(requests[0]?.headers.get("authorization")).toBe("Bearer bridge-test");
120 expect(requests[1]?.headers.get("mcp-session-id")).toBe("session-42");
121 expect(requests[1]?.headers.get("mcp-protocol-version")).toBe("2025-06-18");
122 }),
123 );
124
125 it.effect("acknowledges notifications silently and synthesizes errors for failed requests", () =>
126 Effect.gen(function* () {
127 let call = 0;
128 const { input, written, bridge } = makeHarness(() => {
129 call += 1;
130 return call === 1
131 ? new Response(null, { status: 202 })
132 : new Response("upstream broke", { status: 500 });
133 });
134 const fiber = yield* Effect.forkChild(bridge);
135
136 input.write(`${JSON.stringify({ jsonrpc: "2.0", method: "notifications/initialized" })}\n`);
137 input.write(`${JSON.stringify({ jsonrpc: "2.0", id: 3, method: "tools/list" })}\n`);
138 input.end();
139 yield* Fiber.join(fiber);
140
141 expect(written).toHaveLength(1);
142 expect(JSON.parse(written[0]!)).toMatchObject({
143 id: 3,
144 error: { code: -32603 },
145 });
146 }),
147 );
148
149 it.effect("emits parse errors for malformed JSON-RPC input", () =>
150 Effect.gen(function* () {
151 const { input, written, requests, bridge } = makeHarness(
152 () => new Response(null, { status: 202 }),
153 );
154 const fiber = yield* Effect.forkChild(bridge);
155
156 input.end("{not-json}\n");
157 yield* Fiber.join(fiber);
158
159 expect(requests).toHaveLength(0);
160 expect(JSON.parse(written[0]!)).toEqual({
161 jsonrpc: "2.0",
162 id: null,
163 error: { code: -32700, message: "Parse error" },
164 });
165 }),
166 );
167
168 it.effect("keeps U+2028 and U+2029 inside a message instead of splitting it", () =>
169 Effect.gen(function* () {
170 const { input, written, requests, bridge } = makeHarness(
171 () => Response.json({ jsonrpc: "2.0", id: 4, result: { ok: true } }),
172 // Object mode keeps the two writes as separate chunks instead of one buffered read.
173 new NodeStream.PassThrough({ objectMode: true }),
174 );
175 const fiber = yield* Effect.forkChild(bridge);
176 // JSON.stringify leaves both separators unescaped, as MCP stdio clients send them.
177 const message = JSON.stringify({
178 jsonrpc: "2.0",
179 id: 4,
180 method: "tools/call",
181 params: { name: "echo", arguments: { text: "a\u2028b\u2029c\u{1f30d}" } },
182 });
183
184 // Split inside the emoji's UTF-8 bytes and leave the last line unterminated, as stdin may.
185 const bytes = Buffer.from(message);
186 const split = bytes.indexOf(Buffer.from("\u{1f30d}")) + 2;
187 input.write(bytes.subarray(0, split));
188 input.end(bytes.subarray(split));
189 yield* Fiber.join(fiber);
190
191 expect(requests.map((request) => request.body)).toEqual([message]);
192 expect(written.map((chunk) => JSON.parse(chunk))).toEqual([
193 { jsonrpc: "2.0", id: 4, result: { ok: true } },
194 ]);
195 }),
196 );
197
198 it.effect("forwards SSE messages before the HTTP response closes", () =>
199 Effect.gen(function* () {
200 let resolveResponseController!: (
201 controller: ReadableStreamDefaultController<Uint8Array>,
202 ) => void;
203 const responseControllerReady = new Promise<ReadableStreamDefaultController<Uint8Array>>(
204 (resolve) => {
205 resolveResponseController = resolve;
206 },
207 );
208 let firstWrite: (() => void) | undefined;
209 const firstWritten = new Promise<void>((resolve) => {
210 firstWrite = resolve;
211 });
212 const input = new NodeStream.PassThrough();
213 const written: Array<string> = [];
214 const fiber = yield* Effect.forkChild(
215 runAcpMcpStdioBridge({
216 endpoint: "http://127.0.0.1:1/mcp",
217 authorization: "Bearer bridge-test",
218 input,
219 output: {
220 write: (chunk) => {
221 written.push(chunk);
222 firstWrite?.();
223 },
224 },
225 fetchImplementation: async () =>
226 new Response(
227 new ReadableStream<Uint8Array>({
228 start: (controller) => {
229 resolveResponseController(controller);
230 },
231 }),
232 { headers: { "content-type": "text/event-stream" } },
233 ),
234 }),
235 );
236
237 input.write(`${JSON.stringify({ jsonrpc: "2.0", id: 7, method: "tools/call" })}\n`);
238 const responseController = yield* Effect.promise(() => responseControllerReady);
239 responseController.enqueue(
240 new TextEncoder().encode(
241 'data: {"jsonrpc":"2.0","method":"notifications/progress","params":{}}\n\n',
242 ),
243 );
244 yield* Effect.promise(() => firstWritten);
245 expect(JSON.parse(written[0]!)).toMatchObject({ method: "notifications/progress" });
246
247 responseController.enqueue(
248 new TextEncoder().encode('data: {"jsonrpc":"2.0","id":7,"result":{}}\n\n'),
249 );
250 responseController.close();
251 input.end();
252 yield* Fiber.join(fiber);
253 expect(written).toHaveLength(2);
254 }),
255 );
256
257 it.effect("forwards client responses while an SSE request is still open", () =>
258 Effect.gen(function* () {
259 let responseController: ReadableStreamDefaultController<Uint8Array> | undefined;
260 let serverRequestWritten: (() => void) | undefined;
261 const serverRequestReachedClient = new Promise<void>((resolve) => {
262 serverRequestWritten = resolve;
263 });
264 const input = new NodeStream.PassThrough();
265 const written: Array<string> = [];
266 const fiber = yield* Effect.forkChild(
267 runAcpMcpStdioBridge({
268 endpoint: "http://127.0.0.1:1/mcp",
269 authorization: "Bearer bridge-test",
270 input,
271 output: {
272 write: (chunk) => {
273 written.push(chunk);
274 if (JSON.parse(chunk).id === "server-request-1") serverRequestWritten?.();
275 },
276 },
277 fetchImplementation: async (_url, init) => {
278 const body = JSON.parse(String(init?.body ?? "{}")) as {
279 readonly id?: unknown;
280 readonly method?: unknown;
281 readonly result?: unknown;
282 };
283 if (body.method === "tools/call") {
284 return new Response(
285 new ReadableStream<Uint8Array>({
286 start: (controller) => {
287 responseController = controller;
288 controller.enqueue(
289 new TextEncoder().encode(
290 'data: {"jsonrpc":"2.0","id":"server-request-1","method":"elicitation/create","params":{}}\n\n',
291 ),
292 );
293 },
294 }),
295 { headers: { "content-type": "text/event-stream" } },
296 );
297 }
298 expect(body).toMatchObject({ id: "server-request-1", result: { action: "accept" } });
299 responseController?.enqueue(
300 new TextEncoder().encode('data: {"jsonrpc":"2.0","id":7,"result":{}}\n\n'),
301 );
302 responseController?.close();
303 return new Response(null, { status: 202 });
304 },
305 }),
306 );
307
308 input.write(`${JSON.stringify({ jsonrpc: "2.0", id: 7, method: "tools/call" })}\n`);
309 yield* Effect.promise(() => serverRequestReachedClient);
310 input.write(
311 `${JSON.stringify({ jsonrpc: "2.0", id: "server-request-1", result: { action: "accept" } })}\n`,
312 );
313 input.end();
314 yield* Fiber.join(fiber);
315
316 expect(written.map((line) => JSON.parse(line).id)).toEqual(["server-request-1", 7]);
317 }),
318 );
319
320 it.effect("forwards cancellation while a tool SSE response is still open", () =>
321 Effect.gen(function* () {
322 let resolveResponseController!: (
323 controller: ReadableStreamDefaultController<Uint8Array>,
324 ) => void;
325 const responseControllerReady = new Promise<ReadableStreamDefaultController<Uint8Array>>(
326 (resolve) => {
327 resolveResponseController = resolve;
328 },
329 );
330 const methods: Array<string> = [];
331 const input = new NodeStream.PassThrough();
332 const written: Array<string> = [];
333 const fiber = yield* Effect.forkChild(
334 runAcpMcpStdioBridge({
335 endpoint: "http://127.0.0.1:1/mcp",
336 authorization: "Bearer bridge-test",
337 input,
338 output: { write: (chunk) => written.push(chunk) },
339 fetchImplementation: async (_url, init) => {
340 const body = JSON.parse(String(init?.body ?? "{}")) as {
341 readonly method?: string;
342 };
343 methods.push(body.method ?? "");
344 if (body.method === "tools/call") {
345 return new Response(
346 new ReadableStream<Uint8Array>({
347 start: resolveResponseController,
348 }),
349 { headers: { "content-type": "text/event-stream" } },
350 );
351 }
352 const controller = await responseControllerReady;
353 controller.enqueue(
354 new TextEncoder().encode('data: {"jsonrpc":"2.0","id":7,"result":{}}\n\n'),
355 );
356 controller.close();
357 return new Response(null, { status: 202 });
358 },
359 }),
360 );
361
362 input.write(`${JSON.stringify({ jsonrpc: "2.0", id: 7, method: "tools/call" })}\n`);
363 yield* Effect.promise(() => responseControllerReady);
364 input.write(
365 `${JSON.stringify({ jsonrpc: "2.0", method: "notifications/cancelled", params: { requestId: 7 } })}\n`,
366 );
367 input.end();
368 yield* Fiber.join(fiber);
369
370 expect(methods).toEqual(["tools/call", "notifications/cancelled"]);
371 expect(JSON.parse(written[0]!)).toMatchObject({ id: 7, result: {} });
372 }),
373 );
374
375 it.effect("aborts the pending fetch when a tool call is interrupted", () =>
376 Effect.gen(function* () {
377 const started = yield* Deferred.make<AbortSignal | undefined>();
378 const fiber = yield* mcpToolSession({
379 endpoint: "http://127.0.0.1:1/mcp",
380 authorization: "Bearer bridge-test",
381 clientName: "t3-code-acp-cli",
382 fetchImplementation: pendingFetch(started),
383 }).pipe(
384 Effect.flatMap((session) => session.call("orchestrator_capabilities", {})),
385 Effect.scoped,
386 Effect.forkChild,
387 );
388
389 const signal = yield* Deferred.await(started);
390 yield* Fiber.interrupt(fiber);
391 expect(signal?.aborted).toBe(true);
392 }),
393 );
394
395 it.effect("answers a server ping that arrives before the tool result on its SSE stream", () =>
396 Effect.gen(function* () {
397 const sent: Array<{ id?: unknown; method?: string; result?: unknown }> = [];
398 const result = yield* mcpToolSession({
399 endpoint: "http://127.0.0.1:1/mcp",
400 authorization: "Bearer bridge-test",
401 clientName: "t3-code-acp-cli",
402 fetchImplementation: async (_url, init) => {
403 if (init?.method === "DELETE") return new Response(null, { status: 204 });
404 const message = JSON.parse(String(init?.body)) as (typeof sent)[number];
405 sent.push(message);
406 if (message.method === "initialize") {
407 return Response.json(
408 { jsonrpc: "2.0", id: message.id, result: { protocolVersion: "2025-06-18" } },
409 { headers: { "mcp-session-id": "session-42" } },
410 );
411 }
412 if (message.method !== "tools/call") return new Response(null, { status: 202 });
413 const ping = { jsonrpc: "2.0", id: "server-ping", method: "ping" };
414 const reply = { jsonrpc: "2.0", id: message.id, result: { content: [] } };
415 return new Response(
416 `data: ${JSON.stringify(ping)}\n\ndata: ${JSON.stringify(reply)}\n\n`,
417 { headers: { "content-type": "text/event-stream" } },
418 );
419 },
420 }).pipe(
421 Effect.flatMap((session) => session.call("orchestrator_capabilities", {})),
422 Effect.scoped,
423 );
424
425 expect(result).toEqual({ content: [] });
426 expect(sent).toContainEqual({ jsonrpc: "2.0", id: "server-ping", result: {} });
427 }),
428 );
429
430 it.effect("aborts the pending fetch when the stdio bridge is interrupted", () =>
431 Effect.gen(function* () {
432 const started = yield* Deferred.make<AbortSignal | undefined>();
433 const input = new NodeStream.PassThrough();
434 const fiber = yield* runAcpMcpStdioBridge({
435 endpoint: "http://127.0.0.1:1/mcp",
436 authorization: "Bearer bridge-test",
437 input,
438 output: { write: () => undefined },
439 fetchImplementation: pendingFetch(started),
440 }).pipe(Effect.forkChild);
441
442 input.write(`${JSON.stringify({ jsonrpc: "2.0", id: 1, method: "tools/list" })}\n`);
443 const signal = yield* Deferred.await(started);
444 yield* Fiber.interrupt(fiber);
445 expect(signal?.aborted).toBe(true);
446 }),
447 );
448});