From 1a443ae42ac8bf8cb46bb3bafe2d876115815679 Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Fri, 30 Oct 2026 16:00:13 -0500 Subject: [PATCH] fix(core): frame MCP stdio messages without the SDK read buffer (#60694) --- packages/core/src/mcp/stdio.ts | 78 +++++++++------- packages/core/test/mcp-stdio.test.ts | 130 +++++++++++++++++++++++++++ 2 files changed, 177 insertions(+), 31 deletions(-) create mode 100644 packages/core/test/mcp-stdio.test.ts diff --git a/packages/core/src/mcp/stdio.ts b/packages/core/src/mcp/stdio.ts index 31dd7f224e1..b21c2a717a7 100644 --- a/packages/core/src/mcp/stdio.ts +++ b/packages/core/src/mcp/stdio.ts @@ -1,7 +1,7 @@ export * as McpStdio from "./stdio.js" -import { ReadBuffer, serializeMessage, type JSONRPCMessage, type Transport } from "@modelcontextprotocol/client" -import { Cause, Duration, Effect, Fiber, Option, Queue, Scope, Stream } from "effect" +import { deserializeMessage, serializeMessage, type JSONRPCMessage, type Transport } from "@modelcontextprotocol/client" +import { Cause, Duration, Effect, Fiber, Option, Queue, Result, Scope, Stream } from "effect" import { ChildProcess } from "effect/process" import type { ChildProcessHandle } from "effect/process/ChildProcessSpawner" import { Environment } from "../environment/index.js" @@ -48,11 +48,13 @@ export const make = Effect.fnUntraced(function* (options: Options) { // Outgoing frames are queued rather than written to `handle.stdin` directly: the sink closes the // stream it is run with, and stdin must stay open across the whole session. const outgoing = yield* Queue.bounded(OUTGOING_CAPACITY) - const buffer = new ReadBuffer() const state: { phase: "ready" | "starting" | "open" | "closed"; handle?: ChildProcessHandle } = { phase: "ready" } let startup: Promise | undefined let closing: Promise | undefined - let trailingBytes = 0 + // Bytes of the unterminated line, kept as chunk views and joined once its newline arrives. + let pending: Array = [] + let pendingBytes = 0 + const decoder = new TextDecoder() let stderr = "" const stop = Effect.fnUntraced(function* (handle: ChildProcessHandle) { @@ -66,13 +68,15 @@ export const make = Effect.fnUntraced(function* (options: Options) { (closing ??= Effect.runPromise( Effect.gen(function* () { state.phase = "closed" + pending = [] + pendingBytes = 0 Queue.endUnsafe(outgoing) if (startup) yield* Effect.promise(() => startup!.catch(() => undefined)) const handle = state.handle if (!handle) return state.handle = undefined yield* stop(handle) - }).pipe(Effect.ensuring(Queue.shutdown(outgoing)), Effect.ensuring(Effect.sync(() => buffer.clear()))), + }).pipe(Effect.ensuring(Queue.shutdown(outgoing))), )) const transport: Transport = { @@ -114,32 +118,44 @@ export const make = Effect.fnUntraced(function* (options: Options) { close, } - const deliver = (chunk: Uint8Array) => - Effect.gen(function* () { - for (const byte of chunk) { - trailingBytes = byte === 10 ? 0 : trailingBytes + 1 - if (trailingBytes > MAX_FRAME_BYTES) return yield* Effect.fail(new Error("MCP stdio frame exceeded 16 MiB")) - } - buffer.append(Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength)) - while (true) { - // `undefined` means the frame failed to parse: the buffer has already advanced past it, so - // keep draining. `null` means the buffer holds no complete frame yet. - const message = yield* Effect.try({ - try: () => buffer.readMessage(), - catch: (cause) => (cause instanceof Error ? cause : new Error(String(cause))), - }).pipe( - Effect.catch((error) => - Effect.sync(() => { - transport.onerror?.(error) - return undefined - }), - ), - ) - if (message === undefined) continue - if (message === null) return - transport.onmessage?.(message) - } + const deliver = (chunk: Uint8Array) => { + let start = 0 + let newline = chunk.indexOf(10) + while (newline !== -1 && state.phase !== "closed") { + const line = chunk.subarray(start, newline) + const size = pendingBytes + line.length + if (size > MAX_FRAME_BYTES) return oversized() + receive(pending.length === 0 ? line : Buffer.concat([...pending, line], size)) + pending = [] + pendingBytes = 0 + start = newline + 1 + newline = chunk.indexOf(10, start) + } + if (start === chunk.length || state.phase === "closed") return + pendingBytes += chunk.length - start + if (pendingBytes > MAX_FRAME_BYTES) return oversized() + // A view is enough: the runtime hands out a fresh buffer for every pipe read. + pending.push(start === 0 ? chunk : chunk.subarray(start)) + } + + const receive = (line: Uint8Array) => { + const text = decoder.decode(line.at(-1) === 13 ? line.subarray(0, -1) : line) + if (text.trim() === "") return + const message = Result.try({ + try: () => deserializeMessage(text), + catch: (cause) => new Error(`Invalid MCP stdio message from ${options.server}: ${cause}`, { cause }), }) + if (Result.isFailure(message)) return transport.onerror?.(message.failure) + transport.onmessage?.(message.success) + } + + // Closing fails every pending request at once rather than leaving the one this line answered waiting. + const oversized = () => { + transport.onerror?.( + new Error(`MCP server ${options.server} wrote more than 16 MiB to stdout without a message boundary`), + ) + void close() + } const startOutput = (handle: ChildProcessHandle) => Effect.gen(function* () { @@ -158,7 +174,7 @@ export const make = Effect.fnUntraced(function* (options: Options) { ) yield* Effect.forkScoped( - Stream.runForEach(handle.stdout, deliver).pipe( + Stream.runForEach(handle.stdout, (chunk) => Effect.sync(() => deliver(chunk))).pipe( Effect.tapCause((cause) => Effect.sync(() => { const error = Cause.squash(cause) diff --git a/packages/core/test/mcp-stdio.test.ts b/packages/core/test/mcp-stdio.test.ts new file mode 100644 index 00000000000..7e6d7ea308f --- /dev/null +++ b/packages/core/test/mcp-stdio.test.ts @@ -0,0 +1,130 @@ +import { expect, test } from "bun:test" +import type { JSONRPCMessage } from "@modelcontextprotocol/client" +import { Environment } from "@opencode/core/environment/index" +import { McpStdio } from "@opencode/core/mcp/stdio" +import { Deferred, Effect, Layer, Sink, Stream } from "effect" +import { ChildProcessSpawner } from "effect/process" +import { ExitCode, makeHandle, ProcessId } from "effect/process/ChildProcessSpawner" + +const encoder = new TextEncoder() +const MiB = 1024 * 1024 +const after = encoder.encode('{"jsonrpc":"2.0","method":"after"}\n') + +test("receives a frame between 10 and 16 MiB split into pipe-sized chunks", async () => { + // 12 MiB sits between the SDK read buffer's 10 MiB default and our 16 MiB cap. + const output = await read(chunks(frame(12 * MiB), 64 * 1024)) + + expect(output.errors).toEqual([]) + expect(output.messages).toHaveLength(1) + expect(output.messages[0]).toMatchObject({ jsonrpc: "2.0", id: 1 }) +}) + +test("frames CRLF, blank lines, several messages per chunk and split UTF-8", async () => { + const bytes = encoder.encode( + [ + '{"jsonrpc":"2.0","method":"first","params":{"text":"héllo 🙂"}}\r\n', + "\n", + " \r\n", + "not json\n", + '{"jsonrpc":"2.0","method":"second"}\n{"jsonrpc":"2.0","method":"third"}\r\n', + ].join(""), + ) + + for (const size of [bytes.length, 1, 3]) { + const output = await read(chunks(bytes, size)) + expect(output.messages).toEqual([ + { jsonrpc: "2.0", method: "first", params: { text: "héllo 🙂" } }, + { jsonrpc: "2.0", method: "second" }, + { jsonrpc: "2.0", method: "third" }, + ]) + expect(output.errors).toEqual([expect.stringContaining("Invalid MCP stdio message")]) + } +}) + +test("accepts a frame of exactly 16 MiB", async () => { + // Frame sizes include the newline. + const output = await read(chunks(frame(16 * MiB + 1), 64 * 1024)) + + expect(output.errors).toEqual([]) + expect(output.messages).toEqual([expect.objectContaining({ id: 1, result: expect.anything() })]) +}) + +test("closes the connection on a line one byte over 16 MiB", async () => { + // stdout stays open until the process is killed, so finishing at all means the transport closed + // itself, which is what fails the client's pending requests. + const output = await read([...chunks(frame(16 * MiB + 2), 64 * 1024), after], { open: true }) + + expect(output.errors).toEqual(["MCP server framing wrote more than 16 MiB to stdout without a message boundary"]) + expect(output.messages).toEqual([]) +}) + +function frame(size: number) { + const prefix = '{"jsonrpc":"2.0","id":1,"result":{"text":"' + const suffix = '"}}\n' + return encoder.encode(prefix + "x".repeat(size - prefix.length - suffix.length) + suffix) +} + +function chunks(bytes: Uint8Array, size: number) { + return Array.from({ length: Math.ceil(bytes.length / size) }, (_, index) => + bytes.subarray(index * size, (index + 1) * size), + ) +} + +/** Runs the transport against a process whose stdout yields `output`, then ends or, with `open`, waits to be killed. */ +async function read(output: ReadonlyArray, options?: { open: boolean }) { + const killed = Deferred.makeUnsafe() + const messages: Array = [] + const errors: Array = [] + const driver = Environment.makeMemoryDriver() + const environment = Layer.succeed( + Environment.Service, + Environment.Service.of({ + files: Environment.makeFiles(driver), + spawner: ChildProcessSpawner.make(() => + Effect.succeed( + makeHandle({ + pid: ProcessId(1), + exitCode: Effect.succeed(ExitCode(0)), + isRunning: Effect.succeed(false), + kill: () => Deferred.succeed(killed, undefined), + stdin: Sink.drain, + stdout: Stream.fromIterable(output).pipe( + Stream.concat( + options?.open ? Stream.fromEffect(Deferred.await(killed)).pipe(Stream.drain) : Stream.empty, + ), + ), + stderr: Stream.empty, + all: Stream.empty, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + unref: Effect.succeed(Effect.void), + }), + ), + ), + }), + ) + + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const transport = yield* McpStdio.make({ + server: "framing", + command: "unused", + args: [], + cwd: import.meta.dir, + environment: {}, + }) + const closed = new Promise((resolve) => (transport.onclose = resolve)) + transport.onmessage = (message) => messages.push(message) + // stdout ending reports the exit; only framing errors matter here. + transport.onerror = (error) => { + if (!error.message.startsWith("MCP server process exited")) errors.push(error.message) + } + yield* Effect.promise(() => transport.start()) + yield* Effect.promise(() => closed) + }).pipe(Effect.provide(environment)), + ), + ) + + return { messages, errors } +}