From 0cb3868b03c3d6fd51e0c3ad870a721a70784ead Mon Sep 17 00:00:00 2001 From: Epinephrine Date: Wed, 14 Oct 2026 23:10:33 -0700 Subject: [PATCH 1/6] fix(oauth): bound refresh consumers and preserve retryable capacity Bound unresolved OAuth refresh consumers while preserving stale-flight ownership evidence across refused replacement admission. Report retryable capacity separately from authentication failures, and preserve cancellation priority and lease-release assertions. (cherry picked from commit b722d746f10dafb7eeea2aabcf8b94aec6a6a5db) --- scripts/test-layout/layout.json | 2 + src/oauth/index.ts | 125 +++++++++----- src/server/responses/adapter-dispatch.ts | 11 +- src/server/responses/passthrough-dispatch.ts | 4 +- src/server/responses/request-transport.ts | 5 + src/server/responses/run-turn-execution.ts | 32 ++-- structure/runtime.md | 2 +- structure/transports/responses-failover.md | 2 +- tests/fixtures/test-layout-expected.json | 4 +- .../oauth/oauth-refresh-generic-lock.test.ts | 132 +++++++++++++++ .../oauth-refresh-stale-capacity.test.ts | 152 ++++++++++++++++++ .../kiro/kiro-leased-responses.test.ts | 12 +- .../responses-oauth-refresh-capacity.test.ts | 144 +++++++++++++++++ 13 files changed, 573 insertions(+), 54 deletions(-) create mode 100644 tests/oauth/oauth-refresh-stale-capacity.test.ts create mode 100644 tests/responses/responses-oauth-refresh-capacity.test.ts diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 16227fe5cf1..b8f892ff1c8 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -108,6 +108,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", @@ -1252,6 +1253,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 f3d0af7d047..1cd215aa999 100644 --- a/src/oauth/index.ts +++ b/src/oauth/index.ts @@ -4,6 +4,7 @@ import type { OcxConfig, OcxProviderConfig, RefreshPolicy } from "../types"; import { getDefaultConfig, initializePersistedConfigIfMissing, loadConfig, mutatePersistedConfig, observeInitialConfigState, saveConfig } from "../config"; import { resolveProviderApiKey } from "../providers/key-store"; import { projectEmail } from "../lib/privacy"; +import { createAdmissionGate } from "../lib/admission"; import { KiroTokenRefreshError, environmentKiroRoutingMetadata, loginKiro, refreshKiroToken, settleKiroLoginTransaction } from "./kiro"; import { KiroUserIdentityUnavailableError, kiroReauthIdentityMatches } from "./kiro-account-identity"; import { @@ -131,9 +132,12 @@ export type OAuthActiveTokenObservation = | { readonly kind: "unsupported" }; const MAX_OAUTH_TOKEN_REFRESH_FLIGHTS = 32; +/** Independent of client lifetime; a cancelled caller can still be awaiting a shared refresh. */ +export const MAX_OAUTH_TOKEN_REFRESH_WAITERS = 256; +const tokenRefreshWaiters = createAdmissionGate("oauth_token_refresh_waiters", MAX_OAUTH_TOKEN_REFRESH_WAITERS); 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 } const tokenRefreshes = new Map(); export class OAuthTokenRefreshBusyError extends Error { readonly code = "OAUTH_TOKEN_REFRESH_BUSY"; @@ -157,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; @@ -583,6 +588,28 @@ export function observeActiveOAuthAccessToken( }; } +// 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. +function retireSettledOAuthRefreshFlight(key: string, flight: OAuthTokenRefreshFlight): boolean { + if (!flight.settled) return false; + const [provider, accountId] = key.split("\0"); + try { + if (provider && accountId && isAnthropicOAuthInstance(provider)) { + const intent = readOAuthRefreshIntent(provider, accountId); + if (intent?.uncertain) return false; + 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 false; + } + } + } catch { return false; } + if (tokenRefreshes.get(key) === flight) tokenRefreshes.delete(key); + return true; +} + async function resolveAccessSnapshotForAccount( provider: string, accountId: string, @@ -608,45 +635,69 @@ 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) { + // Keep the owner in the registry until admission succeeds. A full waiter gate must + // not discard the only evidence distinguishing this abort from an unknown rotation. + if (existing.settled && !retireSettledOAuthRefreshFlight(key, existing)) { + throw new OAuthTokenRefreshStaleError(); + } + if (tokenRefreshes.get(key) === existing) tokenRefreshes.delete(key); + existing = undefined; } - 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(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(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/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 1b6cb6f4c44..18a754a7e4d 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, @@ -964,6 +965,10 @@ export async function prepareResponsesTransport( `${redactSecretString(err.message)}. Remove or reconfigure provider '${safeProviderName}' in the OpenCodex configuration.`, ); } + if (err instanceof OAuthTokenRefreshBusyError) { + return (options.abortSignal ?? 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 689730cfae0..ceabea1255e 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"; @@ -436,7 +436,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 => { @@ -450,9 +462,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; } @@ -475,7 +485,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; } @@ -752,13 +764,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) { @@ -897,13 +909,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 2a31215cb52..83b822212fd 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, but rejected admission retains their registry evidence until settlement can classify the matching durable intent. 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. 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 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 535ffb4c842..b59dca8de0f 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 wins with 499. 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 210221f4d50..56572f21478 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -138,7 +138,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", @@ -828,7 +828,7 @@ "oauth-login-summary.test.ts": "oauth", "oauth-manual-code.test.ts": "oauth", "oauth-open-browser-choice.test.ts": "oauth", "oauth-open-url-result.test.ts": "oauth", "oauth-provider-reconcile.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..ea4a7894715 --- /dev/null +++ b/tests/oauth/oauth-refresh-stale-capacity.test.ts @@ -0,0 +1,152 @@ +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); } + }); +} 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..5edeec5abc4 --- /dev/null +++ b/tests/responses/responses-oauth-refresh-capacity.test.ts @@ -0,0 +1,144 @@ +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 })))); +cases.push({ transport: "runTurn", stream: true, cancelled: false, ready: true }); +for (const { transport, stream, cancelled, ready } of cases) { + test(`${transport} fresh auth then 401 maps saturated refresh correctly (stream=${stream}, cancelled=${cancelled}, ready=${ready})`, 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(); + 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(); + } + 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: controller.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) expect(body).toContain("response.failed"); + 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); + } + }); +} From 9ff799102686450b368beb53c002c1d181ac1740 Mon Sep 17 00:00:00 2001 From: JUN Date: Thu, 15 Oct 2026 16:25:54 +0900 Subject: [PATCH 2/6] fix(oauth): retire stale owners before admitting replacements Keep stale registry ownership through definitive-result cleanup and durable retirement regardless of spare consumer capacity. Recheck the installed flight when concurrent replacements resume. Co-authored-by: Epinephrine --- src/oauth/index.ts | 12 +++-- structure/runtime.md | 2 +- .../oauth-refresh-stale-capacity.test.ts | 49 +++++++++++++++++++ 3 files changed, 57 insertions(+), 6 deletions(-) diff --git a/src/oauth/index.ts b/src/oauth/index.ts index 1cd215aa999..35dbe1e7854 100644 --- a/src/oauth/index.ts +++ b/src/oauth/index.ts @@ -647,13 +647,15 @@ async function resolveAccessSnapshotForAccount( if (!waiter) throw new OAuthTokenRefreshBusyError(); try { if (existing && replacedStaleFlight) { - // Keep the owner in the registry until admission succeeds. A full waiter gate must - // not discard the only evidence distinguishing this abort from an unknown rotation. - if (existing.settled && !retireSettledOAuthRefreshFlight(key, existing)) { + // 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(key, existing)) { throw new OAuthTokenRefreshStaleError(); } - if (tokenRefreshes.get(key) === existing) tokenRefreshes.delete(key); - existing = undefined; + // Other admitted replacements may have resumed first and installed the next flight. + existing = tokenRefreshes.get(key); } if (existing) { logOAuthEvent("OAuth refresh joined existing operation", { provider, accountId }); diff --git a/structure/runtime.md b/structure/runtime.md index 83b822212fd..f2e93165e4d 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 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, but rejected admission retains their registry evidence until settlement can classify the matching durable intent. 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. 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 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. +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. 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/tests/oauth/oauth-refresh-stale-capacity.test.ts b/tests/oauth/oauth-refresh-stale-capacity.test.ts index ea4a7894715..46e8df1e55d 100644 --- a/tests/oauth/oauth-refresh-stale-capacity.test.ts +++ b/tests/oauth/oauth-refresh-stale-capacity.test.ts @@ -150,3 +150,52 @@ for (const outcome of ["rejected", "rotated", "marker-io"] as const) { } 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"); + 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); + 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 { marker.mockRestore(); clock.mockRestore(); finish(); await old; await replacement; } + }); +} From 043971280eb7a3e5369c757ecfb7fb4896b3527f Mon Sep 17 00:00:00 2001 From: JUN Date: Thu, 15 Oct 2026 16:25:54 +0900 Subject: [PATCH 3/6] fix(responses): preserve cancellation at refresh capacity Observe both independent cancellation signals at initial refresh refusal and document retry behavior for OAuth capacity. Co-authored-by: Epinephrine --- .../troubleshooting/codex-cannot-sign-in.md | 14 ++++++ src/server/responses/request-transport.ts | 2 +- .../responses-oauth-refresh-capacity.test.ts | 44 +++++++++++++++++++ 3 files changed, 59 insertions(+), 1 deletion(-) 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/src/server/responses/request-transport.ts b/src/server/responses/request-transport.ts index 18a754a7e4d..1142367e0fa 100644 --- a/src/server/responses/request-transport.ts +++ b/src/server/responses/request-transport.ts @@ -966,7 +966,7 @@ export async function prepareResponsesTransport( ); } if (err instanceof OAuthTokenRefreshBusyError) { - return (options.abortSignal ?? req.signal).aborted ? clientCancelledResponse() + return options.abortSignal?.aborted || req.signal.aborted ? clientCancelledResponse() : formatErrorResponse(503, "server_error", err.message, { retryAfter: "1" }); } if (err instanceof OAuthAccountPausedError) { diff --git a/tests/responses/responses-oauth-refresh-capacity.test.ts b/tests/responses/responses-oauth-refresh-capacity.test.ts index 5edeec5abc4..60001efa66f 100644 --- a/tests/responses/responses-oauth-refresh-capacity.test.ts +++ b/tests/responses/responses-oauth-refresh-capacity.test.ts @@ -142,3 +142,47 @@ for (const { transport, stream, cancelled, ready } of cases) { } }); } + +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); } + }); +} From 61df34faa07f90cd43c04e49cd46f2bb7927f2b2 Mon Sep 17 00:00:00 2001 From: JUN Date: Thu, 15 Oct 2026 16:34:43 +0900 Subject: [PATCH 4/6] fix(responses): link both client signals through runTurn recovery Address the hosted Codex P2: request-only cancellation takes priority during refresh refusal, including committed SSE. Remove both listeners on cleanup. Co-authored-by: Epinephrine --- src/server/responses/run-turn-execution.ts | 8 ++++--- structure/transports/responses-failover.md | 2 +- .../responses-oauth-refresh-capacity.test.ts | 23 ++++++++++++------- 3 files changed, 21 insertions(+), 12 deletions(-) diff --git a/src/server/responses/run-turn-execution.ts b/src/server/responses/run-turn-execution.ts index ceabea1255e..da79907c8a9 100644 --- a/src/server/responses/run-turn-execution.ts +++ b/src/server/responses/run-turn-execution.ts @@ -77,7 +77,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, @@ -127,7 +127,7 @@ export async function executeResponsesRunTurn( >, completionPolicy: Pick, ): Promise { - const { options, logCtx, config } = requestContext; + const { req, options, logCtx, config } = requestContext; const { selectionIsCurrent, adapterBindings, @@ -186,7 +186,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 diff --git a/structure/transports/responses-failover.md b/structure/transports/responses-failover.md index b59dca8de0f..7af58b85267 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. -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 wins with 499. 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`. +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/responses/responses-oauth-refresh-capacity.test.ts b/tests/responses/responses-oauth-refresh-capacity.test.ts index 60001efa66f..85553083687 100644 --- a/tests/responses/responses-oauth-refresh-capacity.test.ts +++ b/tests/responses/responses-oauth-refresh-capacity.test.ts @@ -59,10 +59,12 @@ afterEach(() => { 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 })))); -cases.push({ transport: "runTurn", stream: true, cancelled: false, ready: true }); -for (const { transport, stream, cancelled, ready } of cases) { - test(`${transport} fresh auth then 401 maps saturated refresh correctly (stream=${stream}, cancelled=${cancelled}, ready=${ready})`, async () => { + [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, { @@ -70,7 +72,7 @@ for (const { transport, stream, cancelled, ready } of cases) { accountId: "fixture-refresh-capacity-account", }); const snapshot = await oauth.getValidAccessTokenSnapshot(PROVIDER); - const controller = new AbortController(); + const controller = new AbortController(), optionController = new AbortController(); let releaseRefresh!: () => void; let enteredRefresh!: () => void; const blocked = new Promise(resolve => { releaseRefresh = resolve; }); @@ -98,7 +100,7 @@ for (const { transport, stream, cancelled, ready } of cases) { if (error instanceof oauth.OAuthTokenRefreshBusyError) { busyRejections++; // Cancel at the refusal boundary so each transport's catch must prioritize it. - if (cancelled) controller.abort(); + if (cancelled) { controller.abort(); if (!requestOnly) optionController.abort(); } } throw error; } @@ -118,7 +120,7 @@ for (const { transport, stream, cancelled, ready } of cases) { 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: controller.signal }); + }), config, { model: "", provider: "" }, { abortSignal: optionController.signal }); const body = await response.text(); expect(inferenceSends).toBe(1); expect(runTurnSends).toBe(transport === "runTurn" ? 1 : 0); @@ -131,7 +133,12 @@ for (const { transport, stream, cancelled, ready } of cases) { if (!cancelled) expect(body).toContain("OAuth token refresh capacity reached"); expect(body).not.toContain("authentication_error"); expect(body).not.toContain("unauthenticated"); - if (ready) expect(body).toContain("response.failed"); + 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); From 6fba175cac2ff1a55e5c2d92ea85d9e5f2a5c291 Mon Sep 17 00:00:00 2001 From: JUN Date: Thu, 15 Oct 2026 16:34:43 +0900 Subject: [PATCH 5/6] fix(oauth): report deferred refresh retirement without identity Address the CodeRabbit observability finding with one bounded diagnostic per retained flight, a closed failure reason, and no account or credential fields. Co-authored-by: Epinephrine --- src/oauth/index.ts | 15 +++++++++++---- structure/runtime.md | 2 +- tests/oauth/oauth-refresh-stale-capacity.test.ts | 6 +++++- 3 files changed, 17 insertions(+), 6 deletions(-) diff --git a/src/oauth/index.ts b/src/oauth/index.ts index 35dbe1e7854..9c5b001a110 100644 --- a/src/oauth/index.ts +++ b/src/oauth/index.ts @@ -137,7 +137,7 @@ export const MAX_OAUTH_TOKEN_REFRESH_WAITERS = 256; const tokenRefreshWaiters = createAdmissionGate("oauth_token_refresh_waiters", MAX_OAUTH_TOKEN_REFRESH_WAITERS); 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; settled?: boolean } +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"; @@ -594,18 +594,25 @@ export function observeActiveOAuthAccessToken( function retireSettledOAuthRefreshFlight(key: string, flight: OAuthTokenRefreshFlight): 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 false; + 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 false; + if (!retired) return defer("intent_changed"); } } - } catch { return false; } + } catch { return defer("storage_failure"); } if (tokenRefreshes.get(key) === flight) tokenRefreshes.delete(key); return true; } diff --git a/structure/runtime.md b/structure/runtime.md index f2e93165e4d..bb85793ac82 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 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. 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. +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/tests/oauth/oauth-refresh-stale-capacity.test.ts b/tests/oauth/oauth-refresh-stale-capacity.test.ts index 46e8df1e55d..5a0706c80a6 100644 --- a/tests/oauth/oauth-refresh-stale-capacity.test.ts +++ b/tests/oauth/oauth-refresh-stale-capacity.test.ts @@ -171,6 +171,7 @@ for (const outcome of ["rejected", "marker-io"] as const) { 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 { @@ -190,12 +191,15 @@ for (const outcome of ["rejected", "marker-io"] as const) { } 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 { marker.mockRestore(); clock.mockRestore(); finish(); await old; await replacement; } + } finally { info.mockRestore(); marker.mockRestore(); clock.mockRestore(); finish(); await old; await replacement; } }); } From 75a8da42f8d03d6afecdb50761dc7ff92f21d97d Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 16 Oct 2026 05:08:15 +0900 Subject: [PATCH 6/6] refactor(oauth): move refresh-flight retirement into its own module Pure move of retireSettledOAuthRefreshFlight, the refresh-waiter gate and OAuthRefreshFlightEvidence out of src/oauth/index.ts, so this branch merged with current dev stays under the 2000-line file-size ratchet. The registry map is passed explicitly; behavior is unchanged. --- src/oauth/index.ts | 42 +++------------------- src/oauth/refresh-flight-retirement.ts | 48 ++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 37 deletions(-) create mode 100644 src/oauth/refresh-flight-retirement.ts diff --git a/src/oauth/index.ts b/src/oauth/index.ts index b805d753c07..04c63e7e26f 100644 --- a/src/oauth/index.ts +++ b/src/oauth/index.ts @@ -5,7 +5,6 @@ import type { OcxConfig, OcxProviderConfig, RefreshPolicy } from "../types"; import { getDefaultConfig, initializePersistedConfigIfMissing, loadConfig, mutatePersistedConfig, observeInitialConfigState, saveConfig } from "../config"; import { resolveProviderApiKey } from "../providers/key-store"; import { projectEmail } from "../lib/privacy"; -import { createAdmissionGate } from "../lib/admission"; import { KiroTokenRefreshError, environmentKiroRoutingMetadata, loginKiro, refreshKiroToken, settleKiroLoginTransaction } from "./kiro"; import { KiroUserIdentityUnavailableError, kiroReauthIdentityMatches } from "./kiro-account-identity"; import { @@ -60,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,11 +136,7 @@ export type OAuthActiveTokenObservation = | { readonly kind: "unsupported" }; const MAX_OAUTH_TOKEN_REFRESH_FLIGHTS = 32; -/** Independent of client lifetime; a cancelled caller can still be awaiting a shared refresh. */ -export const MAX_OAUTH_TOKEN_REFRESH_WAITERS = 256; -const tokenRefreshWaiters = createAdmissionGate("oauth_token_refresh_waiters", MAX_OAUTH_TOKEN_REFRESH_WAITERS); 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; settled?: boolean; retirementDeferredLogged?: boolean } const tokenRefreshes = new Map(); export class OAuthTokenRefreshBusyError extends Error { @@ -614,35 +611,6 @@ export function observeActiveOAuthAccessToken( }; } -// 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. -function retireSettledOAuthRefreshFlight(key: string, flight: OAuthTokenRefreshFlight): 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; -} - async function resolveAccessSnapshotForAccount( provider: string, accountId: string, @@ -684,7 +652,7 @@ async function resolveAccessSnapshotForAccount( // 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(key, existing)) { + if (tokenRefreshes.get(key) === existing && !retireSettledOAuthRefreshFlight(tokenRefreshes, key, existing)) { throw new OAuthTokenRefreshStaleError(); } // Other admitted replacements may have resumed first and installed the next flight. @@ -696,7 +664,7 @@ async function resolveAccessSnapshotForAccount( } if (tokenRefreshes.size >= MAX_OAUTH_TOKEN_REFRESH_FLIGHTS) { for (const [staleKey, stale] of tokenRefreshes) { - if (stale.abort.signal.reason instanceof OAuthTokenRefreshStaleError) retireSettledOAuthRefreshFlight(staleKey, stale); + if (stale.abort.signal.reason instanceof OAuthTokenRefreshStaleError) retireSettledOAuthRefreshFlight(tokenRefreshes, staleKey, stale); } if (tokenRefreshes.size >= MAX_OAUTH_TOKEN_REFRESH_FLIGHTS) throw new OAuthTokenRefreshBusyError(); } @@ -726,7 +694,7 @@ async function resolveAccessSnapshotForAccount( flight.settled = true; if (tokenRefreshes.get(key) !== flight) return; if (abort.signal.reason instanceof OAuthTokenRefreshStaleError) { - retireSettledOAuthRefreshFlight(key, flight); + retireSettledOAuthRefreshFlight(tokenRefreshes, key, flight); } else tokenRefreshes.delete(key); }); flight.promise = refresh; 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; +} +