From 9e05997a2fc001a4565f8e6973efbe8a02c6afb7 Mon Sep 17 00:00:00 2001
From: Yum-wu <1172989563@qq.com>
Date: Fri, 4 Sep 2026 02:01:33 +0800
Subject: [PATCH 1/3] feat(retry): refetch once on zero-output mid-stream
 socket reset

Pre-stream retry wrappers (fetchWithResetRetry / fetchWithTransientRetry)
only cover fetch() rejecting before response headers. A mid-stream socket
reset (Cloudflare closing idle keep-alive connections while Bun's pool
reuses the half-closed socket) surfaces as a ReadableStream read()
rejection and kills the turn with response.failed/upstream_reset even
when zero bytes were relayed to the client.

Add wrapWithZeroOutputRefetch + refetchOnZeroOutputReset to
src/lib/upstream-retry.ts: a body wrapper that, on a reset-shaped read()
error before the first byte is consumed, transparently refetches ONCE on
a fresh connection (connection-reset recovery init: Connection: close +
keepalive: false). Partial-output failures, clean EOF, non-reset errors,
and failed/empty refetches keep the existing fail-closed tail.

Wire the wrapper into both streaming lanes before any tee/eager/parser
branch, so inspection and client relays need no changes:
- src/server/responses/core.ts: wrap the raw passthrough body before the
  terminal-repair layer (the refetched body then runs through the same
  repair pipeline).
- src/server/chat-native.ts: wrap the native chat SSE body with an inline
  refetch thunk that uses the finalized active request/provider.

Add focused regression tests (12 cases) covering the refetch gate,
zero-output swap, partial-output fail-closed, single-retry ceiling, and
cancel forwarding.
---
 src/lib/upstream-retry.ts                    | 121 ++++++++++
 src/server/chat-native.ts                    |  40 +++-
 src/server/responses/fetch-helpers.ts        |   4 +-
 src/server/responses/passthrough-delivery.ts |  19 +-
 src/server/responses/passthrough-dispatch.ts |  33 +++
 tests/upstream-retry-zero-output.test.ts     | 222 +++++++++++++++++++
 6 files changed, 426 insertions(+), 13 deletions(-)
 create mode 100644 tests/upstream-retry-zero-output.test.ts

diff --git a/src/lib/upstream-retry.ts b/src/lib/upstream-retry.ts
index 3a4fb619a3..55a3058406 100644
--- a/src/lib/upstream-retry.ts
+++ b/src/lib/upstream-retry.ts
@@ -594,3 +594,124 @@ export async function fetchWithTransientRetry(
     opts.onSendsConsumed?.(sent);
   }
 }
+
+/**
+ * Refetch an upstream request ONCE after a mid-stream socket reset that
+ * happened before the caller consumed any response bytes.
+ *
+ * `fetchWithResetRetry` / `fetchWithTransientRetry` only cover pre-stream
+ * failures — `fetch()` rejecting before response headers. Once headers arrive
+ * and the caller starts reading the SSE body, a mid-stream reset (Cloudflare
+ * closing an idle keep-alive connection while Bun's pool reuses the half-closed
+ * socket) surfaces as a ReadableStream read() rejection, outside every
+ * pre-stream retry wrapper. The turn then dies with a terminal
+ * `response.failed / upstream_reset` even though nothing was relayed to the
+ * client.
+ *
+ * This helper closes that gap for the one case where a replay is provably safe:
+ * zero bytes consumed and no protocol terminal seen. `doFetch` must be
+ * replay-safe (string body, same contract as {@link ReplayableFetch}); the
+ * replacement send goes out with the connection-reset recovery init
+ * (`Connection: close` + `keepalive: false`) so the fresh connection never
+ * reuses the pooled half-closed socket. Exactly one replacement send, no
+ * backoff — the pre-stream layers already spent their retry budget reaching
+ * the first headers.
+ *
+ * Returns null (and the caller keeps its existing fail-closed tail) when the
+ * error is not a reset shape, the caller signal is aborted, the refetch itself
+ * throws, or the replacement has no body. Callers must not retry the returned
+ * response's body.
+ */
+export async function refetchOnZeroOutputReset(
+  doFetch: ReplayableFetch,
+  err: unknown,
+  opts: ResetRetryOptions = {},
+): Promise<Response | null> {
+  if (!isConnectionResetError(err)) return null;
+  if (opts.abortSignal?.aborted) return null;
+  let replacement: Response;
+  try {
+    replacement = await doFetch("connection-reset");
+  } catch (retryErr) {
+    console.warn(
+      `[upstream-retry] zero-output reset${opts.label ? ` (${opts.label})` : ""} — refetch failed: ${
+        retryErr instanceof Error ? retryErr.message : String(retryErr)
+      }`,
+    );
+    return null;
+  }
+  if (!replacement.body) {
+    console.warn(
+      `[upstream-retry] zero-output reset${opts.label ? ` (${opts.label})` : ""} — refetch returned no body, keeping original failure`,
+    );
+    try {
+      replacement.arrayBuffer().catch(() => {});
+    } catch { /* body already unusable; original failure stands */ }
+    return null;
+  }
+  console.warn(
+    `[upstream-retry] zero-output reset${opts.label ? ` (${opts.label})` : ""} — refetched on a fresh connection`,
+  );
+  return replacement;
+}
+
+/**
+ * Wrap an upstream SSE body so a mid-stream socket reset before the first byte
+ * is consumed transparently swaps in ONE refetched body (see
+ * {@link refetchOnZeroOutputReset}). Everything downstream — tee inspection
+ * branches, eager or pull relays, SSE parsers — reads the wrapped stream and
+ * never observes the first upstream send dying, so no relay needs changes.
+ *
+ * The gate is deliberately narrow: only a read() rejection matching
+ * {@link isConnectionResetError}, with zero bytes read so far, a live caller
+ * signal, and a single swap per wrapped stream. Partial-output failures, clean
+ * EOF, non-reset errors, and a failed or empty refetch all propagate the
+ * ORIGINAL error untouched, preserving every existing fail-closed tail
+ * (replaying after emitted tool calls would duplicate side effects).
+ */
+export function wrapWithZeroOutputRefetch(
+  body: ReadableStream<Uint8Array>,
+  doFetch: ReplayableFetch,
+  opts: ResetRetryOptions = {},
+): ReadableStream<Uint8Array> {
+  let reader = body.getReader();
+  let bytesRead = 0;
+  let retried = false;
+  return new ReadableStream<Uint8Array>({
+    async pull(controller) {
+      for (;;) {
+        try {
+          const { done, value } = await reader.read();
+          if (done) {
+            controller.close();
+            return;
+          }
+          bytesRead += value.byteLength;
+          controller.enqueue(value);
+          return;
+        } catch (err) {
+          if (!retried && bytesRead === 0 && !opts.abortSignal?.aborted) {
+            retried = true;
+            const replacement = await refetchOnZeroOutputReset(doFetch, err, opts);
+            if (replacement?.body) {
+              try {
+                reader.cancel().catch(() => {});
+              } catch { /* broken reader; the refetch won */ }
+              reader = replacement.body.getReader();
+              continue;
+            }
+          }
+          try {
+            controller.error(err);
+          } catch { /* already torn down */ }
+          return;
+        }
+      }
+    },
+    cancel(reason) {
+      try {
+        reader.cancel(reason).catch(() => {});
+      } catch { /* already torn down */ }
+    },
+  });
+}
diff --git a/src/server/chat-native.ts b/src/server/chat-native.ts
index 791c687bd0..d5521d4677 100644
--- a/src/server/chat-native.ts
+++ b/src/server/chat-native.ts
@@ -25,8 +25,10 @@ import {
   applyUpstreamRecoveryInit,
   fetchWithResetRetry,
   fetchWithTransientRetry,
+  isNonReplayableResponse,
   prepareSameTarget429Wait,
   type UpstreamSendRecovery,
+  wrapWithZeroOutputRefetch,
 } from "../lib/upstream-retry";
 import {
   isTranslatorBudgetExceededError,
@@ -301,7 +303,12 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio
     : Number.POSITIVE_INFINITY;
   const transientSendAvailable = (): boolean => remainingTransientSends() > 0;
 
-  const send = async (request: AdapterRequest, recovery?: "rate-limit-429" | "key-429"): Promise<Response> => {
+  const send = async (
+    request: AdapterRequest,
+    recovery?: "rate-limit-429" | "key-429" | "connection-reset",
+    singleSend = false,
+    sendSignal: AbortSignal = upstream.signal,
+  ): Promise<Response> => {
     try {
       // #2643: opted-in key-auth openai-chat providers retry pre-stream transient statuses on
       // the native chat lane too; everyone else keeps reset-only semantics.
@@ -312,21 +319,24 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio
       const fetchWithPolicy = requestTransientPolicy ? fetchWithTransientRetry : fetchWithResetRetry;
       return await fetchWithPolicy(
         (transportRecovery?: UpstreamSendRecovery) => {
+          const wireRecovery = transportRecovery ?? (recovery === "connection-reset" ? recovery : undefined);
           return fetchWithHeaderTimeout(
             request.url,
             applyUpstreamRecoveryInit({
               method: request.method,
               headers: request.headers,
               body: request.body,
-            }, transportRecovery),
-            upstream.signal,
+            }, wireRecovery),
+            sendSignal,
             connectMs,
             requestedStream,
             providerFetch(activeProvider, undefined, {
+              httpOnly: singleSend,
               providerName: route.providerName,
               modelId: route.modelId,
               dispatchOverride: async (_input, init, execute) => {
                 if (!providerApiKeySelectionIsCurrent(config, route.providerName, activeProvider)) {
+                  if (singleSend) throw new Error("Provider key selection changed before native Chat stream recovery");
                   const current = resolveCurrentProviderApiKeyTransport(config, route.providerName, activeProvider);
                   if (!current || !isNativeChatRouteEligible({ ...route, provider: current }, options.chatBody, config)) {
                     throw new Error("Provider key selection is no longer available for native Chat");
@@ -349,7 +359,7 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio
                 noteProviderAttemptSend(logCtx, route.providerName, activeProvider, logCtx.usageLogInputTokens, transportRecovery ?? recovery);
                 const dispatched = await ((activeProvider as OcxProviderTransport).fetch ?? execute)(request.url, applyUpstreamRecoveryInit({
                   ...init, method: request.method, headers, body: request.body,
-                }, transportRecovery));
+                }, wireRecovery));
                 if (!dispatched.ok) await recordKeyAttemptFailure(logCtx, dispatched, init.signal ?? upstream.signal);
                 return dispatched;
               },
@@ -357,14 +367,14 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio
           );
         },
         {
-          abortSignal: upstream.signal,
+          abortSignal: sendSignal,
           label: safeHostLabel(request.url),
           ...(requestTransientPolicy
             ? {
-              attempts: remaining,
+              attempts: singleSend ? Math.min(1, remaining) : remaining,
               onSendsConsumed: (sends: number) => { transientSendsUsed += Math.max(0, sends); },
             }
-            : {}),
+            : singleSend ? { attempts: 1 } : {}),
         },
       );
     } finally {
@@ -423,9 +433,9 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio
     }
     return fail(502, error instanceof Error ? error.message : String(error), "server_error");
   }
