fix(core): frame MCP stdio messages without the SDK read buffer (#60694)

2026-11-14 20:21:39 +03:00 · 2026-10-30 16:00:13 -05:00
parent 208ea0e0f7
commit 1a443ae42a
2 changed files with 177 additions and 31 deletions
@@ -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<string, Cause.Done>(OUTGOING_CAPACITY)
-  const buffer = new ReadBuffer()
  const state: { phase: "ready" | "starting" | "open" | "closed"; handle?: ChildProcessHandle } = { phase: "ready" }
  let startup: Promise<void> | undefined
  let closing: Promise<void> | undefined
-  let trailingBytes = 0
+  // Bytes of the unterminated line, kept as chunk views and joined once its newline arrives.
+  let pending: Array<Uint8Array> = []
+  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)
@@ -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<Uint8Array>, options?: { open: boolean }) {
+  const killed = Deferred.makeUnsafe<void>()
+  const messages: Array<JSONRPCMessage> = []
+  const errors: Array<string> = []
+  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<void>((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 }
+}