fix(mcp): join chunks of an incomplete stdio message once instead of on every chunk

fixes #11761

packages/mcp/src/transports/lines.ts

@@ -18,7 +18,9 @@ export interface JsonRpcLineReaderOptions {
  */
 export class JsonRpcLineReader {
 	private readonly options: JsonRpcLineReaderOptions;
-	private stdoutBuffer = Buffer.alloc(0);
+	/** The bytes of the incomplete message received so far. */
+	private stdoutChunks: Buffer[] = [];
+	private stdoutPendingBytes = 0;
 	private stderrBuffer = Buffer.alloc(0);
 
 	constructor(options: JsonRpcLineReaderOptions) {
@@ -30,21 +32,38 @@ export class JsonRpcLineReader {
 		return this.stderrBuffer.toString("utf8");
 	}
 
-	/** Takes the next stdout bytes and reports every complete message, or the error parsing it. */
+	/**
+	 * Takes the next stdout bytes and reports every complete message, or the error parsing it. Bytes of an
+	 * incomplete message are kept as chunks and joined once when its newline arrives, so a large message that
+	 * arrives in many chunks is copied and scanned only once.
+	 */
 	readStdout(chunk: Uint8Array): void {
-		this.stdoutBuffer = Buffer.concat([this.stdoutBuffer, chunk]);
+		const data = Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength);
 		const maxMessageBytes = this.options.maxMessageBytes ?? DEFAULT_MAX_MESSAGE_BYTES;
+		let start = 0;
 		while (true) {
-			const newline = this.stdoutBuffer.indexOf(0x0a);
+			const newline = data.indexOf(0x0a, start);
 			if (newline < 0) {
-				if (this.stdoutBuffer.length > maxMessageBytes) {
-					this.stdoutBuffer = Buffer.alloc(0);
+				if (start < data.length) {
+					// Copied, since the caller may reuse the chunk's memory.
+					this.stdoutChunks.push(Buffer.from(data.subarray(start)));
+					this.stdoutPendingBytes += data.length - start;
+				}
+				if (this.stdoutPendingBytes > maxMessageBytes) {
+					this.stdoutChunks = [];
+					this.stdoutPendingBytes = 0;
 					this.options.onError(new Error(`MCP stdio message exceeds ${maxMessageBytes} bytes`));
 				}
 				return;
 			}
-			const line = this.stdoutBuffer.subarray(0, newline);
-			this.stdoutBuffer = this.stdoutBuffer.subarray(newline + 1);
+			let line = data.subarray(start, newline);
+			if (this.stdoutChunks.length > 0) {
+				this.stdoutChunks.push(line);
+				line = Buffer.concat(this.stdoutChunks, this.stdoutPendingBytes + line.length);
+				this.stdoutChunks = [];
+				this.stdoutPendingBytes = 0;
+			}
+			start = newline + 1;
 			if (line.length > maxMessageBytes) {
 				this.options.onError(new Error(`MCP stdio message exceeds ${maxMessageBytes} bytes`));
 				continue;
@@ -69,9 +88,10 @@ export class JsonRpcLineReader {
 
 	/** Ends stdout: a message without its newline is reported as an error and dropped. */
 	end(): void {
-		if (this.stdoutBuffer.toString("utf8").trim()) {
+		if (Buffer.concat(this.stdoutChunks).toString("utf8").trim()) {
 			this.options.onError(new Error("MCP stdio server closed with an incomplete JSON-RPC message"));
 		}
-		this.stdoutBuffer = Buffer.alloc(0);
+		this.stdoutChunks = [];
+		this.stdoutPendingBytes = 0;
 	}
 }

packages/mcp/test/lines.test.ts

@@ -0,0 +1,75 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import type { JsonRpcMessage } from "../src/index.ts";
+import { JsonRpcLineReader } from "../src/transports/lines.ts";
+
+function startReader(maxMessageBytes?: number) {
+	const messages: JsonRpcMessage[] = [];
+	const errors: string[] = [];
+	const reader = new JsonRpcLineReader({
+		maxMessageBytes,
+		onMessage: (message) => messages.push(message),
+		onError: (error) => errors.push(error instanceof Error ? error.message : String(error)),
+	});
+	return { reader, messages, errors };
+}
+
+function feed(reader: JsonRpcLineReader, bytes: Buffer, chunkSize: number): void {
+	for (let i = 0; i < bytes.length; i += chunkSize) reader.readStdout(bytes.subarray(i, i + chunkSize));
+}
+
+describe("JsonRpcLineReader", () => {
+	afterEach(() => {
+		vi.restoreAllMocks();
+	});
+
+	it("joins a message split across many chunks once, when its newline arrives", () => {
+		const { reader, messages, errors } = startReader();
+		const text = "x".repeat(1024 * 1024);
+		const bytes = Buffer.from(`${JSON.stringify({ jsonrpc: "2.0", id: 1, result: { text } })}\n`);
+		const concat = vi.spyOn(Buffer, "concat");
+		feed(reader, bytes, 4096);
+		expect(concat).toHaveBeenCalledTimes(1);
+		expect(errors).toEqual([]);
+		expect(messages).toEqual([{ jsonrpc: "2.0", id: 1, result: { text } }]);
+	});
+
+	it("handles multi-byte characters, CRLF, blank lines, and several messages per chunk at any split", () => {
+		const input = Buffer.from(
+			[
+				JSON.stringify({ jsonrpc: "2.0", id: 1, result: { text: "héllo 🌍" } }),
+				"",
+				`${JSON.stringify({ jsonrpc: "2.0", method: "notifications/x" })}\r`,
+				"not json",
+				JSON.stringify({ jsonrpc: "2.0", id: 2, result: {} }),
+				"",
+			].join("\n"),
+		);
+		for (let chunkSize = 1; chunkSize <= input.length; chunkSize++) {
+			const { reader, messages, errors } = startReader();
+			feed(reader, input, chunkSize);
+			reader.end();
+			expect(messages).toEqual([
+				{ jsonrpc: "2.0", id: 1, result: { text: "héllo 🌍" } },
+				{ jsonrpc: "2.0", method: "notifications/x" },
+				{ jsonrpc: "2.0", id: 2, result: {} },
+			]);
+			expect(errors).toHaveLength(1);
+		}
+	});
+
+	it("drops an oversized message and keeps reading the next one", () => {
+		const { reader, messages, errors } = startReader(64);
+		const next = JSON.stringify({ jsonrpc: "2.0", id: 2, result: {} });
+		feed(reader, Buffer.from(`${"a".repeat(50)}`), 10);
+		feed(reader, Buffer.from(`${"a".repeat(50)}\n${next}\n`), 10);
+		expect(errors[0]).toBe("MCP stdio message exceeds 64 bytes");
+		expect(messages).toEqual([{ jsonrpc: "2.0", id: 2, result: {} }]);
+	});
+
+	it("reports a message without its newline at the end", () => {
+		const { reader, errors } = startReader();
+		feed(reader, Buffer.from('{"jsonrpc":"2.0",'), 4);
+		reader.end();
+		expect(errors).toEqual(["MCP stdio server closed with an incomplete JSON-RPC message"]);
+	});
+});