fix(responses): close passthrough streams at terminal events

Author: L14nY1Wang

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
+ * reports this runtime/config decision; the final Responses dispatch may still
+ * require the eager relay to enforce a terminal boundary. 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

@@ -161,6 +161,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).
@@ -231,16 +232,28 @@ 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;
+        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 (hooks.sawTerminal()) {
+          // 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 (!new TextDecoder().decode(outbound).includes("data: [DONE]")) {
+            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();

src/server/relay.ts

@@ -110,16 +110,49 @@ export function relaySseWithFailedTail(
 ): ReadableStream<Uint8Array> {
   const reader = body.getReader();
   const encoder = new TextEncoder();
+  const terminalInspector = createSseInspector({
+    onTerminal: () => {},
+  });
+  let closed = false;
+  const closeAtTerminal = (controller: ReadableStreamDefaultController<Uint8Array>, value: Uint8Array): boolean => {
+    terminalInspector.feed(value);
+    if (!terminalInspector.reported()) return false;
+
+    // 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 the terminal-bearing chunk, add the conventional sentinel when
+    // it was not coalesced into that same chunk, then stop reading upstream.
+    controller.enqueue(value);
+    if (!new TextDecoder().decode(value).includes("data: [DONE]")) {
+      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(() => {});
+    terminalInspector.dispose();
+    return true;
+  };
   return new ReadableStream<Uint8Array>({
     async pull(controller) {
       try {
         const { done, value } = await reader.read();
         if (done) {
+          terminalInspector.finish();
+          terminalInspector.dispose();
           controller.close();
           return;
         }
+        if (closeAtTerminal(controller, value)) return;
         controller.enqueue(value);
       } catch (err) {
+        terminalInspector.dispose();
+        if (closed) return;
         const payload = buildFailedTailPayload(err);
         try {
           // Leading blank line terminates a partial SSE block so the failed frame parses cleanly.
@@ -130,6 +163,7 @@ export function relaySseWithFailedTail(
       }
     },
     cancel(reason) {
+      terminalInspector.dispose();
       if (onClientGone) onClientGone(reason);
       else upstream.abort(reason);
       reader.cancel(reason).catch(() => {});

src/server/responses/core.ts

@@ -1770,12 +1770,17 @@ async function handleResponsesInner(
       // (Bun#32111 JS-sink segfault — text frames pass, the terminal block is
       // lost). The eager single reader applies the same rewrites inline.
       const win32EagerRewrite = isWin32EagerRewrite(process.platform, needsClientRewrite);
+      // A native tee branch cannot be closed at a Responses protocol terminal
+      // while its upstream HTTP connection remains open. Windows therefore
+      // needs the existing single-reader relay even without payload rewrites;
+      // otherwise Codex keeps thinking after response.completed.
+      const win32TerminalRelay = process.platform === "win32";
       const eagerPath = selectEagerPath(
         process.platform,
         needsClientRewrite,
         config.streamMode ?? "auto",
       );
-      if (eagerPath?.useEagerRelay || win32EagerRewrite) {
+      if (eagerPath?.useEagerRelay || win32EagerRewrite || win32TerminalRelay) {
         const turnAc = new AbortController();
         linkAbortSignal(upstream, turnAc.signal);
         registerTurn(turnAc, options.turnAdmissionLease);
@@ -1811,7 +1816,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) }
             : {}),
@@ -1829,9 +1836,9 @@ 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).
+        // Windows always reaches this relay so response.completed can close a
+        // keep-alive upstream. 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, {
@@ -1898,15 +1905,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,20 @@ 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
+- **Terminal-aware eager bounded relay** (`src/server/relay-eager.ts`). Windows
+  always uses this single-reader shape so `response.completed` closes the client
+  stream even when an upstream keeps HTTP/SSE alive. Darwin uses it for
   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
+  selected by `selectEagerPath` in `src/lib/bun-stream-caps.ts`. Windows ignores
+  the no-rewrite gate at the final server dispatch because protocol termination
+  cannot be implemented by the native tee branch; rewrite traffic continues to
+  apply transforms inline. Darwin 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 must use the terminal-aware single-reader relay even without a
+    // payload rewrite, so response.completed can close a keep-alive upstream.
     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('const win32TerminalRelay = process.platform === "win32"');
+    expect(sseBranch).toContain("eagerPath?.useEagerRelay || win32EagerRewrite || 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).

tests/relay-eager.test.ts

@@ -131,8 +131,10 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => {
     expect(text).toContain("RESTORED");
     expect(text).not.toContain("image_gen__gen");
     expect(text).toContain("response.completed");
-    // A partial trailing block reaches the client verbatim at EOF.
-    expect(text).toContain("trailing-partial");
+    // The protocol terminal ends the client stream; bytes produced after it
+    // belong to the gateway's retained connection and must not hold Codex open.
+    expect(text).not.toContain("trailing-partial");
+    expect(text.endsWith("data: [DONE]\n\n")).toBe(true);
   });
 
   test("identity rewrite preserves framing byte-for-byte", async () => {
@@ -153,7 +155,9 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => {
     up.close();
 
     const text = await reading;
-    expect(text).toBe(new TextDecoder().decode(joinBytes([first, enc.encode(second)])));
+    expect(text).toBe(
+      new TextDecoder().decode(joinBytes([first, enc.encode(second)])) + "data: [DONE]\n\n",
+    );
     // The rewrite actually ran — this is what makes the test red pre-fix.
     expect(rewriteCalls).toBeGreaterThan(0);
   });
@@ -209,7 +213,7 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => {
     expect(budget.snapshot().currentBytes).toBe(0);
   });
 
-  test("blocks without a data field pass through untouched", async () => {
+  test("blocks without a data field pass through untouched before the terminal", async () => {
     const up = controlledUpstream();
     const { hooks } = makeHooks();
     let rewriteCalls = 0;
@@ -220,8 +224,8 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => {
     const relayed = relaySseEagerBounded(up.stream, new AbortController(), hooks);
     const reading = readAll(relayed);
 
-    up.push(enc.encode(`event: response.completed\ndata: ${COMPLETED}\n\n`));
     up.push(enc.encode(`: keepalive comment\n\n`));
+    up.push(enc.encode(`event: response.completed\ndata: ${COMPLETED}\n\n`));
     up.close();
 
     const text = await reading;
@@ -317,7 +321,7 @@ describe("relaySseEagerBounded — side-effect parity", () => {
 
     const clientBytes = await readAllBytes(relayed);
     await settle();
-    expect(clientBytes).toEqual(joinBytes(frames));
+    expect(clientBytes).toEqual(joinBytes([...frames, enc.encode("data: [DONE]\n\n")]));
     const wireText = new TextDecoder().decode(clientBytes);
     expect(wireText).not.toContain('"output":');
     expect(rec.completed).toHaveLength(1);

tests/sse-failed-tail.test.ts

@@ -59,6 +59,35 @@ describe("relaySseWithFailedTail", () => {
     expect(upstream.signal.aborted).toBe(false);
   });
 
+  test("closes at response.completed when the upstream keeps its SSE connection open", async () => {
+    const upstream = new AbortController();
+    let sourceCancelled = false;
+    let sentTerminal = false;
+    const src = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        if (!sentTerminal) {
+          sentTerminal = true;
+          controller.enqueue(encoder.encode(
+            'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed"}}\n\n',
+          ));
+        }
+        // Deliberately never close: several Responses-compatible gateways keep
+        // this connection alive after the protocol terminal event.
+      },
+      cancel() { sourceCancelled = true; },
+    });
+
+    const out = await Promise.race([
+      drain(relaySseWithFailedTail(src, upstream)),
+      new Promise<never>((_, reject) => setTimeout(() => reject(new Error("relay did not close at terminal")), 200)),
+    ]);
+
+    expect(out).toContain("response.completed");
+    expect(out.endsWith("data: [DONE]\n\n")).toBe(true);
+    expect(sourceCancelled).toBe(true);
+    expect(upstream.signal.aborted).toBe(false);
+  });
+
   test("mid-stream error keeps prior bytes and appends a clean failed terminal", async () => {
     const upstream = new AbortController();
     const src = sourceStream(['data: {"type":"response.output_text.delta","delta":"hel', ""], { failAfter: true });