mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
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 <triagent@localhost>
This commit is contained in:
parent
2205a824bb
commit
b64e6badd6
2 changed files with 71 additions and 9 deletions
|
|
@ -64,8 +64,7 @@ export async function consumeSseStream(stream: ReadableStream<Uint8Array>, 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<Uint8Array>, 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) {
|
||||
|
|
|
|||
|
|
@ -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<Uint8Array>({
|
||||
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<Uint8Array>;
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
start(c) {
|
||||
controller = c;
|
||||
},
|
||||
});
|
||||
const received = new Promise<SseEvent>((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;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue