diff --git a/packages/shared/src/mcpResponsePayloads.test.ts b/packages/shared/src/mcpResponsePayloads.test.ts new file mode 100644 index 0000000000000..9a3737ab93f53 --- /dev/null +++ b/packages/shared/src/mcpResponsePayloads.test.ts @@ -0,0 +1,71 @@ +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", + ]) { + 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}', + ); + 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 }]); + } + }), +); 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); } } }