fix(retry): refuse ambiguous reset replay without inviting a client retry (#4741)

docs-site/src/content/docs/reference/configuration/server.md

@@ -50,6 +50,15 @@ frame may already be executing upstream, so the client applies its own retry pol
 it would when connected to the backend directly. Once the response has started, a later drop
 surfaces inside the stream as before. `stallTimeoutSec` is unrelated to this window.
 
+An ordinary HTTP send has a third case. When the connection dies before any response header
+arrives, the proxy cannot tell whether the model already processed the request, so it refuses
+to send it again and answers HTTP 429 with `upstream_reset_replay_refused`. The status is
+deliberate: a 5xx here is an instruction to most clients, including Codex, to send the whole
+turn again, which is the duplicate the refusal exists to prevent. No `Retry-After` is
+attached, and the proxy performs no key rotation, account failover or same-target replay on
+it. Tool-call side requests such as vision and web search are replayed normally, because
+repeating them cannot duplicate a turn.
+
 `noProxy` accepts either a comma-separated string or an array. Both forms add entries without
 replacing an inherited `NO_PROXY`:
 

src/bridge/errors.ts

@@ -1,3 +1,9 @@
+import {
+  isNonReplayableUpstreamCode,
+  isReplayRefusalCode,
+  markResponseNonReplayable,
+  REPLAY_REFUSED_STATUS,
+} from "../lib/upstream-retry";
 import {
   adapterFailureFromMessage,
   classifyError,
@@ -18,17 +24,30 @@ export function formatErrorResponse(
     error.code = CYBER_POLICY_ERROR_CODE;
     error.type = cyberPolicyErrorType(type);
   }
-  const finalStatus = error.code === CYBER_POLICY_ERROR_CODE ? 400 : status;
+  // Only the allowlisted transport verdicts survive this formatter. Do not forward
+  // arbitrary provider codes, and preserve the existing cyber-policy precedence.
+  const replayBlocked = error.code !== CYBER_POLICY_ERROR_CODE
+    && isNonReplayableUpstreamCode(options?.code);
+  if (replayBlocked) error.code = options!.code!;
+  // The replay refusal owns its status as well as its code. A combo or adapter formatter
+  // reaches here holding the upstream-shaped status it was about to report, and inheriting
+  // that would hand the client a 5xx it is configured to retry four times.
+  const finalStatus = error.code === CYBER_POLICY_ERROR_CODE
+    ? 400
+    : isReplayRefusalCode(error.code) ? REPLAY_REFUSED_STATUS : status;
   const headers = new Headers({ "Content-Type": "application/json" });
   const retryAfter = options?.retryAfter?.trim();
   if (error.code !== CYBER_POLICY_ERROR_CODE
+    && !replayBlocked
     && retryAfter
     && retryAfter.length > 0
     && retryAfter.length <= 128) {
     headers.set("Retry-After", retryAfter);
   }
-  return new Response(JSON.stringify({ error }), {
+  const response = new Response(JSON.stringify({ error }), {
     status: finalStatus,
     headers,
   });
+  if (replayBlocked) markResponseNonReplayable(response);
+  return response;
 }

src/images/loop.ts

@@ -615,7 +615,7 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise<Respons
                   signal: headerDeadline.signal,
                 }, retryRecovery));
               },
