fix(http): handle CR line endings in SSE streams

src/core/line-buffer.test.ts

@@ -1,5 +1,6 @@
 import { describe, it, expect } from "vitest";
 import { LineBuffer } from "./line-buffer.js";
+import { MessageTooLargeError } from "./stream-limits.js";
 
 const encoder = new TextEncoder();
 const decoder = new TextDecoder();
@@ -55,4 +56,43 @@ describe("LineBuffer", () => {
 
     expect(pushAll(buffer, ["", "a", "", "\n"])).toEqual(["a"]);
   });
+
+  it("keeps a bare CR inside a line by default", () => {
+    const buffer = new LineBuffer();
+
+    expect(pushAll(buffer, ["a\rb\r\n", "c\r", "\n"])).toEqual(["a\rb", "c"]);
+  });
+
+  describe("with splitOnCarriageReturn", () => {
+    it("splits at CR, LF, and CRLF", () => {
+      const buffer = new LineBuffer(undefined, { splitOnCarriageReturn: true });
+
+      expect(pushAll(buffer, ["a\rb\nc\r\nd\n\re\r\r\nf"])).toEqual([
+        "a",
+        "b",
+        "c",
+        "d",
+        "",
+        "e",
+        "",
+      ]);
+      expect(decoder.decode(buffer.flush())).toBe("f");
+    });
+
+    it("joins a CRLF split across chunks", () => {
+      const buffer = new LineBuffer(undefined, { splitOnCarriageReturn: true });
+
+      expect(
+        pushAll(buffer, ["a\r", "\nb\r", "", "\nc\r", "\r\nd\r", "e", "\n"]),
+      ).toEqual(["a", "b", "c", "", "d", "e"]);
+      expect(buffer.flush()).toBeUndefined();
+    });
+
+    it("applies the line limit without counting the line ending", () => {
+      const buffer = new LineBuffer(3, { splitOnCarriageReturn: true });
+
+      expect(pushAll(buffer, ["abc\r", "\nde", "f\r"])).toEqual(["abc", "def"]);
+      expect(() => pushAll(buffer, ["abcd\r"])).toThrow(MessageTooLargeError);
+    });
+  });
 });

src/core/line-buffer.ts

@@ -5,11 +5,18 @@ import {
 } from "./stream-limits.js";
 
 const newline = 0x0a;
+const carriageReturn = 0x0d;
+
+export interface LineBufferOptions {
+  /** Also end lines at a bare CR, as SSE allows. Defaults to false. */
+  splitOnCarriageReturn?: boolean;
+}
 
 /**
- * Incrementally splits a byte stream into newline-terminated lines.
+ * Incrementally splits a byte stream into LF- or CRLF-terminated lines, and
+ * optionally CR-terminated lines.
  *
- * Only the newly pushed chunk is scanned for newlines, so splitting costs
+ * Only the newly pushed chunk is scanned for line endings, so splitting costs
  * O(total bytes) no matter how many chunks a line spans.
  *
  * Chunks passed to {@link push} must not be mutated while its iterator is
@@ -19,25 +26,57 @@ export class LineBuffer {
   /** Bytes of the current (incomplete) line, carried across chunks. */
   readonly #pending: MessageBuffer;
   readonly #maxMessageBytes: number;
+  readonly #splitOnCarriageReturn: boolean;
+  /** Whether the last line ended at a CR whose LF may start the next chunk. */
+  #afterCarriageReturn = false;
 
