fix(streaming): track progress across chunk boundaries (#13839)

Co-authored-by: Paco Cartones <pacocartones@users.noreply.github.com>

open-sse/utils/progressTracker.ts

@@ -1,5 +1,3 @@
-const decoder = new TextDecoder();
-
 /**
  * Progress Tracker — Phase 9.3
  *
@@ -31,8 +29,14 @@ export function createProgressTransform({
   let startTime = Date.now();
   let intervalId;
   let writer;
+  let pendingLine = "";
 
   const encoder = new TextEncoder();
+  const decoder = new TextDecoder();
+
+  const countDataLines = (text: string) => {
+    tokenCount += text.split("\n").filter((line) => line.startsWith("data: ")).length;
+  };
 
   return new TransformStream(
     {
@@ -68,16 +72,24 @@ export function createProgressTransform({
       },
 
       transform(chunk, controller) {
-        // Count token events in the chunk
-        const text = typeof chunk === "string" ? chunk : decoder.decode(chunk);
-        // Count data lines (each is roughly one token event)
-        const dataLines = text.split("\n").filter((l) => l.startsWith("data: "));
-        tokenCount += dataLines.length;
+        const text =
+          typeof chunk === "string"
+            ? decoder.decode() + chunk
+            : decoder.decode(chunk, { stream: true });
+        pendingLine += text;
+
+        const lastNewline = pendingLine.lastIndexOf("\n");
+        if (lastNewline >= 0) {
+          countDataLines(pendingLine.slice(0, lastNewline + 1));
+          pendingLine = pendingLine.slice(lastNewline + 1);
+        }
         controller.enqueue(chunk);
       },
 
       flush() {
         clearInterval(intervalId);
+        pendingLine += decoder.decode();
+        countDataLines(pendingLine);
         // Final progress event
         if (writer) {
           try {

tests/unit/progress-tracker.test.ts

@@ -101,3 +101,42 @@ test("createProgressTransform clears the interval when aborted", async () => {
     await reader.cancel();
   });
 });
+
+test("createProgressTransform preserves split UTF-8 and counts split data lines per stream", async () => {
+  await withFakeIntervals(async () => {
+    const first = createProgressTransform();
+    const second = createProgressTransform();
+    const firstWriter = first.writable.getWriter();
+    const secondWriter = second.writable.getWriter();
+    const firstOutput = [];
+    const secondOutput = [];
+    const pump = async (transform, output) => {
+      for await (const chunk of transform.readable) output.push(chunk);
+    };
+    const firstPump = pump(first, firstOutput);
+    const secondPump = pump(second, secondOutput);
+    const firstBytes = new TextEncoder().encode("data: one 🚀\n\n");
+    const secondBytes = new TextEncoder().encode("data: two 🧭\n\n");
+
+    await firstWriter.write(firstBytes.slice(0, 2));
+    await secondWriter.write(secondBytes.slice(0, 11));
+    await firstWriter.write(firstBytes.slice(2, 12));
+    await secondWriter.write(secondBytes.slice(11));
+    await firstWriter.write(firstBytes.slice(12));
+    await Promise.all([firstWriter.close(), secondWriter.close()]);
+    await Promise.all([firstPump, secondPump]);
+
+    assert.equal(
+      new TextDecoder().decode(Buffer.concat(firstOutput.slice(0, -1))),
+      "data: one 🚀\n\n"
+    );
+    assert.equal(
+      new TextDecoder().decode(Buffer.concat(secondOutput.slice(0, -1))),
+      "data: two 🧭\n\n"
+    );
+    const firstFinal = decodeChunk(firstOutput.at(-1));
+    const secondFinal = decodeChunk(secondOutput.at(-1));
+    assert.equal(JSON.parse(firstFinal.split("data: ")[1]).tokens_generated, 1);
+    assert.equal(JSON.parse(secondFinal.split("data: ")[1]).tokens_generated, 1);
+  });
+});