-              { abortSignal: headerDeadline.signal, label: "image-bridge-loop" },
+              { replaySafe: true, abortSignal: headerDeadline.signal, label: "image-bridge-loop" },
             );
           }
         } finally {

src/lib/upstream-retry.ts

@@ -2,10 +2,10 @@
  * Retry guard for upstream fetches that die on stale pooled keep-alive sockets.
  *
  * chatgpt.com (Cloudflare) closes idle keep-alive connections server-side; Bun's fetch pool
- * reuses the half-closed socket and the request write fails with ECONNRESET before any
- * response bytes arrive. Retrying on a fresh connection is safe for our replayable
- * (string-body) upstream requests, because fetch() rejects only before response headers —
- * a caught error here means no response was ever received.
+ * reuses the half-closed socket and a request can fail before response headers arrive.
+ * A pre-header rejection does not prove that the origin did not process the request.
+ * Mechanically reusable bytes do not make a model POST idempotent: an ambiguous reset
+ * becomes a terminal, non-replayable response unless the operation is explicitly safe.
  *
  * 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
@@ -40,15 +40,40 @@ export function isNonReplayableResponse(response: Response): boolean {
 export const UPSTREAM_NO_RESPONSE_CODE = "upstream_no_response";
 /** Transport closed after the send, before any response event. */
 export const UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE = "upstream_closed_before_response";
+/**
+ * This proxy refused to replay a pre-header fetch rejection.
+ *
+ * Distinct from {@link UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE}, which the Codex WebSocket
+ * transport settles as a 502 after the create frame was already sent. Both are ambiguous,
+ * but only this one is a refusal this process made before any response existed, so it
+ * follows the send-budget precedent and answers 429: the Codex client is configured with
+ * `retry_429: false` and `retry_5xx: true` over four attempts, so a 5xx here multiplies
+ * the duplicate send the refusal exists to prevent. See
+ * structure/transports/responses.md#ambiguous-connection-reset-replay-boundary.
+ */
+export const UPSTREAM_RESET_REPLAY_REFUSED_CODE = "upstream_reset_replay_refused";
 const NON_REPLAYABLE_UPSTREAM_CODES: ReadonlySet<string> = new Set([
   UPSTREAM_NO_RESPONSE_CODE,
   UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE,
+  UPSTREAM_RESET_REPLAY_REFUSED_CODE,
 ]);
 
 export function isNonReplayableUpstreamCode(code: unknown): boolean {
   return typeof code === "string" && NON_REPLAYABLE_UPSTREAM_CODES.has(code);
 }
 
+/**
+ * True for the one non-replayable code this proxy owns end to end. The status it carries is
+ * a local decision, so a re-wrapping formatter must restate it rather than inherit the
+ * caller's upstream-shaped status.
+ */
+export function isReplayRefusalCode(code: unknown): boolean {
+  return code === UPSTREAM_RESET_REPLAY_REFUSED_CODE;
+}
+
+/** Client-facing status for {@link UPSTREAM_RESET_REPLAY_REFUSED_CODE}. */
+export const REPLAY_REFUSED_STATUS = 429;
+
 // 1 initial + 2 retries: the pool may hold more than one stale socket.
 const RESET_RETRY_MAX_ATTEMPTS = 3;
 const RESET_RETRY_BASE_DELAY_MS = 150;
@@ -352,6 +377,12 @@ export async function fetchWithAttemptDeadline(
 }
 
 export interface ResetRetryOptions {
+  /**
+   * Opt in only when repeating this operation cannot duplicate upstream effects.
+   * This permits reset retries, not extra sends: attempts and onSendsConsumed still
+   * bound and count every physical send. A string body is not replay-safety proof.
+   */
+  replaySafe?: boolean;
   abortSignal?: AbortSignal;
   /** Short host/path label for the retry warn log (no secrets/query strings). */
   label?: string;
@@ -442,9 +473,9 @@ export function applyUpstreamRecoveryInit<T extends RequestInit>(
 }
 
 /**
- * Run `doFetch`, retrying only connection-reset-shaped rejections (see
- * isConnectionResetError) with jittered backoff. The caller's thunk must be replay-safe
- * (string body); every retry is logged so persistent resets stay visible.
+ * Run `doFetch` within one send budget. Connection-reset-shaped rejections are
+ * terminal by default; only an explicitly replay-safe operation receives reset retries
+ * with jittered backoff. HTTP responses retain the caller's existing retry policy.
  */
 export async function fetchWithResetRetry(
   doFetch: ReplayableFetch,
@@ -475,6 +506,19 @@ export async function fetchWithResetRetry(
         if (sawReset) throw new UpstreamRetryEvidenceError([], err, true);
         throw err;
       }
+      if (opts.replaySafe !== true) {
+        // Return evidence instead of throwing a generic transport error: outer catches
+        // otherwise turn it into a replayable 502 and a combo/account recovery resends it.
+        // The WeakSet protects in-process recovery; the code survives JSON re-wrapping.
+        // Never expose the raw exception, which can contain credentials or request data.
+        const response = new Response(JSON.stringify({ error: {
+          type: "upstream_error",
+          code: UPSTREAM_RESET_REPLAY_REFUSED_CODE,
+          message: "The upstream connection closed before a response was received. The request may already have been processed; automatic replay was stopped.",
+        } }), { status: REPLAY_REFUSED_STATUS, headers: { "content-type": "application/json" } });
+        markResponseNonReplayable(response);
+        return response;
+      }
       if (attempt === attempts - 1) throw err;
       sawReset = true;
       lastError = err;
@@ -491,9 +535,9 @@ export async function fetchWithResetRetry(
 }
 
 /**
- * fetchWithResetRetry plus a transient-5xx status retry layer, PRE-STREAM only: a
- * returned Response has by definition not been relayed to the client yet, so replaying
- * the (string-body) request is safe. The failed attempt's body is cancelled before the
+ * fetchWithResetRetry plus the caller-selected transient-5xx policy, PRE-STREAM only.
+ * A received HTTP error follows that policy; an ambiguous reset's non-replayable
+ * verdict always stops it. The failed attempt's body is cancelled before the
  * retry; every returned response (ok, non-transient, aborted, slow, exhausted) keeps
  * its body intact. Honors Retry-After via retryBackoffDelayMs.
  *

src/server/chat-native.ts

@@ -25,6 +25,7 @@ import {
   applyUpstreamRecoveryInit,
   fetchWithResetRetry,
   fetchWithTransientRetry,
+  isNonReplayableResponse,
   prepareSameTarget429Wait,
   type UpstreamSendRecovery,
 } from "../lib/upstream-retry";
@@ -379,6 +380,11 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio
     let retries = 0;
     while (
       response.status === 429
+      // A 429 this proxy synthesized for a refused reset replay is not a provider rate
+      // limit: waiting and re-sending here is exactly the duplicate inference the refusal
+      // exists to stop. It kept the same shape under the old 502 only because 502 never
+      // matched this branch.
+      && !isNonReplayableResponse(response)
       && retryPolicy
       && retries < retryPolicy.attempts
       && transientSendAvailable()
@@ -392,7 +398,9 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio
       if (upstream.signal.aborted) throw upstream.signal.reason;
       response = await send(activeRequest, "rate-limit-429");
     }
-    while (response.status === 429 && hasKeyPoolFailover(activeProvider)) {
+    // Same reason as above, plus a second one: rotating here would write a cooldown against
+    // a key that rate-limited nothing, and that false signal outlives the request.
+    while (response.status === 429 && !isNonReplayableResponse(response) && hasKeyPoolFailover(activeProvider)) {
       const rotated = rotateProviderTransportOn429(config, route.providerName, activeProvider, {
         retryAfter: response.headers.get("retry-after"),
         now: Date.now(),

src/server/responses/adapter-continuation.ts

@@ -25,6 +25,7 @@ import {
   fetchWithTransientRetry,
   fetchWithResetRetry,
   applyUpstreamRecoveryInit,
+  isNonReplayableResponse,
   prepareSameTarget429Wait,
 } from "../../lib/upstream-retry";
 import { redactSecretString } from "../../lib/redact";
@@ -264,6 +265,9 @@ export function createAdapterContinuations(
       // loop; only after the attempts are exhausted does the continuation fail over.
       while (
         response.status === 429
+        // A synthesized replay refusal is not a rate limit; replaying the continuation on
+        // it would re-send a turn whose first send may already have been processed.
+        && !isNonReplayableResponse(response)
         && rateLimitPolicy !== null
         && adapterExchange.rateLimitRetries < rateLimitPolicy.attempts
         // The main recovery loop and the passthrough ladder both consult the shared remainder
@@ -311,7 +315,7 @@ export function createAdapterContinuations(
         }
       }
 
-      if (response.status === 429 && hasKeyPoolFailover(route.provider)) {
+      if (response.status === 429 && !isNonReplayableResponse(response) && hasKeyPoolFailover(route.provider)) {
         const rotated = rotateProviderTransportOn429(config, route.providerName, route.provider, {
           retryAfter: response.headers.get("retry-after"),
           now: Date.now(),
@@ -346,6 +350,7 @@ export function createAdapterContinuations(
       }
       if (
         response.status === 429
+        && !isNonReplayableResponse(response)
         && transportState.anthropicPoolAccountId
         && transportState.anthropicPoolFailovers < ANTHROPIC_POOL_MAX_FAILOVERS_PER_REQUEST
       ) {
@@ -387,6 +392,7 @@ export function createAdapterContinuations(
       // the per-request bound cannot be silently re-armed by reaching a different loop.
       if (
         response.status === 429
+        && !isNonReplayableResponse(response)
         && transportState.genericFailoverAccountId
         && transportState.genericFailovers < GENERIC_OAUTH_MAX_FAILOVERS_PER_REQUEST
         && isGenericOAuthFailoverEnabled(config, route.providerName)

src/server/responses/adapter-dispatch.ts

@@ -1,3 +1,4 @@
+import { isNonReplayableResponse } from "../../lib/upstream-retry";
 import type { ResponsesRequestContext, ResponsesAdmissionState } from "./core-options";
 import type { PreparedResponsesRequest } from "./request-prepare";
 import type { ResponsesTransport } from "./request-transport";
@@ -526,6 +527,12 @@ export async function prepareAdapterExchange(
     };
     // Keep recovery kinds in sync with the native Responses `passthroughRecovery:` loop above.
     recovery: for (;;) {
+      // Preserve the terminal verdict through adapter and combo error formatting.
+      // This also covers a reset reached by a 401/429/413 recovery refetch.
+      if (isNonReplayableResponse(upstreamResponse)) {
+        cleanupUpstreamAbort();
+        return upstreamResponse;
+      }
       if (
         upstreamResponse.status === 401
         && isOAuth401ReplayProvider

src/server/responses/compact.ts

@@ -85,6 +85,7 @@ import {
   fetchWithResetRetry,
   fetchWithTransientRetry,
   applyUpstreamRecoveryInit,
+  isNonReplayableResponse,
   SendBudgetExhaustedError,
   TRANSIENT_RETRY_MAX_ATTEMPTS,
   type UpstreamSendRecovery,
@@ -1079,6 +1080,10 @@ export async function handleResponsesCompact(
     // — reporting exhausted retries while another pool account sat idle (#913).
     if (
       (upstream.status === 429 || upstream.status === 402)
+      // A replay refusal this proxy synthesized carries 429 for the client's benefit only.
+      // It is not pool quota evidence, and the alternate account below is another send of a
+      // compact turn that may already have been processed.
+      && !isNonReplayableResponse(upstream)
       && !storedPool401ReplayAttempted
       && usesCodexForwardPoolAuth(authCtx, route.provider)
       && !authCtx.fixedAccount
@@ -1211,8 +1216,14 @@ export async function handleResponsesCompact(
     const bufferedErrorText = buffered.ok
       ? ""
       : await buffered.clone().text().catch(() => "");
-    const explicitQuotaStatus = buffered.status === 429 || buffered.status === 402;
-    const bodyInferredQuota = !buffered.ok
+    // The client-facing 429 of a synthesized replay refusal says nothing about this
+    // account's quota. Pool accounting keeps reading it as the transport failure it is,
+    // which is also what it recorded before the status was corrected for the client.
+    const replayRefused = isNonReplayableResponse(upstream);
+    const explicitQuotaStatus = !replayRefused
+      && (buffered.status === 429 || buffered.status === 402);
+    const bodyInferredQuota = !replayRefused
+      && !buffered.ok
       && !explicitQuotaStatus
       && isRateLimitOrQuotaFailureMessage(bufferedErrorText);
     const quotaFailure = explicitQuotaStatus || bodyInferredQuota;
@@ -1225,7 +1236,11 @@ export async function handleResponsesCompact(
     // A body-confirmed quota failure can arrive behind a generic 5xx. Record it as
     // quota evidence; otherwise preserve the real upstream status so a local buffering
     // failure after a 200 cannot soft-avoid a healthy account or rotate a thread.
-    recordCompactPoolOutcome(outcomeCtx, bodyInferredQuota ? 429 : upstream.status, { retryAfter, resetAt });
+    recordCompactPoolOutcome(
+      outcomeCtx,
+      bodyInferredQuota ? 429 : replayRefused ? 502 : upstream.status,
+      { retryAfter, resetAt },
+    );
     // Lift usage and response metadata from the buffered upstream JSON into the
     // request log; the routed branch gets the same through handleResponses. The
     // synthetic buffer errors are not upstream bodies and stay uninspected.

src/server/responses/passthrough-dispatch.ts

@@ -102,6 +102,7 @@ import {
   fetchWithTransientRetry,
   applyUpstreamRecoveryInit,
   TRANSIENT_RETRY_MAX_ATTEMPTS,
+  isNonReplayableResponse,
   prepareSameTarget429Wait,
   sleepWithAbort,
 } from "../../lib/upstream-retry";
@@ -1117,6 +1118,10 @@ export async function preparePassthroughExchange(
     // the same quorum, cooldown and request budget here, before any client bytes flow.
     if (
       upstreamResponse.status === 429
+      // Not a provider rate limit when this proxy synthesized it for a refused reset
+      // replay; rotating accounts on it would re-send an inference that may already
+      // have run and would cool down an account that refused nothing.
+      && !isNonReplayableResponse(upstreamResponse)
       && transportState.genericFailoverAccountId
       && transportState.genericFailovers < GENERIC_OAUTH_MAX_FAILOVERS_PER_REQUEST
       && isGenericOAuthFailoverEnabled(config, route.providerName)
@@ -1171,6 +1176,7 @@ export async function preparePassthroughExchange(
     // keep their pool logic below (rateLimitRetryPolicyFor returns null for them).
     while (
       upstreamResponse.status === 429
+      && !isNonReplayableResponse(upstreamResponse)
       && rateLimitPolicy !== null
       && rateLimitRetries < rateLimitPolicy.attempts
       // Checked here rather than inside the helper: prepareSameTarget429Wait releases the 429

src/vision/anthropic-describe.ts

@@ -205,7 +205,7 @@ export async function describeImageAnthropic(
         body: JSON.stringify(body),
         signal: linkedSignal.signal,
       }, recovery)),
-      { abortSignal: linkedSignal.signal, label: "vision-sidecar-anthropic" },
+      { replaySafe: true, abortSignal: linkedSignal.signal, label: "vision-sidecar-anthropic" },
     );
     if (!res.ok) {
       // The body is untrusted and only feeds one auth-failure message, so read a bounded prefix.

src/vision/describe.ts

@@ -108,7 +108,7 @@ export async function describeImage(
         // `session_id`, and `x-codex-turn-metadata` to the redirect target.
         redirect: "manual",
       }, recovery)),
-      { abortSignal: linkedSignal.signal, label: "vision-sidecar" },
+      { replaySafe: true, abortSignal: linkedSignal.signal, label: "vision-sidecar" },
     );
     const detachBodyGuard = cancelBodyOnAbort(res.body, linkedSignal.signal);
     try {

src/web-search/anthropic-executor.ts

@@ -215,7 +215,7 @@ export async function runAnthropicWebSearch(
         body: JSON.stringify(body),
         signal: linkedSignal.signal,
       }, recovery)),
-      { abortSignal: linkedSignal.signal, label: "web-search-sidecar-anthropic" },
+      { replaySafe: true, abortSignal: linkedSignal.signal, label: "web-search-sidecar-anthropic" },
     );
     // Guard before any branch reads the body: the failure branch's `res.text()` ran ahead of
     // the success-path guard, reopening the fetch-resolution-to-reader-attach race

src/web-search/exa-executor.ts

@@ -46,7 +46,7 @@ export async function runExaWebSearch(
         signal: linkedSignal.signal,
         redirect: "manual",
       }, recovery)),
-      { abortSignal: linkedSignal.signal, label: "exa-web-search-sidecar" },
+      { replaySafe: true, abortSignal: linkedSignal.signal, label: "exa-web-search-sidecar" },
     );
     const detachBodyGuard = cancelBodyOnAbort(res.body, linkedSignal.signal);
     try {