From b64e6badd6ddefda17fc7f6286c013467186d8a1 Mon Sep 17 00:00:00 2001 From: "robopipi[bot]" <336283572+robopipi[bot]@users.noreply.github.com> Date: Wed, 4 Nov 2026 09:18:01 +0100 Subject: [PATCH] fix(mcp): recognize CR-only line endings in Streamable HTTP SSE (#11775) `consumeSseStream()` split SSE lines only on `\n` and then stripped one trailing `\r`. That handled LF and CRLF. The SSE spec also allows a lone CR as a line ending, and a stream framed that way never got split. At EOF the whole stream was parsed as a single `id` line, so no event was dispatched, and a tool call answered this way failed with "stream ended without a response". With this change, a line ends at CR, LF, or CRLF. A CR at the end of a chunk ends the line right away, so CR-terminated events are delivered while the stream is still open. A flag skips the LF that may start the next chunk, so a CRLF split across chunks still counts as one line ending. Size limits, dispatch, and error cleanup are unchanged. Testing: - Added two tests to `packages/mcp/test/streamable-http.test.ts`: - LF, CRLF, and CR framing, each sent as one chunk and one character per chunk (which splits every CRLF), plus mixed endings. - A CR-terminated event is dispatched before the stream closes. - Both tests fail without the fix and pass with it. All 127 tests in `packages/mcp` pass. - Feeding `id: r`, `data: {...}`, blank, blank joined by `\r` used to produce no events. It now produces the same event as the LF and CRLF versions. - `npm run check` fails here, but only in `packages/ai` and `packages/chord`, which this change does not touch. It reports nothing in `packages/mcp`; biome and `tsc` are clean on the changed package. Fixes #11764 Requested by @badlogic Co-authored-by: triagent --- .../mcp/src/transports/streamable-http.ts | 41 +++++++++++++++---- packages/mcp/test/streamable-http.test.ts | 39 ++++++++++++++++++ 2 files changed, 71 insertions(+), 9 deletions(-) diff --git a/packages/mcp/src/transports/streamable-http.ts b/packages/mcp/src/transports/streamable-http.ts index 9ecdb2973..85e3f116e 100644 --- a/packages/mcp/src/transports/streamable-http.ts +++ b/packages/mcp/src/transports/streamable-http.ts @@ -64,8 +64,7 @@ export async function consumeSseStream(stream: ReadableStream, optio dataBytes = 0; }; - const processLine = (rawLine: string) => { - const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine; + const processLine = (line: string) => { if (line === "") { dispatch(); return; @@ -86,21 +85,45 @@ export async function consumeSseStream(stream: ReadableStream, optio } else if (field === "retry" && /^\d+$/.test(value)) options.onRetry?.(Number(value)); }; + // SSE lines end with CRLF, LF, or a lone CR. + let skipLeadingLf = false; + const consumeLines = () => { + let start = 0; + for (let i = 0; i < buffered.length; i++) { + const char = buffered[i]; + if (char !== "\r" && char !== "\n") continue; + processLine(buffered.slice(start, i)); + if (char === "\r") { + if (i + 1 < buffered.length) { + if (buffered[i + 1] === "\n") i++; + } else { + skipLeadingLf = true; + } + } + start = i + 1; + } + buffered = buffered.slice(start); + }; + try { while (true) { const { done, value } = await reader.read(); if (done) break; - buffered += decoder.decode(value, { stream: true }); - let newline = buffered.indexOf("\n"); - while (newline >= 0) { - processLine(buffered.slice(0, newline)); - buffered = buffered.slice(newline + 1); - newline = buffered.indexOf("\n"); + let text = decoder.decode(value, { stream: true }); + // A CRLF split across chunks: the CR already ended the line, so drop the LF. + if (skipLeadingLf && text.length > 0) { + if (text.startsWith("\n")) text = text.slice(1); + skipLeadingLf = false; } + buffered += text; + consumeLines(); if (Buffer.byteLength(buffered) > maxEventBytes) throw new Error(`MCP SSE event exceeds ${maxEventBytes} bytes`); } - buffered += decoder.decode(); + let tail = decoder.decode(); + if (skipLeadingLf && tail.startsWith("\n")) tail = tail.slice(1); + buffered += tail; + consumeLines(); if (buffered) processLine(buffered); dispatch(); } catch (error) { diff --git a/packages/mcp/test/streamable-http.test.ts b/packages/mcp/test/streamable-http.test.ts index bfef0210e..0b3656199 100644 --- a/packages/mcp/test/streamable-http.test.ts +++ b/packages/mcp/test/streamable-http.test.ts @@ -104,6 +104,45 @@ describe("consumeSseStream", () => { expect(events).toEqual([{ id: "7", data: '{"one":\n1}' }]); }); + it("recognizes CR, LF, and CRLF line endings, including CRLF split across chunks", async () => { + const encoder = new TextEncoder(); + const parse = async (chunks: string[]) => { + const stream = new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(encoder.encode(chunk)); + controller.close(); + }, + }); + const events: SseEvent[] = []; + await consumeSseStream(stream, { onEvent: (event) => events.push(event) }); + return events; + }; + const expected = [{ id: "1", data: "a\nb" }, { data: "c" }]; + for (const separator of ["\n", "\r\n", "\r"]) { + const text = ["id: 1", "data: a", "data: b", "", "data: c", "", ""].join(separator); + expect(await parse([text])).toEqual(expected); + // One character per chunk splits every CRLF between chunks. + expect(await parse([...text])).toEqual(expected); + } + expect(await parse(["id: 1\rdata: a\r\ndata: b\n\r", "\ndata: c\r\r"])).toEqual(expected); + }); + + it("dispatches CR-terminated events while the stream stays open", async () => { + const encoder = new TextEncoder(); + let controller!: ReadableStreamDefaultController; + const stream = new ReadableStream({ + start(c) { + controller = c; + }, + }); + const received = new Promise((resolve) => { + void consumeSseStream(stream, { onEvent: resolve }).catch(() => {}); + }); + controller.enqueue(encoder.encode("data: first\r\r")); + expect(await received).toEqual({ data: "first" }); + controller.close(); + }); + it("rejects events whose data lines exceed the limit without a blank line", async () => { const encoder = new TextEncoder(); let sent = 0;