[WRONG BRANCH] fix(responses): bound terminal SSE relay and restore Windows native passthrough

src/server/relay.ts

@@ -17,6 +17,7 @@ const nativePassthroughSseResponses = new WeakSet<Response>();
 const eagerRelaySseResponses = new WeakSet<Response>();
 
 export const MAX_INSPECTION_SSE_FRAME_BYTES = 4 * 1024 * 1024;
+export const MAX_CLIENT_SSE_FRAME_BYTES = 4 * 1024 * 1024;
 export const MAX_COMPLETED_OUTPUT_ITEMS = 256;
 export const MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES = 8 * 1024 * 1024;
 export const MAX_TAIL_ERROR_MESSAGE_CHARS = 512;
@@ -150,7 +151,14 @@ export function createSseTerminalOutputBoundary(): SseTerminalOutputBoundary {
     feed(chunk) {
       if (disposed || terminal) return new Uint8Array(0);
       buffer += decoder!.decode(chunk, { stream: true });
-      return process(false);
+      const output = process(false);
+      if (encoder.encode(buffer).byteLength > MAX_CLIENT_SSE_FRAME_BYTES) {
+        // Drop the oversized partial before the relay catch path calls finish();
+        // reflecting it into a synthetic failure would defeat the memory cap.
+        buffer = "";
+        throw new Error("upstream SSE frame exceeded the safe limit");
+      }
+      return output;
     },
     finish() {
       if (disposed || terminal) return new Uint8Array(0);

src/server/responses/core.ts

@@ -2221,7 +2221,12 @@ async function handleResponsesInner(
       const rewrittenBody = clientBlockRewrite !== undefined || payloadRewrites.length > 0
         ? relaySseWithBlockRewrite(nativeBody, clientBlockRewrite ?? payloadRewriteAsBlockRewrite(composeSsePayloadRewrites(...payloadRewrites)), translatorBudget)
         : nativeBody;
-      const clientBody = relaySseWithFailedTail(rewrittenBody, upstream, reason => clientGone.abort(reason));
+      // Known-bad Windows Bun runtimes must keep the no-rewrite client branch
+      // as a native tee relay. A JS pull wrapper reintroduces Bun#32111 when the
+      // client disconnects; fixed runtimes already took the eager path above.
+      const clientBody = process.platform === "win32" && !needsClientRewrite
+        ? nativeBody
+        : relaySseWithFailedTail(rewrittenBody, upstream, reason => clientGone.abort(reason));
       return markNativePassthroughSseResponse(new Response(clientBody, {
         status: upstreamResponse.status,
         headers,

tests/passthrough-abort.test.ts

@@ -62,6 +62,7 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => {
     expect(sseBranch).toContain("rewritePayload: composeSsePayloadRewrites(...payloadRewrites)");
     // Elsewhere the failed-tail relay converts mid-stream resets into a clean response.failed.
     expect(sseBranch).toContain("relaySseWithFailedTail(rewrittenBody, upstream");
+    expect(sseBranch).toContain('process.platform === "win32" && !needsClientRewrite');
     expect(sseBranch).toContain("new Response(clientBody");
     expect(sseBranch).toContain("markNativePassthroughSseResponse");
     // #314/phase 100 two-platform contract: the real core gate delegates to the

tests/relay-eager.test.ts

@@ -284,6 +284,22 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => {
     expect(budget.snapshot().currentBytes).toBe(0);
   });
 
+  test("bounds delimiter-free frames before they can bypass the client queue cap", async () => {
+    const up = controlledUpstream();
+    const upstream = new AbortController();
+    const { hooks } = makeHooks();
+    const relayed = relaySseEagerBounded(up.stream, upstream, hooks, { maxQueueBytes: 1 });
+    const reading = readAll(relayed);
+
+    up.push(new Uint8Array(4 * 1024 * 1024 + 1).fill(120));
+    up.close();
+
+    const text = await reading;
+    expect(text).not.toContain("x".repeat(64));
+    expect(text).toContain("upstream SSE frame exceeded the safe limit");
+    expect(upstream.signal.aborted).toBe(true);
+  });
+
   test("blocks without a data field pass through untouched before the terminal", async () => {
     const up = controlledUpstream();
     const { hooks } = makeHooks();

tests/sse-failed-tail.test.ts

@@ -1,7 +1,7 @@
 import { describe, expect, test } from "bun:test";
 import { relaySseWithFailedTail, relayWithAbort } from "../src/server";
 import { relaySseEagerBounded, type EagerRelayHooks } from "../src/server/relay-eager";
-import { MAX_TAIL_ERROR_MESSAGE_CHARS } from "../src/server/relay";
+import { MAX_CLIENT_SSE_FRAME_BYTES, MAX_TAIL_ERROR_MESSAGE_CHARS } from "../src/server/relay";
 import { TranslatorBudgetExceededError } from "../src/lib/translator-budget";
 
 const encoder = new TextEncoder();
@@ -140,6 +140,17 @@ describe("relaySseWithFailedTail", () => {
     expect(out.endsWith("data: [DONE]\n\n")).toBe(true);
   });
 
+  test("fails closed when an unterminated SSE frame exceeds the client buffer cap", async () => {
+    const upstream = new AbortController();
+    const chunk = "x".repeat(MAX_CLIENT_SSE_FRAME_BYTES / 2);
+    const out = await drain(relaySseWithFailedTail(sourceStream([chunk, chunk, "x"]), upstream));
+
+    expect(out).not.toContain(chunk);
+    expect(out).toContain("upstream SSE frame exceeded the safe limit");
+    expect(out.endsWith("data: [DONE]\n\n")).toBe(true);
+    expect(upstream.signal.aborted).toBe(true);
+  });
+
   test("translator overflow failed tail preserves translation_buffer_limit", async () => {
     const upstream = new AbortController();
     const error = new TranslatorBudgetExceededError("live_transient", 32 * 1024 * 1024);