fix(app): parse multiline SSE events when observing activation

apps/app/src/lib/__tests__/telemetry-sse-regression.test.ts

@@ -0,0 +1,14 @@
+// @vitest-environment node
+import { it, expect } from "vitest";
+import { observeRunResult } from "../launch-telemetry";
+it.each(["\n","\r\n","\r"])("observes multiline events separated by %j", async newline => {
+  const body = ['data: {"type":"TEXT_MESSAGE_CONTENT",','data: "delta":"你好"}',"",'data: {"type":"RUN_FINISHED"}',"",""].join(newline);
+  const bytes = new TextEncoder().encode(body);
+  const stream = new ReadableStream<Uint8Array>({start(c){for(const b of bytes)c.enqueue(new Uint8Array([b])); c.close();}});
+  await expect(observeRunResult(stream)).resolves.toBe(true);
+  expect(stream.locked).toBe(false);
+});
+it("releases the reader after a failed stream", async () => {
+  const stream = new ReadableStream<Uint8Array>({pull(c){c.error(new Error("offline"));}});
+  await expect(observeRunResult(stream)).resolves.toBe(false); expect(stream.locked).toBe(false);
+});

apps/app/src/lib/launch-telemetry.ts

@@ -213,31 +213,42 @@ export async function observeRunResult(stream: ReadableStream<Uint8Array | strin
   let failed = false;
   let finished = false;
 
-  const inspect = (line: string) => {
-    if (!line.startsWith("data:")) return;
+  let dataLines: string[] = [];
+  const inspect = () => {
+    const data = dataLines.join("\n");
+    dataLines = [];
+    if (!data) return;
     let event: { type?: string; delta?: string };
-    try {
-      event = JSON.parse(line.slice(5));
-    } catch {
-      return;
-    }
+    try { event = JSON.parse(data); } catch { return; }
     if (event.type === "TEXT_MESSAGE_CONTENT" && typeof event.delta === "string" && event.delta.trim()) text = true;
     if (event.type === "RUN_ERROR") failed = true;
     if (event.type === "RUN_FINISHED") finished = true;
   };
-
+  const line = (value: string) => {
+    if (!value) inspect();
+    else if (value.startsWith("data:")) dataLines.push(value.slice(5).replace(/^ /, ""));
+  };
+  const consume = (final = false) => {
+    let match: RegExpExecArray | null;
+    while ((match = /\r\n|\r(?!$)|\n/.exec(carry))) {
+      line(carry.slice(0, match.index));
+      carry = carry.slice(match.index + match[0].length);
+    }
+    if (final) { line(carry.replace(/\r$/, "")); inspect(); carry = ""; }
+  };
   try {
     for (;;) {
       const { done, value } = await reader.read();
       if (done) break;
       carry += typeof value === "string" ? value : decoder.decode(value, { stream: true });
-      const lines = carry.split("\n");
-      carry = lines.pop() ?? "";
-      lines.forEach((line) => inspect(line.trim()));
+      consume();
     }
-    inspect(carry.trim());
+    carry += decoder.decode();
+    consume(true);
   } catch {
     return false;
+  } finally {
+    reader.releaseLock();
   }
   return finished && text && !failed;
 }