feat(retry): refetch once on zero-output mid-stream socket reset

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` 설정과는 별개입니다.

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.

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",

src/lib/upstream-retry.ts

@@ -8,13 +8,15 @@
  * 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).
  */
 import { clearableDeadline } from "./abort";
+import { redactSecretString } from "./redact";
 
 /**
  * Responses the origin may already be executing. RFC 9110 §9.2.2 forbids an intermediary
@@ -594,3 +596,152 @@ export async function fetchWithTransientRetry(
     opts.onSendsConsumed?.(sent);
   }
 }
+
+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: ZeroOutputReplayFetch,
+  err: unknown,
+  opts: ZeroOutputRefetchOptions = {},
+): Promise<Response | 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", opts.abortSignal);
+  } catch {
+    console.warn("[upstream-retry] zero-output refetch failed" + label + "; preserving original stream error");
+    return null;
+  }
+  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 mid-stream reset" + label + "; using one replacement stream");
+  return replacement;
+}
+
+/**
+ * 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: 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 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) {
+      while (!closed) {
+        const current = reader;
+        try {
+          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 (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) {
+              reader = replacement.body.getReader();
+              continue;
+            }
+          }
+          closed = true;
+          releaseReplay();
+          detach();
+          retireReader(current, err);
+          controller.error(err);
+          output = undefined;
+          return;
+        }
+      }
+    },
+    cancel(reason) {
+      if (closed) return;
+      closed = true;
+      replayAbort.abort(reason);
+      releaseReplay();
+      detach();
+      retireReader(reader, reason);
+      output = undefined;
+    },
+  }, { highWaterMark: 0 });
+}

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,22 +359,22 @@ 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;
               },
             }),
           );
         },
         {
-          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, {

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`

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,