fix(transport): tighten SSE buffer guards

vincentkoc · vincentkoc · commit b8c7ae1ea8b9 · 2026-06-26T17:45:01.000-07:00
diff --git a/src/agents/provider-transport-fetch.test.ts b/src/agents/provider-transport-fetch.test.ts
@@ -1133,7 +1133,8 @@ describe("buildGuardedModelFetch", () => {
   it("handles a large transport chunk containing many valid small SSE events", async () => {
     // Regression: one TCP read can deliver >64 KiB of already-delimited SSE
     // events; the cap must apply only to the unterminated tail, not the full chunk.
-    const manyEvents = `data: ${JSON.stringify({ ok: true })}\n\n`.repeat(500);
+    const eventCount = 5_000;
+    const manyEvents = `data: ${JSON.stringify({ ok: true })}\n\n`.repeat(eventCount);
     fetchWithSsrFGuardMock.mockResolvedValue({
       response: new Response(manyEvents, {
         headers: { "content-type": "text/event-stream" },
@@ -1156,7 +1157,7 @@ describe("buildGuardedModelFetch", () => {
     for await (const item of Stream.fromSSEResponse(response, new AbortController())) {
       items.push(item);
     }
-    expect(items.length).toBe(500);
+    expect(items.length).toBe(eventCount);
     expect(items[0]).toEqual({ ok: true });
   });
 
diff --git a/src/agents/provider-transport-fetch.ts b/src/agents/provider-transport-fetch.ts
@@ -132,13 +132,14 @@ function sanitizeOpenAISdkSseResponse(
               controller.close();
               return;
             }
-            buffer += decoder.decode(chunk.value, { stream: true });
-            totalBytes += chunk.value.byteLength;
-            if (totalBytes > SSE_SYNTHESIZE_JSON_MAX_BYTES) {
+            const nextTotalBytes = totalBytes + chunk.value.byteLength;
+            if (nextTotalBytes > SSE_SYNTHESIZE_JSON_MAX_BYTES) {
               throw new Error(
                 `Streaming JSON body exceeded ${SSE_SYNTHESIZE_JSON_MAX_BYTES} bytes while synthesizing SSE frames`,
               );
             }
+            totalBytes = nextTotalBytes;
+            buffer += decoder.decode(chunk.value, { stream: true });
           }
         } catch (error) {
           await reader?.cancel(error).catch(() => {});
@@ -191,6 +192,7 @@ function sanitizeOpenAISdkSseResponse(
       if (hasReadableSseData(block)) {
         controller.enqueue(encoder.encode(`${block}${separator}`));
         enqueued += 1;
+        return enqueued;
       }
     }
   };
@@ -202,6 +204,10 @@ function sanitizeOpenAISdkSseResponse(
     async pull(controller) {
       try {
         for (;;) {
+          const pending = enqueueSanitized(controller, "");
+          if (pending > 0) {
+            return;
+          }
           const chunk = await reader?.read();
           if (!chunk || chunk.done) {
             const tail = decoder.decode();