-  constructor(maxMessageBytes?: number) {
+  constructor(
+    maxMessageBytes?: number,
+    { splitOnCarriageReturn = false }: LineBufferOptions = {},
+  ) {
     this.#maxMessageBytes = resolveMaxMessageBytes(maxMessageBytes);
     this.#pending = new MessageBuffer(
       Math.min(Number.MAX_SAFE_INTEGER, this.#maxMessageBytes + 1),
     );
+    this.#splitOnCarriageReturn = splitOnCarriageReturn;
   }
 
   /**
-   * Consumes a chunk, yielding each complete line without its trailing
-   * LF or CRLF.
+   * Consumes a chunk, yielding each complete line without its line ending.
    */
   *push(chunk: Uint8Array): Generator<Uint8Array> {
     let start = 0;
+    // A CRLF may be split across chunks, even with empty chunks in between.
+    if (this.#afterCarriageReturn && chunk.byteLength > 0) {
+      this.#afterCarriageReturn = false;
+      if (chunk[0] === newline) {
+        start = 1;
+      }
+    }
+
     let newlineIndex = chunk.indexOf(newline, start);
-    while (newlineIndex !== -1) {
-      yield this.#takeLine(chunk.subarray(start, newlineIndex));
-      start = newlineIndex + 1;
-      newlineIndex = chunk.indexOf(newline, start);
+    let carriageReturnIndex = this.#splitOnCarriageReturn
+      ? chunk.indexOf(carriageReturn, start)
+      : -1;
+    while (newlineIndex !== -1 || carriageReturnIndex !== -1) {
+      if (
+        carriageReturnIndex === -1 ||
+        (newlineIndex !== -1 && newlineIndex < carriageReturnIndex)
+      ) {
+        yield this.#takeLine(chunk.subarray(start, newlineIndex));
+        start = newlineIndex + 1;
+        newlineIndex = chunk.indexOf(newline, start);
+      } else {
+        yield this.#takeLine(chunk.subarray(start, carriageReturnIndex));
+        start = carriageReturnIndex + 1;
+        if (start === chunk.byteLength) {
+          this.#afterCarriageReturn = true;
+        } else if (chunk[start] === newline) {
+          start++;
+          newlineIndex = chunk.indexOf(newline, start);
+        }
+        carriageReturnIndex = chunk.indexOf(carriageReturn, start);
+      }
     }
     if (start < chunk.byteLength) {
       const tail = chunk.subarray(start);
@@ -51,13 +90,15 @@ export class LineBuffer {
    * undefined if no bytes are buffered.
    */
   flush(): Uint8Array | undefined {
+    this.#afterCarriageReturn = false;
     if (this.#pending.byteLength === 0) {
       return undefined;
     }
     return stripCarriageReturn(this.#pending.take());
   }
 
   clear(): void {
+    this.#afterCarriageReturn = false;
     this.#pending.clear();
   }
 
@@ -74,7 +115,9 @@ export class LineBuffer {
     const lastByte =
       tail.byteLength > 0 ? tail[tail.byteLength - 1] : this.#pending.lastByte;
     const byteLength =
-      this.#pending.byteLength + tail.byteLength - (lastByte === 0x0d ? 1 : 0);
+      this.#pending.byteLength +
+      tail.byteLength -
+      (lastByte === carriageReturn ? 1 : 0);
     if (byteLength > this.#maxMessageBytes) {
       this.clear();
       throw new MessageTooLargeError(this.#maxMessageBytes);
@@ -83,5 +126,7 @@ export class LineBuffer {
 }
 
 function stripCarriageReturn(line: Uint8Array): Uint8Array {
-  return line[line.byteLength - 1] === 0x0d ? line.subarray(0, -1) : line;
+  return line[line.byteLength - 1] === carriageReturn
+    ? line.subarray(0, -1)
+    : line;
 }

src/transports/sse.test.ts

@@ -154,6 +154,58 @@ describe("SSE transport helpers", () => {
     ]);
   });
 
+  it("parses events separated by bare CR blank lines", async () => {
+    const first: AnyMessage = { jsonrpc: "2.0", id: 1, result: { ok: true } };
+    const second: AnyMessage = {
+      jsonrpc: "2.0",
+      method: "session/update",
+      params: { sessionId: "s1" },
+    };
+    const body =
+      `data: ${JSON.stringify(first)}\r\r` +
+      `data: ${JSON.stringify(second)}\r\r`;
+
+    await expect(collectMessages(streamFromChunks([body]))).resolves.toEqual([
+      first,
+      second,
+    ]);
+  });
+
+  it("parses events with mixed line endings split byte by byte", async () => {
+    const first: AnyMessage = { jsonrpc: "2.0", id: 1, result: { ok: true } };
+    const second: AnyMessage = {
+      jsonrpc: "2.0",
+      method: "session/update",
+      params: { sessionId: "s1" },
+    };
+    const body =
+      'data: {"jsonrpc":"2.0",\rdata: "id":1,\r\n' +
+      'data: "result":{"ok":true}}\n\r' +
+      `data: ${JSON.stringify(second)}\r\n\n`;
+
+    await expect(
+      collectMessages(streamFromChunks(chunkBytes(body, 1))),
+    ).resolves.toEqual([first, second]);
+  });
+
+  it("yields a CR-terminated event before the stream ends", async () => {
+    const message: AnyMessage = { jsonrpc: "2.0", id: 1, result: { ok: true } };
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(
+          new TextEncoder().encode(`data: ${JSON.stringify(message)}\r\r`),
+        );
+      },
+    });
+    const iterator = parseSseStream(body)[Symbol.asyncIterator]();
+
+    await expect(iterator.next()).resolves.toEqual({
+      value: message,
+      done: false,
+    });
+    await iterator.return?.();
+  });
+
   it("parses a final event without a trailing blank line", async () => {
     const message: AnyMessage = { jsonrpc: "2.0", id: 1, result: { ok: true } };
 

src/transports/sse.ts

@@ -9,7 +9,7 @@ import {
 
 const dataPrefix = new TextEncoder().encode("data:");
 const dataSeparator = new Uint8Array([0x0a]);
-const maxLineOverhead = 9; // UTF-8 BOM + "data: "; LineBuffer strips LF/CRLF.
+const maxLineOverhead = 9; // UTF-8 BOM + "data: "; LineBuffer strips line endings.
 
 export function serializeSseEvent(msg: AnyMessage): string {
   return serializeSseJson(JSON.stringify(msg));
@@ -30,8 +30,10 @@ export async function* parseSseStream(
 ): AsyncIterable<AnyMessage> {
   const decoder = new TextDecoder("utf-8", { ignoreBOM: true });
   const eventData = new MessageBuffer(maxMessageBytes);
+  // SSE lines may end in CRLF, LF, or a bare CR.
   const lines = new LineBuffer(
     Math.min(Number.MAX_SAFE_INTEGER, maxMessageBytes + maxLineOverhead),
+    { splitOnCarriageReturn: true },
   );
   const reader = body.getReader();
   let hasData = false;