fix(lib): tolerate malformed json in sse stream

Author: ex3ndr

sources/lib/lib.integration.test.ts

@@ -92,4 +92,23 @@ describe("lib integration", () => {
     },
     TEST_TIMEOUT_MS
   );
+
+  it(
+    "sse stream skips malformed and non-json lines",
+    async () => {
+      const runner = createNodeRunner(
+        'console.log("event: connected"); console.log("data: {\\"id\\":1}"); console.log("not-json"); console.log("{\\"id\\":2}"); console.log("data: [DONE]");'
+      );
+
+      const stream = createJsonSseStream<{ id: number }>(runner);
+      const received: Array<{ id: number }> = [];
+
+      for await (const event of stream.events) {
+        received.push(event.data);
+      }
+
+      expect(received).toEqual([{ id: 1 }, { id: 2 }]);
+    },
+    TEST_TIMEOUT_MS
+  );
 });

sources/lib/sse.ts

@@ -77,13 +77,19 @@ export function createJsonSseStream<T = unknown>(
           if (!trimmed) {
             continue;
           }
-          yield parseJsonLine<T>(trimmed);
+          const parsed = tryParseJsonLine<T>(trimmed);
+          if (parsed) {
+            yield parsed;
+          }
         }
       }
 
       const remaining = buffer.trim();
       if (remaining) {
-        yield parseJsonLine<T>(remaining);
+        const parsed = tryParseJsonLine<T>(remaining);
+        if (parsed) {
+          yield parsed;
+        }
       }
 
       const exitCode = await process.exited;
@@ -113,15 +119,35 @@ export function createJsonSseStream<T = unknown>(
   return { events, close, process };
 }
 
-function parseJsonLine<T>(line: string): JsonSseEvent<T> {
+function tryParseJsonLine<T>(line: string): JsonSseEvent<T> | null {
+  const payload = normalizeJsonPayload(line);
+  if (payload === null) {
+    return null;
+  }
+
   try {
-    const data = JSON.parse(line) as T;
+    const data = JSON.parse(payload) as T;
     return { data, raw: line };
-  } catch (error) {
-    const message =
-      error instanceof Error ? error.message : "Unknown JSON parse error";
-    throw new Error(`Invalid JSON from bee stream: ${message}`);
+  } catch {
+    return null;
+  }
+}
+
+function normalizeJsonPayload(line: string): string | null {
+  if (line.startsWith(":")) {
+    return null;
+  }
+  if (line.startsWith("event:") || line.startsWith("id:")) {
+    return null;
+  }
+  if (line.startsWith("data:")) {
+    const payload = line.slice("data:".length).trimStart();
+    if (!payload || payload === "[DONE]") {
+      return null;
+    }
+    return payload;
   }
+  return line;
 }
 
 function readStream(