From 603c8680f8179acf787ee197ee818867ef4a0ef8 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Wed, 16 Sep 2026 18:56:56 +0900 Subject: [PATCH 1/3] fix(retry): stop ambiguous reset replay across recovery boundaries Redesigns luvs01/opencodex#135 on upstream dev. Preserve total-send accounting and provider HTTP retry policy while carrying the terminal verdict through combo recovery and error formatting. Carried from #4621's sibling PR #4741 by a maintainer; original authorship preserved. Co-authored-by: luvs01 <27862058+luvs01@users.noreply.github.com> --- src/bridge/errors.ts | 11 +- src/lib/upstream-retry.ts | 39 ++++-- src/server/responses/adapter-dispatch.ts | 7 ++ structure/transports/responses.md | 20 +++ .../issue-914-transport-attribution.test.ts | 4 +- tests/lib/upstream-retry.test.ts | 115 +++++++++++++++++- .../upstream-transient-retry.test.ts | 4 +- .../responses-send-budget-counts.test.ts | 97 +++++++++++++++ 8 files changed, 278 insertions(+), 19 deletions(-) diff --git a/src/bridge/errors.ts b/src/bridge/errors.ts index 175e3ec451d..f26bd21fbbf 100644 --- a/src/bridge/errors.ts +++ b/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; } diff --git a/src/lib/upstream-retry.ts b/src/lib/upstream-retry.ts index 3a4fb619a37..05109e991f3 100644 --- a/src/lib/upstream-retry.ts +++ b/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( } /** - * 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. * diff --git a/src/server/responses/adapter-dispatch.ts b/src/server/responses/adapter-dispatch.ts index 2863105d7ab..c26b77b19b8 100644 --- a/src/server/responses/adapter-dispatch.ts +++ b/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 diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 6f56a276147..572ecf8bf9c 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -878,6 +878,26 @@ The actual dispatch commits their count and recovery label once; unsent pending is discarded on process exit and is not usage evidence. See [key attribution](../gui-and-management-api.md#upstream-key-account-attribution). Generic refetches record metadata inside each admitted retry callback, retaining the transient recovery reason when present and otherwise the outer recovery reason. + +## 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 diff --git a/tests/codex-integration/issue-914-transport-attribution.test.ts b/tests/codex-integration/issue-914-transport-attribution.test.ts index 6823c7b6e37..a3fdc090d0d 100644 --- a/tests/codex-integration/issue-914-transport-attribution.test.ts +++ b/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"); }); diff --git a/tests/lib/upstream-retry.test.ts b/tests/lib/upstream-retry.test.ts index 4e014c76450..4859221450f 100644 --- a/tests/lib/upstream-retry.test.ts +++ b/tests/lib/upstream-retry.test.ts @@ -1,8 +1,11 @@ +import { formatErrorResponse as formatReplaySafetyError } from "../../src/bridge/errors"; import { afterEach, describe, expect, spyOn, test } from "bun:test"; import { fetchWithResetRetry, fetchWithTransientRetry, isConnectionResetError, + isNonReplayableResponse, + UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE, prepareSameTarget429Wait, releaseResponseBodyBestEffort, retryBackoffDelayMs, @@ -148,7 +151,7 @@ describe("fetchWithResetRetry", () => { test("retries a Bun-shaped reset and returns the second attempt's response", async () => { silenceWarn(); const mock = mockDoFetch([bunResetError(), new Response("ok", { status: 200 })]); - const res = await fetchWithResetRetry(mock.doFetch, { label: "test" }); + const res = await fetchWithResetRetry(mock.doFetch, { label: "test", replaySafe: true }); expect(res.status).toBe(200); expect(await res.text()).toBe("ok"); expect(mock.calls).toHaveLength(2); @@ -161,7 +164,7 @@ describe("fetchWithResetRetry", () => { new Error("The socket connection was closed unexpectedly."), new Response("ok", { status: 200 }), ]); - const res = await fetchWithResetRetry(mock.doFetch); + const res = await fetchWithResetRetry(mock.doFetch, { replaySafe: true }); expect(res.status).toBe(200); expect(mock.calls).toHaveLength(2); }); @@ -190,7 +193,7 @@ describe("fetchWithResetRetry", () => { test("gives up after max attempts and rethrows the last reset error", async () => { silenceWarn(); const mock = mockDoFetch([bunResetError(), bunResetError(), bunResetError(), bunResetError()]); - await expect(fetchWithResetRetry(mock.doFetch)).rejects.toThrow("socket connection was closed unexpectedly"); + await expect(fetchWithResetRetry(mock.doFetch, { replaySafe: true })).rejects.toThrow("socket connection was closed unexpectedly"); expect(mock.calls).toHaveLength(3); expect(warnSpies[0]).toHaveBeenCalledTimes(2); }); @@ -207,7 +210,7 @@ describe("fetchWithResetRetry", () => { silenceWarn(); const ac = new AbortController(); const mock = mockDoFetch([bunResetError(), new Response("ok", { status: 200 })]); - const pending = fetchWithResetRetry(mock.doFetch, { abortSignal: ac.signal }); + const pending = fetchWithResetRetry(mock.doFetch, { abortSignal: ac.signal, replaySafe: true }); // First attempt rejects with a reset synchronously-ish; abort lands mid-backoff. setTimeout(() => ac.abort(new DOMException("client closed", "AbortError")), 10); await expect(pending).rejects.toThrow("client closed"); @@ -428,3 +431,107 @@ describe("prepareSameTarget429Wait", () => { expect(events.every(type => type === "heartbeat")).toBe(true); }); }); + +describe("ambiguous reset safety", () => { + test("a reset is terminal by default, even with a remaining send budget", async () => { + const reports: number[] = []; + const mock = mockDoFetch([bunResetError(), new Response("duplicate")]); + const response = await fetchWithResetRetry(mock.doFetch, { + attempts: 3, onSendsConsumed: count => reports.push(count), + }); + expect(response.status).toBe(502); + expect(isNonReplayableResponse(response)).toBe(true); + expect((await response.json()).error.code).toBe(UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE); + expect(mock.calls).toHaveLength(1); + expect(reports).toEqual([1]); + }); + + test("a 503 followed by a reset stops both retry layers and reports both sends once", async () => { + silenceWarn(); + const reports: number[] = []; + const mock = mockDoFetch([ + new Response("busy", { status: 503 }), bunResetError(), new Response("duplicate"), + ]); + const response = await fetchWithTransientRetry(mock.doFetch, { + attempts: 3, onSendsConsumed: count => reports.push(count), + }); + expect(response.status).toBe(502); + expect(isNonReplayableResponse(response)).toBe(true); + expect((await response.json()).error.code).toBe(UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE); + expect(mock.calls).toHaveLength(2); + expect(reports).toEqual([2]); + }); + + test("an exhausted last send still carries the no-replay verdict", async () => { + const mock = mockDoFetch([bunResetError()]); + const response = await fetchWithResetRetry(mock.doFetch, { attempts: 1 }); + expect(isNonReplayableResponse(response)).toBe(true); + expect(mock.calls).toHaveLength(1); + }); + + test("EPIPE and message-only resets are ambiguous too, without leaking the exception", async () => { + for (const error of [ + Object.assign(new Error("private transport detail"), { code: "EPIPE" }), + new Error("The socket connection was closed unexpectedly. private transport detail"), + ]) { + const mock = mockDoFetch([error]); + const response = await fetchWithResetRetry(mock.doFetch); + expect(isNonReplayableResponse(response)).toBe(true); + expect(await response.text()).not.toContain("private transport detail"); + expect(mock.calls).toHaveLength(1); + } + }); + + test("zero and invalid budgets never dispatch regardless of replay safety", async () => { + for (const replaySafe of [false, true]) { + for (const attempts of [0, -1, 1.5, Number.NaN, Number.POSITIVE_INFINITY]) { + const mock = mockDoFetch([new Response("must not send")]); + await expect(fetchWithResetRetry(mock.doFetch, { attempts, replaySafe })).rejects.toThrow(); + expect(mock.calls).toHaveLength(0); + } + } + }); + + test("explicitly replay-safe resets still share the total budget with 5xx", async () => { + silenceWarn(); + const reports: number[] = []; + const mock = mockDoFetch([ + bunResetError(), new Response("busy", { status: 503 }), new Response("ok"), + ]); + const response = await fetchWithTransientRetry(mock.doFetch, { + attempts: 3, replaySafe: true, onSendsConsumed: count => reports.push(count), + }); + expect(await response.text()).toBe("ok"); + expect(mock.calls).toHaveLength(3); + expect(reports).toEqual([3]); + }); +}); + +describe("ambiguous reset safety through error formatting", () => { + test("both terminal codes survive formatting without advertising Retry-After", async () => { + for (const code of ["upstream_no_response", "upstream_closed_before_response"]) { + const response = formatReplaySafetyError(502, "upstream_error", "closed", { code, retryAfter: "2" }); + expect(isNonReplayableResponse(response)).toBe(true); + expect(response.headers.get("retry-after")).toBeNull(); + expect((await response.json()).error.code).toBe(code); + } + }); + + test("unrecognized upstream codes do not override ordinary error classification", async () => { + const response = formatReplaySafetyError(502, "upstream_error", "failed", { + code: "untrusted_provider_code", retryAfter: "2", + }); + expect(isNonReplayableResponse(response)).toBe(false); + expect(response.headers.get("retry-after")).toBe("2"); + expect((await response.json()).error.code).not.toBe("untrusted_provider_code"); + }); + + test("the cyber-policy hard block retains precedence", async () => { + const response = formatReplaySafetyError(502, "upstream_error", "blocked due to high-risk cybersecurity activity", { + code: "upstream_closed_before_response", retryAfter: "2", + }); + expect(response.status).toBe(400); + expect(response.headers.get("retry-after")).toBeNull(); + expect((await response.json()).error.code).toBe("cyber_policy"); + }); +}); diff --git a/tests/providers/upstream-transient-retry.test.ts b/tests/providers/upstream-transient-retry.test.ts index 77a224eb369..abe2ccf7b3a 100644 --- a/tests/providers/upstream-transient-retry.test.ts +++ b/tests/providers/upstream-transient-retry.test.ts @@ -149,7 +149,7 @@ describe("fetchWithTransientRetry", () => { const err = new Error("socket hang up") as Error & { code?: string }; err.code = "ECONNRESET"; throw err; - }, { attempts: 3, slowAttemptMs: 60_000, onSendsConsumed: n => reported.push(n) })).rejects.toThrow(); + }, { replaySafe: true, attempts: 3, slowAttemptMs: 60_000, onSendsConsumed: n => reported.push(n) })).rejects.toThrow(); expect(reported.length).toBe(1); expect(reported[0]!).toBeGreaterThan(0); }); @@ -166,7 +166,7 @@ describe("fetchWithTransientRetry", () => { throw err; } return bodyResponse(sends === 3 ? 200 : 503); - }, { attempts: 3, slowAttemptMs: 60_000 }); + }, { replaySafe: true, attempts: 3, slowAttemptMs: 60_000 }); expect(sends).toBe(3); expect(res.status).toBe(200); diff --git a/tests/responses/responses-send-budget-counts.test.ts b/tests/responses/responses-send-budget-counts.test.ts index 1fd7635e866..35ad6cc1242 100644 --- a/tests/responses/responses-send-budget-counts.test.ts +++ b/tests/responses/responses-send-budget-counts.test.ts @@ -1,3 +1,6 @@ +import { shouldRetryCodexPoolAccountQuota, shouldRetryCodexPoolAccountTransient } from "../../src/server/responses/core-codex-account"; +import { consumeComboFailure } from "../../src/server/responses/core-combo-failure"; +import { fetchWithResetRetry, isNonReplayableResponse } from "../../src/lib/upstream-retry"; import { afterEach, beforeEach, describe, expect, test } from "bun:test"; import { clearComboSelectionState, clearComboTargetCooldowns } from "../../src/combos"; import { clearKeyCooldowns } from "../../src/providers/key-failover"; @@ -179,3 +182,97 @@ describe("upstream sends per logical request", () => { // cross-pool move are both asserted. Restoring an end-to-end row needs a harness that actually // rotates, which is its own change. }); + +describe("ambiguous reset safety across Responses recovery", () => { + for (const adapter of ["openai-chat", "openai-responses"]) { + for (const combo of [false, true]) { + test(`${adapter}: no replay or target hop after an ambiguous reset (combo=${combo})`, async () => { + const config = comboOverTargets(2); + for (const provider of Object.values(config.providers)) provider.adapter = adapter; + const authorizations: string[] = []; + globalThis.fetch = (async (_input: string | URL | Request, init?: RequestInit) => { + authorizations.push(new Headers(init?.headers).get("authorization") ?? ""); + throw Object.assign(new Error("The socket connection was closed unexpectedly."), { code: "ECONNRESET" }); + }) as typeof fetch; + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponses( + responsesRequest(combo ? "combo/fan" : "t0/model-t0"), config, logCtx, + ); + expect(response.status).toBe(502); + const payload = await response.json(); + expect(payload.error.code).toBe("upstream_closed_before_response"); + expect(authorizations).toEqual(["Bearer sk-t0"]); + expect(totalSends(logCtx)).toBe(1); + }); + } + } + + test("a provider 503 policy is retained, but the following reset cannot reach a combo sibling", async () => { + const authorizations: string[] = []; + globalThis.fetch = (async (_input: string | URL | Request, init?: RequestInit) => { + authorizations.push(new Headers(init?.headers).get("authorization") ?? ""); + if (authorizations.length === 1) { + return new Response(JSON.stringify({ error: { message: "busy" } }), { + status: 503, headers: { "content-type": "application/json" }, + }); + } + throw Object.assign(new Error("connection reset by peer"), { code: "ECONNRESET" }); + }) as typeof fetch; + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponses(responsesRequest("combo/fan"), comboOverTargets(2), logCtx); + expect(response.status).toBe(502); + expect((await response.json()).error.code).toBe("upstream_closed_before_response"); + expect(authorizations).toEqual(["Bearer sk-t0", "Bearer sk-t0"]); + expect(totalSends(logCtx)).toBe(2); + }); + + test("reset-only providers stop too, without opting into the transient policy", async () => { + const config = comboOverTargets(2); + for (const provider of Object.values(config.providers)) delete provider.transientRetryOn5xx; + let sends = 0; + globalThis.fetch = (async () => { + sends += 1; + throw Object.assign(new Error("reset"), { code: "ECONNRESET" }); + }) as typeof fetch; + const response = await handleResponses(responsesRequest("combo/fan"), config, { model: "", provider: "" }); + expect(response.status).toBe(502); + expect((await response.json()).error.code).toBe("upstream_closed_before_response"); + expect(sends).toBe(1); + }); +}); + +describe("ambiguous reset safety after outer recovery", () => { + test("a 429 recovery refetch cannot launder a subsequent reset into a combo hop", async () => { + const config = comboOverTargets(2); + config.providers.t0!.retryOn429 = { attempts: 1 }; + const authorizations: string[] = []; + globalThis.fetch = (async (_input: string | URL | Request, init?: RequestInit) => { + authorizations.push(new Headers(init?.headers).get("authorization") ?? ""); + if (authorizations.length === 1) return new Response("rate limited", { + status: 429, headers: { "retry-after": "0" }, + }); + throw Object.assign(new Error("connection reset by peer"), { code: "ECONNRESET" }); + }) as typeof fetch; + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponses(responsesRequest("combo/fan"), config, logCtx); + expect(response.status).toBe(502); + expect((await response.json()).error.code).toBe("upstream_closed_before_response"); + expect(authorizations).toEqual(["Bearer sk-t0", "Bearer sk-t0"]); + expect(totalSends(logCtx)).toBe(2); + }); + + test("account and combo recovery retain the no-replay verdict after one body read", async () => { + const response = await fetchWithResetRetry(async () => { + throw Object.assign(new Error("reset"), { code: "ECONNRESET" }); + }); + expect(shouldRetryCodexPoolAccountTransient(response)).toBe(false); + expect(await shouldRetryCodexPoolAccountQuota(response)).toBe(false); + const failure = await consumeComboFailure(response); + expect(failure.upstreamCode).toBe("upstream_closed_before_response"); + expect(isNonReplayableResponse(failure.response)).toBe(true); + expect(shouldRetryCodexPoolAccountTransient(failure.response)).toBe(false); + expect(await shouldRetryCodexPoolAccountQuota(failure.response)).toBe(false); + expect(failure.response.headers.get("retry-after")).toBeNull(); + expect((await failure.response.json()).error.code).toBe("upstream_closed_before_response"); + }); +}); From 42907a3514a06bfea823a99c21d5f40033746c27 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Wed, 16 Sep 2026 18:56:56 +0900 Subject: [PATCH 2/3] fix(retry): keep tool-sidecar reset replay and record the ambiguous reserve verdict Co-authored-by: luvs01 <27862058+luvs01@users.noreply.github.com> --- src/images/loop.ts | 2 +- src/vision/anthropic-describe.ts | 2 +- src/vision/describe.ts | 2 +- src/web-search/anthropic-executor.ts | 2 +- src/web-search/exa-executor.ts | 2 +- src/web-search/executor.ts | 2 +- src/web-search/gemini-executor.ts | 2 +- src/web-search/loop.ts | 2 +- src/web-search/ollama-executor.ts | 2 +- src/web-search/xai-executor.ts | 2 +- tests/codex-integration/reserve-dispatch.test.ts | 15 ++++++++++++--- 11 files changed, 22 insertions(+), 13 deletions(-) diff --git a/src/images/loop.ts b/src/images/loop.ts index 6f9eacad1fd..193bbef45f9 100644 --- a/src/images/loop.ts +++ b/src/images/loop.ts @@ -615,7 +615,7 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise { 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(); From 2ea335f11cdc1f60fe201ee60f4c6253850f4ef5 Mon Sep 17 00:00:00 2001 From: JUN Date: Wed, 16 Sep 2026 19:05:52 +0900 Subject: [PATCH 3/3] fix(retry): answer a refused reset replay 429 with its own code An ambiguous pre-header reset is a refusal this proxy made, not something the upstream reported. It was borrowing `upstream_closed_before_response` and its 502, and the Codex client builds its policy from `retry_429: false` / `retry_5xx: true` over `DEFAULT_REQUEST_MAX_RETRIES = 4`, so the status invited up to four more sends of the exact turn the refusal exists to protect. It now carries `upstream_reset_replay_refused` and HTTP 429, following the `request_send_budget_exhausted` precedent. The distinct code keeps it separable from the WebSocket transport's post-send 502/504 verdicts, which are unchanged, and lets `formatErrorResponse` restate the status when a combo or adapter formatter re-wraps it holding an upstream-shaped 502. Because a 429 is no longer sufficient evidence of a provider rate limit, every same-target replay, key rotation, account rotation and pool-quota recorder that keys on 429 now consults `isNonReplayableResponse` first. Without that, correcting the status would have re-created the duplicate send inside native Chat and the continuation loop and written cooldowns against credentials that refused nothing. Compact records the transport outcome rather than the client-facing status, so pool health sees exactly what it saw before. `adapter-dispatch` already had the guard at the top of its recovery loop. Rewrites the owning section of structure/transports/responses.md to separate a pre-header rejection the proxy refuses to replay from an upstream reset seen mid-stream or after a terminal, and records the reclassification as the behaviour change it is. Also repairs the stale `## Upstream reset retry` section, which still described reset retries as the default and pointed at the pre-split `src/server/responses.ts`. Co-authored-by: luvs01 <27862058+luvs01@users.noreply.github.com> --- .../docs/reference/configuration/server.md | 9 ++ src/bridge/errors.ts | 14 ++- src/lib/upstream-retry.ts | 29 +++++- src/server/chat-native.ts | 10 +- src/server/responses/adapter-continuation.ts | 8 +- src/server/responses/compact.ts | 21 ++++- src/server/responses/passthrough-dispatch.ts | 6 ++ structure/transports/responses.md | 92 ++++++++++++++----- .../reserve-dispatch.test.ts | 8 +- tests/lib/upstream-retry.test.ts | 29 ++++-- .../upstream-transient-retry.test.ts | 3 +- .../responses-send-budget-counts.test.ts | 20 ++-- 12 files changed, 195 insertions(+), 54 deletions(-) diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index 3fbf619df47..397f2803eae 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/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`: diff --git a/src/bridge/errors.ts b/src/bridge/errors.ts index f26bd21fbbf..8947eff9d75 100644 --- a/src/bridge/errors.ts +++ b/src/bridge/errors.ts @@ -1,4 +1,9 @@ -import { isNonReplayableUpstreamCode, markResponseNonReplayable } from "../lib/upstream-retry"; +import { + isNonReplayableUpstreamCode, + isReplayRefusalCode, + markResponseNonReplayable, + REPLAY_REFUSED_STATUS, +} from "../lib/upstream-retry"; import { adapterFailureFromMessage, classifyError, @@ -24,7 +29,12 @@ export function formatErrorResponse( 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; + // 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 diff --git a/src/lib/upstream-retry.ts b/src/lib/upstream-retry.ts index 05109e991f3..98e0e5d0151 100644 --- a/src/lib/upstream-retry.ts +++ b/src/lib/upstream-retry.ts @@ -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 = 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; @@ -488,9 +513,9 @@ export async function fetchWithResetRetry( // 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, + 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: 502, headers: { "content-type": "application/json" } }); + } }), { status: REPLAY_REFUSED_STATUS, headers: { "content-type": "application/json" } }); markResponseNonReplayable(response); return response; } diff --git a/src/server/chat-native.ts b/src/server/chat-native.ts index 791c687bd01..21ead2baace 100644 --- a/src/server/chat-native.ts +++ b/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(), diff --git a/src/server/responses/adapter-continuation.ts b/src/server/responses/adapter-continuation.ts index 6d4ffcfb433..fcf2bce705c 100644 --- a/src/server/responses/adapter-continuation.ts +++ b/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) diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 3d7f6557c75..73e8d0a89b6 100644 --- a/src/server/responses/compact.ts +++ b/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. diff --git a/src/server/responses/passthrough-dispatch.ts b/src/server/responses/passthrough-dispatch.ts index 65eb3512da6..19a4de0bac2 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, + 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 diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 572ecf8bf9c..352e4d46b36 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -489,7 +489,10 @@ is unchanged; only the native recovery caller supplies the exact error predicate Both shapes carry the inbound caller-abort signal separately from the turn/shutdown controller. A caller-driven read rejection is 499/client_cancel without pool penalty; -a genuine upstream reset remains synthetic 502. An already received terminal, including +a genuine upstream reset seen while reading the stream remains synthetic 502; the +pre-header case is a different verdict and is covered by +[ambiguous connection-reset replay boundary](#ambiguous-connection-reset-replay-boundary). +An already received terminal, including one completed by the error-path parser flush, retains its real outcome. Eager relays remove the caller listener when done and close signal-cancelled downstream streams even when the response-body cancel hook has not run. @@ -591,13 +594,21 @@ with the same item id. The batch/non-streaming bridge follows the same rule. `src/lib/upstream-retry.ts` guards upstream fetches against stale pooled keep-alive sockets (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. +`ECONNRESET` before any response bytes). `fetchWithResetRetry` never retries on its own +account. A reset-shaped rejection is replayed only when the caller passes `replaySafe: true`, +and then up to 3 total attempts with jittered backoff, warn-logged. Without it the rejection +becomes the terminal refusal described in +[ambiguous connection-reset replay boundary](#ambiguous-connection-reset-replay-boundary). +Reusable request bytes were never the test: a string body makes a send mechanically +repeatable, not idempotent, and a model POST is not idempotent. Timeouts, aborts, +`ECONNREFUSED`, HTTP error statuses, and mid-stream SSE failures are never retried at all. + +The opted-in callers are the sidecars, whose work is a tool call rather than a turn: the +vision describers, the web-search executors and loop, and the image loop. The model-POST +paths — native Responses passthrough, the generic adapter dispatch and its continuation loop, +compact, and native Chat — are deliberately not opted in. Adapters with their own +`fetchResponse` (kiro, cursor, google) keep their own retry policies; kiro imports the shared +abort/sleep helpers from this module. ## Console upload rejection recovery @@ -881,22 +892,55 @@ transient recovery reason when present and otherwise the outer recovery reason. ## 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. +Three failures look alike from the outside — the turn may have executed and we cannot +prove otherwise — and they are answered differently, because the status is an instruction +to the client and the client obeys it. Codex builds its retry policy from +`ApiRetryConfig { retry_429: false, retry_5xx: true, max_attempts: request_max_retries() }` +with `DEFAULT_REQUEST_MAX_RETRIES = 4`. A 5xx is therefore an invitation to send the whole +turn up to four more times, and a 429 is where the client stops. + +**A pre-header fetch rejection this proxy refuses to replay is a refusal this proxy made.** +`src/lib/upstream-retry.ts` returns a marked **429** carrying its own code, +`upstream_reset_replay_refused`. No response headers is not evidence that the model POST +was never processed, so the decision not to replay is ours, made before any response +existed — the same shape as `request_send_budget_exhausted`, and it takes the same status +for the same reason. Only an explicitly replay-safe operation opts into reset retries. + +**An upstream reset observed mid-stream or after a terminal keeps its existing behaviour.** +The passthrough read path still settles a genuine upstream reset as a synthetic 502, and the +Codex WebSocket transport still settles `upstream_closed_before_response` (socket closed +after the create frame) and `upstream_no_response` (origin never produced an event) as 502 +and 504. Those describe something the upstream did after our send, they are the contract the +public server reference already documents, and this release does not move them. + +This reclassification is the recorded behaviour change: before it, the pre-header refusal +borrowed `upstream_closed_before_response` and its 502, which multiplied the duplicate send +the refusal exists to prevent. The distinct code is what keeps the two separable afterwards — +both are non-replayable, but only one is ours to restate. + +Because the refusal now carries 429, a 429 is no longer sufficient evidence of a provider +rate limit. Every same-target replay, key rotation, account rotation and pool-quota recorder +that keys on 429 first asks `isNonReplayableResponse`: +`src/server/responses/adapter-dispatch.ts` (at the top of its recovery loop, which also +covers a reset reached by a 401/429/413 refetch), `src/server/responses/adapter-continuation.ts`, +`src/server/responses/passthrough-dispatch.ts`, `src/server/responses/compact.ts` and +`src/server/chat-native.ts`. Compact additionally records the transport outcome rather than +the client-facing status, so pool health sees exactly what it saw before the correction. +Rotating on a synthetic 429 would both re-send an inference that may already have run and +write a cooldown against a credential that refused nothing — a false signal that outlives the +request, which is the same hazard `rotateRunTurnAdapterOnPreflight429` already guards for the +send budget. + +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/bridge/errors.ts` retains only the allowlisted non-replayable transport codes, +reapplies the in-process marker, attaches no `Retry-After`, and restates 429 for the refusal +code alone so a combo or adapter formatter holding an upstream-shaped 502 cannot hand the +client back a retryable status. Other upstream codes keep the existing classification; +cyber-policy hard blocks retain precedence. The helper, formatter and public Responses count +regressions live in `tests/lib/upstream-retry.test.ts`, +`tests/responses/responses-send-budget-counts.test.ts` and +`tests/codex-integration/reserve-dispatch.test.ts`. ## Combo output headroom diff --git a/tests/codex-integration/reserve-dispatch.test.ts b/tests/codex-integration/reserve-dispatch.test.ts index a8a1834ffd7..d155927bb68 100644 --- a/tests/codex-integration/reserve-dispatch.test.ts +++ b/tests/codex-integration/reserve-dispatch.test.ts @@ -289,7 +289,9 @@ describe("Reserve dispatch-time permission", () => { // 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. + // client is not told to retry. Both therefore answer 429, and the distinct codes are what + // separate them: the revocation names the reserve, the reset names the refused replay. + // 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. @@ -305,8 +307,8 @@ describe("Reserve dispatch-time permission", () => { ? await handleResponsesCompact(request, config(), { model: "", provider: "" }, undefined, loopbackAdmission) : await handleResponses(request, config(), { model: "", provider: "" }, { admission: loopbackAdmission }); if (firstFailure === "reset") { - expect(response.status).toBe(502); - expect(await response.text()).toContain("upstream_closed_before_response"); + expect(response.status).toBe(429); + expect(await response.text()).toContain("upstream_reset_replay_refused"); } else { expect(response.status).toBe(429); expect(await response.text()).toContain("Reserve is unavailable"); diff --git a/tests/lib/upstream-retry.test.ts b/tests/lib/upstream-retry.test.ts index 4859221450f..e8d09b309e3 100644 --- a/tests/lib/upstream-retry.test.ts +++ b/tests/lib/upstream-retry.test.ts @@ -5,7 +5,7 @@ import { fetchWithTransientRetry, isConnectionResetError, isNonReplayableResponse, - UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE, + UPSTREAM_RESET_REPLAY_REFUSED_CODE, prepareSameTarget429Wait, releaseResponseBodyBestEffort, retryBackoffDelayMs, @@ -439,9 +439,11 @@ describe("ambiguous reset safety", () => { const response = await fetchWithResetRetry(mock.doFetch, { attempts: 3, onSendsConsumed: count => reports.push(count), }); - expect(response.status).toBe(502); + // 429, not 502: the Codex client is configured retry_5xx / no-retry-429, so a 5xx here + // would be re-sent four times by the caller this refusal exists to protect. + expect(response.status).toBe(429); expect(isNonReplayableResponse(response)).toBe(true); - expect((await response.json()).error.code).toBe(UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE); + expect((await response.json()).error.code).toBe(UPSTREAM_RESET_REPLAY_REFUSED_CODE); expect(mock.calls).toHaveLength(1); expect(reports).toEqual([1]); }); @@ -455,9 +457,9 @@ describe("ambiguous reset safety", () => { const response = await fetchWithTransientRetry(mock.doFetch, { attempts: 3, onSendsConsumed: count => reports.push(count), }); - expect(response.status).toBe(502); + expect(response.status).toBe(429); expect(isNonReplayableResponse(response)).toBe(true); - expect((await response.json()).error.code).toBe(UPSTREAM_CLOSED_BEFORE_RESPONSE_CODE); + expect((await response.json()).error.code).toBe(UPSTREAM_RESET_REPLAY_REFUSED_CODE); expect(mock.calls).toHaveLength(2); expect(reports).toEqual([2]); }); @@ -508,8 +510,8 @@ describe("ambiguous reset safety", () => { }); describe("ambiguous reset safety through error formatting", () => { - test("both terminal codes survive formatting without advertising Retry-After", async () => { - for (const code of ["upstream_no_response", "upstream_closed_before_response"]) { + test("every terminal code survives formatting without advertising Retry-After", async () => { + for (const code of ["upstream_no_response", "upstream_closed_before_response", "upstream_reset_replay_refused"]) { const response = formatReplaySafetyError(502, "upstream_error", "closed", { code, retryAfter: "2" }); expect(isNonReplayableResponse(response)).toBe(true); expect(response.headers.get("retry-after")).toBeNull(); @@ -517,6 +519,19 @@ describe("ambiguous reset safety through error formatting", () => { } }); + test("only the proxy-owned refusal restates the status; upstream verdicts keep theirs", async () => { + // The formatter is reached from combo and adapter paths holding an upstream-shaped 502. + // The two transport verdicts describe something upstream did and keep it; the refusal is + // this proxy's own decision and carries its own status wherever it is re-wrapped. + const refused = formatReplaySafetyError(502, "upstream_error", "closed", { + code: "upstream_reset_replay_refused", + }); + expect(refused.status).toBe(429); + for (const code of ["upstream_no_response", "upstream_closed_before_response"]) { + expect(formatReplaySafetyError(502, "upstream_error", "closed", { code }).status).toBe(502); + } + }); + test("unrecognized upstream codes do not override ordinary error classification", async () => { const response = formatReplaySafetyError(502, "upstream_error", "failed", { code: "untrusted_provider_code", retryAfter: "2", diff --git a/tests/providers/upstream-transient-retry.test.ts b/tests/providers/upstream-transient-retry.test.ts index abe2ccf7b3a..d7518fc29fd 100644 --- a/tests/providers/upstream-transient-retry.test.ts +++ b/tests/providers/upstream-transient-retry.test.ts @@ -83,9 +83,10 @@ describe("fetchWithTransientRetry", () => { expect(res.status).toBe(504); }); - test("the structured codes name exactly the two post-send verdicts", () => { + test("the structured codes name the post-send verdicts and the proxy's own refusal", () => { expect(isNonReplayableUpstreamCode("upstream_no_response")).toBe(true); expect(isNonReplayableUpstreamCode("upstream_closed_before_response")).toBe(true); + expect(isNonReplayableUpstreamCode("upstream_reset_replay_refused")).toBe(true); expect(isNonReplayableUpstreamCode("upstream_error")).toBe(false); expect(isNonReplayableUpstreamCode(undefined)).toBe(false); }); diff --git a/tests/responses/responses-send-budget-counts.test.ts b/tests/responses/responses-send-budget-counts.test.ts index 35ad6cc1242..5afee9db84f 100644 --- a/tests/responses/responses-send-budget-counts.test.ts +++ b/tests/responses/responses-send-budget-counts.test.ts @@ -198,9 +198,9 @@ describe("ambiguous reset safety across Responses recovery", () => { const response = await handleResponses( responsesRequest(combo ? "combo/fan" : "t0/model-t0"), config, logCtx, ); - expect(response.status).toBe(502); + expect(response.status).toBe(429); const payload = await response.json(); - expect(payload.error.code).toBe("upstream_closed_before_response"); + expect(payload.error.code).toBe("upstream_reset_replay_refused"); expect(authorizations).toEqual(["Bearer sk-t0"]); expect(totalSends(logCtx)).toBe(1); }); @@ -220,8 +220,8 @@ describe("ambiguous reset safety across Responses recovery", () => { }) as typeof fetch; const logCtx: RequestLogContext = { model: "", provider: "" }; const response = await handleResponses(responsesRequest("combo/fan"), comboOverTargets(2), logCtx); - expect(response.status).toBe(502); - expect((await response.json()).error.code).toBe("upstream_closed_before_response"); + expect(response.status).toBe(429); + expect((await response.json()).error.code).toBe("upstream_reset_replay_refused"); expect(authorizations).toEqual(["Bearer sk-t0", "Bearer sk-t0"]); expect(totalSends(logCtx)).toBe(2); }); @@ -235,8 +235,8 @@ describe("ambiguous reset safety across Responses recovery", () => { throw Object.assign(new Error("reset"), { code: "ECONNRESET" }); }) as typeof fetch; const response = await handleResponses(responsesRequest("combo/fan"), config, { model: "", provider: "" }); - expect(response.status).toBe(502); - expect((await response.json()).error.code).toBe("upstream_closed_before_response"); + expect(response.status).toBe(429); + expect((await response.json()).error.code).toBe("upstream_reset_replay_refused"); expect(sends).toBe(1); }); }); @@ -255,8 +255,8 @@ describe("ambiguous reset safety after outer recovery", () => { }) as typeof fetch; const logCtx: RequestLogContext = { model: "", provider: "" }; const response = await handleResponses(responsesRequest("combo/fan"), config, logCtx); - expect(response.status).toBe(502); - expect((await response.json()).error.code).toBe("upstream_closed_before_response"); + expect(response.status).toBe(429); + expect((await response.json()).error.code).toBe("upstream_reset_replay_refused"); expect(authorizations).toEqual(["Bearer sk-t0", "Bearer sk-t0"]); expect(totalSends(logCtx)).toBe(2); }); @@ -268,11 +268,11 @@ describe("ambiguous reset safety after outer recovery", () => { expect(shouldRetryCodexPoolAccountTransient(response)).toBe(false); expect(await shouldRetryCodexPoolAccountQuota(response)).toBe(false); const failure = await consumeComboFailure(response); - expect(failure.upstreamCode).toBe("upstream_closed_before_response"); + expect(failure.upstreamCode).toBe("upstream_reset_replay_refused"); expect(isNonReplayableResponse(failure.response)).toBe(true); expect(shouldRetryCodexPoolAccountTransient(failure.response)).toBe(false); expect(await shouldRetryCodexPoolAccountQuota(failure.response)).toBe(false); expect(failure.response.headers.get("retry-after")).toBeNull(); - expect((await failure.response.json()).error.code).toBe("upstream_closed_before_response"); + expect((await failure.response.json()).error.code).toBe("upstream_reset_replay_refused"); }); });