From e78f8f02ba4b1118e0be274d406e6838971621c3 Mon Sep 17 00:00:00 2001 From: Romil Bijarnia Date: Tue, 3 Nov 2026 10:47:26 +1100 Subject: [PATCH 1/2] fix(mcp): accept CR and mixed line endings in response streams --- .../shared/src/mcpResponsePayloads.test.ts | 49 +++++++++++++++++++ packages/shared/src/mcpResponsePayloads.ts | 14 +++--- 2 files changed, 57 insertions(+), 6 deletions(-) create mode 100644 packages/shared/src/mcpResponsePayloads.test.ts diff --git a/packages/shared/src/mcpResponsePayloads.test.ts b/packages/shared/src/mcpResponsePayloads.test.ts new file mode 100644 index 0000000000000..13aa9e4b83997 --- /dev/null +++ b/packages/shared/src/mcpResponsePayloads.test.ts @@ -0,0 +1,49 @@ +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Stream from "effect/Stream"; + +import { responsePayloads } from "./mcpResponsePayloads.ts"; + +it.effect("reads SSE line endings across byte-chunk boundaries without splitting CRLF", () => + Effect.gen(function* () { + for (const separator of [ + "\n\n", + "\r\r", + "\r\n\r\n", + "\r\n\n", + "\n\r", + "\n\r\n", + "\r\r\n", + "\r\n\r", + ]) { + const bytes = new TextEncoder().encode( + `: keep-alive${separator}data: {"id":1,\r\ndata: "result":{"text":"café😀"}}${separator}` + + 'data: {"id":2,"result":{}}\n\ndata: {"id":3}', + ); + for (let cut = 0; cut <= bytes.length; cut += 1) { + const response = new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(bytes.subarray(0, cut)); + controller.enqueue(bytes.subarray(cut)); + controller.close(); + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ); + const payloads = yield* Stream.runCollect( + responsePayloads(response, (cause) => ({ + _tag: "ResponsePayloadError" as const, + cause, + separator, + cut, + })), + ); + expect(Array.from(payloads), JSON.stringify({ separator, cut })).toEqual([ + { id: 1, result: { text: "café😀" } }, + { id: 2, result: {} }, + ]); + } + } + }), +); diff --git a/packages/shared/src/mcpResponsePayloads.ts b/packages/shared/src/mcpResponsePayloads.ts index fe5a89153d6e0..0d40e8ba7fd33 100644 --- a/packages/shared/src/mcpResponsePayloads.ts +++ b/packages/shared/src/mcpResponsePayloads.ts @@ -2,23 +2,25 @@ import * as Effect from "effect/Effect"; import * as Stream from "effect/Stream"; +const SSE_EVENT_SEPARATOR = /(?:\r\n|\r(?!\n)|\n){2}/u; + async function* sseDataLines(response: Response): AsyncGenerator { if (response.body === null) return; const decoder = new TextDecoder(); let buffered = ""; for await (const chunk of response.body) { buffered += decoder.decode(chunk as Uint8Array, { stream: true }); - let separatorIndex = buffered.search(/\n\n|\r\n\r\n/u); - while (separatorIndex !== -1) { - const rawEvent = buffered.slice(0, separatorIndex); - buffered = buffered.slice(separatorIndex).replace(/^(?:\r?\n){2}/u, ""); + let separator = SSE_EVENT_SEPARATOR.exec(buffered); + while (separator !== null) { + const rawEvent = buffered.slice(0, separator.index); + buffered = buffered.slice(separator.index + separator[0].length); const data = rawEvent - .split(/\r?\n/u) + .split(/\r\n|\r|\n/u) .filter((line) => line.startsWith("data:")) .map((line) => line.slice("data:".length).trimStart()) .join("\n"); if (data.length > 0) yield data; - separatorIndex = buffered.search(/\n\n|\r\n\r\n/u); + separator = SSE_EVENT_SEPARATOR.exec(buffered); } } } From 1daf15be351dafb27d67c71a3199ba190fda0f1f Mon Sep 17 00:00:00 2001 From: Romil Bijarnia Date: Tue, 3 Nov 2026 11:12:29 +1100 Subject: [PATCH 2/2] test(shared): cover open MCP streams and CR data lines --- .../shared/src/mcpResponsePayloads.test.ts | 74 ++++++++++++------- 1 file changed, 48 insertions(+), 26 deletions(-) diff --git a/packages/shared/src/mcpResponsePayloads.test.ts b/packages/shared/src/mcpResponsePayloads.test.ts index 13aa9e4b83997..9a3737ab93f53 100644 --- a/packages/shared/src/mcpResponsePayloads.test.ts +++ b/packages/shared/src/mcpResponsePayloads.test.ts @@ -16,34 +16,56 @@ it.effect("reads SSE line endings across byte-chunk boundaries without splitting "\r\r\n", "\r\n\r", ]) { - const bytes = new TextEncoder().encode( - `: keep-alive${separator}data: {"id":1,\r\ndata: "result":{"text":"café😀"}}${separator}` + - 'data: {"id":2,"result":{}}\n\ndata: {"id":3}', - ); - for (let cut = 0; cut <= bytes.length; cut += 1) { - const response = new Response( - new ReadableStream({ - start(controller) { - controller.enqueue(bytes.subarray(0, cut)); - controller.enqueue(bytes.subarray(cut)); - controller.close(); - }, - }), - { headers: { "content-type": "text/event-stream" } }, - ); - const payloads = yield* Stream.runCollect( - responsePayloads(response, (cause) => ({ - _tag: "ResponsePayloadError" as const, - cause, - separator, - cut, - })), + for (const dataLineSeparator of ["\n", "\r", "\r\n"]) { + const bytes = new TextEncoder().encode( + `: keep-alive${separator}data: {"id":1,${dataLineSeparator}data: "result":{"text":"café😀"}}${separator}` + + 'data: {"id":2,"result":{}}\n\ndata: {"id":3}', ); - expect(Array.from(payloads), JSON.stringify({ separator, cut })).toEqual([ - { id: 1, result: { text: "café😀" } }, - { id: 2, result: {} }, - ]); + for (let cut = 0; cut <= bytes.length; cut += 1) { + const response = new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(bytes.subarray(0, cut)); + controller.enqueue(bytes.subarray(cut)); + controller.close(); + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ); + const payloads = yield* Stream.runCollect( + responsePayloads(response, (cause) => ({ + _tag: "ResponsePayloadError" as const, + cause, + separator, + dataLineSeparator, + cut, + })), + ); + expect( + Array.from(payloads), + JSON.stringify({ separator, dataLineSeparator, cut }), + ).toEqual([ + { id: 1, result: { text: "café😀" } }, + { id: 2, result: {} }, + ]); + } } + const response = new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(`data: {"id":4}${separator}`)); + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ); + const payloads = yield* Stream.runCollect( + responsePayloads(response, (cause) => ({ + _tag: "ResponsePayloadError" as const, + cause, + separator, + })).pipe(Stream.take(1)), + ); + expect(Array.from(payloads)).toEqual([{ id: 4 }]); } }), );