-  releaseRetainedRequest();
 
   if (!response.ok) {
+    releaseRetainedRequest();
     let bodyText = "";
     try {
       const body = await readBoundedResponseBody(response, {
@@ -506,7 +516,18 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio
   if (contentType.includes("text/event-stream") && response.body) {
     if (requestedStream) transferTurnToStream();
     let terminalStatus: number | undefined;
-    const stream = nativeChatSse(response.body, {
+    // Reuse physical-send credential checks, pacing and the same request policy budget.
+    const canRefetch = !isNonReplayableResponse(response);
+    if (!canRefetch) releaseRetainedRequest();
+    const resilientBody = canRefetch
+      ? wrapWithZeroOutputRefetch(response.body, (_recovery, signal) =>
+        send(activeRequest, "connection-reset", true, signal), {
+          abortSignal: upstream.signal, label: safeHostLabel(activeRequest.url),
+          acceptResponse: replacement => replacement.headers.get("content-type")?.toLowerCase().includes("text/event-stream") === true,
+          onReplayUnavailable: releaseRetainedRequest,
+        })
+      : response.body;
+    const stream = nativeChatSse(resilientBody, {
       requestedModel,
       translatorBudget,
       signal: upstream.signal,
@@ -575,6 +596,7 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio
     }
   }
 
+  releaseRetainedRequest();
   let body;
   try {
     body = await readBoundedResponseBody(response, {
diff --git a/src/server/responses/fetch-helpers.ts b/src/server/responses/fetch-helpers.ts
index 00bdbdc0f2..c0b901d925 100644
--- a/src/server/responses/fetch-helpers.ts
+++ b/src/server/responses/fetch-helpers.ts
@@ -53,6 +53,8 @@ export interface PaceAwareFetch {
 export type ProviderFetch = typeof globalThis.fetch & PaceAwareFetch;
 
 export interface ProviderFetchOptions {
+  /** A replay of an HTTP body must not initiate a fresh WebSocket exchange. */
+  httpOnly?: boolean;
   providerName?: string;
   modelId?: string;
   /** One pacing slot was acquired immediately before this fetch wrapper was created. */
@@ -96,7 +98,7 @@ export function providerFetch(
   // else keeps the provider's HTTP fetch. See ws-upstream.ts for the details.
   const unpaced = async (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
     const upstreamWebsocket = provider.upstreamWebsocket === true;
-    if (typeof input === "string" && init && shouldUseCodexWsUpstream(input, init, runtime, upstreamWebsocket)) {
+    if (!options.httpOnly && typeof input === "string" && init && shouldUseCodexWsUpstream(input, init, runtime, upstreamWebsocket)) {
       // The fallback has to be the same HTTP fetch the non-WS branch would have
       // used, protocol pin included: a WS turn that falls back is serving the
       // request over HTTP, and dropping the provider's `upstreamHttpVersion`
diff --git a/src/server/responses/passthrough-delivery.ts b/src/server/responses/passthrough-delivery.ts
index 1c69c0d6d5..d19f16d2ab 100644
--- a/src/server/responses/passthrough-delivery.ts
+++ b/src/server/responses/passthrough-delivery.ts
@@ -42,7 +42,8 @@ import {
   createPassthroughWebSearchBridgeStream,
   createPassthroughWebSearchBridgeExecutor,
 } from "../../web-search/passthrough-bridge";
-import { fetchWithHeaderTimeout, providerFetch } from "./fetch-helpers";
+import { fetchWithHeaderTimeout, providerFetch, safeHostLabel } from "./fetch-helpers";
+import { isNonReplayableResponse, wrapWithZeroOutputRefetch } from "../../lib/upstream-retry";
 import { providerApiKeySelectionIsCurrent } from "../../providers/api-key-selection";
 import { requiresVisionPreprocessing } from "../../vision";
 import { checkOutboundBodySize, describeOutboundBodyRefusal } from "./outbound-body-guard";
@@ -133,6 +134,7 @@ export async function deliverPassthroughResponse(
   nativeExchange: Pick<
     PassthroughExchange,
     | "upstreamResponse"
+    | "refetchZeroOutput"
     | "codexSafetyBufferingOptions"
     | "upstream"
     | "request"
@@ -346,10 +348,21 @@ export async function deliverPassthroughResponse(
       const webSearchBridgeBinding = requestBindings.get(nativeExchange.request);
       // The bridge wraps the RAW upstream body, so terminal repair below still owns the single
       // client-facing terminal — the bridge drops the terminal of every intercepted leg.
+      // Preserve the original HTTP-byte boundary before rewriting or hosted-search work.
+      // A sent WebSocket exchange must not replay even when no SSE bytes were delivered.
+      const rawBody = !isCodexWsUpstreamResponse(upstreamResponse) && !isNonReplayableResponse(upstreamResponse)
+        ? wrapWithZeroOutputRefetch(upstreamResponse.body, nativeExchange.refetchZeroOutput, {
+          abortSignal: upstream.signal, label: safeHostLabel(nativeExchange.request.url),
+          acceptResponse: replacement => {
+            const type = replacement.headers.get("content-type")?.toLowerCase();
+            return type?.includes("text/event-stream") === true || (!type && !passthroughCt);
+          },
+        })
+        : upstreamResponse.body;
       const upstreamSseBody = webSearchBridgePlan
         ? createPassthroughWebSearchBridgeStream({
           plan: webSearchBridgePlan,
-          firstLeg: upstreamResponse.body,
+          firstLeg: rawBody,
           requestBody: nativeExchange.request.body,
           // Continuation legs replay the same built request with the executed search appended.
           // The first leg already passed the recovery ladder, the outbound size ceiling, and the
@@ -389,7 +402,7 @@ export async function deliverPassthroughResponse(
           },
           signal: upstream.signal,
         })
-        : upstreamResponse.body;
+        : rawBody;
       const passthroughSseBody = terminalRepairPolicy
         ? relayResponsesSseWithTerminalRepair(
           upstreamSseBody,
diff --git a/src/server/responses/passthrough-dispatch.ts b/src/server/responses/passthrough-dispatch.ts
index 65eb3512da..0b8cdfef57 100644
--- a/src/server/responses/passthrough-dispatch.ts
+++ b/src/server/responses/passthrough-dispatch.ts
@@ -102,6 +102,7 @@ import {
   fetchWithTransientRetry,
   applyUpstreamRecoveryInit,
   TRANSIENT_RETRY_MAX_ATTEMPTS,
+  type UpstreamSendRecovery,
   prepareSameTarget429Wait,
   sleepWithAbort,
 } from "../../lib/upstream-retry";
@@ -165,6 +166,8 @@ export async function preparePassthroughExchange(
     | "genericFailoverAccountId"
     | "passiveQuotaWriterGeneration"
     | "oauthDispatch"
+    | "selectionIsCurrent"
+    | "requestBindings"
     | "resolveSelectionAdapter"
     | "isOAuth401ReplayProvider"
     | "sentOAuthSnapshot"
@@ -1444,6 +1447,35 @@ export async function preparePassthroughExchange(
     break;
     }
 
+  // Delivery may request one zero-byte HTTP recovery, never a fresh retry allowance.
+  const refetchZeroOutput = (_recovery?: UpstreamSendRecovery, signal: AbortSignal = upstream.signal): Promise<Response> =>
+    fetchWithTransientRetry(() => fetchWithHeaderTimeout(
+      request.url,
+      applyUpstreamRecoveryInit({ method: request.method, headers: request.headers, body: request.body }, "connection-reset"),
+      signal, connectMs, true,
+      providerFetch(route.provider, options.codexWsRuntimeIdentity, {
+        httpOnly: true,
+        providerName: route.providerName, modelId: route.modelId,
+        dispatchOverride: oauthDispatch(request),
+        beforeDispatch: headers => {
+          if (signal.aborted) throw signal.reason;
+          if (!transportState.selectionIsCurrent(transportState.requestBindings.get(request))) {
+            throw new Error("Credential selection changed before zero-output recovery");
+          }
+          if (isCanonicalOpenAiForwardProvider(route.provider)) {
+            createCodexReserveDispatchGuard(admissionState.authCtx, options.codexAuthPolicy ?? config,
+              route.modelId, options.admission, options.visionDescribeTerminal === true)?.(headers);
+          }
+          noteAttemptSend(logCtx.activeAttempt, logCtx.usageLogInputTokens, "connection-reset");
+        },
+      }),
+      route.provider.authMode === "forward",
+    ).then(adoptObservedResponse), {
+      abortSignal: signal, label: safeHostLabel(request.url),
+      attempts: Math.min(1, remainingTransientSendBudget(TRANSIENT_RETRY_MAX_ATTEMPTS)),
+      onSendsConsumed: noteTransientSends,
+    });
+
   return {
     codexSafetyBufferingOptions,
     imageGenCallAliases,
@@ -1482,6 +1514,7 @@ export async function preparePassthroughExchange(
     upstream,
     connectMs,
     upstreamResponse,
+    refetchZeroOutput,
   };
 }
 
diff --git a/tests/upstream-retry-zero-output.test.ts b/tests/upstream-retry-zero-output.test.ts
new file mode 100644
index 0000000000..f01c69f7d7
--- /dev/null
+++ b/tests/upstream-retry-zero-output.test.ts
@@ -0,0 +1,222 @@
+import { afterEach, describe, expect, spyOn, test } from "bun:test";
+import {
+  refetchOnZeroOutputReset,
+  wrapWithZeroOutputRefetch,
+} from "../src/lib/upstream-retry";
+
+function resetError(): Error {
+  // Shape of Bun's fetch rejection on a stale pooled socket.
+  const err = new Error("The socket connection was closed unexpectedly. For more information, pass `verbose: true` in the second argument to fetch()");
+  (err as Error & { code: string }).code = "ECONNRESET";
+  return err;
+}
+
+function streamOf(chunks: Uint8Array[]): ReadableStream<Uint8Array> {
+  let i = 0;
+  return new ReadableStream<Uint8Array>({
+    pull(controller) {
+      if (i < chunks.length) controller.enqueue(chunks[i++]!);
+      else controller.close();
+    },
+  });
+}
+
+function failingStream(err: Error): ReadableStream<Uint8Array> {
+  return new ReadableStream<Uint8Array>({
+    pull(controller) {
+      controller.error(err);
+    },
+  });
+}
+
+async function collect(stream: ReadableStream<Uint8Array>): Promise<Uint8Array[]> {
+  const out: Uint8Array[] = [];
+  const reader = stream.getReader();
+  for (;;) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    out.push(value);
+  }
+  return out;
+}
+
+function responseWith(body: ReadableStream<Uint8Array>): Response {
+  return new Response(body);
+}
+
+function noBodyResponse(): Response {
+  return new Response(null);
+}
+
+const warnSpies: Array<ReturnType<typeof spyOn>> = [];
+function silenceWarn(): void {
+  warnSpies.push(spyOn(console, "warn").mockImplementation(() => {}));
+}
+
+afterEach(() => {
+  for (const spy of warnSpies.splice(0)) spy.mockRestore();
+});
+
+describe("refetchOnZeroOutputReset", () => {
+  test("refetches once on a reset-shaped error with a live signal", async () => {
+    silenceWarn();
+    const calls: string[] = [];
+    const doFetch = async (recovery?: string): Promise<Response> => {
+      calls.push(recovery ?? "none");
+      return responseWith(streamOf([new TextEncoder().encode("replacement")]));
+    };
+    const result = await refetchOnZeroOutputReset(doFetch, resetError(), {});
+    expect(result).not.toBeNull();
+    expect(calls).toEqual(["connection-reset"]);
+  });
+
+  test("returns null for non-reset errors", async () => {
+    let calls = 0;
+    const result = await refetchOnZeroOutputReset(async () => { calls += 1; return responseWith(streamOf([])); }, new Error("something else"), {});
+    expect(result).toBeNull();
+    expect(calls).toBe(0);
+  });
+
+  test("returns null when the caller signal is aborted", async () => {
+    let calls = 0;
+    const controller = new AbortController();
+    controller.abort();
+    const result = await refetchOnZeroOutputReset(
+      async () => { calls += 1; return responseWith(streamOf([])); },
+      resetError(),
+      { abortSignal: controller.signal },
+    );
+    expect(result).toBeNull();
+    expect(calls).toBe(0);
+  });
+
+  test("returns null when the refetch throws", async () => {
+    silenceWarn();
+    const result = await refetchOnZeroOutputReset(
+      async () => { throw new Error("refetch failed"); },
+      resetError(),
+      {},
+    );
+    expect(result).toBeNull();
+  });
+
+  test("returns null when the replacement has no body", async () => {
+    silenceWarn();
+    const result = await refetchOnZeroOutputReset(
+      async () => noBodyResponse(),
+      resetError(),
+      {},
+    );
+    expect(result).toBeNull();
+  });
+});
+
+describe("wrapWithZeroOutputRefetch", () => {
+  test("swaps in the refetched stream on a zero-byte reset", async () => {
+    silenceWarn();
+    const original = failingStream(resetError());
+    const refetchCalls: string[] = [];
+    const doFetch = async (recovery?: string): Promise<Response> => {
+      refetchCalls.push(recovery ?? "none");
+      return responseWith(streamOf([new TextEncoder().encode("ok")]));
+    };
+    const wrapped = wrapWithZeroOutputRefetch(original, doFetch, {});
+    const chunks = await collect(wrapped);
+    expect(new TextDecoder().decode(chunks[0])).toBe("ok");
+    expect(refetchCalls).toEqual(["connection-reset"]);
+  });
+
+  test("does not refetch after bytes were already consumed", async () => {
+    silenceWarn();
+    const good = new TextEncoder().encode("partial");
+    let delivered = false;
+    const stream = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        if (!delivered) {
+          delivered = true;
+          controller.enqueue(good);
+          return;
+        }
+        controller.error(resetError());
+      },
+    });
+    let refetchCalls = 0;
+    const doFetch = async (): Promise<Response> => {
+      refetchCalls += 1;
+      return responseWith(streamOf([new TextEncoder().encode("never")]));
+    };
+    const wrapped = wrapWithZeroOutputRefetch(stream, doFetch, {});
+    const reader = wrapped.getReader();
+    const first = await reader.read();
+    // Partial output was delivered, then the original error propagated: the
+    // wrapper must NOT mask a partial-output failure with a replay.
+    expect(new TextDecoder().decode(first.value)).toBe("partial");
+    await expect(reader.read()).rejects.toThrow(/socket connection was closed/i);
+    expect(refetchCalls).toBe(0);
+  });
+
+  test("propagates the original error on a non-reset failure", async () => {
+    let refetchCalls = 0;
+    const wrapped = wrapWithZeroOutputRefetch(
+      failingStream(new Error("boom")),
+      async () => { refetchCalls += 1; return responseWith(streamOf([])); },
+      {},
+    );
+    const reader = wrapped.getReader();
+    await expect(reader.read()).rejects.toThrow("boom");
+    expect(refetchCalls).toBe(0);
+  });
+
+  test("propagates the original error when the refetch itself fails", async () => {
+    silenceWarn();
+    const wrapped = wrapWithZeroOutputRefetch(
+      failingStream(resetError()),
+      async () => { throw new Error("refetch failed"); },
+      {},
+    );
+    const reader = wrapped.getReader();
+    await expect(reader.read()).rejects.toThrow(/socket connection was closed/i);
+  });
+
+  test("propagates the original error when the refetch returns a bodyless response", async () => {
+    silenceWarn();
+    const wrapped = wrapWithZeroOutputRefetch(
+      failingStream(resetError()),
+      async () => noBodyResponse(),
+      {},
+    );
+    const reader = wrapped.getReader();
+    await expect(reader.read()).rejects.toThrow(/socket connection was closed/i);
+  });
+
+  test("retries at most once: a second zero-byte reset on the replacement propagates", async () => {
+    silenceWarn();
+    const replacement = failingStream(resetError());
+    let calls = 0;
+    const wrapped = wrapWithZeroOutputRefetch(
+      failingStream(resetError()),
+      async () => { calls += 1; return responseWith(replacement); },
+      {},
+    );
+    const reader = wrapped.getReader();
+    await expect(reader.read()).rejects.toThrow(/socket connection was closed/i);
+    expect(calls).toBe(1);
+  });
+
+  test("forwards cancellation to the active reader", async () => {
+    const cancelled: string[] = [];
+    const original = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        controller.enqueue(new TextEncoder().encode("x"));
+      },
+      cancel(reason) {
+        cancelled.push(String(reason));
+      },
+    });
+    const wrapped = wrapWithZeroOutputRefetch(original, async () => responseWith(streamOf([])), {});
+    const reader = wrapped.getReader();
+    await reader.read();
+    await reader.cancel("stop");
+    expect(cancelled.length).toBe(1);
+  });
+});

From 98df7f3c52b33f581529ae094a491b09bc832849 Mon Sep 17 00:00:00 2001
From: Yum-wu <1172989563@qq.com>
Date: Fri, 4 Sep 2026 09:35:50 +0800
Subject: [PATCH 2/3] fix(retry): reject non-ok replacements and redact refetch
 error logs
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Address CodeRabbit review on #3389 (2 Major / quick-win findings):

1. Sensitive-data exposure (CWE-532): the refetch-failure warn logged the raw
   error message, which can carry provider-returned URLs/headers/tokens.
   Redact via redactSecretString (leaf module, no new dependency) before it
   reaches the log.

2. Missing replacement.ok gate: a body-bearing 401/429/5xx/redirect
   replacement was accepted and its body relayed as SSE stream data under the
   original 200 response status — a malformed 'successful' stream. Treat any
   non-ok replacement as a failed refetch (cancel body, return null) so the
   wrapper propagates the original reset failure.

Regression tests added: refetchOnZeroOutputReset rejects a body-bearing 503;
wrapWithZeroOutputRefetch propagates the original error for body-bearing 503
and 401 replacements; refetch error messages are redacted before logging.

Verified: tests/upstream-retry-zero-output.test.ts 16 pass, adjacent
upstream-retry (23) and upstream-transient-retry (15) suites green,
tsc --noEmit clean.
---
 src/lib/upstream-retry.ts                | 22 ++++++----
 tests/upstream-retry-zero-output.test.ts | 51 ++++++++++++++++++++++++
 2 files changed, 66 insertions(+), 7 deletions(-)

diff --git a/src/lib/upstream-retry.ts b/src/lib/upstream-retry.ts
index 55a3058406..8be4d7b876 100644
--- a/src/lib/upstream-retry.ts
+++ b/src/lib/upstream-retry.ts
@@ -15,6 +15,7 @@
  * the shared abort helpers from here).
  */
 import { clearableDeadline } from "./abort";
+import { redactSecretString } from "./redact";
 
 /**
  * Responses the origin may already be executing. RFC 9110 §9.2.2 forbids an intermediary
@@ -619,8 +620,8 @@ export async function fetchWithTransientRetry(
  *
  * Returns null (and the caller keeps its existing fail-closed tail) when the
  * error is not a reset shape, the caller signal is aborted, the refetch itself
- * throws, or the replacement has no body. Callers must not retry the returned
- * response's body.
+ * throws, or the replacement is not successful (non-2xx) or has no body.
+ * Callers must not retry the returned response's body.
  */
 export async function refetchOnZeroOutputReset(
   doFetch: ReplayableFetch,
@@ -633,16 +634,23 @@ export async function refetchOnZeroOutputReset(
   try {
     replacement = await doFetch("connection-reset");
   } catch (retryErr) {
+    // The replacement rejection may carry provider-returned text (URLs, headers,
+    // tokens) in its message; redact before it reaches the log.
+    const detail = retryErr instanceof Error ? retryErr.message : String(retryErr);
     console.warn(
-      `[upstream-retry] zero-output reset${opts.label ? ` (${opts.label})` : ""} — refetch failed: ${
-        retryErr instanceof Error ? retryErr.message : String(retryErr)
-      }`,
+      `[upstream-retry] zero-output reset${opts.label ? ` (${opts.label})` : ""} — refetch failed: ${redactSecretString(detail)}`,
     );
     return null;
   }
-  if (!replacement.body) {
+  // A non-success replacement (401/429/5xx/redirect) with a body must NOT be
+  // relayed as stream data: the caller keeps the original 200 response status,
+  // so a 503 body would surface as a malformed "successful" SSE stream. Treat
+  // any non-ok replacement as a failed refetch and preserve the original error.
+  if (!replacement.ok || !replacement.body) {
     console.warn(
-      `[upstream-retry] zero-output reset${opts.label ? ` (${opts.label})` : ""} — refetch returned no body, keeping original failure`,
+      `[upstream-retry] zero-output reset${opts.label ? ` (${opts.label})` : ""} — refetch returned ${
+        replacement.ok ? "no body" : `non-ok status ${replacement.status}`
+      }, keeping original failure`,
     );
     try {
       replacement.arrayBuffer().catch(() => {});
diff --git a/tests/upstream-retry-zero-output.test.ts b/tests/upstream-retry-zero-output.test.ts
index f01c69f7d7..df9712a6c2 100644
--- a/tests/upstream-retry-zero-output.test.ts
+++ b/tests/upstream-retry-zero-output.test.ts
@@ -109,6 +109,31 @@ describe("refetchOnZeroOutputReset", () => {
     );
     expect(result).toBeNull();
   });
+
+  test("returns null when the replacement is non-ok even with a body (503 regression)", async () => {
+    silenceWarn();
+    // CodeRabbit review: a body-bearing 503 replacement must be treated as a
+    // failed refetch, not relayed as stream data under the original 200 status.
+    const result = await refetchOnZeroOutputReset(
+      async () => new Response(new TextEncoder().encode("service unavailable"), { status: 503 }),
+      resetError(),
+      {},
+    );
+    expect(result).toBeNull();
+  });
+
+  test("redacts the refetch error message before logging", async () => {
+    const warns: string[] = [];
+    const spy = spyOn(console, "warn").mockImplementation((msg: string) => { warns.push(String(msg)); });
+    warnSpies.push(spy);
+    await refetchOnZeroOutputReset(
+      async () => { throw new Error("fetch failed: https://api.example.com/v1?api_key=sk-secret123"); },
+      resetError(),
+      {},
+    );
+    expect(warns.some(w => w.includes("sk-secret123"))).toBe(false);
+    expect(warns.some(w => w.includes("refetch failed"))).toBe(true);
+  });
 });
 
 describe("wrapWithZeroOutputRefetch", () => {
@@ -189,6 +214,32 @@ describe("wrapWithZeroOutputRefetch", () => {
     await expect(reader.read()).rejects.toThrow(/socket connection was closed/i);
   });
 
+  test("propagates the original error when the refetch returns a body-bearing 503 (not relayed as SSE)", async () => {
+    silenceWarn();
+    // CodeRabbit review: without a replacement.ok gate, a 503 body would be
+    // relayed as stream data under the original 200 status, surfacing as a
+    // malformed "successful" SSE stream. The wrapper must reject it and keep
+    // the original reset failure.
+    const wrapped = wrapWithZeroOutputRefetch(
+      failingStream(resetError()),
+      async () => new Response(new TextEncoder().encode("service unavailable"), { status: 503 }),
+      {},
+    );
+    const reader = wrapped.getReader();
+    await expect(reader.read()).rejects.toThrow(/socket connection was closed/i);
+  });
+
+  test("propagates the original error when the refetch returns a body-bearing 401", async () => {
+    silenceWarn();
+    const wrapped = wrapWithZeroOutputRefetch(
+      failingStream(resetError()),
+      async () => new Response(new TextEncoder().encode("unauthorized"), { status: 401 }),
+      {},
+    );
+    const reader = wrapped.getReader();
+    await expect(reader.read()).rejects.toThrow(/socket connection was closed/i);
+  });
+
   test("retries at most once: a second zero-byte reset on the replacement propagates", async () => {
     silenceWarn();
     const replacement = failingStream(resetError());

From 5c997629d70b1c861c014f5761c9c447fd0a4e55 Mon Sep 17 00:00:00 2001
From: JUN <bitkyc08@gmail.com>
Date: Wed, 16 Sep 2026 11:43:15 +0900
Subject: [PATCH 3/3] fix(retry): preserve dispatch budgets and cancellation
 after modular rebase

Port zero-byte HTTP recovery to split dispatch/delivery owners without reviving core.ts. Preserve native Chat terminal precedence, current credential selection, HTTP-only dispatch and shared send accounting. Cancel rejected or late replacement bodies instead of draining them. Register moved regression tests and document the billable replay boundary. Connected-host tests, builds and typecheck were not run.

Co-authored-by: Yum-wu <1172989563@qq.com>
---
 .../docs/ko/reference/configuration/server.md |   8 +
 .../docs/reference/configuration/server.md    |   9 +
 scripts/test-layout/layout.json               |   3 +-
 src/lib/upstream-retry.ts                     | 196 ++++++++++--------
 structure/adapters/registry.md                |   2 +
 structure/catalog.md                          |   2 +
 structure/clients/claude-desktop.md           |   2 +
 structure/clients/integrations.md             |   2 +
 structure/data-planes/images.md               |   2 +
 structure/data-planes/inbound-compat.md       |   2 +
 structure/gui-and-management-api.md           |   2 +
 structure/ops/docs-and-release.md             |   2 +
 structure/ops/service-and-sidecars.md         |   2 +
 structure/overview.md                         |   2 +
 structure/providers/xai-grok.md               |   2 +
 structure/runtime.md                          |   2 +
 structure/subagents.md                        |   2 +
 structure/transports/byte-accounting.md       |   2 +
 structure/transports/inventory.md             |   2 +
 structure/transports/responses.md             |  11 +-
 structure/transports/streaming-health.md      |  18 ++
 tests/fixtures/test-layout-expected.json      |   3 +-
 .../upstream-retry-zero-output.test.ts        | 105 +++++++++-
 23 files changed, 288 insertions(+), 95 deletions(-)
 rename tests/{ => lib}/upstream-retry-zero-output.test.ts (65%)

diff --git a/docs-site/src/content/docs/ko/reference/configuration/server.md b/docs-site/src/content/docs/ko/reference/configuration/server.md
index 5727ba13d4..ecc2cfe00b 100644
--- a/docs-site/src/content/docs/ko/reference/configuration/server.md
+++ b/docs-site/src/content/docs/ko/reference/configuration/server.md
@@ -235,3 +235,11 @@ Anthropic OAuth 사이드카는 opencodex의 기존 Claude Code OAuth fingerprin
 ## Codex 할당량 네트워크 진단
 
 메인 Codex 계정 행의 `quotaRefresh`는 할당량 조회 결과를 분류하는 진단값입니다. 남은 할당량이나 모델 접근 권한을 뜻하지 않으며, 캐시를 쓰거나 조회하지 않았다면 생략될 수 있습니다. 요청은 명령을 입력한 터미널이 아니라 실행 중인 프록시 서비스의 환경을 따릅니다. `proxy`를 지정하지 않으면 기존 환경을 유지하고, `"auto"`는 시작할 때 Windows의 정적 프록시 설정만 읽습니다. PAC/WPAD, SOCKS 전용 설정과 실행 중 변경은 자동으로 반영하지 않습니다. TUN에서 성공했다고 HTTP 프록시 경로도 정상이라는 뜻은 아닙니다. 명령과 상태값은 [네트워크 진단(영문)](/reference/configuration/server/#codex-quota-network-diagnostics)에서 확인하세요.
+
+## 첫 바이트 전 스트림 복구
+
+네이티브 Chat과 Responses는 HTTP 응답 헤더를 받은 뒤, 원본 응답 바이트를 하나도 읽지 못한 상태에서
+연결 재설정 오류가 나면 요청을 한 번 더 보낼 수 있습니다. 추가 전송은 남은 요청 한도와 현재 자격 증명을
+사용합니다. 취소, 일부 출력, 정상 EOF 또는 이미 전송한 WebSocket 요청은 재전송하지 않습니다.
+대체 응답도 기존 스트림 형식을 유지해야 합니다. 바이트를 받지 못했다고 공급자가 작업하지 않은 것은
+아니므로 추가 비용이 발생할 수 있습니다. `emptyCompletionRetry` 설정과는 별개입니다.
diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md
index 3fbf619df4..97037af0d3 100644
--- a/docs-site/src/content/docs/reference/configuration/server.md
+++ b/docs-site/src/content/docs/reference/configuration/server.md
@@ -560,3 +560,12 @@ A hub that serves its own local clients also sets
 [`unauthenticatedLoopbackListener`](#local-clients-that-cannot-receive-the-token). Its port-less
 companion form is what makes a hub a single-port deployment, and it is refused on a loopback or
 wildcard `hostname`, where the public listener already holds `127.0.0.1:<port>`.
+
+## Zero-byte stream recovery
+
+Native Chat and Responses may retry an HTTP request once after response headers if its body fails
+with a connection reset before any raw response byte is read. The extra send uses the remaining
+request allowance and the current credential; cancellation, partial output, clean EOF and
+already-sent WebSocket exchanges do not trigger it. A replacement must preserve the stream
+format. Zero observed bytes do not guarantee the provider did no work, so a replay may be billable.
+This is separate from the `emptyCompletionRetry` setting.
diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json
index 30bcc28b47..b497314ce7 100644
--- a/scripts/test-layout/layout.json
+++ b/scripts/test-layout/layout.json
@@ -1466,7 +1466,8 @@
     "execution-budget-permits.test.ts": "lib",
     "spend-instrumentation-log.test.ts": "server",
     "codex-pool-refresh-backoff.test.ts": "codex-integration",
-    "responses-account-change-scrub.test.ts": "responses"
+    "responses-account-change-scrub.test.ts": "responses",
+    "upstream-retry-zero-output.test.ts": "lib"
   },
   "migrated": [
     "adapters",
diff --git a/src/lib/upstream-retry.ts b/src/lib/upstream-retry.ts
index 8be4d7b876..c812b141b6 100644
--- a/src/lib/upstream-retry.ts
+++ b/src/lib/upstream-retry.ts
@@ -8,8 +8,9 @@
  * a caught error here means no response was ever received.
  *
  * Deliberately narrow: timeouts, aborts, ECONNREFUSED/DNS/TLS failures, and HTTP error
- * statuses (returned as Response, never thrown) are NOT retried. Mid-stream SSE resets are
- * out of scope — the response has already resolved by then.
+ * statuses (returned as Response, never thrown) are NOT retried by the reset-only helper.
+ * The separate zero-byte body wrapper below permits one HTTP replacement through its
+ * caller's existing send budget; partial output and sent WebSocket exchanges never replay.
  *
  * MUST stay a leaf module: imports nothing from server.ts or adapters (kiro-retry imports
  * the shared abort helpers from here).
@@ -596,130 +597,151 @@ export async function fetchWithTransientRetry(
   }
 }
 
-/**
- * Refetch an upstream request ONCE after a mid-stream socket reset that
- * happened before the caller consumed any response bytes.
- *
- * `fetchWithResetRetry` / `fetchWithTransientRetry` only cover pre-stream
- * failures — `fetch()` rejecting before response headers. Once headers arrive
- * and the caller starts reading the SSE body, a mid-stream reset (Cloudflare
- * closing an idle keep-alive connection while Bun's pool reuses the half-closed
- * socket) surfaces as a ReadableStream read() rejection, outside every
- * pre-stream retry wrapper. The turn then dies with a terminal
- * `response.failed / upstream_reset` even though nothing was relayed to the
- * client.
- *
- * This helper closes that gap for the one case where a replay is provably safe:
- * zero bytes consumed and no protocol terminal seen. `doFetch` must be
- * replay-safe (string body, same contract as {@link ReplayableFetch}); the
- * replacement send goes out with the connection-reset recovery init
- * (`Connection: close` + `keepalive: false`) so the fresh connection never
- * reuses the pooled half-closed socket. Exactly one replacement send, no
- * backoff — the pre-stream layers already spent their retry budget reaching
- * the first headers.
- *
- * Returns null (and the caller keeps its existing fail-closed tail) when the
- * error is not a reset shape, the caller signal is aborted, the refetch itself
- * throws, or the replacement is not successful (non-2xx) or has no body.
- * Callers must not retry the returned response's body.
- */
+export type ZeroOutputReplayFetch = (
+  recovery?: UpstreamSendRecovery,
+  signal?: AbortSignal,
+) => Promise<Response>;
+
+export interface ZeroOutputRefetchOptions extends ResetRetryOptions {
+  /** The replacement must match the response contract already sent to the client. */
+  acceptResponse?: (response: Response) => boolean;
+  /** Release retained request material when no further replay is possible. */
+  onReplayUnavailable?: () => void;
+}
+
+/** One replacement attempt; the caller retains physical-send admission and accounting. */
 export async function refetchOnZeroOutputReset(
-  doFetch: ReplayableFetch,
+  doFetch: ZeroOutputReplayFetch,
   err: unknown,
-  opts: ResetRetryOptions = {},
+  opts: ZeroOutputRefetchOptions = {},
 ): Promise<Response | null> {
-  if (!isConnectionResetError(err)) return null;
-  if (opts.abortSignal?.aborted) return null;
+  if (!isConnectionResetError(err) || opts.abortSignal?.aborted || opts.attempts === 0) return null;
+  const label = opts.label
+    ? " (" + redactSecretString(opts.label).replace(/[\r\n\u0000-\u001f\u007f]/g, "").slice(0, 128) + ")"
+    : "";
   let replacement: Response;
   try {
-    replacement = await doFetch("connection-reset");
-  } catch (retryErr) {
-    // The replacement rejection may carry provider-returned text (URLs, headers,
-    // tokens) in its message; redact before it reaches the log.
-    const detail = retryErr instanceof Error ? retryErr.message : String(retryErr);
-    console.warn(
-      `[upstream-retry] zero-output reset${opts.label ? ` (${opts.label})` : ""} — refetch failed: ${redactSecretString(detail)}`,
-    );
+    replacement = await doFetch("connection-reset", opts.abortSignal);
+  } catch {
+    console.warn("[upstream-retry] zero-output refetch failed" + label + "; preserving original stream error");
     return null;
   }
-  // A non-success replacement (401/429/5xx/redirect) with a body must NOT be
-  // relayed as stream data: the caller keeps the original 200 response status,
-  // so a 503 body would surface as a malformed "successful" SSE stream. Treat
-  // any non-ok replacement as a failed refetch and preserve the original error.
-  if (!replacement.ok || !replacement.body) {
-    console.warn(
-      `[upstream-retry] zero-output reset${opts.label ? ` (${opts.label})` : ""} — refetch returned ${
-        replacement.ok ? "no body" : `non-ok status ${replacement.status}`
-      }, keeping original failure`,
-    );
-    try {
-      replacement.arrayBuffer().catch(() => {});
-    } catch { /* body already unusable; original failure stands */ }
+  const body = replacement.body;
+  let accepted = !opts.abortSignal?.aborted && replacement.ok && body !== null
+    && !replacement.bodyUsed && !body.locked && !isNonReplayableResponse(replacement);
+  try { if (accepted && opts.acceptResponse) accepted = opts.acceptResponse(replacement); }
+  catch { accepted = false; }
+  if (!accepted || opts.abortSignal?.aborted || body?.locked) {
+    // Do not drain an unbounded error/JSON response or await an uncooperative cancellation.
+    try { void body?.cancel().catch(() => {}); } catch { /* already locked or closed */ }
+    console.warn("[upstream-retry] zero-output refetch rejected" + label + "; preserving original stream error");
     return null;
   }
-  console.warn(
-    `[upstream-retry] zero-output reset${opts.label ? ` (${opts.label})` : ""} — refetched on a fresh connection`,
-  );
+  console.warn("[upstream-retry] zero-output mid-stream reset" + label + "; using one replacement stream");
   return replacement;
 }
 
 /**
- * Wrap an upstream SSE body so a mid-stream socket reset before the first byte
- * is consumed transparently swaps in ONE refetched body (see
- * {@link refetchOnZeroOutputReset}). Everything downstream — tee inspection
- * branches, eager or pull relays, SSE parsers — reads the wrapped stream and
- * never observes the first upstream send dying, so no relay needs changes.
- *
- * The gate is deliberately narrow: only a read() rejection matching
- * {@link isConnectionResetError}, with zero bytes read so far, a live caller
- * signal, and a single swap per wrapped stream. Partial-output failures, clean
- * EOF, non-reset errors, and a failed or empty refetch all propagate the
- * ORIGINAL error untouched, preserving every existing fail-closed tail
- * (replaying after emitted tool calls would duplicate side effects).
+ * Recover at most once before the first upstream byte. Zero observed bytes do not prove
+ * the origin performed no work; replay can still be billable. EOF and partial output never retry.
  */
 export function wrapWithZeroOutputRefetch(
   body: ReadableStream<Uint8Array>,
-  doFetch: ReplayableFetch,
-  opts: ResetRetryOptions = {},
+  doFetch: ZeroOutputReplayFetch,
+  opts: ZeroOutputRefetchOptions = {},
 ): ReadableStream<Uint8Array> {
+  const { abortSignal, label, acceptResponse } = opts;
+  const replayAbort = new AbortController();
   let reader = body.getReader();
+  let replay: ZeroOutputReplayFetch | undefined = opts.attempts === 0 ? undefined : doFetch;
+  let onReplayUnavailable = opts.onReplayUnavailable;
   let bytesRead = 0;
-  let retried = false;
+  let closed = false;
+  let output: ReadableStreamDefaultController<Uint8Array> | undefined;
+  const releaseReplay = (): void => {
+    replay = undefined;
+    const release = onReplayUnavailable;
+    onReplayUnavailable = undefined;
+    try { release?.(); } catch { /* bookkeeping cannot fail the stream */ }
+  };
+  const retireReader = (target: ReadableStreamDefaultReader<Uint8Array>, reason?: unknown): void => {
+    try { void target.cancel(reason).catch(() => {}); } catch { /* already closed */ }
+    try { target.releaseLock(); } catch { /* already released */ }
+  };
+  const detach = (): void => { abortSignal?.removeEventListener("abort", onAbort); };
+  const onAbort = (): void => {
+    if (closed) return;
+    closed = true;
+    const reason = abortSignal?.reason ?? new DOMException("The operation was aborted.", "AbortError");
+    replayAbort.abort(reason);
+    releaseReplay();
+    detach();
+    retireReader(reader, reason);
+    output?.error(reason);
+    output = undefined;
+  };
   return new ReadableStream<Uint8Array>({
+    start(controller) {
+      output = controller;
+      abortSignal?.addEventListener("abort", onAbort, { once: true });
+      if (abortSignal?.aborted) onAbort();
+    },
     async pull(controller) {
-      for (;;) {
+      while (!closed) {
+        const current = reader;
         try {
-          const { done, value } = await reader.read();
+          const { done, value } = await current.read();
+          if (closed) return;
           if (done) {
+            closed = true;
+            releaseReplay();
+            detach();
+            current.releaseLock();
             controller.close();
+            output = undefined;
             return;
           }
           bytesRead += value.byteLength;
+          if (bytesRead > 0) releaseReplay();
           controller.enqueue(value);
           return;
         } catch (err) {
-          if (!retried && bytesRead === 0 && !opts.abortSignal?.aborted) {
-            retried = true;
-            const replacement = await refetchOnZeroOutputReset(doFetch, err, opts);
+          if (closed) return;
+          const refetch = replay;
+          if (refetch && bytesRead === 0 && !replayAbort.signal.aborted && isConnectionResetError(err)) {
+            replay = undefined;
+            retireReader(current, err);
+            const replacement = await refetchOnZeroOutputReset(refetch, err, {
+              abortSignal: replayAbort.signal, label, acceptResponse,
+            });
+            releaseReplay();
+            if (closed || replayAbort.signal.aborted) {
+              try { void replacement?.body?.cancel().catch(() => {}); } catch { /* best effort */ }
+              return;
+            }
             if (replacement?.body) {
-              try {
-                reader.cancel().catch(() => {});
-              } catch { /* broken reader; the refetch won */ }
               reader = replacement.body.getReader();
               continue;
             }
           }
-          try {
-            controller.error(err);
-          } catch { /* already torn down */ }
+          closed = true;
+          releaseReplay();
+          detach();
+          retireReader(current, err);
+          controller.error(err);
+          output = undefined;
           return;
         }
       }
     },
     cancel(reason) {
-      try {
-        reader.cancel(reason).catch(() => {});
-      } catch { /* already torn down */ }
+      if (closed) return;
+      closed = true;
+      replayAbort.abort(reason);
+      releaseReplay();
+      detach();
+      retireReader(reader, reason);
+      output = undefined;
     },
-  });
+  }, { highWaterMark: 0 });
 }
diff --git a/structure/adapters/registry.md b/structure/adapters/registry.md
index 9c77eaac24..1597674a57 100644
--- a/structure/adapters/registry.md
+++ b/structure/adapters/registry.md
@@ -183,3 +183,5 @@ implement legacy call/result pairing. Modern tool-image carriers are unchanged.
 raw passthrough; `tests/responses/chat-media-translation.test.ts` reaches the real HTTP
 translation boundary and verifies that rejection sends no upstream request.
 Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged.
+
+[Zero-byte HTTP recovery](../transports/streaming-health.md#zero-byte-http-stream-recovery) does not resolve another adapter; it reuses the current physical-send boundary and validates the existing credential binding.
diff --git a/structure/catalog.md b/structure/catalog.md
index 427a67d876..51ff2957ba 100644
--- a/structure/catalog.md
+++ b/structure/catalog.md
@@ -430,3 +430,5 @@ Exact [model input declarations](config.md#explicit-per-model-capability-declara
 ## Renamed destination reasoning metadata
 
 `src/providers/derive.ts` fills missing reasoning tables for renamed providers accepted by the existing fixed-key destination matcher. Model entries are cloned and explicit user entries (including empty arrays) win. Provider-wide effort defaults fill only when undefined; Command Code unknown models therefore keep the registry's empty picker policy unless overridden. Identity, transport and other capability axes are unchanged. The gathered row drives client exports; this metadata contract does not prove arbitrary gateway routing.
+
+Model selection is not repeated by [zero-byte HTTP recovery](transports/streaming-health.md#zero-byte-http-stream-recovery); the selected request and its existing allowance remain authoritative.
diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md
index c7a8e7c8a5..4e3e248e23 100644
--- a/structure/clients/claude-desktop.md
+++ b/structure/clients/claude-desktop.md
@@ -168,3 +168,5 @@ The [explicit model-capability contract](../config.md#explicit-per-model-capabil
 Exact [model input declarations](../config.md#explicit-per-model-capability-declarations) now feed text-only eligibility and catalog hints; existing image-description/omission handling consumes them before the main upstream send.
 
 Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior.
+
+Desktop routing continues through the existing ingress; [zero-byte HTTP recovery](../transports/streaming-health.md#zero-byte-http-stream-recovery) does not alter client setup or retry a sent WebSocket exchange.
diff --git a/structure/clients/integrations.md b/structure/clients/integrations.md
index 3f2d3bf103..4b46ed888d 100644
--- a/structure/clients/integrations.md
+++ b/structure/clients/integrations.md
@@ -213,3 +213,5 @@ existing explicit confirmation. The journal endpoint evaluates Undo against the
 Recovery reads commit history and ownership through strict store methods. Unreadable or malformed
 metadata is uncertainty, never evidence that a transaction did not commit. Pending records validate
 complete ownership, exact Cline paths and result fingerprints before either native file is replaced.
+
+Client configuration writers remain separate from [zero-byte HTTP stream recovery](../transports/streaming-health.md#zero-byte-http-stream-recovery), which changes only an eligible in-flight body reader.
diff --git a/structure/data-planes/images.md b/structure/data-planes/images.md
index a73fcc6f07..1a495583a1 100644
--- a/structure/data-planes/images.md
+++ b/structure/data-planes/images.md
@@ -121,3 +121,5 @@ Live sideband admission and its bounded upstream handshake follow the [runtime c
 The [explicit model-capability contract](../config.md#explicit-per-model-capability-declarations) preserves operator declarations through provider storage and catalog capture; it does not infer upstream capability or change this surface's routing behavior.
 
 Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior.
+
+Image-loop retry policy is separate from [zero-byte HTTP stream recovery](../transports/streaming-health.md#zero-byte-http-stream-recovery); the latter wraps only the native Chat or Responses first body leg.
diff --git a/structure/data-planes/inbound-compat.md b/structure/data-planes/inbound-compat.md
index 0ad47b07e2..b93e36d4e4 100644
--- a/structure/data-planes/inbound-compat.md
+++ b/structure/data-planes/inbound-compat.md
@@ -331,3 +331,5 @@ Modern `tool` images continue through the existing following-user carrier. These
 an OpenCodex conversion limit, not a provider capability claim. Final Responses-to-adapter
 admission follows the [registry contract](../adapters/registry.md#untranslated-input-media).
 Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged.
+
+Native Chat preserves terminal/cancellation precedence during [zero-byte HTTP recovery](../transports/streaming-health.md#zero-byte-http-stream-recovery); partial output is never replayed.
diff --git a/structure/gui-and-management-api.md b/structure/gui-and-management-api.md
index e3a479a3da..320580eeb6 100644
--- a/structure/gui-and-management-api.md
+++ b/structure/gui-and-management-api.md
@@ -652,3 +652,5 @@ The [explicit model-capability contract](config.md#explicit-per-model-capability
 Exact [model input declarations](config.md#explicit-per-model-capability-declarations) now feed text-only eligibility and catalog hints; existing image-description/omission handling consumes them before the main upstream send.
 
 The raw provider editor round-trips `autoReviewModel` and `autoReviewModelOverrides` through editor-owned DTO fields. POST/PATCH/PUT share validation; PUT copies schema-normalized values into the persisted and live candidate before adoption. Canonical `openai` rejects these fields, including clear forms. Field-masked writes (PATCH, editor PUT, reload) pin every registry-seed key and ignore operator overlays the seed never defines, most commonly `selectedModels`; POST keeps the exact-key comparison. Canonical `openai` still rejects `allowPrivateNetwork`, which must not short-circuit destination DNS checks on the ChatGPT forward row. Existing authentication, origin checks and stale-baseline protection still govern the writes. See [reviewer projection](catalog.md#provider-scoped-approval-reviewer).
+
+A [zero-byte HTTP recovery](transports/streaming-health.md#zero-byte-http-stream-recovery) records its physical send as `connection-reset`; logs do not include the failed refetch response body.
diff --git a/structure/ops/docs-and-release.md b/structure/ops/docs-and-release.md
index a4ab2ae1e0..b8a5e3edbc 100644
--- a/structure/ops/docs-and-release.md
+++ b/structure/ops/docs-and-release.md
@@ -392,3 +392,5 @@ Exact [model input declarations](../config.md#explicit-per-model-capability-decl
 Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior.
 
 Renamed fixed-key providers receive [missing reasoning metadata](../catalog.md#renamed-destination-reasoning-metadata) during derivation; explicit per-model entries and provider defaults retain precedence.
+
+The zero-byte retry regression resides at `tests/lib/upstream-retry-zero-output.test.ts`, registered in both test-layout manifests. The [runtime contract](../transports/streaming-health.md#zero-byte-http-stream-recovery) distinguishes one HTTP replacement from a new retry budget.
diff --git a/structure/ops/service-and-sidecars.md b/structure/ops/service-and-sidecars.md
index db1a8653a1..66eceea215 100644
--- a/structure/ops/service-and-sidecars.md
+++ b/structure/ops/service-and-sidecars.md
@@ -182,3 +182,5 @@ Live sideband admission and its bounded upstream handshake follow the [runtime c
 The [explicit model-capability contract](../config.md#explicit-per-model-capability-declarations) preserves operator declarations through provider storage and catalog capture; it does not infer upstream capability or change this surface's routing behavior.
 
 Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior.
+
+[Zero-byte HTTP recovery](../transports/streaming-health.md#zero-byte-http-stream-recovery) is request-local; it starts no service timer and does not change sidecar-owned retry policies.
diff --git a/structure/overview.md b/structure/overview.md
index 0151115adc..6967b4aaab 100644
--- a/structure/overview.md
+++ b/structure/overview.md
@@ -147,3 +147,5 @@ Translated Chat request construction uses the [inline-image budget](transports/s
 The [explicit model-capability contract](config.md#explicit-per-model-capability-declarations) preserves operator declarations through provider storage and catalog capture; it does not infer upstream capability or change this surface's routing behavior.
 
 Provider-scoped approval reviewer settings are projected by the [catalog owner](catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior.
+
+HTTP body recovery retains the existing request boundary; see [zero-byte stream recovery](transports/streaming-health.md#zero-byte-http-stream-recovery) for its one-send, no-partial-output contract.
diff --git a/structure/providers/xai-grok.md b/structure/providers/xai-grok.md
index d146bc53fe..06d9ffeed8 100644
--- a/structure/providers/xai-grok.md
+++ b/structure/providers/xai-grok.md
@@ -143,3 +143,5 @@ Account quota surfaces use [safe probe diagnostics](../transports/inventory.md#a
 Live sideband admission and its bounded upstream handshake follow the [runtime contract](../runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate.
 
 Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate.
+
+Native xAI Responses uses the shared [zero-byte HTTP recovery](../transports/streaming-health.md#zero-byte-http-stream-recovery) without bypassing the selected OAuth binding or refreshing a new account after headers.
diff --git a/structure/runtime.md b/structure/runtime.md
index 37cee672c1..7573cf45fe 100644
--- a/structure/runtime.md
+++ b/structure/runtime.md
@@ -492,3 +492,5 @@ stamps the configured key selected for the physical request. `src/server/request
 retains per-key attempt usage, and `src/usage/log.ts` validates and persists labels. The
 [account attribution contract](gui-and-management-api.md#upstream-key-account-attribution)
 defines identity, unknown records, and aggregation boundaries.
+
+Native Chat and Responses retain physical-send credential admission during [zero-byte HTTP recovery](transports/streaming-health.md#zero-byte-http-stream-recovery); cancellation does not leave a replacement send running.
diff --git a/structure/subagents.md b/structure/subagents.md
index 051031f512..189f3c447b 100644
--- a/structure/subagents.md
+++ b/structure/subagents.md
@@ -384,3 +384,5 @@ Exact [model input declarations](config.md#explicit-per-model-capability-declara
 Provider-scoped approval reviewer settings are projected by the [catalog owner](catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior.
 
 Renamed fixed-key providers receive [missing reasoning metadata](catalog.md#renamed-destination-reasoning-metadata) during derivation; explicit per-model entries and provider defaults retain precedence.
+
+A child Responses request retains its parent workflow charge during [zero-byte HTTP recovery](transports/streaming-health.md#zero-byte-http-stream-recovery); the body wrapper cannot create a fresh fan-out allowance.
diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md
index b9b3980bf9..f320986733 100644
--- a/structure/transports/byte-accounting.md
+++ b/structure/transports/byte-accounting.md
@@ -41,3 +41,5 @@ Translated audio/file admission follows the [final-adapter input contract](../ad
 Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged.
 
 Upstream API-key usage follows the [physical-attempt account attribution contract](../gui-and-management-api.md#upstream-key-account-attribution), independently of subscription quota observations.
+
+During [zero-byte HTTP recovery](streaming-health.md#zero-byte-http-stream-recovery), native Chat retains request material only while replay remains possible. Rejected replacement bodies are cancelled, not accumulated with `arrayBuffer()`.
diff --git a/structure/transports/inventory.md b/structure/transports/inventory.md
index 91d527750a..bec8bd6082 100644
--- a/structure/transports/inventory.md
+++ b/structure/transports/inventory.md
@@ -150,3 +150,5 @@ Renamed fixed-key providers receive [missing reasoning metadata](../catalog.md#r
 
 Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate.
 Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged.
+
+The shared fetch executor has an HTTP-only mode for [zero-byte HTTP recovery](streaming-health.md#zero-byte-http-stream-recovery); pacing, dispatch overrides and credential checks still precede the physical send.
diff --git a/structure/transports/responses.md b/structure/transports/responses.md
index 6f56a27614..55f8491e92 100644
--- a/structure/transports/responses.md
+++ b/structure/transports/responses.md
@@ -593,11 +593,12 @@ with the same item id. The batch/non-streaming bridge follows the same rule.
 (Cloudflare closes idle connections; Bun's fetch reuses the dead socket and rejects with
 `ECONNRESET` before any response bytes). `fetchWithResetRetry` retries only
 connection-reset-shaped rejections (up to 3 total attempts, jittered backoff, warn-logged);
-timeouts, aborts, `ECONNREFUSED`, HTTP error statuses, and mid-stream SSE failures are never
-retried. Guarded paths: the ChatGPT passthrough and generic adapter fetch in
-`src/server/responses.ts`, the vision/web-search sidecars, and the web-search loop's direct-fetch
-fallback. Adapters with their own `fetchResponse` (kiro, cursor, google) keep their own retry
-policies; kiro imports the shared abort/sleep helpers from this module.
+timeouts, aborts, `ECONNREFUSED` and HTTP error statuses are not reset-retried. These helpers
+finish when headers arrive. The separate [zero-byte HTTP stream recovery](streaming-health.md#zero-byte-http-stream-recovery)
+keeps one replacement within existing send accounting and excludes already-sent WebSocket exchanges.
+Guarded fetch paths include `src/server/responses/passthrough-dispatch.ts`,
+`src/server/responses/adapter-dispatch.ts`, vision/web-search sidecars and direct-search fallback.
+Adapters with their own `fetchResponse` keep their own retry policies.
 
 ## Console upload rejection recovery
 
diff --git a/structure/transports/streaming-health.md b/structure/transports/streaming-health.md
index 06783ad651..cecdc1aa3f 100644
--- a/structure/transports/streaming-health.md
+++ b/structure/transports/streaming-health.md
@@ -252,3 +252,21 @@ Live sideband admission and its bounded upstream handshake follow the [runtime c
 The [explicit model-capability contract](../config.md#explicit-per-model-capability-declarations) preserves operator declarations through provider storage and catalog capture; it does not infer upstream capability or change this surface's routing behavior.
 
 Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior.
+
+## Zero-byte HTTP stream recovery
+
+`src/lib/upstream-retry.ts` can replace an HTTP stream once when its reader fails with a
+connection reset before the first raw response byte. Clean EOF, partial output, cancellation and
+other errors do not replay. A sent Codex WebSocket exchange or non-replayable response is excluded.
+`src/server/responses/passthrough-dispatch.ts` reuses the remaining request allowance and
+workflow accounting; `passthrough-delivery.ts` wraps the raw first leg before hosted-search,
+terminal repair or rewriting. Native Chat reuses its selected-key send path and configured
+transient policy. Both physical send boundaries recheck credential selection after pacing and
+use an HTTP-only executor; neither recovery selects a replacement account or grants a new budget.
+Cancellation aborts a pending refetch and discards a late body. Non-success, non-readable and
+content-type-incompatible replacements are cancelled without draining them into memory; the
+original stream error reaches the existing failed-tail handling. Request retention is released
+when replay becomes impossible. Zero observed bytes do not prove the origin did no work, so the
+additional request may be billable. This path is separate from the empty-completion opt-in.
+`tests/lib/upstream-retry-zero-output.test.ts` covers helper settlement, HTTP-only dispatch
+and the split-owner budget/credential wiring.
diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json
index 5553d5a7e1..afbbb90980 100644
--- a/tests/fixtures/test-layout-expected.json
+++ b/tests/fixtures/test-layout-expected.json
@@ -1298,5 +1298,6 @@
   "execution-budget-permits.test.ts": "lib",
   "spend-instrumentation-log.test.ts": "server",
   "codex-pool-refresh-backoff.test.ts": "codex-integration",
-  "responses-account-change-scrub.test.ts": "responses"
+  "responses-account-change-scrub.test.ts": "responses",
+  "upstream-retry-zero-output.test.ts": "lib"
 }
diff --git a/tests/upstream-retry-zero-output.test.ts b/tests/lib/upstream-retry-zero-output.test.ts
similarity index 65%
rename from tests/upstream-retry-zero-output.test.ts
rename to tests/lib/upstream-retry-zero-output.test.ts
index df9712a6c2..f6ac24b038 100644
--- a/tests/upstream-retry-zero-output.test.ts
+++ b/tests/lib/upstream-retry-zero-output.test.ts
@@ -1,8 +1,12 @@
+import { readFileSync } from "node:fs";
+import { repoPath } from "../helpers/repo-root";
+import { providerFetch } from "../../src/server/responses/fetch-helpers";
+import type { OcxProviderConfig } from "../../src/types";
 import { afterEach, describe, expect, spyOn, test } from "bun:test";
 import {
   refetchOnZeroOutputReset,
   wrapWithZeroOutputRefetch,
-} from "../src/lib/upstream-retry";
+} from "../../src/lib/upstream-retry";
 
 function resetError(): Error {
   // Shape of Bun's fetch rejection on a stale pooled socket.
@@ -271,3 +275,102 @@ describe("wrapWithZeroOutputRefetch", () => {
     expect(cancelled.length).toBe(1);
   });
 });
+
+
+describe("zero-output rebase boundary regressions", () => {
+  test("cancellation aborts a pending refetch and disposes its late response", async () => {
+    silenceWarn();
+    const entered = Promise.withResolvers<void>();
+    const replacement = Promise.withResolvers<Response>();
+    const discarded = Promise.withResolvers<void>();
+    let signal: AbortSignal | undefined;
+    let releases = 0;
+    const reader = wrapWithZeroOutputRefetch(failingStream(resetError()), async (_recovery, nextSignal) => {
+      signal = nextSignal;
+      entered.resolve();
+      return replacement.promise;
+    }, { onReplayUnavailable: () => { releases++; } }).getReader();
+    const pending = reader.read();
+    await entered.promise;
+    await reader.cancel("stop");
+    expect(signal?.aborted).toBe(true);
+    replacement.resolve(new Response(new ReadableStream<Uint8Array>({
+      cancel() { discarded.resolve(); },
+    }, { highWaterMark: 0 })));
+    await discarded.promise;
+    expect((await pending).done).toBe(true);
+    expect(releases).toBe(1);
+  });
+
+  test("an error replacement is cancelled rather than drained", async () => {
+    silenceWarn();
+    let reads = 0;
+    let cancels = 0;
+    const replacement = new Response(new ReadableStream<Uint8Array>({
+      pull() { reads++; },
+      cancel() { cancels++; return new Promise<void>(() => {}); },
+    }, { highWaterMark: 0 }), { status: 503 });
+    expect(await refetchOnZeroOutputReset(async () => replacement, resetError())).toBeNull();
+    expect(reads).toBe(0);
+    expect(cancels).toBe(1);
+  });
+
+  test("a JSON replacement cannot enter a committed SSE response", async () => {
+    silenceWarn();
+    const replacement = Response.json({ error: "not an SSE stream" });
+    expect(await refetchOnZeroOutputReset(async () => replacement, resetError(), {
+      acceptResponse: response => response.headers.get("content-type") === "text/event-stream",
+    })).toBeNull();
+    expect(replacement.bodyUsed).toBe(true);
+  });
+
+  test("zero remaining allowance and clean EOF never refetch", async () => {
+    silenceWarn();
+    let calls = 0;
+    const execute = async () => { calls++; return new Response("unused"); };
+    await expect(collect(wrapWithZeroOutputRefetch(failingStream(resetError()), execute, { attempts: 0 })))
+      .rejects.toThrow(/socket connection was closed/i);
+    expect(await collect(wrapWithZeroOutputRefetch(streamOf([]), execute))).toEqual([]);
+    expect(calls).toBe(0);
+  });
+
+  test("the HTTP-only executor retains physical-send hooks without dialing WebSocket", async () => {
+    const original = globalThis.WebSocket;
+    let dials = 0;
+    let sends = 0;
+    let admissions = 0;
+    let overrides = 0;
+    globalThis.WebSocket = class {
+      constructor() { dials++; throw new Error("HTTP recovery must not dial WebSocket"); }
+    } as unknown as typeof WebSocket;
+    try {
+      const provider = { fetch: (async () => { sends++; return new Response("ok"); }) as typeof fetch } as OcxProviderConfig;
+      const execute = providerFetch(provider, "1.4.0", {
+        httpOnly: true,
+        beforeDispatch: () => { admissions++; },
+        dispatchOverride: (input, init, send) => { overrides++; return send(input, init); },
+      });
+      const response = await execute("https://chatgpt.com/backend-api/codex/responses", {
+        method: "POST", body: JSON.stringify({ model: "fixture", stream: true }),
+      });
+      expect(await response.text()).toBe("ok");
+      expect({ dials, sends, admissions, overrides }).toEqual({ dials: 0, sends: 1, admissions: 1, overrides: 1 });
+    } finally { globalThis.WebSocket = original; }
+  });
+
+  test("replay stays in split owners and does not buy a fresh request allowance", () => {
+    const dispatch = readFileSync(repoPath("src/server/responses/passthrough-dispatch.ts"), "utf8");
+    const delivery = readFileSync(repoPath("src/server/responses/passthrough-delivery.ts"), "utf8");
+    const chat = readFileSync(repoPath("src/server/chat-native.ts"), "utf8");
+    const core = readFileSync(repoPath("src/server/responses/core.ts"), "utf8");
+    expect(dispatch).toContain("Math.min(1, remainingTransientSendBudget(TRANSIENT_RETRY_MAX_ATTEMPTS))");
+    expect(dispatch).toContain("onSendsConsumed: noteTransientSends");
+    expect(dispatch).toContain("transportState.selectionIsCurrent(transportState.requestBindings.get(request))");
+    expect(delivery).toContain("!isCodexWsUpstreamResponse(upstreamResponse) && !isNonReplayableResponse(upstreamResponse)");
+    expect(delivery).toContain("firstLeg: rawBody");
+    expect(chat).toContain("singleSend ? Math.min(1, remaining) : remaining");
+    expect(chat).toContain("send(activeRequest, \"connection-reset\", true, signal)");
+    expect(chat).toContain("let terminalStatus: number | undefined");
+    expect(core).not.toContain("wrapWithZeroOutputRefetch");
+  });
+});