diff --git a/docs-site/src/content/docs/troubleshooting/codex-cannot-sign-in.md b/docs-site/src/content/docs/troubleshooting/codex-cannot-sign-in.md index 9bff3814fdb..e1f84d88b7c 100644 --- a/docs-site/src/content/docs/troubleshooting/codex-cannot-sign-in.md +++ b/docs-site/src/content/docs/troubleshooting/codex-cannot-sign-in.md @@ -114,3 +114,17 @@ ocx account login openai --device See [Codex Integration](/guides/codex-integration/) for what the injection writes and how routing is chosen. + +## OAuth refresh capacity reached + +If a Responses request reports `OAuth token refresh capacity reached`, wait at +least one second before retrying. The proxy limits pending OAuth refresh work +to 256 consumers per process, including requests waiting on the same refresh. +A disconnected request keeps its refresh reservation until that work settles; +requests using fresh cached credentials can still proceed. + +Before a response starts, this temporary refusal returns HTTP 503 with +`Retry-After: 1`. A stream that has already started instead ends with a server +error. Capacity refusal does not mark the account as needing reauthentication, +so signing in again is unnecessary for this error. A refresh that never settles +can continue occupying capacity; the limit does not impose a refresh deadline. diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index dcfc094acfe..bc12111f4f8 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -114,6 +114,7 @@ "worker-embed-coverage.test.ts": "lib", "jev-stats.test.ts": "usage", "responses-devin-401-replay.test.ts": "responses", + "responses-oauth-refresh-capacity.test.ts": "responses", "responses-grok-devin-preflight.test.ts": "responses", "deepseek-quota-currency.test.ts": "providers", "mimo-token-plan-capacity.test.ts": "providers", @@ -1271,6 +1272,7 @@ "oauth-public-surface.test.ts": "oauth", "oauth-reauth-bind.test.ts": "oauth", "oauth-refresh-generic-lock.test.ts": "oauth", + "oauth-refresh-stale-capacity.test.ts": "oauth", "oauth-refresh-lock-multiprocess.test.ts": "oauth", "oauth-refresh.test.ts": "oauth", "oauth-status-privacy.test.ts": "oauth", diff --git a/src/oauth/index.ts b/src/oauth/index.ts index 4caa9aa6920..04c63e7e26f 100644 --- a/src/oauth/index.ts +++ b/src/oauth/index.ts @@ -59,6 +59,8 @@ import { providerModelsUrl, resolveProviderModelDiscoveryUrl } from "../provider import { resolveProviderTransport } from "../providers/xai-transport"; import { detectGrokCliToken, hasComparableGrokIdentity, isSameGrokIdentity, shouldAdoptGrokGeneration } from "./local-token-detect"; import { logOAuthEvent } from "./log"; +import { MAX_OAUTH_TOKEN_REFRESH_WAITERS, retireSettledOAuthRefreshFlight, tokenRefreshWaiters, type OAuthRefreshFlightEvidence } from "./refresh-flight-retirement"; +export { MAX_OAUTH_TOKEN_REFRESH_WAITERS }; import { captureConfigGeneration, sweepExpiredOnWrite } from "../lib/state-store-sweeper"; import { admitLoginFlow, classifyLoginCodeError, clearManualCodeSlot, CodeLoginUnsupportedError, ensureManualCodeSlot, kiroLoginSettling, latestLoginFlowKey, loginAbort, loginFlowProvider, loginState, recordStartedLoginFlow, settleLoginFlow, waitForManualLoginCode, type LoginCodeError, type OAuthLoginHint } from "./login-flow-state"; export { cancelLoginFlow, clearLoginState, CodeLoginUnsupportedError, isCodeLoginFlow, loginFlowAccountId, reconcileOAuthFlowState, submitLoginCode, submitManualLoginCode, waitForLoginSettled } from "./login-flow-state"; @@ -135,8 +137,7 @@ export type OAuthActiveTokenObservation = const MAX_OAUTH_TOKEN_REFRESH_FLIGHTS = 32; const OAUTH_TOKEN_REFRESH_FLIGHT_STALE_MS = 120_000; -interface OAuthRefreshFlightEvidence { flightId: string; dispatched: boolean } -interface OAuthTokenRefreshFlight extends OAuthRefreshFlightEvidence { promise: Promise; startedAt: number; abort: AbortController } +interface OAuthTokenRefreshFlight extends OAuthRefreshFlightEvidence { promise: Promise; startedAt: number; abort: AbortController; settled?: boolean; retirementDeferredLogged?: boolean } const tokenRefreshes = new Map(); export class OAuthTokenRefreshBusyError extends Error { readonly code = "OAUTH_TOKEN_REFRESH_BUSY"; @@ -160,7 +161,8 @@ export function seedOAuthTokenRefreshFlightsForTests(rows: Array<{ key: string; const promise = new Promise((_resolve, reject) => { abort.signal.addEventListener("abort", () => reject(abort.signal.reason), { once: true }); }); - const flight = { promise, startedAt: startedAt ?? Date.now(), abort, flightId: flightId ?? randomUUID(), dispatched: dispatched ?? false }; + const flight: OAuthTokenRefreshFlight = { promise, startedAt: startedAt ?? Date.now(), abort, flightId: flightId ?? randomUUID(), dispatched: dispatched ?? false }; + abort.signal.addEventListener("abort", () => { flight.settled = true; }, { once: true }); tokenRefreshes.set(key, flight); inserted.push(flight); return promise; @@ -634,45 +636,71 @@ async function resolveAccessSnapshotForAccount( const key = `${provider}\u0000${accountId}`; let existing = tokenRefreshes.get(key); let replacedStaleFlight: OAuthRefreshFlightEvidence | undefined; - if (existing && Date.now() - existing.startedAt <= OAUTH_TOKEN_REFRESH_FLIGHT_STALE_MS) { - logOAuthEvent("OAuth refresh joined existing operation", { provider, accountId }); - return existing.promise; - } - if (existing) { + if (existing && (existing.abort.signal.reason instanceof OAuthTokenRefreshStaleError + || Date.now() - existing.startedAt > OAUTH_TOKEN_REFRESH_FLIGHT_STALE_MS)) { replacedStaleFlight = { flightId: existing.flightId, dispatched: existing.dispatched }; existing.abort.abort(new OAuthTokenRefreshStaleError()); - if (tokenRefreshes.get(key) === existing) tokenRefreshes.delete(key); - existing = undefined; } - if (tokenRefreshes.size >= MAX_OAUTH_TOKEN_REFRESH_FLIGHTS) throw new OAuthTokenRefreshBusyError(); - - const abort = new AbortController(); - const flight: OAuthTokenRefreshFlight = { - promise: undefined as unknown as Promise, - startedAt: Date.now(), - abort, - flightId: randomUUID(), - dispatched: false, - }; - const refresh = (async (): Promise => { - const accessToken = await refreshAndPersistAccessToken(provider, accountId, def, cred, abort.signal, flight, replacedStaleFlight); - const persistedRow = getAccountCredentialWithStatus(provider, accountId); - if (!persistedRow) throw new OAuthLoginRequiredError(provider); - if (persistedRow.paused === true) throw new OAuthAccountPausedError(); - const persisted = persistedRow.credential; - if (persisted.access !== accessToken) { - throw new Error(`OAuth refresh persisted an unexpected access token for ${provider}`); + // Count every consumer, including same-flight joins, until its promise settles. Cancelling + // a client cannot recycle this permit. Stale-flight abort above still runs under pressure; + // no replacement starts until the old consumers have actually released enough capacity. + const waiter = tokenRefreshWaiters.tryAcquire(); + if (!waiter) throw new OAuthTokenRefreshBusyError(); + try { + if (existing && replacedStaleFlight) { + // A replacement would wait for this owner's lock anyway. Keep its registry slot + // through definitive-result cleanup and durable retirement, even with spare capacity. + // In particular, an ignored abort must not mark a late explicit rejection as uncertain. + await existing.promise.catch(() => {}); + if (tokenRefreshes.get(key) === existing && !retireSettledOAuthRefreshFlight(tokenRefreshes, key, existing)) { + throw new OAuthTokenRefreshStaleError(); + } + // Other admitted replacements may have resumed first and installed the next flight. + existing = tokenRefreshes.get(key); } - return accessSnapshot(provider, accountId, persisted, oauthProvider); - })().catch(error => { - if (abort.signal.reason instanceof OAuthTokenRefreshStaleError) throw abort.signal.reason; - throw error; - }).finally(() => { - if (tokenRefreshes.get(key) === flight) tokenRefreshes.delete(key); - }); - flight.promise = refresh; - tokenRefreshes.set(key, flight); - return refresh; + if (existing) { + logOAuthEvent("OAuth refresh joined existing operation", { provider, accountId }); + return await existing.promise; + } + if (tokenRefreshes.size >= MAX_OAUTH_TOKEN_REFRESH_FLIGHTS) { + for (const [staleKey, stale] of tokenRefreshes) { + if (stale.abort.signal.reason instanceof OAuthTokenRefreshStaleError) retireSettledOAuthRefreshFlight(tokenRefreshes, staleKey, stale); + } + if (tokenRefreshes.size >= MAX_OAUTH_TOKEN_REFRESH_FLIGHTS) throw new OAuthTokenRefreshBusyError(); + } + + const abort = new AbortController(); + const flight: OAuthTokenRefreshFlight = { + promise: undefined as unknown as Promise, + startedAt: Date.now(), + abort, + flightId: randomUUID(), + dispatched: false, + }; + const refresh = (async (): Promise => { + const accessToken = await refreshAndPersistAccessToken(provider, accountId, def, cred, abort.signal, flight, replacedStaleFlight); + const persistedRow = getAccountCredentialWithStatus(provider, accountId); + if (!persistedRow) throw new OAuthLoginRequiredError(provider); + if (persistedRow.paused === true) throw new OAuthAccountPausedError(); + const persisted = persistedRow.credential; + if (persisted.access !== accessToken) { + throw new Error(`OAuth refresh persisted an unexpected access token for ${provider}`); + } + return accessSnapshot(provider, accountId, persisted, oauthProvider); + })().catch(error => { + if (abort.signal.reason instanceof OAuthTokenRefreshStaleError) throw abort.signal.reason; + throw error; + }).finally(() => { + flight.settled = true; + if (tokenRefreshes.get(key) !== flight) return; + if (abort.signal.reason instanceof OAuthTokenRefreshStaleError) { + retireSettledOAuthRefreshFlight(tokenRefreshes, key, flight); + } else tokenRefreshes.delete(key); + }); + flight.promise = refresh; + tokenRefreshes.set(key, flight); + return await refresh; + } finally { waiter.release(); } } export async function getValidAccessTokenSnapshot( diff --git a/src/oauth/refresh-flight-retirement.ts b/src/oauth/refresh-flight-retirement.ts new file mode 100644 index 00000000000..c431e7a0cb6 --- /dev/null +++ b/src/oauth/refresh-flight-retirement.ts @@ -0,0 +1,48 @@ +import { createAdmissionGate } from "../lib/admission"; +import { isAnthropicOAuthInstance } from "../providers/anthropic-instance"; +import { clearObservedAnthropicRefreshIntent } from "./anthropic-oauth-definitions"; +import { logOAuthEvent } from "./log"; +import { markOAuthRefreshIntentStaleOwner, readOAuthRefreshIntent } from "./store"; + +/** Independent of client lifetime; a cancelled caller can still be awaiting a shared refresh. */ +export const MAX_OAUTH_TOKEN_REFRESH_WAITERS = 256; +export const tokenRefreshWaiters = createAdmissionGate("oauth_token_refresh_waiters", MAX_OAUTH_TOKEN_REFRESH_WAITERS); + +export interface OAuthRefreshFlightEvidence { flightId: string; dispatched: boolean } + +export interface RetirableOAuthRefreshFlight extends OAuthRefreshFlightEvidence { + settled?: boolean; + retirementDeferredLogged?: boolean; +} + +// Retire only after the old attempt has finished its definitive-result cleanup. A provider +// may ignore abort; marking its intent sooner would poison a late, safely retryable rejection. +// Failed durable retirement retains the same bounded registry slot for a later retry. +export function retireSettledOAuthRefreshFlight( + tokenRefreshes: Map, key: string, flight: F, +): boolean { + if (!flight.settled) return false; + const [provider, accountId] = key.split("\0"); + const defer = (reason: "uncertain_intent" | "intent_changed" | "storage_failure"): false => { + if (!flight.retirementDeferredLogged) { + flight.retirementDeferredLogged = true; + logOAuthEvent("OAuth refresh flight retirement deferred", { provider: provider ?? "unknown", reason }); + } + return false; + }; + try { + if (provider && accountId && isAnthropicOAuthInstance(provider)) { + const intent = readOAuthRefreshIntent(provider, accountId); + if (intent?.uncertain) return defer("uncertain_intent"); + if (intent?.flightId === flight.flightId && !intent.staleOwner && !intent.cleanupPending) { + const retired = flight.dispatched + ? markOAuthRefreshIntentStaleOwner(provider, accountId, intent.generation, flight.flightId) + : clearObservedAnthropicRefreshIntent(provider, accountId, intent); + if (!retired) return defer("intent_changed"); + } + } + } catch { return defer("storage_failure"); } + if (tokenRefreshes.get(key) === flight) tokenRefreshes.delete(key); + return true; +} + diff --git a/src/server/responses/adapter-dispatch.ts b/src/server/responses/adapter-dispatch.ts index 8c75b514d2c..6f418529a3f 100644 --- a/src/server/responses/adapter-dispatch.ts +++ b/src/server/responses/adapter-dispatch.ts @@ -54,7 +54,7 @@ import { describeUpstreamConnectFailure } from "./upstream-error"; import type { OpaqueBlobRecoveryGuard } from "./core-opaque-recovery"; import type { AttemptRecoveryKind } from "../../usage/log"; import type { OAuthAccessSnapshot } from "../../oauth"; -import { OAuthAccountPausedError, OAuthLoginRequiredError, publicOAuthAuthenticationErrorMessage } from "../../oauth"; +import { OAuthAccountPausedError, OAuthLoginRequiredError, OAuthTokenRefreshBusyError, publicOAuthAuthenticationErrorMessage } from "../../oauth"; import { getAccountSet, markAccountNeedsReauthIfGeneration } from "../../oauth/store"; import { tryKiroAlternateAfterTerminalRefresh } from "../../oauth/kiro-terminal-failover"; import { classifyKiroRefusal } from "../../adapters/kiro-refusal"; @@ -932,6 +932,15 @@ export async function prepareAdapterExchange( try { refreshed = await refreshResolvedOAuthSelection(transportState.sentOAuthSnapshot); } catch (err) { + const cancelled = options.abortSignal?.aborted || req.signal.aborted; + if (cancelled || err instanceof OAuthTokenRefreshBusyError) { + cleanupUpstreamAbort(); + upstream.abort(); + try { void upstreamResponse.body?.cancel().catch(() => {}); } catch { /* already closed */ } + releaseCodexAuthContextProbeLease(admissionState.authCtx); + return cancelled ? clientCancelledResponse() + : formatErrorResponse(503, "server_error", publicOAuthAuthenticationErrorMessage(err), { retryAfter: "1" }); + } const failed = transportState.sentOAuthSnapshot; if (route.providerName === "google-antigravity" && !antigravityAuthRotationAttempted && err instanceof OAuthLoginRequiredError && failed diff --git a/src/server/responses/passthrough-dispatch.ts b/src/server/responses/passthrough-dispatch.ts index 58a1d8be395..71e9885aad5 100644 --- a/src/server/responses/passthrough-dispatch.ts +++ b/src/server/responses/passthrough-dispatch.ts @@ -141,7 +141,7 @@ import { resolveWireProtocolOverride } from "../adapter-resolve"; import { refreshPoolForwardAuth, refreshNativeMainForwardAuth, withClaudeNativeSession } from "./core-auth"; import { bindRouteReasoningReplayScope } from "./core-replay"; import type { OAuthAccessSnapshot } from "../../oauth"; -import { OAuthAccountPausedError, OAuthLoginRequiredError, publicOAuthAuthenticationErrorMessage } from "../../oauth"; +import { OAuthAccountPausedError, OAuthLoginRequiredError, OAuthTokenRefreshBusyError, publicOAuthAuthenticationErrorMessage } from "../../oauth"; import { AnthropicAccountCooldownError } from "../../oauth/anthropic-routing"; import { resolveCopilotApiBaseUrl } from "../../oauth/github-copilot"; import { @@ -1358,6 +1358,8 @@ export async function preparePassthroughExchange( } catch (err) { upstream.abort(); releaseCodexAuthContextProbeLease(admissionState.authCtx); + if (options.abortSignal?.aborted || req.signal.aborted) return clientCancelledResponse(); + if (err instanceof OAuthTokenRefreshBusyError) return formatErrorResponse(503, "server_error", publicOAuthAuthenticationErrorMessage(err), { retryAfter: "1" }); return formatErrorResponse(401, "authentication_error", publicOAuthAuthenticationErrorMessage(err)); } if (route.provider.googleMode === "cloud-code-assist" && !refreshed.projectId) { diff --git a/src/server/responses/request-transport.ts b/src/server/responses/request-transport.ts index 8d9adbe1f82..f370c0e9fdd 100644 --- a/src/server/responses/request-transport.ts +++ b/src/server/responses/request-transport.ts @@ -31,6 +31,7 @@ import { } from "../../oauth/anthropic-routing"; import { OAuthAccountPausedError, + OAuthTokenRefreshBusyError, getValidAccessSnapshotForAccount, forceRefreshOAuthAccessSnapshot, getValidAccessTokenSnapshot, @@ -965,6 +966,10 @@ export async function prepareResponsesTransport( `${redactSecretString(err.message)}. Remove or reconfigure provider '${safeProviderName}' in the OpenCodex configuration.`, ); } + if (err instanceof OAuthTokenRefreshBusyError) { + return options.abortSignal?.aborted || req.signal.aborted ? clientCancelledResponse() + : formatErrorResponse(503, "server_error", err.message, { retryAfter: "1" }); + } if (err instanceof OAuthAccountPausedError) { return formatErrorResponse(403, "permission_error", publicOAuthAuthenticationErrorMessage(err)); } diff --git a/src/server/responses/run-turn-execution.ts b/src/server/responses/run-turn-execution.ts index 73d0dc8ade9..6eddb05a73d 100644 --- a/src/server/responses/run-turn-execution.ts +++ b/src/server/responses/run-turn-execution.ts @@ -30,7 +30,7 @@ import { rotateGenericOAuthAccountOn429, failoverAccountSnapshot, } from "../../oauth/generic-account-failover"; -import { OAuthAccountPausedError, publicOAuthAuthenticationErrorMessage, type OAuthAccessSnapshot } from "../../oauth/index"; +import { OAuthAccountPausedError, OAuthTokenRefreshBusyError, publicOAuthAuthenticationErrorMessage, type OAuthAccessSnapshot } from "../../oauth/index"; import { tryAlternateAfterTerminalRefresh } from "../../oauth/kiro-terminal-failover"; import { resolveWireProtocolOverride } from "../adapter-resolve"; import { formatErrorResponse, bridgeToResponsesSSE, buildResponseJSON } from "../../bridge"; @@ -79,7 +79,7 @@ const RUNTURN_WS_ROUTE_STATE_KEYS = [ /** One responsibility of the Responses request pipeline; state owners are explicit. */ export async function executeResponsesRunTurn( - requestContext: Pick, + requestContext: Pick, admissionState: ResponsesAdmissionState, requestState: Pick< PreparedResponsesRequest, @@ -130,7 +130,7 @@ export async function executeResponsesRunTurn( >, completionPolicy: Pick, ): Promise { - const { options, logCtx, config } = requestContext; + const { req, options, logCtx, config } = requestContext; const { selectionIsCurrent, adapterBindings, @@ -216,7 +216,9 @@ export async function executeResponsesRunTurn( } const runTurnAbort = new AbortController(); - const cleanupRunTurnAbort = linkAbortSignal(runTurnAbort, options.abortSignal); + const cleanupRequestAbort = linkAbortSignal(runTurnAbort, req.signal); + const cleanupOptionAbort = linkAbortSignal(runTurnAbort, options.abortSignal); + const cleanupRunTurnAbort = (): void => { cleanupRequestAbort(); cleanupOptionAbort(); }; const devinProducers = new Set(); const orderMessagesOutput = (source: AsyncIterable): AsyncIterable => inboundWire === "anthropic" && transportState.runTurnAdapter.name === "devin" && !routedCompaction @@ -466,7 +468,19 @@ export async function executeResponsesRunTurn( // selected; the turn then moves to a surviving stored account, or, with none, the client // gets the login instruction instead of an opaque upstream 401. let oauth401ReplayAttempted = false; - let pausedAuthRecovery = false; + let authRecoveryRefusal: Response | undefined; + const refuseLocalAuthRecovery = (error: Extract, cause: unknown): boolean => { + const cancelled = runTurnAbort.signal.aborted; + if (!cancelled && !(cause instanceof OAuthTokenRefreshBusyError) && !(cause instanceof OAuthAccountPausedError)) return false; + const status = cancelled ? 499 : cause instanceof OAuthTokenRefreshBusyError ? 503 : 403; + const errorType = cancelled ? "client_cancelled" : status === 503 ? "server_error" : "permission_error"; + const message = cancelled ? "Client cancelled request" : publicOAuthAuthenticationErrorMessage(cause); + // If a readiness heartbeat already committed SSE, the mapped event remains retryable + // capacity rather than retaining the rejected credential's authentication error/code. + Object.assign(error, { status, errorType, message, code: undefined, retryable: status === 503 }); + authRecoveryRefusal = formatErrorResponse(status, errorType, message, status === 503 ? { retryAfter: "1" } : undefined); + return true; + }; const recoverRunTurnAdapterOnPreflight401 = async ( error: Extract, ): Promise => { @@ -480,9 +494,7 @@ export async function executeResponsesRunTurn( try { admitted = await applyFailoverSnapshot(await refreshResolvedOAuthSelection(sent)); } catch (err) { - if (err instanceof OAuthAccountPausedError) { - pausedAuthRecovery = true; - Object.assign(error, { status: 403, errorType: "permission_error", message: publicOAuthAuthenticationErrorMessage(err) }); + if (refuseLocalAuthRecovery(error, err)) { hop.permit?.release(); return false; } @@ -505,7 +517,9 @@ export async function executeResponsesRunTurn( } catch (err) { // Applying the alternate can still fail, e.g. when a newer manual selection points back // at the flagged account; the client then needs the login instruction, not the raw 401. - Object.assign(error, { errorType: "authentication_error", message: publicOAuthAuthenticationErrorMessage(err) }); + if (!refuseLocalAuthRecovery(error, err)) { + Object.assign(error, { errorType: "authentication_error", message: publicOAuthAuthenticationErrorMessage(err) }); + } hop.permit?.release(); return false; } @@ -787,13 +801,13 @@ export async function executeResponsesRunTurn( // Preflight holds only heartbeats and the first meaningful event. A first-event 429 can be // replayed transparently; after any output reaches the bridge, a later error stays terminal. eventSource = await preflightRunTurnFailover(eventSource, wsFirstParsed, preflightDeadlineAt); - if (pausedAuthRecovery) { + if (authRecoveryRefusal) { cancelResponseCompletion(); runTurnAbort.abort(); queue.close(); cleanupRunTurnAbort(); releaseSearchProbeLease(); - return formatErrorResponse(403, "permission_error", publicOAuthAuthenticationErrorMessage(new OAuthAccountPausedError())); + return authRecoveryRefusal; } } if (grokDevinPreflight) { @@ -936,13 +950,13 @@ export async function executeResponsesRunTurn( for await (const event of await preflightRunTurnFailover( (async function* () { yield* firstAttemptEvents; })(), )) runTurnEvents.push(event); - if (pausedAuthRecovery) { + if (authRecoveryRefusal) { cancelResponseCompletion(); runTurnAbort.abort(); queue.close(); cleanupRunTurnAbort(); releaseSearchProbeLease(); - return formatErrorResponse(403, "permission_error", publicOAuthAuthenticationErrorMessage(new OAuthAccountPausedError())); + return authRecoveryRefusal; } } if (grokDevinPreflight) { diff --git a/structure/runtime.md b/structure/runtime.md index 94dc59e6a74..f4ca3262769 100644 --- a/structure/runtime.md +++ b/structure/runtime.md @@ -30,7 +30,7 @@ Responses admission and finalization are composed through the [core module owner Catalog HTTP acquisition follows the [proxy-routing contract](catalog.md#remote-catalog-http-proxy-routing). CLI model lists expose shared catalog estimates as `price`; `models price` retains the saved override in `cost` and adds `effectiveCost`. Explicit zero overrides win; unknown prices remain null, and automatic defaults are never persisted to `modelCosts`. Covered by `tests/cli/cli-models.test.ts` and `tests/cli/cli-models-price.test.ts`. Ordinary model discovery, startup sync, and explicit sync apply the [new-arrival catalog contract](catalog.md#shared-catalog) before publishing rows. The scoped persisted decision also updates the live config merge baseline, so subsequent visibility writes preserve manual re-enablement. A file-loaded inventory that already differs from disk can publish HTTP rows through a detached policy projection without adding unsaved live disables; retained catalog writes still refuse stale inventory, and its load provenance follows the [config contract](config.md). -OAuth refresh coordination follows the [refresh-lock identity contract](catalog.md#accounts-namespaces-and-pool-rotation): a fresh unreadable lock remains held, and release requires matching descriptor identity. A failed path-identity probe preserves the refresh callback outcome. Cooperating lock metadata changes serialize through the existing SQLite mutation transaction; release keeps the descriptor open through identity comparison and any unlink, then closes it. Failed metadata writes remove only a matching owned path after successful coordination; unknown identity, failed probes or unavailable coordination retain the path for stale recovery. Async refresh work holds no metadata transaction. +OAuth refresh resolution in `src/oauth/index.ts` admits at most 256 pending consumers across the process, including callers joining an existing refresh. The separate 32-flight ceiling remains. Consumers retain their admission until the shared promise settles, even if their client already cancelled; fresh credential snapshots do not take this admission. Exhaustion raises the existing retryable, account-free refresh-capacity error, and initial Responses preparation and upstream-401 forced-refresh consumers answer 503 with a short Retry-After before response commitment; cancellation takes priority. An already-committed runTurn stream emits retryable capacity rather than an authentication failure. Stale owners are aborted even under waiter pressure; rejected admission retains their registry evidence, and admitted replacements wait for settlement and durable retirement before taking the registry slot. Post-dispatch uncertainty remains retryable without replay or false reauthentication; pre-dispatch intent can be cleared exactly. Retirement IO failure retains the bounded slot for a later retry, including after a backward clock adjustment. Deferred durable retirement logs once per flight with a closed reason and provider only; no account or credential identity is emitted. Retirement waits for definitive provider-result cleanup, preserving late successful rotation or explicit rejection. This bounds pending resolution count, not request bytes, refresh duration, or arbitrary cancelled work. The release-on-cancel turn contract is unchanged. `tests/oauth/oauth-refresh-generic-lock.test.ts` covers cancelled consumers, rejection cleanup and cached credentials under pressure. `tests/oauth/oauth-refresh-stale-capacity.test.ts` covers saturated and unsaturated stale-owner retirement and late settlement; transport coverage is in `tests/responses/responses-oauth-refresh-capacity.test.ts`. OAuth refresh coordination follows the [refresh-lock identity contract](catalog.md#accounts-namespaces-and-pool-rotation): a fresh unreadable lock remains held, and release requires matching descriptor identity. A failed path-identity probe preserves the refresh callback outcome. Cooperating lock metadata changes serialize through the existing SQLite mutation transaction; release keeps the descriptor open through identity comparison and any unlink, then closes it. Failed metadata writes remove only a matching owned path after successful coordination; unknown identity, failed probes or unavailable coordination retain the path for stale recovery. Async refresh work holds no metadata transaction. The configuration-only [plaintext V2 contract](subagents.md#plaintext-v2-agent-messages) is scoped to canonical ChatGPT Responses forwarding; other source-area behavior described here is unchanged. Cursor's localized native-shell names follow the [routing-commentary guard contract](providers/cursor.md#cursor-native-exec). diff --git a/structure/transports/responses-failover.md b/structure/transports/responses-failover.md index 0fd76873559..f873d18f175 100644 --- a/structure/transports/responses-failover.md +++ b/structure/transports/responses-failover.md @@ -245,7 +245,7 @@ the failure on the current target. `src/server/responses/run-turn-execution.ts` handles a structured pre-output 401 for `isOAuth401ReplayProvider` runTurn routes with one generation-fenced forced refresh and an `auth-recovery` hop. A terminal refresh may admit a surviving account; otherwise the client gets the login instruction. Devin quota `permission_denied` (429) and plain permission denial (403) do not trigger this path. An `oauth`-source Devin key rejected with 401 needs reauthentication because Cognition has no refresh endpoint. A `local-cli` slot with a stored account ID or email can adopt a changed CLI key after host and bounded identity validation; [OAuth/Devin ownership](../providers-and-adapters.md) defines the locked store check across aliases. An identity-less CLI import instead needs an explicit `ocx login devin` after rotation. Unreadable files and transient probes leave the account unflagged. A pause during refresh returns 403 without retry, reauth, or replacing the stored key. Kiro's terminal alternate requires `OAuthLoginRequiredError`; transient refresh failures do not enter it. -Tests: `tests/responses/responses-devin-401-replay.test.ts` and `tests/server/server-kiro-refusal-e2e.test.ts`. +Forced-refresh capacity exhaustion on adapter, passthrough and runTurn paths is a local retryable refusal: before response commitment it returns 503 with Retry-After: 1, and cancellation from either the request signal or the option signal wins with 499. The runTurn controller links and cleans up both signals. It does not rotate an account or require login. A runTurn readiness heartbeat that already committed SSE leaves HTTP 200 in place but emits a retryable server-error capacity event without the rejected credential's authentication code. Tests: `tests/responses/responses-devin-401-replay.test.ts`, `tests/responses/responses-oauth-refresh-capacity.test.ts` and `tests/server/server-kiro-refusal-e2e.test.ts`. ## Optional client transport hints diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index e880d102fd5..37b16875436 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -156,7 +156,7 @@ "openai-chat-tool-call-id-remint.test.ts": "adapters/openai", "coding-agent-json-lines-scaling.test.ts": "providers", "usage-snapshot-digest-reuse.test.ts": "usage", "worker-embed.test.ts": "lib", "worker-embed-coverage.test.ts": "lib", "jev-stats.test.ts": "usage", - "responses-devin-401-replay.test.ts": "responses", "responses-grok-devin-preflight.test.ts": "responses", + "responses-devin-401-replay.test.ts": "responses", "responses-oauth-refresh-capacity.test.ts": "responses", "responses-grok-devin-preflight.test.ts": "responses", "abort-idle-deadline.test.ts": "lib", "tool-envelope-echo-whole-line.test.ts": "adapters", "abort-race.test.ts": "adapters", "account-import.test.ts": "server", "account-pool-management-api.test.ts": "server", "link-listener-admission.test.ts": "server", @@ -858,7 +858,7 @@ "oauth-open-browser-choice.test.ts": "oauth", "oauth-open-url-result.test.ts": "oauth", "oauth-provider-reconcile.test.ts": "oauth", "oauth-reconcile-operator-maps.test.ts": "oauth", "oauth-public-surface.test.ts": "oauth", - "oauth-reauth-bind.test.ts": "oauth", "oauth-refresh-generic-lock.test.ts": "oauth", + "oauth-reauth-bind.test.ts": "oauth", "oauth-refresh-generic-lock.test.ts": "oauth", "oauth-refresh-stale-capacity.test.ts": "oauth", "oauth-refresh-lock-multiprocess.test.ts": "oauth", "oauth-refresh.test.ts": "oauth", "oauth-status-privacy.test.ts": "oauth", "oauth-store-multi.test.ts": "oauth", "oauth-tos-warning.test.ts": "gui", "oauth-upsert-preserves-api-key.test.ts": "oauth", diff --git a/tests/oauth/oauth-refresh-generic-lock.test.ts b/tests/oauth/oauth-refresh-generic-lock.test.ts index 63cd499e03a..c857b4a6cfa 100644 --- a/tests/oauth/oauth-refresh-generic-lock.test.ts +++ b/tests/oauth/oauth-refresh-generic-lock.test.ts @@ -4,6 +4,10 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { getValidAccessTokenForAccount, + getValidAccessTokenSnapshot, + getValidAccessSnapshotForAccount, + MAX_OAUTH_TOKEN_REFRESH_WAITERS, + OAuthTokenRefreshBusyError, refreshGenericAccountWithLock, OAuthLoginRequiredError, OAuthTokenRefreshStaleError, @@ -13,6 +17,9 @@ import type { OAuthCredentials } from "../../src/oauth/types"; import { ChatGptTokenError, ChatGptTokenRequestError, refreshChatGPTToken } from "../../src/oauth/chatgpt"; import { getAccountCredential, getAccountSet, saveCredential } from "../../src/oauth/store"; import { removeTreeWithRetry } from "../helpers/remove-tree"; +import { getActiveTurnCount, tryAdmitTurn } from "../../src/server/lifecycle"; +import { handleResponses } from "../../src/server/responses"; +import type { OcxConfig } from "../../src/types"; // Gate/CAS refresh races can exceed the 5s default under windows-latest contention // (same flake class as kiro-oauth / oauth queue budgets). @@ -403,3 +410,128 @@ describe("ChatGPT refresh through the registered provider", () => { }); }); + +/** Real resolver calls retain refresh admission after their client turn has cancelled. */ +describe("OAuth refresh consumer admission", () => { + test("cancelled callers cannot grow one pending refresh beyond the consumer limit", async () => { + const accountId = await seedExpiredKimi(); + let finish!: () => void; + let started!: () => void; + const entered = new Promise(resolve => { started = resolve; }); + const blocked = new Promise(resolve => { finish = resolve; }); + const tracker = stubKimiRefresh(async () => { + started(); + await blocked; + return { access: "fixture-refreshed", refresh: "fixture-rotated", expires: Date.now() + 3_600_000 }; + }); + const waiting: Promise[] = []; + const rejected: unknown[] = []; + const activeBefore = getActiveTurnCount(); + try { + for (let i = 0; i < MAX_OAUTH_TOKEN_REFRESH_WAITERS + 44; i++) { + const controller = new AbortController(); + const lease = tryAdmitTurn(undefined, controller.signal); + expect(lease).not.toBeNull(); + const promise = i % 2 === 0 ? getValidAccessTokenSnapshot("kimi") + : getValidAccessSnapshotForAccount("kimi", accountId); + waiting.push(promise.catch(error => { rejected.push(error); })); + controller.abort(new Error("fixture disconnected")); + lease?.release(); + } + await entered; + await Promise.resolve(); + await Promise.resolve(); + expect(getActiveTurnCount()).toBe(activeBefore); + expect(rejected).toHaveLength(44); + expect(rejected.every(error => error instanceof OAuthTokenRefreshBusyError)).toBe(true); + expect(tracker.calls()).toBe(1); + expect(getAccountCredential("kimi", accountId)?.access).toBe("kimi-old"); + expect(getAccountSet("kimi")?.accounts.find(row => row.id === accountId)?.needsReauth).not.toBe(true); + } finally { + finish(); + await Promise.all(waiting); + } + await expect(getValidAccessTokenForAccount("kimi", accountId)).resolves.toBe("fixture-refreshed"); + expect(tracker.calls()).toBe(1); + }); + + test("rejected shared refresh releases all consumers and permits one later retry", async () => { + const accountId = await seedExpiredKimi(); + let fail!: () => void; + let started!: () => void; + const entered = new Promise(resolve => { started = resolve; }); + const blocked = new Promise((_resolve, reject) => { fail = () => reject(new Error("fixture transient failure")); }); + stubKimiRefresh(async () => { started(); await blocked; throw new Error("unreachable"); }); + const waiting = Array.from({ length: MAX_OAUTH_TOKEN_REFRESH_WAITERS }, () => + getValidAccessTokenForAccount("kimi", accountId).catch(error => error)); + try { + await entered; + // Do not await an over-budget control without a deadline: the unpatched resolver waits. + let refused: unknown; + const overflow = getValidAccessTokenForAccount("kimi", accountId).catch(error => { refused = error; }); + waiting.push(overflow); + await Promise.resolve(); await Promise.resolve(); await Promise.resolve(); + expect(refused).toBeInstanceOf(OAuthTokenRefreshBusyError); + } finally { fail(); await Promise.all(waiting); } + const tracker = stubKimiRefresh(async () => ({ + access: "fixture-retry", refresh: "fixture-next", expires: Date.now() + 3_600_000, + })); + await expect(getValidAccessTokenForAccount("kimi", accountId)).resolves.toBe("fixture-retry"); + expect(tracker.calls()).toBe(1); + expect(getAccountSet("kimi")?.accounts.find(row => row.id === accountId)?.needsReauth).not.toBe(true); + }); + + test("fresh snapshots remain usable while refresh consumers are full", async () => { + const expiredId = await seedExpiredKimi(); + await saveCredential("kimi", { access: "fixture-cached", refresh: "fixture-cached-refresh", + expires: Date.now() + 3_600_000, accountId: "fixture-cached-account" }); + const freshId = getAccountSet("kimi")!.activeAccountId; + let finish!: () => void; + let started!: () => void; + const entered = new Promise(resolve => { started = resolve; }); + const blocked = new Promise(resolve => { finish = resolve; }); + const tracker = stubKimiRefresh(async () => { started(); await blocked; return { + access: "fixture-finished", refresh: "fixture-finished-refresh", expires: Date.now() + 3_600_000, + }; }); + const waiting = Array.from({ length: MAX_OAUTH_TOKEN_REFRESH_WAITERS }, () => + getValidAccessTokenForAccount("kimi", expiredId).catch(error => error)); + try { + await entered; + await expect(getValidAccessTokenForAccount("kimi", freshId)).resolves.toBe("fixture-cached"); + expect(tracker.calls()).toBe(1); + } finally { finish(); await Promise.all(waiting); } + }); +}); + + +test("Responses maps exhausted OAuth refresh consumers to retryable capacity, not login failure", async () => { + const accountId = await seedExpiredKimi(); + let finish!: () => void; + let started!: () => void; + const entered = new Promise(resolve => { started = resolve; }); + const blocked = new Promise(resolve => { finish = resolve; }); + stubKimiRefresh(async () => { started(); await blocked; return { + access: "fixture-finished", refresh: "fixture-rotated", expires: Date.now() + 3_600_000, + }; }); + const waiting = Array.from({ length: MAX_OAUTH_TOKEN_REFRESH_WAITERS }, () => + getValidAccessTokenForAccount("kimi", accountId).catch(error => error)); + let inferenceSends = 0; + const config = { port: 0, defaultProvider: "kimi", providers: { kimi: { + adapter: "openai-chat", authMode: "oauth", baseUrl: "https://fixture.invalid/v1", models: ["fixture-model"], + fetch: async () => { inferenceSends++; throw new Error("unexpected fixture send"); }, + } } } as OcxConfig; + try { + await entered; + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "kimi/fixture-model", input: "fixture", stream: false }), + }), config, { model: "", provider: "" }); + expect(response.status).toBe(503); + expect(response.headers.get("retry-after")).toBe("1"); + const body = await response.text(); + expect(body).toContain("OAuth token refresh capacity reached"); + expect(body).not.toContain("kimi-old"); + expect(inferenceSends).toBe(0); + expect(getAccountSet("kimi")?.accounts.find(row => row.id === accountId)?.needsReauth).not.toBe(true); + } finally { finish(); await Promise.all(waiting); } +}); diff --git a/tests/oauth/oauth-refresh-stale-capacity.test.ts b/tests/oauth/oauth-refresh-stale-capacity.test.ts new file mode 100644 index 00000000000..5a0706c80a6 --- /dev/null +++ b/tests/oauth/oauth-refresh-stale-capacity.test.ts @@ -0,0 +1,205 @@ +import { afterEach, beforeEach, expect, spyOn, test } from "bun:test"; +import { mkdirSync, mkdtempSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + getValidAccessTokenForAccount, + MAX_OAUTH_TOKEN_REFRESH_WAITERS, + OAuthTokenRefreshBusyError, + OAuthTokenRefreshStaleError, + OAUTH_PROVIDERS, + seedOAuthTokenRefreshFlightsForTests, +} from "../../src/oauth"; +import { credentialGeneration, getAccountCredential, getAccountSet, readOAuthRefreshIntent, + saveCredential, writeOAuthRefreshIntent } from "../../src/oauth/store"; +import { AnthropicTokenError } from "../../src/oauth/anthropic"; +import * as store from "../../src/oauth/store"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; + +const original = { home: process.env.HOME, ocx: process.env.OPENCODEX_HOME, + claude: process.env.CLAUDE_CONFIG_DIR, fetch: globalThis.fetch, + anthropic: OAUTH_PROVIDERS.anthropic!.refresh, kimi: OAUTH_PROVIDERS.kimi!.refresh }; +let home: string; +beforeEach(() => { + home = mkdtempSync(join(tmpdir(), "oauth-stale-capacity-")); + process.env.HOME = home; + process.env.OPENCODEX_HOME = join(home, "ocx"); + process.env.CLAUDE_CONFIG_DIR = join(home, "claude"); + mkdirSync(process.env.CLAUDE_CONFIG_DIR); + writeFileSync(join(process.env.CLAUDE_CONFIG_DIR, ".credentials.json"), "{}"); + globalThis.fetch = (async () => Response.json({ account: { uuid: "synthetic-stale-account" } })) as unknown as typeof fetch; +}); +afterEach(() => { + OAUTH_PROVIDERS.anthropic!.refresh = original.anthropic; + OAUTH_PROVIDERS.kimi!.refresh = original.kimi; + globalThis.fetch = original.fetch; + for (const [key, value] of [["HOME", original.home], ["OPENCODEX_HOME", original.ocx], ["CLAUDE_CONFIG_DIR", original.claude]]) { + if (value === undefined) delete process.env[key!]; else process.env[key!] = value; + } + removeTreeWithRetry(home); +}); +async function seed(provider: "anthropic" | "kimi") { + await saveCredential(provider, { access: "fixture-old", refresh: "fixture-refresh", expires: 1, accountId: "fixture-account" }); + return getAccountSet(provider)!.activeAccountId; +} + +// A rejected admission must not discard the stale owner's generation-bound intent evidence. +test("saturated stale post-dispatch owner remains retryable after all 256 consumers settle", async () => { + const id = await seed("anthropic"); + let started!: () => void; + const entered = new Promise(resolve => { started = resolve; }); + let calls = 0; + OAUTH_PROVIDERS.anthropic!.refresh = async (_token, signal) => { + calls++; + started(); + return new Promise((_, reject) => signal!.addEventListener("abort", () => reject(signal!.reason), { once: true })); + }; + const waiting = Array.from({ length: MAX_OAUTH_TOKEN_REFRESH_WAITERS }, () => + getValidAccessTokenForAccount("anthropic", id).catch(error => error)); + await entered; + const intent = readOAuthRefreshIntent("anthropic", id); + expect(intent?.flightId).toBeString(); + const clock = spyOn(Date, "now").mockReturnValue(Date.now() + 120_001); + try { + await expect(getValidAccessTokenForAccount("anthropic", id)).rejects.toBeInstanceOf(OAuthTokenRefreshBusyError); + expect((await Promise.all(waiting)).every(error => error instanceof OAuthTokenRefreshStaleError)).toBe(true); + await expect(getValidAccessTokenForAccount("anthropic", id)).rejects.toBeInstanceOf(OAuthTokenRefreshStaleError); + expect(calls).toBe(1); + expect(getAccountSet("anthropic")!.accounts.find(a => a.id === id)?.needsReauth).not.toBe(true); + expect(readOAuthRefreshIntent("anthropic", id)).toMatchObject({ ...intent, staleOwner: true }); + } finally { clock.mockRestore(); await Promise.all(waiting); } +}); + +test("saturated stale pre-dispatch owner retains evidence for one safe replacement after capacity releases", async () => { + const id = await seed("anthropic"); + const kimiId = await seed("kimi"); + let started!: () => void, finish!: () => void; + const entered = new Promise(resolve => { started = resolve; }); + const blocked = new Promise(resolve => { finish = resolve; }); + OAUTH_PROVIDERS.kimi!.refresh = async () => { + started(); await blocked; + return { access: "fixture-kimi-new", refresh: "fixture-kimi-next", expires: Date.now() + 3_600_000 }; + }; + let calls = 0; + OAUTH_PROVIDERS.anthropic!.refresh = async () => { + calls++; + return { access: "fixture-anthropic-new", refresh: "fixture-anthropic-next", expires: Date.now() + 3_600_000 }; + }; + const waiting = Array.from({ length: MAX_OAUTH_TOKEN_REFRESH_WAITERS }, () => + getValidAccessTokenForAccount("kimi", kimiId).catch(error => error)); + await entered; + const flightId = "fixture-pre-dispatch-owner"; + writeOAuthRefreshIntent("anthropic", id, credentialGeneration(getAccountCredential("anthropic", id)!), Date.now() - 120_001, flightId); + const stale = seedOAuthTokenRefreshFlightsForTests([{ key: `anthropic\0${id}`, startedAt: Date.now() - 120_001, flightId, dispatched: false }]); + const stopped = stale.promises[0]!.catch(error => error); + try { + await expect(getValidAccessTokenForAccount("anthropic", id)).rejects.toBeInstanceOf(OAuthTokenRefreshBusyError); + expect(await stopped).toBeInstanceOf(OAuthTokenRefreshStaleError); + finish(); await Promise.all(waiting); + await expect(getValidAccessTokenForAccount("anthropic", id)).resolves.toBe("fixture-anthropic-new"); + expect(calls).toBe(1); + expect(getAccountSet("anthropic")!.accounts.find(a => a.id === id)?.needsReauth).not.toBe(true); + expect(readOAuthRefreshIntent("anthropic", id)).toBeUndefined(); + } finally { finish(); await Promise.all(waiting); stale.cleanup(); } +}); + +for (const outcome of ["rejected", "rotated", "marker-io"] as const) { + test(`saturated stale owner preserves late ${outcome} settlement`, async () => { + const id = await seed("anthropic"); + let started!: () => void, finish!: () => void; + const entered = new Promise(resolve => { started = resolve; }); + const blocked = new Promise(resolve => { finish = resolve; }); + let calls = 0; + OAUTH_PROVIDERS.anthropic!.refresh = async () => { + calls++; + if (calls === 1) { + started(); await blocked; // Deliberately ignore abort, like the real Anthropic adapter. + if (outcome === "rejected") throw new AnthropicTokenError("fixture unavailable", 503, undefined); + if (outcome === "marker-io") throw new OAuthTokenRefreshStaleError(); + } + return { access: "fixture-late-new", refresh: "fixture-late-next", expires: Date.now() + 3_600_000 }; + }; + const waiting = Array.from({ length: MAX_OAUTH_TOKEN_REFRESH_WAITERS }, () => + getValidAccessTokenForAccount("anthropic", id).catch(error => error)); + await entered; + const intent = readOAuthRefreshIntent("anthropic", id); + const clock = spyOn(Date, "now").mockReturnValue(Date.now() + 120_001); + const marker = outcome === "marker-io" + ? spyOn(store, "markOAuthRefreshIntentStaleOwner").mockImplementation(() => { throw new Error("fixture retirement IO failure"); }) + : undefined; + try { + await expect(getValidAccessTokenForAccount("anthropic", id)).rejects.toBeInstanceOf(OAuthTokenRefreshBusyError); + expect(readOAuthRefreshIntent("anthropic", id)).toEqual(intent); + finish(); + const settled = await Promise.all(waiting); + expect(settled.every(result => outcome === "rotated" ? result === "fixture-late-new" : result instanceof OAuthTokenRefreshStaleError)).toBe(true); + if (outcome === "marker-io") { + expect(marker).toHaveBeenCalledTimes(1); + expect(readOAuthRefreshIntent("anthropic", id)).toEqual(intent); + marker!.mockRestore(); + clock.mockRestore(); // Retained aborted owners remain stale even if wall-clock time moves back. + await expect(getValidAccessTokenForAccount("anthropic", id)).rejects.toBeInstanceOf(OAuthTokenRefreshStaleError); + expect(readOAuthRefreshIntent("anthropic", id)).toMatchObject({ ...intent, staleOwner: true }); + expect(calls).toBe(1); + } else { + expect(readOAuthRefreshIntent("anthropic", id)).toBeUndefined(); + await expect(getValidAccessTokenForAccount("anthropic", id)).resolves.toBe("fixture-late-new"); + expect(calls).toBe(outcome === "rejected" ? 2 : 1); + } + expect(getAccountSet("anthropic")!.accounts.find(a => a.id === id)?.needsReauth).not.toBe(true); + } finally { marker?.mockRestore(); clock.mockRestore(); finish(); await Promise.all(waiting); } + }); +} + +for (const outcome of ["rejected", "marker-io"] as const) { + test(`unsaturated stale owner retains late ${outcome} evidence until settlement`, async () => { + const id = await seed("anthropic"); + let started!: () => void, finish!: () => void; + const entered = new Promise(resolve => { started = resolve; }); + const blocked = new Promise(resolve => { finish = resolve; }); + let calls = 0; + OAUTH_PROVIDERS.anthropic!.refresh = async () => { + if (++calls === 1) { + started(); await blocked; + if (outcome === "rejected") throw new AnthropicTokenError("fixture unavailable", 503, undefined); + throw new OAuthTokenRefreshStaleError(); + } + return { access: "fixture-unsaturated-new", refresh: "fixture-unsaturated-next", expires: Date.now() + 3_600_000 }; + }; + const old = getValidAccessTokenForAccount("anthropic", id).catch(error => error); + await entered; + const intent = readOAuthRefreshIntent("anthropic", id); + const clock = spyOn(Date, "now").mockReturnValue(Date.now() + 120_001); + const marker = spyOn(store, "markOAuthRefreshIntentStaleOwner"); + const info = spyOn(console, "info").mockImplementation(() => {}); + if (outcome === "marker-io") marker.mockImplementation(() => { throw new Error("fixture retirement IO failure"); }); + let replacement: Promise | undefined; + try { + replacement = Promise.all(Array.from({ length: 8 }, () => + getValidAccessTokenForAccount("anthropic", id).catch(error => error))); + await Promise.resolve(); await Promise.resolve(); + // The old attempt still holds the lock and has no definitive result. No marker yet. + expect(readOAuthRefreshIntent("anthropic", id)).toEqual(intent); + expect(marker).not.toHaveBeenCalled(); + finish(); + expect(await old).toBeInstanceOf(OAuthTokenRefreshStaleError); + const results = await replacement; + if (outcome === "rejected") { + expect(results).toEqual(Array(8).fill("fixture-unsaturated-new")); + expect(calls).toBe(2); + expect(readOAuthRefreshIntent("anthropic", id)).toBeUndefined(); + } else { + expect(results.every(result => result instanceof OAuthTokenRefreshStaleError)).toBe(true); + expect(readOAuthRefreshIntent("anthropic", id)).toEqual(intent); + const notices = info.mock.calls.map(([message]) => message) + .filter(message => String(message).includes("OAuth refresh flight retirement deferred")); + expect(notices).toEqual(["[opencodex] OAuth refresh flight retirement deferred provider=anthropic reason=storage_failure"]); + marker.mockRestore(); clock.mockRestore(); + await expect(getValidAccessTokenForAccount("anthropic", id)).rejects.toBeInstanceOf(OAuthTokenRefreshStaleError); + expect(readOAuthRefreshIntent("anthropic", id)).toMatchObject({ ...intent, staleOwner: true }); + expect(calls).toBe(1); + } + expect(getAccountSet("anthropic")!.accounts.find(row => row.id === id)?.needsReauth).not.toBe(true); + } finally { info.mockRestore(); marker.mockRestore(); clock.mockRestore(); finish(); await old; await replacement; } + }); +} diff --git a/tests/providers/kiro/kiro-leased-responses.test.ts b/tests/providers/kiro/kiro-leased-responses.test.ts index bf3f10c618a..49ed1720d93 100644 --- a/tests/providers/kiro/kiro-leased-responses.test.ts +++ b/tests/providers/kiro/kiro-leased-responses.test.ts @@ -175,6 +175,8 @@ test("an abandoned response body releases its lease when the request aborts", as test("cancellation during Kiro credential recovery cannot install a replacement lease", async () => { const [id] = await seed(); const controller = new AbortController(); + let inferenceSends = 0; + let refreshCalls = 0; let refreshStarted!: () => void; let finishRefresh!: () => void; const started = new Promise(resolve => { refreshStarted = resolve; }); @@ -182,10 +184,12 @@ test("cancellation during Kiro credential recovery cannot install a replacement globalThis.fetch = (async input => { const url = input instanceof Request ? input.url : String(input); if (url.endsWith("/refreshToken")) { + refreshCalls++; refreshStarted(); await refreshGate; return Response.json({ accessToken: "refreshed-access", refreshToken: "load-refresh-0", expiresIn: 3600 }); } + inferenceSends++; return new Response("expired", { status: 401 }); }) as typeof fetch; @@ -196,8 +200,14 @@ test("cancellation during Kiro credential recovery cannot install a replacement expect(accountInFlight("kiro", id!)).toBe(0); finishRefresh(); const response = await pending; - expect(response.status).toBe(401); + expect(response.status).toBe(499); expect(accountInFlight("kiro", id!)).toBe(0); + expect(response.headers.get("retry-after")).toBeNull(); + expect(await response.json()).toMatchObject({ error: { + type: "client_cancelled", code: "client_cancelled", message: "Client cancelled request", + } }); + expect(inferenceSends).toBe(1); + expect(refreshCalls).toBe(1); }); test("a reactive rotation onto a full account leaves the store selection and request state unchanged", async () => { diff --git a/tests/responses/responses-oauth-refresh-capacity.test.ts b/tests/responses/responses-oauth-refresh-capacity.test.ts new file mode 100644 index 00000000000..85553083687 --- /dev/null +++ b/tests/responses/responses-oauth-refresh-capacity.test.ts @@ -0,0 +1,195 @@ +import { afterAll, afterEach, beforeEach, expect, mock, spyOn, test } from "bun:test"; +import type { ProviderAdapter } from "../../src/adapters/base"; +import type { AdapterEvent, OcxConfig, OcxProviderConfig } from "../../src/types"; +import * as oauth from "../../src/oauth"; +import { getAccountSet, saveCredential } from "../../src/oauth/store"; +import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { createTempHome } from "../helpers/temp-home"; + +const PROVIDER = "orcarouter-oauth"; +const ACCESS = "fixture-refresh-capacity-access"; +const originalRefresh = oauth.OAUTH_PROVIDERS[PROVIDER]!.refresh; +const originalForceRefresh = oauth.forceRefreshOAuthAccessSnapshot; +const resolver = await import("../../src/server/adapter-resolve"); +const originalResolver = { ...resolver }; +let rejectInference: () => Promise; +let inferenceSends = 0; +let transportMode: "adapter" | "passthrough" | "runTurn"; +let runTurnSends = 0; +let runTurnReady = false; +let httpPaths: string[] = []; +mock.module("../../src/server/adapter-resolve", () => ({ ...originalResolver, + resolveAdapter(provider: OcxProviderConfig, cache?: "none" | "short" | "long", providerId?: string) { + if (provider.baseUrl !== "https://fixture.invalid/v1") return originalResolver.resolveAdapter(provider, cache, providerId); + if (transportMode !== "runTurn") return originalResolver.resolveAdapter({ ...provider, + adapter: transportMode === "passthrough" ? "openai-responses" : "openai-chat", + }, cache, providerId); + return { + name: "devin", + buildRequest: () => ({ url: provider.baseUrl, method: "POST", headers: {}, body: "" }), + async *parseStream() { yield { type: "done" } as AdapterEvent; }, + async runTurn(_parsed, _incoming, emit) { + expect(provider.apiKey).toBe(ACCESS); + inferenceSends++; + runTurnSends++; + if (runTurnReady) emit({ type: "heartbeat", preflightReady: true }); + await rejectInference(); + emit({ type: "error", status: 401, errorType: "authentication_error", + code: "unauthenticated", retryable: false, message: "fixture upstream rejected credential" }); + }, + } satisfies ProviderAdapter; + }, +})); +const { handleResponses } = await import("../../src/server/responses"); + +let home: ReturnType; +let releaseSpendHome: () => void; +beforeEach(() => { + home = createTempHome("ocx-refresh-capacity-"); + releaseSpendHome = acquireOwnedSpendHome(); + inferenceSends = 0; + runTurnSends = 0; + httpPaths = []; +}); +afterEach(() => { + oauth.OAUTH_PROVIDERS[PROVIDER]!.refresh = originalRefresh; + releaseSpendHome(); + home.remove(); +}); +afterAll(() => { mock.module("../../src/server/adapter-resolve", () => originalResolver); }); + +const cases = (["adapter", "passthrough", "runTurn"] as const).flatMap(transport => + [false, true].flatMap(stream => [false, true].map(cancelled => ({ transport, stream, cancelled, ready: false, requestOnly: false })))); +cases.push({ transport: "runTurn", stream: true, cancelled: false, ready: true, requestOnly: false }); +for (const stream of [false, true]) cases.push({ transport: "runTurn", stream, cancelled: true, ready: false, requestOnly: true }); +cases.push({ transport: "runTurn", stream: true, cancelled: true, ready: true, requestOnly: true }); +for (const { transport, stream, cancelled, ready, requestOnly } of cases) { + test(`${transport} fresh auth then 401 maps saturated refresh correctly (stream=${stream}, cancelled=${cancelled}, ready=${ready}, requestOnly=${requestOnly})`, async () => { + transportMode = transport; + runTurnReady = ready; + await saveCredential(PROVIDER, { + access: ACCESS, refresh: "fixture-refresh-capacity-token", expires: Date.now() + 3_600_000, + accountId: "fixture-refresh-capacity-account", + }); + const snapshot = await oauth.getValidAccessTokenSnapshot(PROVIDER); + const controller = new AbortController(), optionController = new AbortController(); + let releaseRefresh!: () => void; + let enteredRefresh!: () => void; + const blocked = new Promise(resolve => { releaseRefresh = resolve; }); + const entered = new Promise(resolve => { enteredRefresh = resolve; }); + let refreshCalls = 0; + oauth.OAUTH_PROVIDERS[PROVIDER]!.refresh = async () => { + refreshCalls++; + enteredRefresh(); + await blocked; + return { access: "fixture-refreshed-access", refresh: "fixture-rotated-refresh", expires: Date.now() + 3_600_000 }; + }; + let waiting: Promise[] = []; + rejectInference = async () => { + // Initial auth really used the fresh token. Only once that inference is in flight do + // independent callers occupy every refresh-consumer slot for its rejected generation. + expect(refreshCalls).toBe(0); + waiting = Array.from({ length: oauth.MAX_OAUTH_TOKEN_REFRESH_WAITERS }, () => + originalForceRefresh(snapshot).catch(error => error)); + await entered; + }; + let busyRejections = 0; + const forceRefresh = spyOn(oauth, "forceRefreshOAuthAccessSnapshot").mockImplementation(async sent => { + try { return await originalForceRefresh(sent); } + catch (error) { + if (error instanceof oauth.OAuthTokenRefreshBusyError) { + busyRejections++; + // Cancel at the refusal boundary so each transport's catch must prioritize it. + if (cancelled) { controller.abort(); if (!requestOnly) optionController.abort(); } + } + throw error; + } + }); + const config = { port: 0, defaultProvider: PROVIDER, providers: { [PROVIDER]: { + adapter: transport === "runTurn" ? "devin" : transport === "passthrough" ? "openai-responses" : "openai-chat", + authMode: "oauth", baseUrl: "https://fixture.invalid/v1", models: ["fixture-model"], + fetch: async (_input: Parameters[0], init?: RequestInit) => { + httpPaths.push(new URL(_input instanceof Request ? _input.url : String(_input)).pathname); + expect(new Headers(init?.headers).get("authorization")).toBe(`Bearer ${ACCESS}`); + inferenceSends++; + await rejectInference(); + return Response.json({ error: { type: "authentication_error", message: "fixture upstream rejected credential" } }, { status: 401 }); + }, + } } } as OcxConfig; + try { + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", headers: { "content-type": "application/json" }, signal: controller.signal, + body: JSON.stringify({ model: `${PROVIDER}/fixture-model`, input: "fixture", stream }), + }), config, { model: "", provider: "" }, { abortSignal: optionController.signal }); + const body = await response.text(); + expect(inferenceSends).toBe(1); + expect(runTurnSends).toBe(transport === "runTurn" ? 1 : 0); + expect(httpPaths).toEqual(transport === "runTurn" ? [] : [transport === "passthrough" ? "/v1/responses" : "/v1/chat/completions"]); + expect(refreshCalls).toBe(1); + expect(busyRejections).toBe(1); + expect(response.status).toBe(ready ? 200 : cancelled ? 499 : 503); + expect(response.headers.get("retry-after")).toBe(cancelled || ready ? null : "1"); + expect(body).toContain(cancelled ? "client_cancelled" : "server_error"); + if (!cancelled) expect(body).toContain("OAuth token refresh capacity reached"); + expect(body).not.toContain("authentication_error"); + expect(body).not.toContain("unauthenticated"); + if (ready) { + const failure = body.split("\n").filter(line => line.startsWith("data: {")) + .map(line => JSON.parse(line.slice(6))).find(frame => frame.type === "response.failed"); + expect(failure?.response.retryable).toBe(!cancelled); + expect(failure?.response.error.type).toBe(cancelled ? "client_cancelled" : "server_error"); + } + expect(body).not.toContain(ACCESS); + expect(getAccountSet(PROVIDER)?.activeAccountId).toBe(snapshot.accountId); + expect(getAccountSet(PROVIDER)?.accounts.find(row => row.id === snapshot.accountId)?.needsReauth).not.toBe(true); + } finally { + forceRefresh.mockRestore(); + releaseRefresh(); + await Promise.all(waiting); + } + }); +} + +for (const cancelledSignal of ["request", "option"] as const) { + test(`initial refresh capacity prioritizes independent ${cancelledSignal} cancellation`, async () => { + transportMode = "adapter"; + await saveCredential(PROVIDER, { access: ACCESS, refresh: "fixture-initial-refresh", expires: 1, + accountId: "fixture-initial-account" }); + const accountId = getAccountSet(PROVIDER)!.activeAccountId; + let finish!: () => void, started!: () => void; + const blocked = new Promise(resolve => { finish = resolve; }); + const entered = new Promise(resolve => { started = resolve; }); + oauth.OAUTH_PROVIDERS[PROVIDER]!.refresh = async () => { + started(); await blocked; + return { access: "fixture-initial-new", refresh: "fixture-initial-next", expires: Date.now() + 3_600_000 }; + }; + const waiting = Array.from({ length: oauth.MAX_OAUTH_TOKEN_REFRESH_WAITERS }, () => + oauth.getValidAccessTokenForAccount(PROVIDER, accountId).catch(error => error)); + await entered; + const requestAbort = new AbortController(), optionAbort = new AbortController(); + const originalResolve = oauth.getValidAccessTokenSnapshot; + const resolve = spyOn(oauth, "getValidAccessTokenSnapshot").mockImplementation(async (...args) => { + try { return await originalResolve(...args); } + catch (error) { + expect(error).toBeInstanceOf(oauth.OAuthTokenRefreshBusyError); + (cancelledSignal === "request" ? requestAbort : optionAbort).abort(); + throw error; + } + }); + const config = { port: 0, defaultProvider: PROVIDER, providers: { [PROVIDER]: { + adapter: "openai-chat", authMode: "oauth", baseUrl: "https://fixture.invalid/v1", models: ["fixture-model"], + fetch: async () => { throw new Error("unexpected inference dispatch"); }, + } } } as OcxConfig; + try { + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", headers: { "content-type": "application/json" }, signal: requestAbort.signal, + body: JSON.stringify({ model: `${PROVIDER}/fixture-model`, input: "fixture", stream: false }), + }), config, { model: "", provider: "" }, { abortSignal: optionAbort.signal }); + expect(resolve).toHaveBeenCalledTimes(1); + expect(response.status).toBe(499); + expect(response.headers.get("retry-after")).toBeNull(); + expect(await response.text()).toContain("client_cancelled"); + expect(getAccountSet(PROVIDER)?.accounts.find(row => row.id === accountId)?.needsReauth).not.toBe(true); + } finally { resolve.mockRestore(); finish(); await Promise.all(waiting); } + }); +}