diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index ab3bed226da..4640fd80a2f 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -744,7 +744,7 @@ "combo-stream-preflight.test.ts": "routing", "combo-workspace-data.test.ts": "gui", "combo-workspace-jev-decision.test.ts": "gui", - "combos.test.ts": "codex-integration", + "combos.test.ts": "codex-integration", "combos-kiro-profile-hop.test.ts": "codex-integration", "command-code-error-finish.test.ts": "providers", "command-code-efforts.test.ts": "providers", "command-code-project-context.test.ts": "providers", diff --git a/src/adapters/kiro-refusal.ts b/src/adapters/kiro-refusal.ts index 21fdb55317b..fbfc0ee784c 100644 --- a/src/adapters/kiro-refusal.ts +++ b/src/adapters/kiro-refusal.ts @@ -1,8 +1,13 @@ /** Account-scoped Kiro refusals. Unknown data never convicts an account. */ import { BOUNDED_BODY_MAX_BYTES } from "../lib/bounded-body"; -export type KiroRefusalKind = "rate" | "monthly_quota" | "suspended" | "other"; +export type KiroRefusalKind = "rate" | "monthly_quota" | "suspended" | "profile" | "other"; export interface KiroRefusal { kind: KiroRefusalKind; resetAt?: number } +/** The vendor's own name for the body validator that rejects a request for this identity. */ +const VALIDATION_EXCEPTION = "ValidationException"; +/** Kiro's wording for that rejection; the code alone is also what a genuinely bad payload gets. */ +const IMPROPER_REQUEST_WORD = /improperly formed request/i; + const SUSPENSION_WORDS = [ "temporarily suspended", "temporarily is suspended", "locked your account", "locked it as a", @@ -22,9 +27,44 @@ export function classifyKiroRefusal(status: number, bodyText: string): KiroRefus } catch { /* Unrecognised bodies cannot establish an account verdict. */ } if ((status === 400 || status === 429) && reason === "MONTHLY_REQUEST_COUNT") return { kind: "monthly_quota" }; + // A 400 whose own validator calls the body improperly formed is not proof that the request is + // impossible: Kiro answers this shape when the identity a request was routed under cannot serve + // it. Treating it as `other` leaves the account selected and the same 400 reaches the client. + // Both vendor tokens are required, so a bad payload or a caller echoing one token stays terminal. + if (status === 400 && typeof message === "string" && IMPROPER_REQUEST_WORD.test(message) + && (reason === VALIDATION_EXCEPTION || message.includes(VALIDATION_EXCEPTION))) + return { kind: "profile" }; if (status === 403 && (reason === "TEMPORARILY_SUSPENDED" || (typeof message === "string" && SUSPENSION_WORDS.some(word => message.toLowerCase().includes(word))))) return { kind: "suspended" }; if (status === 429) return { kind: "rate" }; return { kind: "other" }; } + +/** The adapter's own prefix for a normalized 400 (`normalizedKiroErrorMessage` in kiro-errors.ts). */ +const KIRO_INVALID_REQUEST_PREFIX = "Kiro invalid request: "; + +/** + * The same verdict at the combo layer. There the text is either the adapter's normalized message + * ("Kiro invalid request: ValidationException: Improperly formed request", what a combo child's + * failure carries) or a raw JSON body, each possibly behind the "Provider error 400: " frame. + * JSON goes through the parsed `reason`/`message` rules of `classifyKiroRefusal` (tokens in + * unrelated fields never count); prose counts only behind the adapter's own prefix. Callers must + * also know the failed target is a Kiro adapter. + */ +export function isKiroAccountProfileRefusal(status: number, bodyText: string): boolean { + if (status !== 400 || Buffer.byteLength(bodyText, "utf8") > BOUNDED_BODY_MAX_BYTES) return false; + const text = bodyText.replace(/^Provider error \d{3}: /, ""); + if (text.startsWith(KIRO_INVALID_REQUEST_PREFIX)) { + const detail = text.slice(KIRO_INVALID_REQUEST_PREFIX.length); + return detail.includes(VALIDATION_EXCEPTION) && IMPROPER_REQUEST_WORD.test(detail); + } + try { + const row: unknown = JSON.parse(text); + if (!row || typeof row !== "object" || Array.isArray(row)) return false; + const envelope = (row as Record).error; + const body = envelope && typeof envelope === "object" && !Array.isArray(envelope) ? envelope : row; + const { reason, message } = body as Record; + return classifyKiroRefusal(400, JSON.stringify({ reason, message })).kind === "profile"; + } catch { return false; } +} diff --git a/src/combos/failover.ts b/src/combos/failover.ts index a586ccb1c58..0bdcc0ece2b 100644 --- a/src/combos/failover.ts +++ b/src/combos/failover.ts @@ -2,6 +2,7 @@ import { parseResetCooldownMs } from "../codex/routing"; import { classifyError, isCyberPolicyCode } from "../lib/errors"; import { isNonReplayableUpstreamCode } from "../lib/upstream-retry"; import { isAnthropicBillingFailure } from "../providers/anthropic-billing-refusal"; +import { isKiroAccountProfileRefusal } from "../adapters/kiro-refusal"; import type { OcxComboTarget } from "../types"; import { targetKey } from "./types"; import { @@ -772,7 +773,7 @@ export function codexAccountModelRefusal( export function comboFailureDecision( status: number, message: string, - options?: { code?: string | null; codexModelRefusal?: CodexAccountModelRefusal }, + options?: { code?: string | null; codexModelRefusal?: CodexAccountModelRefusal; failedAdapter?: string }, ): ComboFailureDecision { if (status === 499) return "stop"; if (message.toLowerCase().includes("origin_rejected")) return "stop"; @@ -856,6 +857,12 @@ export function comboFailureDecision( // could have served the request (#4903). if (isResponseFormatCapabilityRefusal(status, message, options?.code)) return "hop"; if (isAnthropicBillingFailure(status, message, options?.code)) return "hop"; + // Kiro's body validator rejects the request for the identity it was routed under. The account + // layer rotates logins first; this is the no-alternate case, so the next combo target is the + // only remaining chance to serve the turn. Only a caller that knows the failed target is a Kiro + // adapter grants it (the words alone prove nothing for another provider); it must precede the + // generic invalid_request_error stop. + if (options?.failedAdapter === "kiro" && isKiroAccountProfileRefusal(status, message)) return "hop"; if (["origin_rejected", "context_length_exceeded", "invalid_request_error"].includes(error.code ?? "")) { return "stop"; } diff --git a/src/server/messages-native.ts b/src/server/messages-native.ts index 192a8d92d23..1c4b9fd5ac0 100644 --- a/src/server/messages-native.ts +++ b/src/server/messages-native.ts @@ -849,6 +849,17 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions) } releaseRetainedRequest(); + // The last pool account refused on spent quota and rotation found no replacement: answer the + // usage limit with its reset (#6991), as the bridge lane does, not the upstream "try again later". + const poolUsageLimit = response.status === 429 && oauthBinding + ? anthropicRoutingFor(nativeInstance!).anthropicPoolUsageLimitAfterRefusal(Date.now(), oauthBinding.routeDecision, route.modelId) + : null; + if (poolUsageLimit) { + discard(response); + cleanupAbort(); + logCtx.errorCode = "usage_limit_reached"; + return retainUpstreamMessagesRequestId(localOAuthFailure(poolUsageLimit)!, response.headers); + } if (!response.ok) { let bodyText = ""; try { diff --git a/src/server/request-log.ts b/src/server/request-log.ts index 62530b72be9..089e0a8a77a 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 @@ -1850,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"; @@ -1876,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/src/server/responses/core-combo.ts b/src/server/responses/core-combo.ts index f390955f010..64e8d7943dc 100644 --- a/src/server/responses/core-combo.ts +++ b/src/server/responses/core-combo.ts @@ -1165,6 +1165,9 @@ export async function executeComboResponses( : comboFailureDecision(failure.response.status, failure.classificationText, { code: failure.upstreamCode, codexModelRefusal: failure.codexModelRefusal, + // The adapter the failed leg actually resolved: a configured `kiro` provider can route a + // model through another wire via `modelAdapters`, and an unresolved leg grants no hop. + failedAdapter: childLog.providerAdapter, }); const wantsStream = (rawBody as { stream?: unknown } | null)?.stream === true; // Local byte admission has its own diagnostic; do not relabel it as an upstream refusal. diff --git a/src/server/responses/fetch-helpers.ts b/src/server/responses/fetch-helpers.ts index 008bbc9ec8b..df3f900468b 100644 --- a/src/server/responses/fetch-helpers.ts +++ b/src/server/responses/fetch-helpers.ts @@ -248,6 +248,20 @@ export function sendWithConnectionPolicy( }); } +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; +}; +/** + * 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; /** @@ -330,7 +344,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 +356,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 +373,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 +413,15 @@ 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), + { 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`. + return options.dispatchOverride ? options.dispatchOverride(input, init, wsSend, reacquirePacing) : wsSend(input, init); } return httpFetch(input, init, reacquirePacing); }; diff --git a/src/server/responses/passthrough-dispatch.ts b/src/server/responses/passthrough-dispatch.ts index 849001a6798..e33d91206e2 100644 --- a/src/server/responses/passthrough-dispatch.ts +++ b/src/server/responses/passthrough-dispatch.ts @@ -1,4 +1,4 @@ -import { recoverAutoPausedAccount } from "./account-auto-pause-recovery"; +import { recoverAutoPausedAccount, rotatePassthroughKeyOn429 } from "./passthrough-key-rotation"; import { adoptCodexWsStage, capturePassthroughAffinity } from "./passthrough-observations"; import { spendLedgerStorageErrorResponse } from "./spend-storage-error"; import { requestPacingOverloadResponse } from "./pacing-overload"; @@ -1606,6 +1606,7 @@ export async function preparePassthroughExchange( return transportFailureResponse(err); } } + const keyRotation = await rotatePassthroughKeyOn429({ config, route, parsed, response: upstreamResponse, adapterName: transportState.adapter.name, budgetSpent: () => sendBudgetExhausted(transientSendAttempts()), refetch: () => 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..0314159f65d --- /dev/null +++ b/src/server/responses/passthrough-key-rotation.ts @@ -0,0 +1,40 @@ +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"; + +// Passthrough's credential moves share one import site so passthrough-dispatch.ts stays under its line cap. +export { recoverAutoPausedAccount } from "./account-auto-pause-recovery"; + +/** + * 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. A spent send budget is checked before any + * key is cooled, so it keeps the real 429 body. Returns undefined when nothing rotated. + */ +export async function rotatePassthroughKeyOn429(options: { + config: OcxConfig; + route: Pick; + parsed: OcxParsedRequest; + response: Response; + adapterName: string; + budgetSpent: () => boolean; + refetch: () => Promise; +}): Promise { + const { config, route, parsed, response, adapterName, refetch } = options; + if (response.status !== 429 || isNonReplayableResponse(response) || !hasKeyPoolFailover(route.provider) + || route.provider._apiKeyAttempt?.distributed || options.budgetSpent()) 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/src/server/responses/request-transport.ts b/src/server/responses/request-transport.ts index 42ca8b7723a..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 { 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; @@ -798,7 +799,10 @@ 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. + // 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/structure/providers/kiro.md b/structure/providers/kiro.md index 285c9aa94dc..e0e26d4ff12 100644 --- a/structure/providers/kiro.md +++ b/structure/providers/kiro.md @@ -76,7 +76,14 @@ observed ID still informs routing preference. `src/adapters/kiro-refusal.ts` recognizes an exact monthly reason on HTTP 400/429 and a confirmed suspension on HTTP 403; ordinary 400/403 remains an error without an account -verdict. `src/providers/kiro-usage.ts` records monthly exhaustion only for the sent +verdict. A 400 whose parsed body carries `ValidationException` (as `reason`, or inside `message`) +and the message "Improperly formed request" is the `profile` kind: the account layer rotates to a +sibling login like any other account-scoped refusal, and both tokens are required in the parsed +fields. `isKiroAccountProfileRefusal` applies the same rule to combo failure text (adapter-normalized +`Kiro invalid request: ...` prose or a JSON body, with or without the `Provider error 400: ` frame). +It is only consulted when the adapter the failed combo leg resolved is `kiro` (see +[combo failover](../transports/responses-failover.md)); coverage is +`tests/providers/kiro/kiro-refusal.test.ts` and `tests/server/server-kiro-refusal-e2e.test.ts`. `src/providers/kiro-usage.ts` records monthly exhaustion only for the sent credential generation and login identity, independently of quota observation time. A completed response from that same live credential clears an older verdict after disk hydration. Suspension is a process-local quarantine; rate refusals use a short cooldown. diff --git a/structure/transports/responses-failover.md b/structure/transports/responses-failover.md index 51d2a00c128..3649a480ad3 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. @@ -566,3 +566,16 @@ The resend charges the root workflow send counter once without a second request- - **Attempt ownership:** Each attempt captures its adapter binding at the physical-send observation. Recovery passes that binding's key and `_apiKeyAttempt` to the existing API-key 401/429 helpers, so a late failure cannot cool or overwrite a newer manual selection. A replacement rebuilds the adapter and replay scope before dispatch. Failed credentials are not resent in the request; the initial pool size and shared budget bound attempts. - **Cooldown:** Only an eligible 429 API-key recovery enters this cooldown path. When no reset time is known, the failed key receives the existing default cooldown. Nonrecoverable quota classifications do not create a key cooldown through this path; no daily reset is inferred. - **Regression coverage:** `tests/providers/qoder-adapter.test.ts` covers classification; `tests/server/server-key-failover-e2e.test.ts` covers mocked/local key recovery and replay boundaries; `tests/oauth/generic-oauth-failover.test.ts` guards recovery-site accounting. + +## Kiro account-profile hop + +`comboFailureDecision` takes an optional `failedAdapter` (the adapter the failed leg actually resolved; +`executeComboResponses` passes the child log's `providerAdapter`, so a `kiro` provider whose model is routed +through another wire by `modelAdapters` does not qualify, and an unresolved leg passes nothing). Only `failedAdapter === "kiro"` +may hop on an HTTP 400 that `isKiroAccountProfileRefusal` recognises (`ValidationException` plus +"Improperly formed request" in the parsed Kiro body, reached after the account layer has no sibling login +left to rotate to). Any other provider, or a caller that does not pass the adapter, keeps the generic +terminal `invalid_request_error` stop, so an invalid payload is never resent to another declared target. +Coverage: `tests/codex-integration/combos-kiro-profile-hop.test.ts` (unit, with non-Kiro and unrelated-field negatives) and +`tests/server/server-kiro-refusal-e2e.test.ts` (account rotation, combo exhaustion hop, non-Kiro 400 stop, +malformed-payload stop, `modelAdapters`-rerouted Kiro provider stop). 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/claude-integration/messages-native-oauth.test.ts b/tests/claude-integration/messages-native-oauth.test.ts index 2170f9f2b30..8cd48110a5a 100644 --- a/tests/claude-integration/messages-native-oauth.test.ts +++ b/tests/claude-integration/messages-native-oauth.test.ts @@ -819,6 +819,33 @@ test("shared quota exhaustion is bounded and never resends an already-tried alte expect(new Set(sent.map(entry => entry.headers.get("authorization"))).size).toBe(sent.length); }); +test("a pool whose every account refuses on spent quota answers the usage limit and its reset (#6991 parity)", async () => { + await seed(2); + const config = fixtureConfig(); + config.anthropicAccountPool = { enabled: true }; + const resetSeconds = Math.floor(Date.now() / 1000) + 7_200; + config.providers.anthropic!.fetch = (async (input, init) => { + sent.push({ url: String(input), headers: new Headers(init?.headers), body: JSON.parse(String(init?.body)) }); + return Response.json({ type: "error", error: { type: "rate_limit_error", message: "fixture window spent" } }, { + status: 429, + headers: { + "anthropic-ratelimit-unified-7d-status": "rejected", "anthropic-ratelimit-unified-7d-reset": String(resetSeconds), + "retry-after": "7200", "request-id": "req_fixture_spent", + }, + }); + }) as typeof fetch; + const { response, text, row } = await send(config, { ...BODY, stream: false }); + expect(response.status).toBe(429); + expect(sent).toHaveLength(2); + expect(JSON.parse(text).error).toMatchObject({ + type: "rate_limit_error", + message: `All Anthropic OAuth accounts have reached their usage limit; the first one resets at ${new Date(resetSeconds * 1000).toISOString()}`, + }); + expect(Number(response.headers.get("retry-after"))).toBeGreaterThan(7_000); + expect(response.headers.get("request-id")).toBe("req_fixture_spent"); + expect(row.protocolTrace?.mode).toBe("native"); +}); + test("family lease releases when spend admission refuses without a physical send", async () => { await seed(1); const config = fixtureConfig(); diff --git a/tests/codex-integration/combos-kiro-profile-hop.test.ts b/tests/codex-integration/combos-kiro-profile-hop.test.ts new file mode 100644 index 00000000000..31b7d6d0409 --- /dev/null +++ b/tests/codex-integration/combos-kiro-profile-hop.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, test } from "bun:test"; +import { comboFailureDecision } from "../../src/combos/failover"; + +// Moved byte for byte out of combos.test.ts ("combo failure policy and advancement"), which sits at +// the 2000-line ratchet threshold. +describe("combo failure policy: Kiro account-profile refusal", () => { + test("only a Kiro target hops on its improperly-formed validator 400", () => { + // A Kiro 400 that names its own validator rejection is account-scoped: after the account layer + // has no login left to rotate to, the next combo target is the only chance to serve the turn. + const improper = JSON.stringify({ message: "ValidationException: Improperly formed request" }); + const kiro = { failedAdapter: "kiro" }; + expect(comboFailureDecision(400, improper, kiro)).toBe("hop"); + expect(comboFailureDecision(400, `Provider error 400: ${improper}`, kiro)).toBe("hop"); + expect(comboFailureDecision(400, JSON.stringify({ message: "Improperly formed request" }), kiro)).toBe("stop"); + expect(comboFailureDecision(400, "ValidationException: model not found", kiro)).toBe("stop"); + // Tokens in unrelated fields, or a failed target that is not Kiro (or unknown), never hop. + expect(comboFailureDecision(400, JSON.stringify({ metadata: "ValidationException", message: "Improperly formed request" }), kiro)).toBe("stop"); + expect(comboFailureDecision(400, JSON.stringify({ error: { message: "Kiro invalid request: ValidationException: Improperly formed request" } }), kiro)).toBe("hop"); + expect(comboFailureDecision(400, "Kiro invalid request: ValidationException: Improperly formed request", kiro)).toBe("hop"); + expect(comboFailureDecision(400, "Kiro invalid request: ValidationException: Improperly formed request", { failedAdapter: "openai-chat" })).toBe("stop"); + expect(comboFailureDecision(400, improper)).toBe("stop"); + expect(comboFailureDecision(400, improper, { failedAdapter: "openai-chat" })).toBe("stop"); + expect(comboFailureDecision(400, improper, { failedAdapter: "anthropic" })).toBe("stop"); + }); +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 8ce291f4639..5c1046973c6 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -586,7 +586,7 @@ "combo-last-resort.test.ts": "codex-integration", "combo-management-api.test.ts": "routing", "combo-vision-enrollment.test.ts": "routing", "combo-stream-preflight.test.ts": "routing", "combo-workspace-data.test.ts": "gui", - "combo-workspace-jev-decision.test.ts": "gui", "combos.test.ts": "codex-integration", + "combo-workspace-jev-decision.test.ts": "gui", "combos.test.ts": "codex-integration", "combos-kiro-profile-hop.test.ts": "codex-integration", "command-code-error-finish.test.ts": "providers", "command-code-efforts.test.ts": "providers", "command-code-project-context.test.ts": "providers", "command-code-provider.test.ts": "providers", "command-code-quota.test.ts": "providers", "command-code-tool-text.test.ts": "providers", 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/providers/kiro/kiro-refusal.test.ts b/tests/providers/kiro/kiro-refusal.test.ts index 82739301b33..6b0e6f2353b 100644 --- a/tests/providers/kiro/kiro-refusal.test.ts +++ b/tests/providers/kiro/kiro-refusal.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test"; -import { classifyKiroRefusal } from "../../../src/adapters/kiro-refusal"; +import { classifyKiroRefusal, isKiroAccountProfileRefusal } from "../../../src/adapters/kiro-refusal"; import { kiroEvidenceIdentity } from "../../../src/providers/kiro-account-state-disk"; import type { ProviderAccount } from "../../../src/oauth/types"; @@ -23,6 +23,46 @@ describe("Kiro refusal classification", () => { } }); + test("kiro refusal: a 400 the vendor's validator calls improperly formed convicts the account", () => { + const improper = "ValidationException: Improperly formed request"; + for (const body of [ + { message: improper }, + { reason: "ValidationException", message: "Improperly formed request." }, + ]) expect(classified(400, body).kind).toBe("profile"); + expect(classified(429, { message: improper }).kind).toBe("rate"); + expect(classified(403, { message: improper }).kind).toBe("other"); + // Both tokens are required, so a bad payload and a caller echoing one token stay terminal. + expect(classified(400, { message: "Improperly formed request" }).kind).toBe("other"); + expect(classified(400, { message: "ValidationException: model not found" }).kind).toBe("other"); + expect(classified(400, { message: improper, padding: "x".repeat(70_000) }).kind).toBe("other"); + }); + + test("kiro refusal: the combo layer recognises the same shape through a wrapped frame", () => { + const improper = JSON.stringify({ message: "ValidationException: Improperly formed request" }); + expect(isKiroAccountProfileRefusal(400, improper)).toBe(true); + expect(isKiroAccountProfileRefusal(400, `Provider error 400: ${improper}`)).toBe(true); + expect(isKiroAccountProfileRefusal(429, improper)).toBe(false); + expect(isKiroAccountProfileRefusal(400, "Improperly formed request")).toBe(false); + // The adapter's normalized envelope is what the combo layer actually receives. + const envelope = JSON.stringify({ error: { message: "Kiro invalid request: ValidationException: Improperly formed request", + type: "invalid_request_error", code: "invalid_request_error" } }); + expect(isKiroAccountProfileRefusal(400, envelope)).toBe(true); + expect(isKiroAccountProfileRefusal(400, JSON.stringify({ error: { message: "Kiro invalid request: model not found" } }))).toBe(false); + expect(isKiroAccountProfileRefusal(400, JSON.stringify({ error: { metadata: "ValidationException", message: "Improperly formed request" } }))).toBe(false); + // The adapter's normalized prose, bare or framed; the same words without its prefix do not count. + const prose = "Kiro invalid request: ValidationException: Improperly formed request"; + expect(isKiroAccountProfileRefusal(400, prose)).toBe(true); + expect(isKiroAccountProfileRefusal(400, `Provider error 400: ${prose}`)).toBe(true); + expect(isKiroAccountProfileRefusal(400, "Kiro invalid request: model not found")).toBe(false); + // Split-field case: both tokens present, but not in the fields the account layer reads. + const split = JSON.stringify({ metadata: "ValidationException", message: "Improperly formed request" }); + expect(classified(400, JSON.parse(split)).kind).toBe("other"); + expect(isKiroAccountProfileRefusal(400, split)).toBe(false); + expect(isKiroAccountProfileRefusal(400, `Provider error 400: ${split}`)).toBe(false); + // Unparseable prose that merely contains both tokens is not the vendor's body. + expect(isKiroAccountProfileRefusal(400, "upstream said ValidationException: Improperly formed request")).toBe(false); + }); + test("kiro refusal: unknown, nested, malformed and oversized shapes never quarantine", () => { for (const body of ["not-json", "{", JSON.stringify({ error: { reason: "TEMPORARILY_SUSPENDED" } }), JSON.stringify({ reason: "INVALID_MODEL_ID" }), 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-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()); +}); diff --git a/tests/responses/responses-spend-anchor.test.ts b/tests/responses/responses-spend-anchor.test.ts index bcab4614f34..917d23b2e38 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(); @@ -567,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(); diff --git a/tests/server/server-kiro-refusal-e2e.test.ts b/tests/server/server-kiro-refusal-e2e.test.ts index eb0b42b5450..7ea09925cdf 100644 --- a/tests/server/server-kiro-refusal-e2e.test.ts +++ b/tests/server/server-kiro-refusal-e2e.test.ts @@ -10,6 +10,7 @@ import { createRequestExecutionBudget } from "../../src/lib/request-execution-bu import { ADAPTER_REGISTRY } from "../../src/adapters/registry"; import { encodeMessage } from "../../src/lib/eventstream-decoder"; import { KIRO_COMPLETION_TOOL_NAME } from "../../src/adapters/kiro-constants"; +import { clearComboSelectionState, clearComboTargetCooldowns } from "../../src/combos"; import { clearGenericFailoverHealth } from "../../src/oauth/generic-account-failover"; import { clearAccountQuotaCache } from "../../src/providers/quota"; import { kiroAccountEvidence, noteKiroMonthlyRefusal } from "../../src/providers/kiro-usage"; @@ -32,11 +33,15 @@ beforeEach(() => { process.env.KIRO_REGION = "us-east-1"; clearAccountQuotaCache(); clearGenericFailoverHealth(); + clearComboSelectionState(); + clearComboTargetCooldowns(); }); afterEach(() => { globalThis.fetch = realFetch; clearAccountQuotaCache(); clearGenericFailoverHealth(); + clearComboSelectionState(); + clearComboTargetCooldowns(); if (oldHome === undefined) delete process.env.OPENCODEX_HOME; else process.env.OPENCODEX_HOME = oldHome; if (oldRegion === undefined) delete process.env.KIRO_REGION; @@ -219,6 +224,135 @@ describe("Kiro refusal recovery through Responses", () => { } + // A 400 the vendor's validator calls improperly formed: account B can serve what A cannot. + const IMPROPER = JSON.stringify({ message: "ValidationException: Improperly formed request" }); + const KIRO_URL = (url: string) => url.startsWith("https://runtime.") && url.endsWith(".kiro.dev/"); + + test("Kiro improperly-formed 400 rotates to the sibling account and is served", async () => { + await seed(); saveConfig(config()); + const auths: Array = []; + globalThis.fetch = (async (input, init) => { + const url = input instanceof Request ? input.url : String(input); + if (!KIRO_URL(url)) return realFetch(input, init); + auths.push(new Headers(init?.headers).get("authorization")); + return auths.length === 1 ? new Response(IMPROPER, { status: 400 }) : answer("served by b"); + }) as typeof fetch; + const server = startServer(0); + try { + const response = await post(server, true); + expect(response.status).toBe(200); + expect(await response.text()).toContain("served by b"); + expect(auths).toEqual(["Bearer access-a", "Bearer access-b"]); + } finally { await server.stop(true); } + }); + + const comboConfig = (primary: "kiro" | "other"): OcxConfig => ({ + ...config(), + providers: { ...config().providers, + other: { adapter: "openai-chat", baseUrl: "https://other.example/v1", authMode: "key", + apiKey: "sk-other", models: ["model-other"] }, + fallback: { adapter: "openai-chat", baseUrl: "https://fallback.example/v1", authMode: "key", + apiKey: "sk-fallback", models: ["model-fallback"] } }, + combos: { fan: { strategy: "failover", targets: [ + primary === "kiro" ? { provider: "kiro", model: "claude-sonnet-4.5" } : { provider: "other", model: "model-other" }, + { provider: "fallback", model: "model-fallback" }, + ] } }, + }) as unknown as OcxConfig; + const comboPost = (server: ReturnType) => realFetch(new URL("/v1/responses", server.url), { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "combo/fan", input: "hello", stream: false }) }); + const fallbackOk = () => Response.json({ id: "chatcmpl-1", object: "chat.completion", model: "model-fallback", + choices: [{ index: 0, finish_reason: "stop", message: { role: "assistant", content: "served by fallback" } }], + usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 } }); + + test("combo hops to the next target once every Kiro account has refused as improperly formed", async () => { + await seed(); saveConfig(comboConfig("kiro")); + let kiroSends = 0; let fallbackSends = 0; + globalThis.fetch = (async (input, init) => { + const url = input instanceof Request ? input.url : String(input); + if (KIRO_URL(url)) { kiroSends++; return new Response(IMPROPER, { status: 400 }); } + if (url.startsWith("https://fallback.example/")) { fallbackSends++; return fallbackOk(); } + return realFetch(input, init); + }) as typeof fetch; + const server = startServer(0); + try { + const response = await comboPost(server); + const text = await response.text(); + expect(response.status).toBe(200); + expect(text).toContain("served by fallback"); + expect(kiroSends).toBe(2); + expect(fallbackSends).toBe(1); + } finally { await server.stop(true); } + }); + + test("combo never forwards a non-Kiro target's identical 400 to the next target", async () => { + await seed(); saveConfig(comboConfig("other")); + let otherSends = 0; let fallbackSends = 0; + globalThis.fetch = (async (input, init) => { + const url = input instanceof Request ? input.url : String(input); + if (url.startsWith("https://other.example/")) { + otherSends++; + return Response.json({ error: { message: "ValidationException: Improperly formed request", + type: "invalid_request_error" } }, { status: 400 }); + } + if (url.startsWith("https://fallback.example/")) { fallbackSends++; return fallbackOk(); } + return realFetch(input, init); + }) as typeof fetch; + const server = startServer(0); + try { + const response = await comboPost(server); + await response.text(); + expect(response.status).toBe(400); + expect(otherSends).toBe(1); + expect(fallbackSends).toBe(0); + } finally { await server.stop(true); } + }); + + test("combo stops on a Kiro 400 that is a malformed payload, not an account refusal", async () => { + await seed(); saveConfig(comboConfig("kiro")); + let fallbackSends = 0; + globalThis.fetch = (async (input, init) => { + const url = input instanceof Request ? input.url : String(input); + if (KIRO_URL(url)) return new Response(JSON.stringify({ message: "Improperly formed request" }), { status: 400 }); + if (url.startsWith("https://fallback.example/")) { fallbackSends++; return fallbackOk(); } + return realFetch(input, init); + }) as typeof fetch; + const server = startServer(0); + try { + const response = await comboPost(server); + await response.text(); + expect(response.status).toBe(400); + expect(fallbackSends).toBe(0); + } finally { await server.stop(true); } + }); + + test("combo never hops a Kiro-configured target whose model is routed through another adapter", async () => { + await seed(); + const routed = comboConfig("kiro"); + routed.providers.kiro = { ...routed.providers.kiro!, modelAdapters: { "claude-sonnet-4.5": "openai-chat" } }; + saveConfig(routed); + let routedSends = 0; let fallbackSends = 0; + globalThis.fetch = (async (input, init) => { + const url = input instanceof Request ? input.url : String(input); + if (url.startsWith("https://runtime.us-east-1.kiro.dev/") && !KIRO_URL(url)) { + routedSends++; + return new Response(IMPROPER, { status: 400, headers: { "content-type": "application/json" } }); + } + if (url.startsWith("https://fallback.example/")) { fallbackSends++; return fallbackOk(); } + return realFetch(input, init); + }) as typeof fetch; + const server = startServer(0); + try { + const response = await comboPost(server); + await response.text(); + // Same-provider account moves may still run; only the cross-target hop is fenced here. + expect(routedSends).toBeGreaterThan(0); + expect(response.status).toBe(400); + expect(fallbackSends).toBe(0); + } finally { await server.stop(true); } + }); + + test("Kiro web-search sidecar completion clears B's older verdict after rotated refusal", async () => { const [a, b] = await seed(); saveConfig({ ...config(), webSearchSidecar: { backend: "exa", exaApiKey: "synthetic" } } as OcxConfig); 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] }; diff --git a/tests/usage/request-log-attempt-identity.test.ts b/tests/usage/request-log-attempt-identity.test.ts index e6ca57f4c1c..c70f0ed8951 100644 --- a/tests/usage/request-log-attempt-identity.test.ts +++ b/tests/usage/request-log-attempt-identity.test.ts @@ -56,4 +56,40 @@ 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"); + }); + + 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, + }); + }); }); 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 },