fix: treat SSE comment lines as comments, not end-of-stream

packages/fetch/src/stream.test.ts

@@ -21,6 +21,23 @@ function createMockResponse(sseLines: string[]): Response {
   } as unknown as Response;
 }
 
+function createMockResponseFromRawChunks(chunks: string[]): Response {
+  const stream = new Readable({
+    read() {
+      for (const chunk of chunks) {
+        this.push(chunk);
+      }
+      this.push(null);
+    },
+  }) as any;
+
+  return {
+    status: 200,
+    body: stream,
+    text: async () => "",
+  } as unknown as Response;
+}
+
 describe("streamSse", () => {
   it("yields parsed SSE data objects that ends with `data:[DONE]`", async () => {
     const sseLines = [
@@ -54,6 +71,19 @@ describe("streamSse", () => {
     expect(results).toEqual([{ foo: "bar" }, { baz: 42 }]);
   });
 
+  it("ignores comment keepalive lines that share a chunk with data lines", async () => {
+    const response = createMockResponseFromRawChunks([
+      'data: {"foo": "bar"}\n\n: ping\ndata: {"baz": 42}\n\ndata: [DONE]\n\n',
+    ]);
+
+    const results = [];
+    for await (const data of streamSse(response)) {
+      results.push(data);
+    }
+
+    expect(results).toEqual([{ foo: "bar" }, { baz: 42 }]);
+  });
+
   it("throws on malformed JSON", async () => {
     const sseLines = ['data: {"foo": "bar"', "data:[DONE]"];
     const response = createMockResponse(sseLines);

packages/fetch/src/stream.ts

@@ -115,9 +115,8 @@ function parseSseLine(line: string): { done: boolean; data: any } {
   if (line.startsWith("data:")) {
     return { done: false, data: parseDataLine(line) };
   }
-  if (line.startsWith(": ping")) {
-    return { done: true, data: undefined };
-  }
+  // Lines starting with ":" are SSE comments (e.g. `: ping` keepalives) and
+  // must be skipped without interrupting the rest of the stream
   return { done: false, data: undefined };
 }