fix(streaming): bound translated SSE inspection (#1112)

src/server/relay.ts

@@ -6,7 +6,6 @@ import {
   addFinalRequestLog,
   httpStatusForRequestLogTerminal,
   inspectResponseLogJson,
-  inspectResponseLogSsePayload,
   inspectResponseLogSsePayloadParsed,
   recordFirstOutput,
   type RequestLogContext,
@@ -359,53 +358,42 @@ export function trackSseForRequestLog(
   onFirstOutput?: () => void,
 ): ReadableStream<Uint8Array> {
   const reader = body.getReader();
-  const decoder = new TextDecoder();
-  let buffer = "";
   let terminalReported = false;
-  const reportFirstOutput = createFirstOutputReporter(onFirstOutput);
 
   const reportTerminal = (status: ResponsesTerminalStatus) => {
     if (terminalReported) return;
     terminalReported = true;
     onTerminal(status);
   };
-
-  const inspectPayload = (payload: string | null) => {
-    if (!payload) return;
-    if (logCtx) inspectResponseLogSsePayload(logCtx, payload);
-    reportFirstOutput.payload(payload);
-    const status = terminalStatusFromSsePayload(payload);
-    if (status) reportTerminal(status);
-  };
-
-  const inspectChunk = (value: Uint8Array) => {
-    buffer += decoder.decode(value, { stream: true });
-    let next: { block: string; rest: string } | null;
-    while ((next = nextSseBlock(buffer))) {
-      buffer = next.rest;
-      inspectPayload(sseDataPayload(next.block));
-    }
-  };
+  // Reuse the byte-bounded inspector so translated responses cannot retain an
+  // unterminated upstream frame or parse the same event once per observer.
+  const inspector = createSseInspector({
+    onTerminal: reportTerminal,
+    logCtx,
+    onFirstOutput,
+  });
 
   return new ReadableStream<Uint8Array>({
     async pull(controller) {
       try {
         const { done, value } = await reader.read();
         if (done) {
-          buffer += decoder.decode();
-          if (buffer.trim()) inspectPayload(sseDataPayload(buffer));
+          inspector.finish();
           if (!terminalReported) reportTerminal("incomplete");
+          inspector.dispose();
           controller.close();
           return;
         }
-        inspectChunk(value);
+        inspector.feed(value);
         controller.enqueue(value);
       } catch (err) {
         if (!terminalReported) reportTerminal("incomplete");
+        inspector.dispose();
         try { controller.error(err); } catch { /* already torn down */ }
       }
     },
     cancel(reason) {
+      inspector.dispose();
       onCancel();
       reader.cancel(reason).catch(() => {});
     },
@@ -516,38 +504,23 @@ export function relaySseWithHeartbeat(
 ): ReadableStream<Uint8Array> | null {
   if (!body) return null;
   const reader = body.getReader();
-  const decoder = new TextDecoder();
   const heartbeat = new TextEncoder().encode(": opencodex keepalive\n\n");
   let timer: ReturnType<typeof setInterval> | undefined;
   let closed = false;
   let clientCancelled = false;
   let terminalReported = false;
-  let buffer = "";
 
   const reportTerminal = (status: ResponsesTerminalStatus) => {
     if (terminalReported || clientCancelled || closed) return;
     terminalReported = true;
     onTerminal?.(status);
   };
-
-  const inspectPayload = (payload: string | null) => {
-    if (!payload) return;
-    const status = terminalStatusFromSsePayload(payload);
-    if (status) reportTerminal(status);
-  };
-
-  const inspectChunk = (value: Uint8Array) => {
-    buffer += decoder.decode(value, { stream: true });
-    let next: { block: string; rest: string } | null;
-    while ((next = nextSseBlock(buffer))) {
-      buffer = next.rest;
-      inspectPayload(sseDataPayload(next.block));
-    }
-  };
+  const inspector = createSseInspector({ onTerminal: reportTerminal });
 
   const cleanup = () => {
     if (closed) return;
     closed = true;
+    inspector.dispose();
     if (timer) clearInterval(timer);
     timer = undefined;
     options?.onDone?.();
@@ -569,14 +542,13 @@ export function relaySseWithHeartbeat(
       try {
         const { done, value } = await reader.read();
         if (done) {
-          buffer += decoder.decode();
-          if (buffer.trim()) inspectPayload(sseDataPayload(buffer));
+          inspector.finish();
           if (!terminalReported && !clientCancelled) reportTerminal("incomplete");
           cleanup();
           controller.close();
           return;
         }
-        inspectChunk(value);
+        inspector.feed(value);
         controller.enqueue(value);
       } catch (err) {
         if (!clientCancelled) reportTerminal("incomplete");

structure/04_transports-and-sidecars.md

@@ -61,6 +61,11 @@ The two-shape contract is mirror-commented in `src/server/index.ts`; the real
 and the platform matrix lives in `tests/bun-stream-caps.test.ts`. Keep all three
 in lockstep with any passthrough-policy change.
 
+Translated response request-log tracking and the heartbeat relay also reuse
+`createSseInspector`. This keeps every client-facing SSE observation path on
+the same byte-bounded, discard-and-resynchronize frame policy and ensures the
+request-log, first-output, and terminal observers share one payload parse.
+
 ## Standalone Search and exact account selectors
 
 `POST /v1/alpha/search` retains the selected model in its request body. When that value is an

tests/sse-inspector-bounds.test.ts

@@ -4,7 +4,9 @@ import {
   getInspectionCounters,
   MAX_COMPLETED_OUTPUT_ITEMS,
   MAX_INSPECTION_SSE_FRAME_BYTES,
+  relaySseWithHeartbeat,
   resetInspectionCountersForTest,
+  trackSseForRequestLog,
 } from "../src/server/relay";
 import type { RequestLogContext } from "../src/server/request-log";
 
@@ -379,3 +381,66 @@ describe("createSseInspector parse-once", () => {
     }
   });
 });
+
+describe("client-facing SSE wrapper bounds", () => {
+  test("translated request-log tracking discards an oversized frame, resynchronizes, and preserves bytes", async () => {
+    const oversized = encoder.encode(`data: ${"x".repeat(MAX_INSPECTION_SSE_FRAME_BYTES)}x`);
+    const delimiter = encoder.encode("\n\n");
+    const terminal = frame(completedEvent("translated-after-cap"));
+    const chunks = [oversized, delimiter, terminal];
+    const terminals: string[] = [];
+    const tracked = trackSseForRequestLog(
+      streamFromChunks(chunks),
+      status => terminals.push(status),
+      () => {},
+      {} as RequestLogContext,
+      () => {},
+    );
+
+    expect(await readAllBytes(tracked)).toEqual(joinBytes(chunks));
+    expect(terminals).toEqual(["completed"]);
+    expect(getInspectionCounters().frameCapOverflows).toBe(1);
+    expect(getInspectionCounters().frameBufferHighWaterBytes)
+      .toBeLessThanOrEqual(MAX_INSPECTION_SSE_FRAME_BYTES);
+  });
+
+  test("translated request-log tracking parses a complete payload once for all observers", async () => {
+    const originalParse = JSON.parse;
+    let parses = 0;
+    JSON.parse = ((text: string) => {
+      parses += 1;
+      return originalParse(text);
+    }) as typeof JSON.parse;
+    try {
+      const tracked = trackSseForRequestLog(
+        streamFromChunks([frame(completedEvent("translated-parse-once"))]),
+        () => {},
+        () => {},
+        {} as RequestLogContext,
+        () => {},
+      );
+      await readAllBytes(tracked);
+      expect(parses).toBe(1);
+    } finally {
+      JSON.parse = originalParse;
+    }
+  });
+
+  test("heartbeat relay applies the same frame bound without changing upstream bytes", async () => {
+    const oversized = encoder.encode(`data: ${"x".repeat(MAX_INSPECTION_SSE_FRAME_BYTES)}x`);
+    const delimiter = encoder.encode("\r\n\r\n");
+    const terminal = frame(completedEvent("heartbeat-after-cap"));
+    const chunks = [oversized, delimiter, terminal];
+    const terminals: string[] = [];
+    const relayed = relaySseWithHeartbeat(
+      streamFromChunks(chunks),
+      new AbortController(),
+      60_000,
+      status => terminals.push(status),
+    )!;
+
+    expect(await readAllBytes(relayed)).toEqual(joinBytes(chunks));
+    expect(terminals).toEqual(["completed"]);
+    expect(getInspectionCounters().frameCapOverflows).toBe(1);
+  });
+});