fix(retry): stop ambiguous reset replay across recovery boundaries

src/bridge/errors.ts

@@ -1,3 +1,4 @@
+import { isNonReplayableUpstreamCode, markResponseNonReplayable } from "../lib/upstream-retry";
 import {
   adapterFailureFromMessage,
   classifyError,
@@ -18,17 +19,25 @@ export function formatErrorResponse(
     error.code = CYBER_POLICY_ERROR_CODE;
     error.type = cyberPolicyErrorType(type);
   }
+  // 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!;
   const finalStatus = error.code === CYBER_POLICY_ERROR_CODE ? 400 : 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
@@ -352,6 +352,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 +448,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 +481,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_CLOSED_BEFORE_RESPONSE_CODE,
+          message: "The upstream connection closed before a response was received. The request may already have been processed; automatic replay was stopped.",
+        } }), { status: 502, headers: { "content-type": "application/json" } });
+        markResponseNonReplayable(response);
+        return response;
+      }
       if (attempt === attempts - 1) throw err;
       sawReset = true;
       lastError = err;
@@ -491,9 +510,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/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";
@@ -505,6 +506,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/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 {

src/web-search/executor.ts

@@ -94,7 +94,7 @@ export async function runWebSearch(
         // `session_id`, and `x-codex-turn-metadata` to the redirect target.
         redirect: "manual",
       }, recovery), forwardProvider)),
-      { abortSignal: linkedSignal.signal, label: "web-search-sidecar" },
+      { replaySafe: true, abortSignal: linkedSignal.signal, label: "web-search-sidecar" },
     );
     // Attach the body guard before ANY branch reads it. The success path guarded itself below,
     // but the failure branch's `res.text()` runs first, so a cancel landing between fetch

src/web-search/gemini-executor.ts

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

src/web-search/loop.ts

@@ -512,7 +512,7 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise<Respons
                   signal: headerDeadline.signal,
                 }, retryRecovery));
               },
-              { abortSignal: headerDeadline.signal, label: "web-search-loop" },
+              { replaySafe: true, abortSignal: headerDeadline.signal, label: "web-search-loop" },
             );
           }
         } finally {

src/web-search/ollama-executor.ts

@@ -57,7 +57,7 @@ export async function runOllamaWebSearch(
         // Bun forwards custom headers across redirects, so a redirect would leak the key.
         redirect: "manual",
       }, recovery)),
-      { abortSignal: linkedSignal.signal, label: "ollama-web-search-bridge" },
+      { replaySafe: true, abortSignal: linkedSignal.signal, label: "ollama-web-search-bridge" },
     );
     const detachBodyGuard = cancelBodyOnAbort(res.body, linkedSignal.signal);
     try {

src/web-search/xai-executor.ts

@@ -110,7 +110,7 @@ export async function runXaiWebSearch(
         // Credential-bearing: never follow a redirect off the pinned origin.
         redirect: "manual",
       }, recovery)),
-      { abortSignal: linkedSignal.signal, label: "xai-web-search-sidecar" },
+      { replaySafe: true, abortSignal: linkedSignal.signal, label: "xai-web-search-sidecar" },
     );
     const detachBodyGuard = cancelBodyOnAbort(res.body, linkedSignal.signal);
     if (!res.ok) {

structure/transports/responses.md

@@ -756,6 +756,25 @@ nor releases. That is not a lost send; it is a send the request never made, spen
 later recovery in the same request then cannot have. `tests/lib/execution-budget-permits.test.ts`
 pins both ladder shapes against exactly that.
 
+## Ambiguous connection-reset replay boundary
+
+`src/lib/upstream-retry.ts` returns a marked 502 with
+`upstream_closed_before_response` when a fetch rejects with an ambiguous connection
+reset. No response headers is not evidence that a model POST was never processed.
+Only an explicitly replay-safe operation opts into reset retries. The existing
+provider HTTP-status policy and the shared physical-send budget remain independent:
+zero refuses dispatch, invalid counts fail, and a stopped send is counted once.
+`src/server/responses/adapter-dispatch.ts` preserves this verdict instead of formatting
+it as a generic replayable upstream error. The existing account and combo guards stop
+on the marker or structured code. The helper and public Responses count regressions
+live in `tests/lib/upstream-retry.test.ts` and
+`tests/responses/responses-send-budget-counts.test.ts`.
+
+`src/bridge/errors.ts` retains only the two allowlisted non-replayable transport
+codes, reapplies the in-process marker, and does not attach `Retry-After` to them.
+Other upstream codes keep the existing classification; cyber-policy hard blocks
+retain precedence. The combo, 429-refetch, and account-guard tests cover this boundary.
+
 ## Combo output headroom
 
 A combo child is admitted against two budgets, not one. `resolveInputCeiling` in

tests/codex-integration/issue-914-transport-attribution.test.ts

@@ -147,14 +147,14 @@ describe("issue #914 — pre-connection failures never touch account health", ()
     const err = await fetchWithResetRetry(async recovery => {
       if (!recovery) throw coded("reset", "ECONNRESET");
       throw rejection;
-    }).catch((e: unknown) => e);
+    }, { replaySafe: true }).catch((e: unknown) => e);
     expect(classifyTransportFailureKind(err)).toBe("connect_error");
   });
 
   test("a plain reachability rejection classifies neutral end to end", async () => {
     const err = await fetchWithTransientRetry(async () => {
       throw coded("refused", "ECONNREFUSED");
-    }).catch((e: unknown) => e);
+    }, { replaySafe: true }).catch((e: unknown) => e);
     expect(classifyTransportFailureKind(err)).toBe("connect_neutral");
   });
 

tests/codex-integration/reserve-dispatch.test.ts

@@ -286,7 +286,11 @@ describe("Reserve dispatch-time permission", () => {
 
   for (const endpoint of ["responses", "compact"] as const) {
     for (const firstFailure of ["reset", "502"] as const) {
-      test(`${endpoint}: ${firstFailure} then revoked proof maps to429 without a second inference or health mutation`, async () => {
+      // A received 502 proves the request reached the origin and was answered, so a revocation
+      // observed afterwards is authoritative and maps to the local 429. A connection reset proves
+      // nothing: the inference may already have run, so the ambiguous-reset verdict wins and the
+      // client is not told to retry. Neither case may send a second inference or mutate health.
+      test(`${endpoint}: ${firstFailure} then revoked proof is terminal without a second inference or health mutation`, async () => {
         inference = () => {
           // Permission changes after the first real attempt, before the retry wrapper dispatches.
           revoke();
@@ -300,8 +304,13 @@ describe("Reserve dispatch-time permission", () => {
         const response = endpoint === "compact"
           ? await handleResponsesCompact(request, config(), { model: "", provider: "" }, undefined, loopbackAdmission)
           : await handleResponses(request, config(), { model: "", provider: "" }, { admission: loopbackAdmission });
-        expect(response.status).toBe(429);
-        expect(await response.text()).toContain("Reserve is unavailable");
+        if (firstFailure === "reset") {
+          expect(response.status).toBe(502);
+          expect(await response.text()).toContain("upstream_closed_before_response");
+        } else {
+          expect(response.status).toBe(429);
+          expect(await response.text()).toContain("Reserve is unavailable");
+        }
         expect(inferenceSends).toBe(1);
         expect(usageReads).toBe(1);
         expect(getCodexUpstreamHealth("__main__")).toBeNull();