From f7160e06d6444b690f8c713d3e83772261feabed Mon Sep 17 00:00:00 2001 From: Vadym O Date: Sun, 18 Oct 2026 23:51:08 +0800 Subject: [PATCH 01/13] fix(responses): run the dispatch override on the Codex WS branch providerFetch applied `dispatchOverride` only on the HTTP path. The WebSocket branch called codexWsUpstreamFetch directly, so a key-auth WebSocket send (api.openai.com with upstreamWebsocket) never reached commitKeyAttemptSend or rebindPhysicalSend: its attempt row reported sendCount 0 for a frame that left. The WS send now runs through the same override; the override hands a marked WS executor its send directly instead of HTTP connection policy, and the HTTP fallback inside it skips the override so one logical send is counted once. Co-Authored-By: Claude Code --- src/server/responses/fetch-helpers.ts | 29 ++++++++++++++----- src/server/responses/request-transport.ts | 5 ++-- structure/transports/responses.md | 2 +- .../responses-dispatch-receipt.test.ts | 8 +++-- .../responses/responses-spend-anchor.test.ts | 2 ++ 5 files changed, 33 insertions(+), 13 deletions(-) diff --git a/src/server/responses/fetch-helpers.ts b/src/server/responses/fetch-helpers.ts index 008bbc9ec8b..ca5ab93197e 100644 --- a/src/server/responses/fetch-helpers.ts +++ b/src/server/responses/fetch-helpers.ts @@ -248,6 +248,14 @@ export function sendWithConnectionPolicy( }); } +const codexWsExecutors = new WeakSet(); +const codexWsExecutor = (send: (input: Parameters[0], init?: RequestInit) => Promise) => { + codexWsExecutors.add(send); + return send as typeof globalThis.fetch; +}; +/** True for the executor a `dispatchOverride` receives on the WS branch: it must not get HTTP-only send policy. */ +export const isCodexWsExecutor = (execute: unknown): boolean => typeof execute === "function" && codexWsExecutors.has(execute); + export interface ProviderFetchOptions { accountAutoPause?: (name: string, provider: OcxProviderConfig, headers: Headers) => AccountAutoPauseObserver | undefined; /** @@ -330,7 +338,9 @@ export function providerFetch( ) as typeof globalThis.fetch); const httpFetch = Object.assign( async (input: Parameters[0], init?: RequestInit, - reacquirePacing?: (provider: OcxProviderConfig, modelId?: string) => Promise) => { + reacquirePacing?: (provider: OcxProviderConfig, modelId?: string) => Promise, + // False only for the HTTP fallback of a WS send the override already wrapped: one logical send. + override: ProviderFetchOptions["dispatchOverride"] | null = options.dispatchOverride) => { // Refuse before any dispatch side effect where that is sound. `beforeDispatch` commits // attempt accounting and consumes admission state, so a refusal firing after it would // charge an attempt for a send that never happens, and a throwing hook would mask the @@ -340,7 +350,7 @@ export function providerFetch( // full decision can be made now. With an override, only the configured value is checked: // the override may rebuild against a different host and select a different transport, and // refusing on this destination would reject a request whose real route is fine. - if (options.dispatchOverride) egressFor(input); + if (override) egressFor(input); else providerEgressSendInit(egressBinding, transport, input); // The hook inspects the outgoing headers and refuses the send by throwing; it is not a // mutator, and the copy it receives is deliberately not threaded onward. `Connection` @@ -357,8 +367,8 @@ export function providerFetch( else if (!notified) { options.onPhysicalDispatch!(); notified = true; } } } : {}), }; - return options.dispatchOverride - ? options.dispatchOverride(input, dispatchInit, dispatch, reacquirePacing) + return override + ? override(input, dispatchInit, dispatch, reacquirePacing) : dispatch(input, dispatchInit); }, { preconnect }, @@ -397,9 +407,14 @@ export function providerFetch( options.onPhysicalDispatch?.(); }; const fallback = ((url: Parameters[0], fallbackInit?: RequestInit) => - httpFetch(url, { ...fallbackInit, [WS_FALLBACK_RECEIPT]: notifyOnce } as RequestInit, reacquirePacing)) as typeof globalThis.fetch; - return codexWsUpstreamFetch(input, init, fallback, runtime, options.onCodexWsQuota, options.beforeDispatch, options.nativeControl, - async () => { (await waitForPacing(init.signal ?? undefined))?.release(); }, notifyOnce, options.preOutputRefusalHold); + httpFetch(url, { ...fallbackInit, [WS_FALLBACK_RECEIPT]: notifyOnce } as RequestInit, reacquirePacing, null)) as typeof globalThis.fetch; + const wsSend = codexWsExecutor((url: Parameters[0], wsInit?: RequestInit) => codexWsUpstreamFetch(String(url), wsInit ?? init, + fallback, runtime, options.onCodexWsQuota, options.beforeDispatch, options.nativeControl, + async () => { (await waitForPacing(init.signal ?? undefined))?.release(); }, notifyOnce, options.preOutputRefusalHold)); + // A WS frame is a physical send like any other, so it runs the same override: that is where a + // key-auth send is committed to its attempt row and the send budget is rebound to the + // credential that left. Without it a key-auth WS turn logged `sendCount: 0`. + return options.dispatchOverride ? options.dispatchOverride(input, init, wsSend, reacquirePacing) : wsSend(input, init); } return httpFetch(input, init, reacquirePacing); }; diff --git a/src/server/responses/request-transport.ts b/src/server/responses/request-transport.ts index 42ca8b7723a..3b507b1828c 100644 --- a/src/server/responses/request-transport.ts +++ b/src/server/responses/request-transport.ts @@ -71,7 +71,7 @@ import { resolveCurrentProviderApiKeyTransport, } from "../../providers/api-key-selection"; import { resolveAdapter, resolveWireProtocolOverride } from "../adapter-resolve"; -import { physicalSendObserverInit, providerFetch, sendWithConnectionPolicy } from "./fetch-helpers"; +import { isCodexWsExecutor, physicalSendObserverInit, providerFetch, sendWithConnectionPolicy } from "./fetch-helpers"; import type { ProviderFetchOptions } from "./fetch-helpers"; import { captureConfigGeneration } from "../../lib/state-store-sweeper"; import { recordAnthropicAccountQuotaFromHeadersForInstance, hasPassiveAccountQuota } from "../../providers/quota"; @@ -798,7 +798,8 @@ export async function prepareResponsesTransport( passiveQuotaWriterGeneration = writerGeneration; } commitKeyAttemptSend(); - response = await sendWithConnectionPolicy( + // The WS executor dials its own socket: HTTP connection policy and body coding do not apply. + response = isCodexWsExecutor(execute) ? await execute(destination, dispatchInit) : await sendWithConnectionPolicy( distributedKeyPhysicalFetch(anthropicBillingPhysicalFetch(fetchImpl, route.providerName, route.provider), route.providerName, route.provider, target), destination, { ...dispatchInit, redirect: "manual", diff --git a/structure/transports/responses.md b/structure/transports/responses.md index c07f0b9e4b7..df61c3976e5 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -51,7 +51,7 @@ performs the physical send, after a dispatch override has selected or rebuilt th matching follows the URL sent on the wire rather than the URL supplied before credential revalidation; prepaid initial-send receipts follow the [spend contract](responses-spend.md#prepaid-initial-sends). At this final HTTP boundary, native ChatGPT Responses and compact JSON strings of at least 1 MiB (UTF-8) become byte buffers to avoid Bun's large-string upload resets. Content, headers, abort signals and retry policy are preserved; WebSocket selection still receives the original string. Other destinations, small strings and existing byte/stream bodies retain their representation. The one exception is the opt-in `upstreamRequestCompression` (`"gzip"` / `"zstd"`, #7247), owned by `src/server/responses/request-compression.ts`. When set, the same canonical destinations' string bodies from 1 KiB to 16 MiB are compressed at this boundary instead: the send's private header copy gains `Content-Encoding` and loses any stale `Content-Length`, and the body becomes the coded bytes. A body that already declares a coding, a non-string body (so a nested pass never codes twice), any other destination, a plugin rewrite away from `chatgpt.com`, and a codec failure all fall through to the representation above. Compression is synchronous so no await separates dispatch admission and the receipt from the physical send; the 16 MiB ceiling bounds the event-loop block. The setting is read per send from the live server config object, which `setLiveStateStoreConfig` installs at startup; detached configs never bind it. Direct-fetch sends that bypass this boundary (`agent-task-recovery.ts`, `codex/warmup.ts`) stay uncompressed. -Opt-in per-key pacing and round-robin use HTTP/SSE so their final dispatch override cannot be bypassed by a WebSocket upgrade. Transport-owned response markers and stage metadata survive owned pacing body rewraps. Replacement-key pacing admission retains client cancellation but pauses the header deadline; the full header interval restarts after admission. The wrapped executor alone is not that boundary. An override that revalidates credentials re-reads +Opt-in per-key pacing and round-robin use HTTP/SSE so their final dispatch override cannot be bypassed by a WebSocket upgrade. An unpaced WebSocket send still runs the dispatch override, so a key-auth frame is committed to its attempt row and rebinds the send budget; the override hands the marked WS executor its send directly instead of the HTTP connection policy, and the HTTP fallback inside that exchange does not run the override a second time. Transport-owned response markers and stage metadata survive owned pacing body rewraps. Replacement-key pacing admission retains client cancellation but pauses the header deadline; the full header interval restarts after admission. The wrapped executor alone is not that boundary. An override that revalidates credentials re-reads `route.provider.fetch` at send time, because reselection can install a different provider transport after the wrapper was built, and then calls that implementation instead of the executor. Both production overrides do this -- `oauthDispatch` in `request-transport.ts`, native Chat in diff --git a/tests/responses/responses-dispatch-receipt.test.ts b/tests/responses/responses-dispatch-receipt.test.ts index 53cf5bc62c5..0b10a9aa58e 100644 --- a/tests/responses/responses-dispatch-receipt.test.ts +++ b/tests/responses/responses-dispatch-receipt.test.ts @@ -105,19 +105,21 @@ test("normal WebSocket frame receives exactly one receipt", async () => { test("D7: accepted WS receipt remains charged if send and local HTTP admission both fail", async () => { globalThis.WebSocket = FakeSocket as unknown as typeof WebSocket; - let receipts = 0, refunds = 0, httpSends = 0, overrides = 0; + let receipts = 0, refunds = 0, httpSends = 0, overrides = 0, socketFailed = false; const budget = createRequestExecutionBudget(undefined, undefined, { charge: () => true, refund: () => { refunds++; }, }); const booking = budget.reserveDispatch({ sendClass: "initial", targetKey: "chatgpt/codex", countedExternally: true }); if (!booking.allowed) throw new Error("fixture booking denied"); - FakeSocket.onSend = () => { expect(receipts).toBe(1); throw new Error("fixture socket send failure"); }; + FakeSocket.onSend = () => { expect(receipts).toBe(1); socketFailed = true; throw new Error("fixture socket send failure"); }; const fetcher = providerFetch(wsProvider((async () => { httpSends++; return new Response("unexpected"); }) as typeof fetch), "1.4.0", { onPhysicalDispatch: () => { receipts++; expect(booking.permit.use()).toBe(true); }, - dispatchOverride: async () => { overrides++; throw new Error("fixture local HTTP refusal"); }, + // The override wraps the WS send once; its HTTP fallback must not re-run it (one logical send). + dispatchOverride: async (input, init, execute) => { overrides++; return execute(input, init); }, + beforeDispatch: () => { if (socketFailed) throw new Error("fixture local HTTP refusal"); }, }); await expect(fetcher(wsUrl, streamingInit())).rejects.toThrow("fixture local HTTP refusal"); booking.permit.release(); diff --git a/tests/responses/responses-spend-anchor.test.ts b/tests/responses/responses-spend-anchor.test.ts index bcab4614f34..87f05ab7069 100644 --- a/tests/responses/responses-spend-anchor.test.ts +++ b/tests/responses/responses-spend-anchor.test.ts @@ -556,6 +556,8 @@ for (const mode of ["normal", "capacity", "exhausted-limit"] as const) { expect(SpendFixtureSocket.instances[0]!.sent).toHaveLength(1); expect(JSON.parse(SpendFixtureSocket.instances[0]!.sent[0]!).type).toBe("response.create"); expect(budget.physicalStarted).toBe(1); + // The key-auth WS frame is a send, so its attempt row counts it like an HTTP send. + expect(ctx.activeAttempt?.sendCount).toBe(1); } else expect(response.status).toBe(429); tracker.settle(undefined); await ledger.waitForReporterDrain(); From 35ba6726ff049f81fca768f6faf49cfaa7f47ca4 Mon Sep 17 00:00:00 2001 From: Vadym O Date: Mon, 19 Oct 2026 00:58:30 +0800 Subject: [PATCH 02/13] fix(responses): send a dispatch-time reselection over HTTP, not the WS WS eligibility (egress, pacing, URL) is checked against the provider the WS executor was built for. The dispatch override can replace route.provider before it calls the executor; that replacement was never checked, so it could ride a WebSocket its own config rules out (e.g. an explicit proxy). The WS executor now records the provider it was admitted for and its HTTP fallback; when the override serves a different provider it sends through that fallback under the replacement's own egress route. Co-Authored-By: Claude Code --- src/server/responses/fetch-helpers.ts | 19 +++++++---- src/server/responses/request-transport.ts | 9 ++++-- .../responses/responses-spend-anchor.test.ts | 32 +++++++++++++++++++ 3 files changed, 51 insertions(+), 9 deletions(-) diff --git a/src/server/responses/fetch-helpers.ts b/src/server/responses/fetch-helpers.ts index ca5ab93197e..df3f900468b 100644 --- a/src/server/responses/fetch-helpers.ts +++ b/src/server/responses/fetch-helpers.ts @@ -248,13 +248,19 @@ export function sendWithConnectionPolicy( }); } -const codexWsExecutors = new WeakSet(); -const codexWsExecutor = (send: (input: Parameters[0], init?: RequestInit) => Promise) => { - codexWsExecutors.add(send); +interface CodexWsExecutor { provider: OcxProviderConfig; http: typeof globalThis.fetch } +const codexWsExecutors = new WeakMap(); +const codexWsExecutor = (send: (input: Parameters[0], init?: RequestInit) => Promise, binding: CodexWsExecutor) => { + codexWsExecutors.set(send, binding); return send as typeof globalThis.fetch; }; -/** True for the executor a `dispatchOverride` receives on the WS branch: it must not get HTTP-only send policy. */ -export const isCodexWsExecutor = (execute: unknown): boolean => typeof execute === "function" && codexWsExecutors.has(execute); +/** + * Set for the executor a `dispatchOverride` receives on the WS branch: it must not get HTTP-only + * send policy. `provider` is the one WS eligibility was checked against; a dispatch-time + * replacement was not, so it sends through `http` (the same fallback) instead. + */ +export const codexWsExecutorBinding = (execute: unknown): CodexWsExecutor | undefined => + typeof execute === "function" ? codexWsExecutors.get(execute) : undefined; export interface ProviderFetchOptions { accountAutoPause?: (name: string, provider: OcxProviderConfig, headers: Headers) => AccountAutoPauseObserver | undefined; @@ -410,7 +416,8 @@ export function providerFetch( httpFetch(url, { ...fallbackInit, [WS_FALLBACK_RECEIPT]: notifyOnce } as RequestInit, reacquirePacing, null)) as typeof globalThis.fetch; const wsSend = codexWsExecutor((url: Parameters[0], wsInit?: RequestInit) => codexWsUpstreamFetch(String(url), wsInit ?? init, fallback, runtime, options.onCodexWsQuota, options.beforeDispatch, options.nativeControl, - async () => { (await waitForPacing(init.signal ?? undefined))?.release(); }, notifyOnce, options.preOutputRefusalHold)); + async () => { (await waitForPacing(init.signal ?? undefined))?.release(); }, notifyOnce, options.preOutputRefusalHold), + { provider, http: markEgressTransparentExecutor(fallback) }); // A WS frame is a physical send like any other, so it runs the same override: that is where a // key-auth send is committed to its attempt row and the send budget is rebound to the // credential that left. Without it a key-auth WS turn logged `sendCount: 0`. diff --git a/src/server/responses/request-transport.ts b/src/server/responses/request-transport.ts index 3b507b1828c..c8f2433cde5 100644 --- a/src/server/responses/request-transport.ts +++ b/src/server/responses/request-transport.ts @@ -71,7 +71,7 @@ import { resolveCurrentProviderApiKeyTransport, } from "../../providers/api-key-selection"; import { resolveAdapter, resolveWireProtocolOverride } from "../adapter-resolve"; -import { isCodexWsExecutor, physicalSendObserverInit, providerFetch, sendWithConnectionPolicy } from "./fetch-helpers"; +import { codexWsExecutorBinding, physicalSendObserverInit, providerFetch, sendWithConnectionPolicy } from "./fetch-helpers"; import type { ProviderFetchOptions } from "./fetch-helpers"; import { captureConfigGeneration } from "../../lib/state-store-sweeper"; import { recordAnthropicAccountQuotaFromHeadersForInstance, hasPassiveAccountQuota } from "../../providers/quota"; @@ -758,7 +758,8 @@ export async function prepareResponsesTransport( pacingChanged = false; } if (selectionIsCurrent(requestBindings.get(wireRequest)) && copilotSessionExpiresAt > Date.now() + 1000) { - const fetchImpl = (route.provider as OcxProviderConfig & { fetch?: typeof globalThis.fetch }).fetch ?? execute; + const ws = codexWsExecutorBinding(execute); + const fetchImpl = (route.provider as OcxProviderConfig & { fetch?: typeof globalThis.fetch }).fetch ?? ws?.http ?? execute; const binding = requestBindings.get(wireRequest); const snapshot = anthropicInstance !== undefined && binding?.kind === "oauth" ? binding.snapshot : undefined; @@ -799,7 +800,9 @@ export async function prepareResponsesTransport( } commitKeyAttemptSend(); // The WS executor dials its own socket: HTTP connection policy and body coding do not apply. - response = isCodexWsExecutor(execute) ? await execute(destination, dispatchInit) : await sendWithConnectionPolicy( + // WS eligibility (egress, pacing, URL) was checked for the provider it was built for; a + // provider reselected above was not, so it takes the HTTP fallback under its own route. + response = ws?.provider === route.provider ? await execute(destination, dispatchInit) : await sendWithConnectionPolicy( distributedKeyPhysicalFetch(anthropicBillingPhysicalFetch(fetchImpl, route.providerName, route.provider), route.providerName, route.provider, target), destination, { ...dispatchInit, redirect: "manual", diff --git a/tests/responses/responses-spend-anchor.test.ts b/tests/responses/responses-spend-anchor.test.ts index 87f05ab7069..917d23b2e38 100644 --- a/tests/responses/responses-spend-anchor.test.ts +++ b/tests/responses/responses-spend-anchor.test.ts @@ -569,6 +569,38 @@ for (const mode of ["normal", "capacity", "exhausted-limit"] as const) { }, mode === "capacity" ? 1 : 8)); } +test("a key reselected at the WS dispatch boundary is not sent over the WS its predecessor was cleared for", async () => withExecutorHome(async () => { + const originalSocket = globalThis.WebSocket; + SpendFixtureSocket.instances = []; + globalThis.WebSocket = SpendFixtureSocket as unknown as typeof WebSocket; + const proxies: unknown[] = []; + globalThis.fetch = (async (_input, init) => { proxies.push((init as { proxy?: unknown } | undefined)?.proxy); return new Response("fixture", { status: 400 }); }) as typeof fetch; + // The key and its egress change together once the dispatch boundary rechecks the selection, so + // the route admitted to WS carried neither and the replacement requires an explicit proxy. + let rotated = false; + const row: Record = { adapter: "openai-responses", authMode: "key", baseUrl: "https://api.openai.com/v1", upstreamWebsocket: true }; + Object.defineProperty(row, "apiKey", { enumerable: true, get() { + rotated ||= new Error().stack!.includes("matchesConfiguredSelection"); + return rotated ? "fixture-rotated" : "fixture-only"; + } }); + Object.defineProperty(row, "proxy", { enumerable: true, get: () => rotated ? "http://proxy.fixture.test:8080" : undefined }); + const translatorBudget = createTranslatorBudget(); + try { + await (await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "P/model", input: "hello", stream: true }), + }), { port: 0, defaultProvider: "P", providers: { P: row } } as unknown as OcxConfig, { model: "", provider: "" }, + { translatorBudget, codexWsRuntimeIdentity: "1.4.0" })).text(); + expect(rotated).toBe(true); + expect(SpendFixtureSocket.instances).toHaveLength(0); + expect(proxies).toEqual(["http://proxy.fixture.test:8080/"]); + } finally { + for (const socket of SpendFixtureSocket.instances) socket.close(); + globalThis.WebSocket = originalSocket; + translatorBudget.dispose(); + } +})); + test("server stop waits for a native Messages transport reporter after listener stop", async () => withExecutorHome(async (ledger, home) => { const lifecycle = acquireSpendLedgerServerLifecycle(home); const wireStarted = Promise.withResolvers(); From fce3e5b0df426f80e16b2b0dbd6d6249c2220108 Mon Sep 17 00:00:00 2001 From: Vadym O Date: Sun, 18 Oct 2026 23:43:34 +0800 Subject: [PATCH 03/13] fix(responses): rotate a pooled API key on a native passthrough 429 The native Responses passthrough returns before the adapter recovery loop that rotates pooled keys, so a key-pool provider's 429 was returned to the client while the same refusal rotates on the adapter and chat-native wires. Rotate to the next key before output, inside the shared send budget. Distributed round-robin selection keeps its cool-without-replay behaviour. Co-Authored-By: Claude Code --- src/server/responses/passthrough-dispatch.ts | 3 ++ .../responses/passthrough-key-rotation.ts | 35 +++++++++++++++++++ structure/transports/responses-failover.md | 2 +- tests/helpers/responses-core-source.ts | 1 + tests/oauth/generic-oauth-failover.test.ts | 7 ++-- .../responses-native-key-cooldown.test.ts | 8 +++++ 6 files changed, 52 insertions(+), 4 deletions(-) create mode 100644 src/server/responses/passthrough-key-rotation.ts diff --git a/src/server/responses/passthrough-dispatch.ts b/src/server/responses/passthrough-dispatch.ts index 849001a6798..4ba9c2b9d64 100644 --- a/src/server/responses/passthrough-dispatch.ts +++ b/src/server/responses/passthrough-dispatch.ts @@ -1,5 +1,6 @@ import { recoverAutoPausedAccount } from "./account-auto-pause-recovery"; import { adoptCodexWsStage, capturePassthroughAffinity } from "./passthrough-observations"; +import { rotatePassthroughKeyOn429 } from "./passthrough-key-rotation"; import { spendLedgerStorageErrorResponse } from "./spend-storage-error"; import { requestPacingOverloadResponse } from "./pacing-overload"; import { recordDistributedResponseFailure } from "../../providers/api-key-response-ownership"; @@ -1606,6 +1607,8 @@ export async function preparePassthroughExchange( return transportFailureResponse(err); } } + const keyRotation = sendBudgetExhausted(transientSendAttempts()) ? undefined : await rotatePassthroughKeyOn429(config, route, parsed, upstreamResponse, transportState.adapter.name, () => rebuildAndRefetch("key-429")); + if (keyRotation) { if ("failed" in keyRotation) return keyRotation.failed; upstreamResponse = keyRotation; continue passthroughRecovery; } const captureAffinityResponse = ( response: Response, captureAuthCtx: CodexAuthContext = admissionState.authCtx, diff --git a/src/server/responses/passthrough-key-rotation.ts b/src/server/responses/passthrough-key-rotation.ts new file mode 100644 index 00000000000..2f02feb6370 --- /dev/null +++ b/src/server/responses/passthrough-key-rotation.ts @@ -0,0 +1,35 @@ +import { isNonReplayableResponse } from "../../lib/upstream-retry"; +import { hasKeyPoolFailover, rotateProviderTransportOn429 } from "../../providers/key-failover"; +import type { RouteResult } from "../../router"; +import type { OcxConfig, OcxParsedRequest } from "../../types"; +import { bindRouteReasoningReplayScope } from "./core-replay"; + +/** + * Key-pool failover for a native passthrough 429 before any output. This wire returns before the + * adapter loop's key-429 arm, so without it a pooled key's 429 reached the client while the same + * refusal rotates on the adapter and chat-native wires. Distributed (round-robin) selection already + * cools the key on the send and deliberately answers without a replay. A synthesized replay refusal + * is not a provider rate limit and must not cool a key. The caller checks the send budget first so + * a spent budget keeps the real 429 body. Returns undefined when nothing rotated. + */ +export async function rotatePassthroughKeyOn429( + config: OcxConfig, + route: Pick, + parsed: OcxParsedRequest, + response: Response, + adapterName: string, + refetch: () => Promise, +): Promise { + if (response.status !== 429 || isNonReplayableResponse(response) || !hasKeyPoolFailover(route.provider) + || route.provider._apiKeyAttempt?.distributed) return; + const rotated = rotateProviderTransportOn429(config, route.providerName, route.provider, { + retryAfter: response.headers.get("retry-after"), + attemptedKey: route.provider.apiKey, + promptCacheKey: parsed.options.promptCacheKey, + }); + if (!rotated) return; + try { void response.body?.cancel().catch(() => {}); } catch { /* already closed */ } + route.provider = rotated; + bindRouteReasoningReplayScope({ parsed, providerName: route.providerName, provider: rotated, adapterName }); + return refetch(); +} diff --git a/structure/transports/responses-failover.md b/structure/transports/responses-failover.md index 51d2a00c128..f9f6ef772d3 100644 --- a/structure/transports/responses-failover.md +++ b/structure/transports/responses-failover.md @@ -156,7 +156,7 @@ The shared Responses path follows the [bounded multipart recovery contract](../s ## Upstream key attempt accounting -Key identity is sealed at guarded physical dispatch after queued selections are rebuilt. Distributed selections retain their request-local entry and revision. Permanent refusal auto-pause can disable that exact entry independently of the active key; recovery selects a healthy sibling under the same pacing and physical-send bounds. A rebuild after selection drift releases the old per-key lease and reacquires admission for the refreshed key; the send guard checks again after waiting. Turn-scoped adapters follow the same rule before invocation. Native Chat, Messages and keyed compact all rebuild stale queued selections before dispatch, including a manual active-key switch under per-key limits; they release the old lease, readmit the new key and honor cancellation while its capacity is occupied. Direct compact and Images record genuine distributed 401/429 cooldowns without replaying paid POSTs; synthetic replay refusals never cool a key. Native Messages revisits mixed distributed 401/429 outcomes with its existing counters. Native Chat and Messages require the physical outgoing credential to match the selected distributed key before pooling a rejection; Messages honors the configured `x-api-key` or Bearer transport while rejecting a conflicting credential header; configured credential overrides cannot cool unused entries. Replacement admission pauses the response-header deadline and retains client cancellation; the header deadline restarts once admission completes. Native Responses captures distributed-key ownership after connection-policy rewrites in `src/providers/api-key-response-ownership.ts`; terminal 401/429 responses settle cooldown only after same-key retry policy, and a failed replay admission or synthetic replay refusal retains the last genuine refusal. Failure ownership accepts an already cooling, still-configured credential so concurrent late 401s extend earlier short cooldowns; physical-send eligibility still refuses cooling credentials. Reused responses, credential/destination overrides, stale revisions and synthetic refusals cannot confer ownership. Local pacing errors retain 429 and Retry-After at the passthrough catch boundary. Unpaced dispatch without a readmission owner fails closed. Replay notification errors release acquired slots before propagating, including pre-acquired-slot dispatch. Translated Anthropic billing recovery uses the [immutable physical-send ownership contract](../providers/anthropic-key-billing.md). Its initial send, credential hop and later mixed-class refusals share the request budget. A denied hop retains the upstream refusal, and a leased pacing body rewrap carries the existing marker rather than attributing the response to rebuilt headers. +Key identity is sealed at guarded physical dispatch after queued selections are rebuilt. Distributed selections retain their request-local entry and revision. Permanent refusal auto-pause can disable that exact entry independently of the active key; recovery selects a healthy sibling under the same pacing and physical-send bounds. A rebuild after selection drift releases the old per-key lease and reacquires admission for the refreshed key; the send guard checks again after waiting. Turn-scoped adapters follow the same rule before invocation. Native Chat, Messages and keyed compact all rebuild stale queued selections before dispatch, including a manual active-key switch under per-key limits; they release the old lease, readmit the new key and honor cancellation while its capacity is occupied. Direct compact and Images record genuine distributed 401/429 cooldowns without replaying paid POSTs; synthetic replay refusals never cool a key. Native Messages revisits mixed distributed 401/429 outcomes with its existing counters. Native Chat and Messages require the physical outgoing credential to match the selected distributed key before pooling a rejection; Messages honors the configured `x-api-key` or Bearer transport while rejecting a conflicting credential header; configured credential overrides cannot cool unused entries. Replacement admission pauses the response-header deadline and retains client cancellation; the header deadline restarts once admission completes. Native Responses captures distributed-key ownership after connection-policy rewrites in `src/providers/api-key-response-ownership.ts`; terminal 401/429 responses settle cooldown only after same-key retry policy, and a failed replay admission or synthetic replay refusal retains the last genuine refusal. A non-distributed (active-key) pool rotates a pre-output native Responses 429 to the next healthy key through `src/server/responses/passthrough-key-rotation.ts`, within the request send budget. Failure ownership accepts an already cooling, still-configured credential so concurrent late 401s extend earlier short cooldowns; physical-send eligibility still refuses cooling credentials. Reused responses, credential/destination overrides, stale revisions and synthetic refusals cannot confer ownership. Local pacing errors retain 429 and Retry-After at the passthrough catch boundary. Unpaced dispatch without a readmission owner fails closed. Replay notification errors release acquired slots before propagating, including pre-acquired-slot dispatch. Translated Anthropic billing recovery uses the [immutable physical-send ownership contract](../providers/anthropic-key-billing.md). Its initial send, credential hop and later mixed-class refusals share the request budget. A denied hop retains the upstream refusal, and a leased pacing body rewrap carries the existing marker rather than attributing the response to rebuilt headers. Before a turn-scoped retry readmits a refreshed selection, it rejects a binding already recorded as failed for this request and returns the last key failure. Eligible replacement keys still release the old lease and reacquire their own pacing capacity before invocation, rechecking after every admission. CLI-backed turns carry a synchronous selection fence through asynchronous tool setup to the physical spawn boundary. Raw adapter terminal usage is recorded before continuation, search, or image loops merge it; repeated parsing of one physical response does not count it twice. Key changes preserve the previous attempt while retaining the active attempt object shared by streaming/combo callbacks. diff --git a/tests/helpers/responses-core-source.ts b/tests/helpers/responses-core-source.ts index 43eb30964c9..1043a65cff2 100644 --- a/tests/helpers/responses-core-source.ts +++ b/tests/helpers/responses-core-source.ts @@ -55,6 +55,7 @@ export const RESPONSES_CORE_MODULES = [ "passthrough-execution.ts", "passthrough-dispatch.ts", "passthrough-observations.ts", + "passthrough-key-rotation.ts", // Reached from passthrough/adapter dispatch, compact and runTurn: local spend-ledger refusal mapping. "spend-storage-error.ts", // Reached from passthrough dispatch: local pacing refusals retain 429 and Retry-After. diff --git a/tests/oauth/generic-oauth-failover.test.ts b/tests/oauth/generic-oauth-failover.test.ts index 63a253f5c60..79adc14f434 100644 --- a/tests/oauth/generic-oauth-failover.test.ts +++ b/tests/oauth/generic-oauth-failover.test.ts @@ -571,8 +571,9 @@ describe("sidecar on429 wiring", () => { // wire/pool remain unchanged. // anthropic = 3: its existing HTTP recovery sites; the self-authenticated Claude // CLI does not authorize API-key recovery through runTurn. - // key = 7: hasKeyPoolFailover guards the three existing HTTP sites (two 429 - // response loops and the pre-stream 401 recovery site), the pre-output + // key = 8: hasKeyPoolFailover guards the three existing HTTP sites (two 429 + // response loops and the pre-stream 401 recovery site), the native + // Responses passthrough 429 arm for non-distributed pools, the pre-output // Anthropic credit-balance loop (#6958), plus runTurn preflight activation, // its explicitly permitted API-key recovery arm, and the streaming // preflight stall bound. The billing loop needs only the key rotator: an @@ -586,7 +587,7 @@ describe("sidecar on429 wiring", () => { // updating the matching number. That decision is the thing this test exists to force. expect(counts.generic).toBe(6); expect(counts.anthropic).toBe(3); - expect(counts.key).toBe(7); + expect(counts.key).toBe(8); }); test("the verify-account arm moves via auth-refusal rotation, never rate-limit", () => { diff --git a/tests/responses/responses-native-key-cooldown.test.ts b/tests/responses/responses-native-key-cooldown.test.ts index 4adac4ab92a..9852075d624 100644 --- a/tests/responses/responses-native-key-cooldown.test.ts +++ b/tests/responses/responses-native-key-cooldown.test.ts @@ -225,3 +225,11 @@ test("a synthetic replay refusal retains the preceding genuine key rejection", a expect(getKeyCooldownUntil(NAME, "k1")).toBeGreaterThan(Date.now()); unchanged(config); }); +test("native Responses rotates a non-distributed key pool on 429 before output", async () => { + const config = fixture(); delete config.providers[NAME]!.requestPacing; saveConfig(config); + intercept(headers => headers.get("authorization") === `Bearer ${KEYS[0]}` + ? Response.json({ error: { message: "fixture capacity" } }, { status: 429, headers: { "retry-after": "45" } }) : completed()); + expect((await start(config)).response.status).toBe(200); + expect(seen).toEqual([`Bearer ${KEYS[0]}`, `Bearer ${KEYS[1]}`]); + expect(getKeyCooldownUntil(NAME, "k1")).toBeGreaterThan(Date.now()); +}); From 9d27e77bd0dbcada90588b5e3c902fcddec36a92 Mon Sep 17 00:00:00 2001 From: Vadym O Date: Sun, 18 Oct 2026 23:24:15 +0800 Subject: [PATCH 04/13] fix(usage): seal a credential-rotation attempt with the status it answered When an account rotation splits the attempt row, the sealed attempt with no recorded status fell back to 401 for key-401, 429 for any *-429 kind, and 502 for everything else. An `oauth-account-403` rotation was therefore logged as a 502 server error, and `oauth-401` / `oauth-account-402` likewise, although the recovery kind names the status the account actually returned. Map those kinds to 401, 403 and 402 too. A status already recorded on the attempt still wins. Co-Authored-By: Claude Code --- src/server/request-log.ts | 4 +++- tests/usage/key-attribution.test.ts | 11 +++++++++++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/src/server/request-log.ts b/src/server/request-log.ts index 62530b72be9..dd8bbe9621d 100644 --- a/src/server/request-log.ts +++ b/src/server/request-log.ts @@ -1720,8 +1720,10 @@ export function noteProviderAttemptSend( if (attempt && attempt.sendCount > 0 && attemptAccountChanged(previous, next, attempt.provider, logCtx.provider)) { // An input estimate is not evidence that a failed send used that many tokens. delete attempt.inputTokenEstimate; + // An account-scoped recovery kind names the status the sealed account answered. finishRequestAttempt(attempt, attempt.status >= 100 ? attempt.status - : recovery === "key-401" ? 401 : recovery?.includes("429") ? 429 : 502, + : recovery === "key-401" || recovery === "oauth-401" ? 401 : recovery === "oauth-account-403" ? 403 + : recovery === "oauth-account-402" ? 402 : recovery?.includes("429") ? 429 : 502, Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now()), attempt.usage); // This attempt is being sealed because a NAMED recovery rejected it, so the recovery kind // is direct evidence here rather than an inference from history. Without this the sealed diff --git a/tests/usage/key-attribution.test.ts b/tests/usage/key-attribution.test.ts index 960963885a9..092bd9f8a2f 100644 --- a/tests/usage/key-attribution.test.ts +++ b/tests/usage/key-attribution.test.ts @@ -245,6 +245,17 @@ describe("key attempt accounting", () => { expect(active).toMatchObject({ ordinal: 2, provider: secondProvider, sendCount: 1, usage: { inputTokens: 50, outputTokens: 5 } }); }); + test.each([["oauth-account-403", 403], ["oauth-401", 401], ["oauth-account-402", 402]] as const)( + "an account rotation after %s seals the first row with the status it answered", (recovery, status) => { + const oauth = { adapter: "anthropic-messages", authMode: "oauth" as const, baseUrl: "https://example.test" }; + const active = beginRequestAttempt(1, "anthropic", "model", "anthropic-messages"); + const ctx: RequestLogContext = { provider: formatAnthropicProviderForLog("anthropic", "account-a"), + model: "model", activeAttempt: active, activeAttemptStartedAt: Date.now(), attempts: [active] }; + noteProviderAttemptSend(ctx, "anthropic", oauth, undefined); + ctx.provider = formatAnthropicProviderForLog("anthropic", "account-b"); + noteProviderAttemptSend(ctx, "anthropic", oauth, undefined, recovery); + expect(ctx.attempts?.[0]).toMatchObject({ ordinal: 1, status }); + }); test("wire snapshots replace the current send against a pre-send baseline", () => { const active = beginRequestAttempt(1, "test", "model", "openai-chat"); const ctx: RequestLogContext = { provider: "test", model: "model", activeAttempt: active, attempts: [active] }; From 339767447c0438ce0a770cec70db3b08d6d5e6a1 Mon Sep 17 00:00:00 2001 From: Vadym O Date: Mon, 19 Oct 2026 02:03:12 +0800 Subject: [PATCH 05/13] fix(usage): a leg refused before any send does not make the row unreported A multi-attempt row takes its usage status from aggregateAttemptUsage, which let any usage-less attempt mark the whole row "unreported". A combo leg the gateway refused locally (input admission, a local cooldown) never reached upstream and carries no usage, so a combo that then served the turn on its next leg with fully reported usage was still logged as unreported. The status now ignores usage-less attempts that never sent or were answered locally -- the same "spent nothing" rule summary.ts already prices by. An attempt that was sent and reported nothing still leaves the row unreported. The existing "folds partial and unsupported" test built its unreported 503 without a send; it now notes the send, which is the case it describes. Co-Authored-By: Claude Code --- src/server/request-log.ts | 14 ++++++++----- .../request-log-attempt-identity.test.ts | 21 +++++++++++++++++++ tests/usage/request-log.test.ts | 8 +++---- 3 files changed, 33 insertions(+), 10 deletions(-) diff --git a/src/server/request-log.ts b/src/server/request-log.ts index dd8bbe9621d..346029f090b 100644 --- a/src/server/request-log.ts +++ b/src/server/request-log.ts @@ -1852,16 +1852,20 @@ export function finishRequestAttempt( export function aggregateAttemptUsage( attempts: readonly PersistedUsageAttempt[], ): FinalizedUsageResult { - const status: UsageStatus = attempts.length > 0 - && attempts.every(attempt => attempt.usageStatus === "unsupported") + // A usage-less attempt that never reached upstream spent nothing (summary.ts `spentNothing`), + // so it cannot leave the row's usage unknown. + const measured = attempts.filter(attempt => attempt.usage + || (attempt.sendCount !== 0 && attempt.locallyAnswered !== true)); + const status: UsageStatus = measured.length > 0 + && measured.every(attempt => attempt.usageStatus === "unsupported") ? "unsupported" - : attempts.some(attempt => ( + : measured.some(attempt => ( attempt.usageStatus === "unreported" || attempt.usageStatus === "unsupported" )) ? "unreported" - : attempts.some(attempt => attempt.usageStatus === "estimated") + : measured.some(attempt => attempt.usageStatus === "estimated") ? "estimated" - : attempts.length > 0 + : measured.length > 0 ? "reported" : "unreported"; diff --git a/tests/usage/request-log-attempt-identity.test.ts b/tests/usage/request-log-attempt-identity.test.ts index e6ca57f4c1c..2c01b4f7c82 100644 --- a/tests/usage/request-log-attempt-identity.test.ts +++ b/tests/usage/request-log-attempt-identity.test.ts @@ -56,4 +56,25 @@ describe("request log attempt identity", () => { }, }); }); + + test("a combo leg refused before any send does not make the served row unreported", () => { + // core-combo seals an input-admission refusal like this: no send, no usage. + const refused = beginRequestAttempt(1, "first", "model-a", "openai-chat"); + finishRequestAttempt(refused, 400, 1); + const served = beginRequestAttempt(2, "second", "model-b", "openai-chat"); + noteAttemptSend(served, undefined); + finishRequestAttempt(served, 200, 8, { inputTokens: 10, outputTokens: 2 }); + + expect(refused.usageStatus).toBe("unreported"); + expect(aggregateAttemptUsage([refused, served])).toEqual({ + status: "reported", + totalTokens: 12, + usage: { inputTokens: 10, outputTokens: 2, totalTokens: 12 }, + }); + // A leg that was sent and reported nothing still leaves the total unknown. + const sentSilent = beginRequestAttempt(1, "first", "model-a", "openai-chat"); + noteAttemptSend(sentSilent, undefined); + finishRequestAttempt(sentSilent, 502, 1); + expect(aggregateAttemptUsage([sentSilent, served]).status).toBe("unreported"); + }); }); diff --git a/tests/usage/request-log.test.ts b/tests/usage/request-log.test.ts index d3e5fa34ed1..29a7c38ee1f 100644 --- a/tests/usage/request-log.test.ts +++ b/tests/usage/request-log.test.ts @@ -547,11 +547,9 @@ describe("request log metadata", () => { 1, { inputTokens: 4, outputTokens: 1 }, ); - const unreported = finishRequestAttempt( - beginRequestAttempt(2, "b", "m2", "openai-chat"), - 503, - 1, - ); + const sent = beginRequestAttempt(2, "b", "m2", "openai-chat"); + noteAttemptSend(sent, undefined); + const unreported = finishRequestAttempt(sent, 503, 1); expect(aggregateAttemptUsage([reported, unreported])).toMatchObject({ status: "unreported", usage: { inputTokens: 4, outputTokens: 1 }, From 35f68a55e59c20379f8c40570c313a271f8dfccb Mon Sep 17 00:00:00 2001 From: Vadym O Date: Mon, 19 Oct 2026 02:40:54 +0800 Subject: [PATCH 06/13] fix(usage): sum cache reads across attempts that report different fields aggregateAttemptUsage summed `cachedInputTokens` and `cacheReadInputTokens` independently. Anthropic usage sets both; an OpenAI-shaped attempt sets only `cachedInputTokens`. Row-level readers take `cacheReadInputTokens ?? cachedInputTokens` (request-log, management/shared.ts), so a row aggregating an Anthropic leg (60 read) and an OpenAI-shaped leg (40 cached) reported 60 cache reads instead of 100. Cost is priced per attempt and is not affected. Once any attempt reports `cacheReadInputTokens`, the aggregate's read field now counts each attempt's `cacheReadInputTokens ?? cachedInputTokens`. Rows whose attempts never report the read field are unchanged. Co-Authored-By: Claude Code --- src/server/request-log.ts | 5 ++++- tests/usage/request-log-attempt-identity.test.ts | 15 +++++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/src/server/request-log.ts b/src/server/request-log.ts index 346029f090b..089e0a8a77a 100644 --- a/src/server/request-log.ts +++ b/src/server/request-log.ts @@ -1882,7 +1882,10 @@ export function aggregateAttemptUsage( return present.length > 0 ? present.reduce((sum, value) => sum + value, 0) : undefined; }; const cachedInputTokens = sumOptional("cachedInputTokens"); - const cacheReadInputTokens = sumOptional("cacheReadInputTokens"); + // Readers take `cacheReadInputTokens ?? cachedInputTokens` (usage/cost.ts), so once any attempt + // reports the read field, an attempt that only reports `cachedInputTokens` must count in it too. + const cacheReadInputTokens = sumOptional("cacheReadInputTokens") === undefined ? undefined + : usages.reduce((sum, usage) => sum + (usage.cacheReadInputTokens ?? usage.cachedInputTokens ?? 0), 0); const cacheCreationInputTokens = sumOptional("cacheCreationInputTokens"); const cacheCreation1hInputTokens = sumOptional("cacheCreation1hInputTokens"); const reasoningOutputTokens = sumOptional("reasoningOutputTokens"); diff --git a/tests/usage/request-log-attempt-identity.test.ts b/tests/usage/request-log-attempt-identity.test.ts index 2c01b4f7c82..c70f0ed8951 100644 --- a/tests/usage/request-log-attempt-identity.test.ts +++ b/tests/usage/request-log-attempt-identity.test.ts @@ -77,4 +77,19 @@ describe("request log attempt identity", () => { finishRequestAttempt(sentSilent, 502, 1); expect(aggregateAttemptUsage([sentSilent, served]).status).toBe("unreported"); }); + + test("cache reads sum across attempts that report them under different fields", () => { + // An Anthropic leg reports both fields; an OpenAI-shaped leg only cachedInputTokens. + const anthropic = beginRequestAttempt(1, "anthropic", "model-a", "anthropic"); + noteAttemptSend(anthropic, undefined); + finishRequestAttempt(anthropic, 502, 1, { inputTokens: 100, outputTokens: 0, cachedInputTokens: 60, cacheReadInputTokens: 60 }); + const openai = beginRequestAttempt(2, "openai", "model-b", "openai-chat"); + noteAttemptSend(openai, undefined); + finishRequestAttempt(openai, 200, 1, { inputTokens: 100, outputTokens: 5, cachedInputTokens: 40 }); + + expect(aggregateAttemptUsage([anthropic, openai]).usage).toMatchObject({ + cachedInputTokens: 100, + cacheReadInputTokens: 100, + }); + }); }); From 10eb255d2093219a7f00db25633cf2b8aa719f4d Mon Sep 17 00:00:00 2001 From: Vadym O Date: Mon, 19 Oct 2026 01:38:42 +0800 Subject: [PATCH 07/13] fix(messages-native): answer a spent Anthropic pool as a usage limit with its reset #6991 made the bridge lane answer a 429 as "All Anthropic OAuth accounts have reached their usage limit; the first one resets at