fix(responses): close passthrough streams at terminal events

src/lib/bun-stream-caps.ts

@@ -6,7 +6,8 @@
  * PR #32120, merged 2026-06-21). No RELEASED Bun version is proven to carry
  * that fix yet, so `MIN_FIXED_BUN_VERSION` is null: every runtime is
  * "known-bad" until a bundle-bump commit sets it. Windows no-rewrite traffic
- * follows this runtime/config decision. Darwin no-rewrite traffic stays on tee
+ * follows this runtime/config decision, preserving the explicit legacy-tee
+ * safety pin. Darwin no-rewrite traffic stays on tee
  * for `auto` regardless of runtime capability and reaches eager relay only via
  * explicit `streamMode: "eager-relay"` opt-in (see
  * devlog/_plan/260731_macos_rss_retention/100_darwin_eager_optin.md).

src/server/index.ts

@@ -250,18 +250,15 @@ function attachLiveSidebandUpstream(ws: ServerWebSocket<WsData>): void {
 // if (isEventStream && upstreamResponse.body) {
 // const repairConfig = route.provider.responsesItemIdRepair;
 // const needsClientRewrite = imageGenCallAliases.size > 0
-// #314 gated shape: win32 no-rewrite traffic follows runtime/config policy; darwin no-rewrite
-// traffic requires explicit config-eager opt-in (`auto` always stays tee on darwin). Default OFF
-// on the bundled known-bad runtime; policy lives in 260731_macos_rss_retention phase 100.
+// #314 gated shape: win32 always uses the terminal-aware eager relay so a keep-alive
+// upstream cannot hold Codex open after response.completed; darwin no-rewrite traffic
+// requires explicit config-eager opt-in (`auto` always stays tee on darwin).
 // selectEagerPath(process.platform, needsClientRewrite, config.streamMode ?? "auto")
 // relaySseEagerBounded(upstreamResponse.body, turnAc,
 // new Response(eagerBody,
 // Default shape (tee + background inspection):
 // upstreamResponse.body.tee()
 // const repairedBody = hasResponsesItemIdRepair(repairConfig)
-// process.platform === "win32"
-// && !needsClientRewrite
-// ? nativeBody
 // relaySseWithFailedTail(repairedBody, upstream)
 // new Response(clientBody
 // markNativePassthroughSseResponse

src/server/relay-eager.ts

@@ -24,7 +24,7 @@
  * up to the drain window.
  */
 
-import { buildFailedTailPayload } from "./relay";
+import { buildFailedTailPayload, createSseTerminalOutputBoundary } from "./relay";
 import {
   nextSseBlock,
   replaceSseDataPayload,
@@ -92,6 +92,7 @@ export function relaySseEagerBounded(
   const now = opts?.now ?? Date.now;
 
   const reader = body.getReader();
+  const terminalBoundary = createSseTerminalOutputBoundary();
   const rewrite = hooks.rewritePayload;
   const rewriteDecoder = rewrite ? new TextDecoder() : null;
   const rewriteEncoder = rewrite ? new TextEncoder() : null;
@@ -161,6 +162,7 @@ export function relaySseEagerBounded(
   let queuedBytes = 0;
   let cancelled = false;
   let done = false;
+  const terminalSentinel = new TextEncoder().encode("data: [DONE]\n\n");
   // Pause gate: resolved by client pull, client cancel, or upstream abort so a
   // paused producer ALWAYS resumes (audit blocker 2 — no deadlock; onDone and
   // turn unregistration stay reachable, drainAndShutdown never hangs).
@@ -216,12 +218,17 @@ export function relaySseEagerBounded(
         if (upstream.signal.aborted) break;
         if (upstreamDone) {
           hooks.finishInspection();
+          const boundedTail = terminalBoundary.finish();
           if (rewrite) {
-            const tail = flushRewriteTail();
+            const rewritten = rewriteOutbound(boundedTail);
+            const tail = joinUint8Arrays(rewritten, flushRewriteTail());
             if (tail.byteLength > 0 && !cancelled) {
               queuedBytes += tail.byteLength;
               try { controllerRef?.enqueue(tail); } catch { /* client already gone */ }
             }
+          } else if (boundedTail.byteLength > 0 && !cancelled) {
+            queuedBytes += boundedTail.byteLength;
+            try { controllerRef?.enqueue(boundedTail); } catch { /* client already gone */ }
           }
           if (!hooks.sawTerminal() && !cancelled && !upstream.signal.aborted) {
             syntheticKind = "incomplete";
@@ -237,17 +244,30 @@ export function relaySseEagerBounded(
           }
           continue;
         }
-        const outbound = rewrite ? rewriteOutbound(value) : value;
-        if (outbound.byteLength === 0) continue;
-        queuedBytes += outbound.byteLength;
-        try {
-          controllerRef?.enqueue(outbound);
-        } catch {
-          // Controller already torn down (client went away without cancel()).
-          cancelled = true;
-          drainDeadline = now() + drainMs;
-          armDrainTimer();
-          continue;
+        const terminalBounded = terminalBoundary.feed(value);
+        const outbound = rewrite ? rewriteOutbound(terminalBounded) : terminalBounded;
+        if (outbound.byteLength > 0) {
+          queuedBytes += outbound.byteLength;
+          try {
+            controllerRef?.enqueue(outbound);
+          } catch {
+            // Controller already torn down (client went away without cancel()).
+            cancelled = true;
+            drainDeadline = now() + drainMs;
+            armDrainTimer();
+            continue;
+          }
+        }
+        if (terminalBoundary.terminalSeen()) {
+          // The Responses terminal event ends the turn even when a compatible
+          // gateway keeps its HTTP connection alive. Add the conventional
+          // sentinel and stop the single-reader relay at that protocol boundary.
+          if (!terminalBoundary.doneSeen()) {
+            queuedBytes += terminalSentinel.byteLength;
+            try { controllerRef?.enqueue(terminalSentinel); } catch { /* client already gone */ }
+          }
+          reader.cancel("Responses terminal event received").catch(() => {});
+          break;
         }
         while (queuedBytes > maxQueueBytes && !cancelled && !upstream.signal.aborted) {
           await paused();
@@ -279,6 +299,7 @@ export function relaySseEagerBounded(
         try { rewriteBudget.releaseRetained(frameBufferBytes, { kind: "live_transient" }); } catch { /* teardown must not throw */ }
         frameBufferBytes = 0;
       }
+      terminalBoundary.dispose();
       if (syntheticKind) hooks.onSynthetic(syntheticKind);
       if (cancelled && !hooks.sawTerminal()) {
         hooks.onClientCancel();
@@ -318,3 +339,12 @@ export function relaySseEagerBounded(
     },
   });
 }
+
+function joinUint8Arrays(first: Uint8Array, second: Uint8Array): Uint8Array {
+  if (first.byteLength === 0) return second;
+  if (second.byteLength === 0) return first;
+  const joined = new Uint8Array(first.byteLength + second.byteLength);
+  joined.set(first);
+  joined.set(second, first.byteLength);
+  return joined;
+}

src/server/relay.ts

@@ -95,6 +95,79 @@ export function buildFailedTailPayload(err: unknown): string {
   });
 }
 
+export type SseTerminalOutputBoundary = {
+  feed(chunk: Uint8Array): Uint8Array;
+  finish(): Uint8Array;
+  terminalSeen(): boolean;
+  doneSeen(): boolean;
+  dispose(): void;
+};
+
+/**
+ * Frame-aware client output boundary shared by both native Responses relays.
+ * It buffers only the current incomplete SSE block, forwards complete blocks
+ * through the first Responses terminal, and drops every later block/byte.
+ */
+export function createSseTerminalOutputBoundary(): SseTerminalOutputBoundary {
+  let decoder: TextDecoder | null = new TextDecoder();
+  const encoder = new TextEncoder();
+  let buffer = "";
+  let terminal = false;
+  let done = false;
+  let disposed = false;
+
+  const process = (flush: boolean): Uint8Array => {
+    if (disposed || terminal) return new Uint8Array(0);
+    let output = "";
+    let responsesTerminal = false;
+    for (;;) {
+      const next = nextSseBlock(buffer);
+      if (!next) break;
+      buffer = next.rest;
+      const payload = sseDataPayload(next.block);
+      if (!responsesTerminal) output += next.block + next.delimiter;
+      if (payload === "[DONE]") {
+        done = true;
+        if (responsesTerminal) output += next.block + next.delimiter;
+        continue;
+      }
+      if (!responsesTerminal && payload && terminalStatusFromSsePayload(payload)) {
+        responsesTerminal = true;
+      }
+    }
+    if (responsesTerminal) {
+      terminal = true;
+      buffer = "";
+    }
+    if (flush && !terminal && buffer.length > 0) {
+      output += buffer;
+      buffer = "";
+    }
+    return encoder.encode(output);
+  };
+
+  return {
+    feed(chunk) {
+      if (disposed || terminal) return new Uint8Array(0);
+      buffer += decoder!.decode(chunk, { stream: true });
+      return process(false);
+    },
+    finish() {
+      if (disposed || terminal) return new Uint8Array(0);
+      buffer += decoder!.decode();
+      return process(true);
+    },
+    terminalSeen: () => terminal,
+    doneSeen: () => done,
+    dispose() {
+      if (disposed) return;
+      disposed = true;
+      decoder = null;
+      buffer = "";
+    },
+  };
+}
+
 /**
  * Relay a passthrough SSE body like relayWithAbort, but convert a MID-STREAM failure (upstream
  * reset after headers) into a clean terminal: any partial block is closed off, then a synthetic
@@ -110,18 +183,57 @@ export function relaySseWithFailedTail(
 ): ReadableStream<Uint8Array> {
   const reader = body.getReader();
   const encoder = new TextEncoder();
+  const terminalBoundary = createSseTerminalOutputBoundary();
+  let closed = false;
+  const relayChunk = (
+    controller: ReadableStreamDefaultController<Uint8Array>,
+    value: Uint8Array,
+  ): "terminal" | "output" | "buffered" => {
+    const outbound = terminalBoundary.feed(value);
+    if (outbound.byteLength > 0) controller.enqueue(outbound);
+    if (!terminalBoundary.terminalSeen()) return outbound.byteLength > 0 ? "output" : "buffered";
+
+    // A Responses terminal frame is the protocol boundary. Some compatible
+    // gateways leave the HTTP connection open after response.completed, which
+    // otherwise leaves Codex waiting forever even though the model turn is done.
+    // Preserve through the terminal block only, add the conventional sentinel
+    // when there was no real [DONE] data event, then stop reading upstream.
+    if (!terminalBoundary.doneSeen()) {
+      controller.enqueue(encoder.encode("data: [DONE]\n\n"));
+    }
+    closed = true;
+    controller.close();
+    const reason = "Responses terminal event received";
+    // Notify the tee inspection branch as well. It has already received the
+    // same terminal-bearing upstream chunk, so its bounded drain records the
+    // real terminal and then releases the turn/upstream keep-alive connection.
+    onClientGone?.(reason);
+    reader.cancel(reason).catch(() => {});
+    terminalBoundary.dispose();
+    return "terminal";
+  };
   return new ReadableStream<Uint8Array>({
     async pull(controller) {
       try {
-        const { done, value } = await reader.read();
-        if (done) {
-          controller.close();
-          return;
+        for (;;) {
+          const { done, value } = await reader.read();
+          if (done) {
+            const tail = terminalBoundary.finish();
+            if (tail.byteLength > 0) controller.enqueue(tail);
+            terminalBoundary.dispose();
+            controller.close();
+            return;
+          }
+          const result = relayChunk(controller, value);
+          if (result !== "buffered") return;
         }
-        controller.enqueue(value);
       } catch (err) {
+        const partial = terminalBoundary.finish();
+        terminalBoundary.dispose();
+        if (closed) return;
         const payload = buildFailedTailPayload(err);
         try {
+          if (partial.byteLength > 0) controller.enqueue(partial);
           // Leading blank line terminates a partial SSE block so the failed frame parses cleanly.
           controller.enqueue(encoder.encode(`\n\nevent: response.failed\ndata: ${payload}\n\ndata: [DONE]\n\n`));
           controller.close();
@@ -130,18 +242,20 @@ export function relaySseWithFailedTail(
       }
     },
     cancel(reason) {
+      terminalBoundary.dispose();
       if (onClientGone) onClientGone(reason);
       else upstream.abort(reason);
       reader.cancel(reason).catch(() => {});
     },
   });
 }
 
-export function nextSseBlock(buffer: string): { block: string; rest: string } | null {
+export function nextSseBlock(buffer: string): { block: string; delimiter: string; rest: string } | null {
   const match = buffer.match(/\r?\n\r?\n/);
   if (!match || match.index === undefined) return null;
   return {
     block: buffer.slice(0, match.index),
+    delimiter: match[0],
     rest: buffer.slice(match.index + match[0].length),
   };
 }

src/server/responses/core.ts

@@ -1870,7 +1870,9 @@ async function handleResponsesInner(
           inspectChunk: chunk => inspector.feed(chunk),
           finishInspection: () => inspector.finish(),
           disposeInspection: () => inspector.dispose(),
-          sawTerminal: () => inspector.reported(),
+          // Stream lifetime follows the protocol terminal even when this request
+          // has no outcome callback configured (reported() would stay false).
+          sawTerminal: () => inspector.terminalSeen(),
           ...(win32EagerRewrite
             ? { rewritePayload: composeSsePayloadRewrites(...payloadRewrites) }
             : {}),
@@ -1888,9 +1890,10 @@ async function handleResponsesInner(
           onClientCancel: () => options.onNativePassthroughCancel?.(),
           onDone: () => unregisterTurn(turnAc),
         }, win32EagerRewrite ? { rewriteBudget: translatorBudget } : undefined);
-        // selectEagerPath admits only no-rewrite traffic on both eligible platforms;
-        // win32 rewrite traffic reaches this relay too, but with the payload rewrite
-        // applied inline — never via an image/item-id JS pull wrapper (#32111, #864).
+        // When selected, this relay closes response.completed even if upstream
+        // keeps the connection alive. Windows rewrite traffic applies its
+        // payload transform inline — never via the Bun#32111-unsafe
+        // tee()+JS-pull chain (#864).
         if (!headers.has("content-type")) headers.set("content-type", "text/event-stream");
         return markEagerRelaySseResponse(
           markNativePassthroughSseResponse(new Response(eagerBody, {
@@ -1957,15 +1960,13 @@ async function handleResponsesInner(
         );
       }
       if (!headers.has("content-type")) headers.set("content-type", "text/event-stream");
-      // win32 must keep the pure native relay (Bun#32111 JS-sink segfault); elsewhere a JS pull
-      // relay is established practice (relayWithAbort, relaySseWithHeartbeat) and lets a
-      // mid-stream reset end with a clean response.failed terminal instead of a raw socket error.
+      // Windows was handled by the eager terminal-aware branch above. Remaining
+      // tee traffic can use the JS relay to close on a protocol terminal and to
+      // convert a mid-stream reset into a clean response.failed event.
       const rewrittenBody = payloadRewrites.length > 0
         ? relaySseWithPayloadRewrite(nativeBody, composeSsePayloadRewrites(...payloadRewrites), translatorBudget)
         : nativeBody;
-      const clientBody = process.platform === "win32" && !needsClientRewrite
-        ? nativeBody
-        : relaySseWithFailedTail(rewrittenBody, upstream, reason => clientGone.abort(reason));
+      const clientBody = relaySseWithFailedTail(rewrittenBody, upstream, reason => clientGone.abort(reason));
       return markNativePassthroughSseResponse(new Response(clientBody, {
         status: upstreamResponse.status,
         headers,

structure/04_transports-and-sidecars.md

@@ -38,19 +38,19 @@ to GUI static serving.
 Native passthrough SSE has TWO shapes, selected per request in
 `src/server/responses/core.ts`:
 
-- **Default: tee + background inspection.** `upstreamResponse.body.tee()` sends
-  branch[0] to the client (pure native relay on win32 without any client-facing
-  rewrite — the Bun#32111 crash workaround; a JS relay elsewhere) while branch[1] is
+- **Default outside Windows: tee + background inspection.** `upstreamResponse.body.tee()` sends
+  branch[0] through a terminal-aware client relay while branch[1] is
   drained eagerly by `consumeForInspection`/`consumeForResponseLogMetadata`
   for terminal-outcome recording, quota, the passthrough continuation cache,
   and request logs. This remains the default shape on bundled Bun 1.3.14.
-- **Gated: eager bounded relay** (`src/server/relay-eager.ts`). win32 and darwin
-  no-client-rewrite traffic only (neither image-gen aliases nor item-id repair),
-  selected by `selectEagerPath` in `src/lib/bun-stream-caps.ts`. Windows `auto`
-  becomes eager only on runtimes proven to carry the Bun#32111 fix
-  (`MIN_FIXED_BUN_VERSION`, null until a bundle bump), while explicit
-  `streamMode: "eager-relay"` opts in today. Darwin is explicit-only: `auto`
-  stays tee even after a future threshold bump. One eager reader + byte-bounded
+- **Terminal-aware eager bounded relay** (`src/server/relay-eager.ts`). Windows
+  uses this single-reader shape for rewrite traffic and for no-rewrite traffic
+  selected by `selectEagerPath` in `src/lib/bun-stream-caps.ts`; the latter keeps
+  `legacy-tee` and known-bad-runtime `auto` on tee as documented. When selected,
+  `response.completed` closes the client stream even if upstream keeps HTTP/SSE
+  alive. Darwin uses it for no-client-rewrite traffic only (neither image-gen
+  aliases nor item-id repair) and is explicit-only: `auto` stays tee even after
+  a future threshold bump. One eager reader + byte-bounded
   client queue + post-cancel bounded discard-drain replaces the tee and goes
   directly to the response without a JS rewrite wrapper, preserving the full
   inspection side-effect set (shared `createSseInspector` factory in `relay.ts`)

tests/passthrough-abort.test.ts

@@ -47,15 +47,14 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => {
     );
 
     expect(sseBranch).toContain("upstreamResponse.body.tee()");
-    // win32 must receive the tee'd body untouched when no client rewrite is required — no JS pull
-    // wrapper on the default path (Bun#32111 segfault).
+    // Windows no-rewrite traffic must honor the stream-mode/runtime gate so
+    // legacy-tee remains a safety escape hatch for Bun#32111.
     expect(sseBranch).toContain("const repairConfig = route.provider.responsesItemIdRepair;");
     expect(sseBranch).toContain("const needsClientRewrite = imageGenCallAliases.size > 0");
     expect(sseBranch).toContain("new Response(eagerBody");
     expect(sseBranch).toContain("const rewrittenBody = payloadRewrites.length > 0");
-    expect(sseBranch).toContain('process.platform === "win32"');
-    expect(sseBranch).toContain("&& !needsClientRewrite");
-    expect(sseBranch).toContain("? nativeBody");
+    expect(sseBranch).toContain("eagerPath?.useEagerRelay || win32EagerRewrite");
+    expect(sseBranch).not.toContain("win32TerminalRelay");
     // #864: win32 traffic that DOES need a client rewrite takes the eager single
     // reader with the payload rewrite applied inline — never the tee()+JS-pull
     // chain that loses the terminal block on Windows (Bun#32111).