1// The harness asserts raw JSON-RPC wire strings, mirroring the bridge's
2// schema-free passthrough.
3// @effect-diagnostics nodeBuiltinImport:off globalTimers:off
4import * as NodeStream from "node:stream";
5
6import { describe, expect, it } from "@effect/vitest";
7import * as Effect from "effect/Effect";
8import * as Fiber from "effect/Fiber";
9
10import { AcpMcpBridgeError, callAcpMcpTool, runAcpMcpStdioBridge } from "./AcpMcpStdioBridge.ts";
11
12function makeHarness(responder: (request: Request) => Promise<Response> | Response) {
13 const input = new NodeStream.PassThrough();
14 const written: Array<string> = [];
15 const requests: Array<{ readonly headers: Headers; readonly body: string }> = [];
16 const bridge = runAcpMcpStdioBridge({
17 endpoint: "http://127.0.0.1:1/mcp",
18 authorization: "Bearer bridge-test",
19 input,
20 output: {
21 write: (chunk: string) => {
22 written.push(chunk);
23 },
24 },
25 fetchImplementation: async (_url, init) => {
26 const request = new Request("http://127.0.0.1:1/mcp", init);
27 requests.push({ headers: request.headers, body: String(init?.body ?? "") });
28 return responder(request);
29 },
30 });
31 return { input, written, requests, bridge };
32}
33
34describe("AcpMcpStdioBridge", () => {
35 it.effect("preserves the original transport failure as the typed error cause", () =>
36 Effect.gen(function* () {
37 const source = new Error("fetch failed");
38 const failure = yield* callAcpMcpTool({
39 endpoint: "http://127.0.0.1:1/mcp",
40 authorization: "Bearer bridge-test",
41 tool: "orchestrator_capabilities",
42 arguments: {},
43 fetchImplementation: () => Promise.reject(source),
44 }).pipe(Effect.asVoid, Effect.flip);
45
46 expect(failure).toBeInstanceOf(AcpMcpBridgeError);
47 expect(failure.cause).toBe(source);
48 }),
49 );
50
51 it.effect("calls one tool through a fresh authenticated MCP session", () =>
52 Effect.gen(function* () {
53 const requests: Array<{ readonly body: unknown; readonly headers: Headers }> = [];
54 const result = yield* callAcpMcpTool({
55 endpoint: "http://127.0.0.1:1/mcp",
56 authorization: "Bearer bridge-test",
57 tool: "orchestrator_capabilities",
58 arguments: {},
59 fetchImplementation: async (_url, init) => {
60 const headers = new Headers(init?.headers);
61 const body: unknown = JSON.parse(String(init?.body ?? "{}"));
62 requests.push({ body, headers });
63 const request = body as { readonly id?: unknown; readonly method?: unknown };
64 if (request.method === "notifications/initialized") {
65 return new Response(null, { status: 202 });
66 }
67 return new Response(
68 JSON.stringify({
69 jsonrpc: "2.0",
70 id: request.id,
71 result:
72 request.method === "initialize"
73 ? { protocolVersion: "2025-06-18", capabilities: {}, serverInfo: {} }
74 : { structuredContent: { providers: ["codex"] } },
75 }),
76 {
77 headers: { "content-type": "application/json", "mcp-session-id": "session-42" },
78 },
79 );
80 },
81 });
82
83 expect(result).toEqual({ structuredContent: { providers: ["codex"] } });
84 expect(requests).toHaveLength(3);
85 expect(requests[0]?.headers.get("authorization")).toBe("Bearer bridge-test");
86 expect(requests[1]?.headers.get("mcp-session-id")).toBe("session-42");
87 expect(requests[2]?.headers.get("mcp-protocol-version")).toBe("2025-06-18");
88 expect(requests[2]?.body).toMatchObject({
89 method: "tools/call",
90 params: { name: "orchestrator_capabilities", arguments: {} },
91 });
92 }),
93 );
94
95 it.effect("forwards requests and replays the session id and protocol version", () =>
96 Effect.gen(function* () {
97 const { input, written, requests, bridge } = makeHarness((request) => {
98 const body = JSON.parse(String(requests.at(-1)?.body ?? "{}")) as { id?: number };
99 void request;
100 return new Response(
101 JSON.stringify({
102 jsonrpc: "2.0",
103 id: body.id ?? null,
104 result: body.id === 1 ? { protocolVersion: "2025-06-18" } : { ok: true },
105 }),
106 {
107 status: 200,
108 headers: { "content-type": "application/json", "mcp-session-id": "session-42" },
109 },
110 );
111 });
112 const fiber = yield* Effect.forkChild(bridge);
113
114 input.write(`${JSON.stringify({ jsonrpc: "2.0", id: 1, method: "initialize" })}\n`);
115 input.write(`${JSON.stringify({ jsonrpc: "2.0", id: 2, method: "tools/list" })}\n`);
116 input.end();
117 yield* Fiber.join(fiber);
118
119 expect(written).toHaveLength(2);
120 expect(JSON.parse(written[0]!)).toMatchObject({ id: 1 });
121 expect(JSON.parse(written[1]!)).toMatchObject({ id: 2, result: { ok: true } });
122 expect(requests[0]?.headers.get("authorization")).toBe("Bearer bridge-test");
123 expect(requests[1]?.headers.get("mcp-session-id")).toBe("session-42");
124 expect(requests[1]?.headers.get("mcp-protocol-version")).toBe("2025-06-18");
125 }),
126 );
127
128 it.effect("acknowledges notifications silently and synthesizes errors for failed requests", () =>
129 Effect.gen(function* () {
130 let call = 0;
131 const { input, written, bridge } = makeHarness(() => {
132 call += 1;
133 return call === 1
134 ? new Response(null, { status: 202 })
135 : new Response("upstream broke", { status: 500 });
136 });
137 const fiber = yield* Effect.forkChild(bridge);
138
139 input.write(`${JSON.stringify({ jsonrpc: "2.0", method: "notifications/initialized" })}\n`);
140 input.write(`${JSON.stringify({ jsonrpc: "2.0", id: 3, method: "tools/list" })}\n`);
141 input.end();
142 yield* Fiber.join(fiber);
143
144 expect(written).toHaveLength(1);
145 expect(JSON.parse(written[0]!)).toMatchObject({
146 id: 3,
147 error: { code: -32603 },
148 });
149 }),
150 );
151
152 it.effect("emits parse errors for malformed JSON-RPC input", () =>
153 Effect.gen(function* () {
154 const { input, written, requests, bridge } = makeHarness(
155 () => new Response(null, { status: 202 }),
156 );
157 const fiber = yield* Effect.forkChild(bridge);
158
159 input.end("{not-json}\n");
160 yield* Fiber.join(fiber);
161
162 expect(requests).toHaveLength(0);
163 expect(JSON.parse(written[0]!)).toEqual({
164 jsonrpc: "2.0",
165 id: null,
166 error: { code: -32700, message: "Parse error" },
167 });
168 }),
169 );
170
171 it.effect("forwards SSE messages before the HTTP response closes", () =>
172 Effect.gen(function* () {
173 let resolveResponseController!: (
174 controller: ReadableStreamDefaultController<Uint8Array>,
175 ) => void;
176 const responseControllerReady = new Promise<ReadableStreamDefaultController<Uint8Array>>(
177 (resolve) => {
178 resolveResponseController = resolve;
179 },
180 );
181 let firstWrite: (() => void) | undefined;
182 const firstWritten = new Promise<void>((resolve) => {
183 firstWrite = resolve;
184 });
185 const input = new NodeStream.PassThrough();
186 const written: Array<string> = [];
187 const fiber = yield* Effect.forkChild(
188 runAcpMcpStdioBridge({
189 endpoint: "http://127.0.0.1:1/mcp",
190 authorization: "Bearer bridge-test",
191 input,
192 output: {
193 write: (chunk) => {
194 written.push(chunk);
195 firstWrite?.();
196 },
197 },
198 fetchImplementation: async () =>
199 new Response(
200 new ReadableStream<Uint8Array>({
201 start: (controller) => {
202 resolveResponseController(controller);
203 },
204 }),
205 { headers: { "content-type": "text/event-stream" } },
206 ),
207 }),
208 );
209
210 input.write(`${JSON.stringify({ jsonrpc: "2.0", id: 7, method: "tools/call" })}\n`);
211 const responseController = yield* Effect.promise(() => responseControllerReady);
212 responseController.enqueue(
213 new TextEncoder().encode(
214 'data: {"jsonrpc":"2.0","method":"notifications/progress","params":{}}\n\n',
215 ),
216 );
217 yield* Effect.promise(() => firstWritten);
218 expect(JSON.parse(written[0]!)).toMatchObject({ method: "notifications/progress" });
219
220 responseController.enqueue(
221 new TextEncoder().encode('data: {"jsonrpc":"2.0","id":7,"result":{}}\n\n'),
222 );
223 responseController.close();
224 input.end();
225 yield* Fiber.join(fiber);
226 expect(written).toHaveLength(2);
227 }),
228 );
229
230 it.effect("forwards client responses while an SSE request is still open", () =>
231 Effect.gen(function* () {
232 let responseController: ReadableStreamDefaultController<Uint8Array> | undefined;
233 let serverRequestWritten: (() => void) | undefined;
234 const serverRequestReachedClient = new Promise<void>((resolve) => {
235 serverRequestWritten = resolve;
236 });
237 const input = new NodeStream.PassThrough();
238 const written: Array<string> = [];
239 const fiber = yield* Effect.forkChild(
240 runAcpMcpStdioBridge({
241 endpoint: "http://127.0.0.1:1/mcp",
242 authorization: "Bearer bridge-test",
243 input,
244 output: {
245 write: (chunk) => {
246 written.push(chunk);
247 if (JSON.parse(chunk).id === "server-request-1") serverRequestWritten?.();
248 },
249 },
250 fetchImplementation: async (_url, init) => {
251 const body = JSON.parse(String(init?.body ?? "{}")) as {
252 readonly id?: unknown;
253 readonly method?: unknown;
254 readonly result?: unknown;
255 };
256 if (body.method === "tools/call") {
257 return new Response(
258 new ReadableStream<Uint8Array>({
259 start: (controller) => {
260 responseController = controller;
261 controller.enqueue(
262 new TextEncoder().encode(
263 'data: {"jsonrpc":"2.0","id":"server-request-1","method":"elicitation/create","params":{}}\n\n',
264 ),
265 );
266 },
267 }),
268 { headers: { "content-type": "text/event-stream" } },
269 );
270 }
271 expect(body).toMatchObject({ id: "server-request-1", result: { action: "accept" } });
272 responseController?.enqueue(
273 new TextEncoder().encode('data: {"jsonrpc":"2.0","id":7,"result":{}}\n\n'),
274 );
275 responseController?.close();
276 return new Response(null, { status: 202 });
277 },
278 }),
279 );
280
281 input.write(`${JSON.stringify({ jsonrpc: "2.0", id: 7, method: "tools/call" })}\n`);
282 yield* Effect.promise(() => serverRequestReachedClient);
283 input.write(
284 `${JSON.stringify({ jsonrpc: "2.0", id: "server-request-1", result: { action: "accept" } })}\n`,
285 );
286 input.end();
287 yield* Fiber.join(fiber);
288
289 expect(written.map((line) => JSON.parse(line).id)).toEqual(["server-request-1", 7]);
290 }),
291 );
292
293 it.effect("forwards cancellation while a tool SSE response is still open", () =>
294 Effect.gen(function* () {
295 let resolveResponseController!: (
296 controller: ReadableStreamDefaultController<Uint8Array>,
297 ) => void;
298 const responseControllerReady = new Promise<ReadableStreamDefaultController<Uint8Array>>(
299 (resolve) => {
300 resolveResponseController = resolve;
301 },
302 );
303 const methods: Array<string> = [];
304 const input = new NodeStream.PassThrough();
305 const written: Array<string> = [];
306 const fiber = yield* Effect.forkChild(
307 runAcpMcpStdioBridge({
308 endpoint: "http://127.0.0.1:1/mcp",
309 authorization: "Bearer bridge-test",
310 input,
311 output: { write: (chunk) => written.push(chunk) },
312 fetchImplementation: async (_url, init) => {
313 const body = JSON.parse(String(init?.body ?? "{}")) as {
314 readonly method?: string;
315 };
316 methods.push(body.method ?? "");
317 if (body.method === "tools/call") {
318 return new Response(
319 new ReadableStream<Uint8Array>({
320 start: resolveResponseController,
321 }),
322 { headers: { "content-type": "text/event-stream" } },
323 );
324 }
325 const controller = await responseControllerReady;
326 controller.enqueue(
327 new TextEncoder().encode('data: {"jsonrpc":"2.0","id":7,"result":{}}\n\n'),
328 );
329 controller.close();
330 return new Response(null, { status: 202 });
331 },
332 }),
333 );
334
335 input.write(`${JSON.stringify({ jsonrpc: "2.0", id: 7, method: "tools/call" })}\n`);
336 yield* Effect.promise(() => responseControllerReady);
337 input.write(
338 `${JSON.stringify({ jsonrpc: "2.0", method: "notifications/cancelled", params: { requestId: 7 } })}\n`,
339 );
340 input.end();
341 yield* Fiber.join(fiber);
342
343 expect(methods).toEqual(["tools/call", "notifications/cancelled"]);
344 expect(JSON.parse(written[0]!)).toMatchObject({ id: 7, result: {} });
345 }),
346 );
347});