From 4a7016be69140b0edf04cf43dc7cb81184829e93 Mon Sep 17 00:00:00 2001 From: iceteaSA <171169159+iceteaSA@users.noreply.github.com> Date: Thu, 21 May 2026 20:19:45 +0200 Subject: [PATCH 1/5] feat(core): add QuotaManager as the single quota path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Unified quota cache + API gateway for main and fallback quota. All consumers share one QuotaManager instance for consistent caching, inflight dedup, and 429 backoff. - One quota path: FallbackAccountManager.refreshAccountQuota() now routes through QuotaManager.refreshFallback() (dedup + backoff) instead of calling fetchOAuthQuotaSnapshot() directly. Falls back to a direct fetch only when no QuotaManager is wired. - 401 is treated as auth/token expiry, not rate-limiting: it no longer arms quota backoff, so the caller's token-refresh-and-retry (and other accounts) are not blocked. - /claude-quota forces a real refresh for main (refreshMain) and all fallbacks (refreshQuotaForAllAccounts({ force: true })) and PERSISTS the refreshed fallback snapshots to anthropic-auth.json, clearing stale errors. 429 backoff still respected (returns last cached numbers when backed off). - Main quota refresh is wired to the request path: needsRefresh(requestCount) drives an every-N background refresh (quota.refreshEveryNRequests), matching the existing time-staleness trigger. - Fallback quota refreshes proactively too: the existing background timer ticks refreshQuotaForDueAccounts(), and the active fallback route refreshes on the same every-N cadence at its markUsed() point (idle fallbacks untouched). - getUsableFallbackAccounts() skips the request-time refresh when an account's quota API is backed off, mirroring the background path (no API hammering). - Seeds from persisted storage on construction to avoid cold-start API calls. The main-account fail-closed killswitch (synthetic 429 when quota API is backed off) is intentionally NOT in this base — it moves to feat/killswitch where it belongs, keeping quota-manager a pure routing optimization that never blocks the main request. --- packages/core/src/accounts.ts | 174 +++++++- packages/core/src/index.ts | 1 + packages/core/src/quota-manager.ts | 399 ++++++++++++++++++ packages/opencode/src/index.ts | 207 ++++----- packages/opencode/src/tests/accounts.test.ts | 41 ++ packages/opencode/src/tests/index.test.ts | 112 ++++- .../opencode/src/tests/quota-manager.test.ts | 324 ++++++++++++++ 7 files changed, 1136 insertions(+), 122 deletions(-) create mode 100644 packages/core/src/quota-manager.ts create mode 100644 packages/opencode/src/tests/quota-manager.test.ts diff --git a/packages/core/src/accounts.ts b/packages/core/src/accounts.ts index b4dd5e12f..f6d89c848 100644 --- a/packages/core/src/accounts.ts +++ b/packages/core/src/accounts.ts @@ -75,8 +75,12 @@ export type AccountStorage = { quota?: { enabled?: boolean checkIntervalMinutes?: number + refreshEveryNRequests?: number minimumRemaining?: Partial> failClosedOnUnknownQuota?: boolean + mainQuota?: OAuthQuotaSnapshot + mainQuotaCheckedAt?: number + mainLastQuotaApiError?: AccountOperationError } claudeCache?: { enabled?: boolean @@ -121,6 +125,7 @@ export type AccountManagerOptions = { now?: () => number fetchImpl?: typeof fetch configPath?: string + quotaManager?: import('./quota-manager.ts').QuotaManager } export type AccountRefreshError = { @@ -135,6 +140,9 @@ const DEFAULT_REFRESH_INTERVAL_MINUTES = 10 const MIN_REFRESH_RETRY_DELAY_MS = 5 * 60_000 const MAX_REFRESH_RETRY_DELAY_MS = 60 * 60_000 const NON_TRANSIENT_REFRESH_RETRY_DELAY_MS = 24 * 60 * 60_000 +const MIN_QUOTA_RETRY_DELAY_MS = 60_000 +const MAX_QUOTA_RETRY_DELAY_MS = 15 * 60_000 +const NON_TRANSIENT_QUOTA_RETRY_DELAY_MS = 5 * 60_000 const DEFAULT_QUOTA_CHECK_INTERVAL_MINUTES = 5 const DEFAULT_MINIMUM_REMAINING: Record = { five_hour: 0, @@ -673,12 +681,76 @@ export function formatRefreshBackoffMessage( return `Claude OAuth refresh is backed off for ${seconds}s after: ${error.message}` } +export function buildQuotaOperationError(input: { + error: unknown + now: number + previous?: AccountOperationError +}): AccountOperationError { + const previousRetryCount = input.previous?.retryCount ?? 0 + const retryCount = previousRetryCount + 1 + const delay = isTransientQuotaError(input.error) + ? Math.min( + MAX_QUOTA_RETRY_DELAY_MS, + MIN_QUOTA_RETRY_DELAY_MS * 2 ** Math.min(retryCount - 1, 6), + ) + : NON_TRANSIENT_QUOTA_RETRY_DELAY_MS + return { + message: formatErrorMessage(input.error), + checkedAt: input.now, + nextRetryAt: input.now + delay, + retryCount, + } +} + +export function quotaBackoffActive( + error: AccountOperationError | undefined, + now: number, +): boolean { + if (!error?.nextRetryAt || error.nextRetryAt <= now) return false + return true +} + +export function formatQuotaBackoffMessage( + error: AccountOperationError, + now: number, +): string { + const seconds = Math.max( + 1, + Math.ceil(((error.nextRetryAt ?? now) - now) / 1000), + ) + return `Quota API backed off for ${seconds}s after: ${error.message}` +} + export function getQuotaCheckIntervalMs(storage: AccountStorage | null) { const minutes = storage?.quota?.checkIntervalMinutes ?? DEFAULT_QUOTA_CHECK_INTERVAL_MINUTES return Math.max(1, minutes) * 60_000 } +export function getPersistedMainQuota( + storage: AccountStorage | null, +): { quota: OAuthQuotaSnapshot; checkedAt: number } | null { + if (!storage?.quota?.mainQuota || !storage.quota.mainQuotaCheckedAt) + return null + return { + quota: storage.quota.mainQuota, + checkedAt: storage.quota.mainQuotaCheckedAt, + } +} + +/** + * How often (in requests) to force a quota refresh, independent of the timer. + * Returns 0 when disabled (default). + */ +export function getQuotaRefreshEveryNRequests( + storage: AccountStorage | null, +): number { + const n = storage?.quota?.refreshEveryNRequests + return typeof n === 'number' && Number.isFinite(n) && n > 0 + ? Math.floor(n) + : 0 +} + function failClosedOnUnknownQuota(storage: AccountStorage | null) { return ( storage?.quota?.failClosedOnUnknownQuota ?? @@ -875,10 +947,11 @@ function recordQuotaRefreshError( error: unknown, now: number, ) { - account.lastQuotaRefreshError = { - message: formatErrorMessage(error), - checkedAt: now, - } + account.lastQuotaRefreshError = buildQuotaOperationError({ + error, + now, + previous: account.lastQuotaRefreshError, + }) if (error instanceof ClaudeOAuthRefreshError) { recordRefreshError(account, error, now) } @@ -898,11 +971,37 @@ export class FallbackAccountManager { private readonly refreshPromises = new Map>() private refreshTimer: ReturnType | null = null private quotaTimer: ReturnType | null = null + readonly quotaManager: import('./quota-manager.ts').QuotaManager | null constructor(options: AccountManagerOptions = {}) { this.now = options.now ?? Date.now this.fetchImpl = options.fetchImpl ?? fetch this.configPath = options.configPath ?? getAccountStoragePath() + this.quotaManager = options.quotaManager ?? null + } + + /** + * Seed QuotaManager from persisted account.quota if no cache entry exists + * yet. Prevents unnecessary API calls when the on-disk snapshot is fresh. + */ + private seedFallbackQuota( + account: OAuthAccount, + storage: AccountStorage, + ): void { + if (!this.quotaManager) return + if (this.quotaManager.getFallback(account.id)) return + if (!account.quota) return + const checkedAt = Math.max( + account.quota.five_hour?.checkedAt ?? 0, + account.quota.seven_day?.checkedAt ?? 0, + ) + if (checkedAt <= 0) return + const checkInterval = getQuotaCheckIntervalMs(storage) + this.quotaManager.setFallback(account.id, { + quota: account.quota, + refreshAfter: checkedAt + checkInterval, + checkedAt, + }) } async load() { @@ -958,7 +1057,18 @@ export class FallbackAccountManager { next = await this.refreshAccount(next, storage) changed = true } - if (quotaIsStale(next, storage, this.now())) { + this.seedFallbackQuota(next, storage) + const stale = this.quotaManager + ? this.quotaManager.isFallbackStale(next.id) + : quotaIsStale(next, storage, this.now()) + // Skip the request-time refresh when this account's quota API is + // backed off (recent 429/5xx). Hitting it again would extend the + // backoff; evaluate policy on the cached/seeded quota instead. Mirrors + // the background refreshQuotaForDueAccounts() guard. + if ( + stale && + !quotaBackoffActive(next.lastQuotaRefreshError, this.now()) + ) { next = await this.refreshAccountQuota(next, storage) changed = true } @@ -1073,7 +1183,16 @@ export class FallbackAccountManager { next = await this.refreshAccount(next, storage) changed = true } - if (!quotaIsStale(next, storage, this.now())) continue + if (quotaBackoffActive(next.lastQuotaRefreshError, this.now())) { + continue + } + this.seedFallbackQuota(next, storage) + // Use QuotaManager staleness when available (shared cache); + // fall back to per-account on-disk staleness otherwise. + const stale = this.quotaManager + ? this.quotaManager.isFallbackStale(next.id) + : quotaIsStale(next, storage, this.now()) + if (!stale) continue await this.refreshAccountQuota(next, storage) changed = true } catch (error) { @@ -1086,10 +1205,11 @@ export class FallbackAccountManager { if (changed) await this.save(storage) } - async refreshQuotaForAllAccounts() { + async refreshQuotaForAllAccounts(options: { force?: boolean } = {}) { const storage = await this.load() const errors: AccountRefreshError[] = [] if (!storage || !quotaEnabled(storage)) return { storage, errors } + const force = options.force ?? false let changed = false for (const account of storage.accounts) { if (account.enabled === false) continue @@ -1108,7 +1228,10 @@ export class FallbackAccountManager { next = await this.refreshAccount(next, storage) changed = true } - if (!quotaIsStale(next, storage, this.now())) { + // force (manual /claude-quota) bypasses the staleness skip to fetch + // fresh numbers on demand. refreshAccountQuota still respects 429 + // backoff via QuotaManager.refreshFallback. + if (!force && !quotaIsStale(next, storage, this.now())) { if (next.lastQuotaRefreshError) { next.lastQuotaRefreshError = undefined updateStoredAccount(storage, next) @@ -1296,12 +1419,19 @@ export class FallbackAccountManager { if (!target.access) { throw new Error(`Fallback account ${account.id} has no access token`) } + // Unify on the shared QuotaManager when present: it adds inflight + // deduplication and 429 backoff gating around the same quota API. Fall back + // to a direct fetch only when no QuotaManager is wired (e.g. in isolation). + const fetchSnapshot = (accessToken: string) => + this.quotaManager + ? this.quotaManager.refreshFallback(target.id, accessToken) + : fetchOAuthQuotaSnapshot({ + accessToken, + fetchImpl: this.fetchImpl, + now: this.now, + }) try { - target.quota = await fetchOAuthQuotaSnapshot({ - accessToken: target.access, - fetchImpl: this.fetchImpl, - now: this.now, - }) + target.quota = await fetchSnapshot(target.access) } catch (error) { const message = error instanceof Error ? error.message : String(error) if (!message.includes('Claude quota check failed: 401')) throw error @@ -1309,14 +1439,22 @@ export class FallbackAccountManager { force: true, }) if (!target.access) throw error - target.quota = await fetchOAuthQuotaSnapshot({ - accessToken: target.access, - fetchImpl: this.fetchImpl, - now: this.now, - }) + // 401 does not arm QuotaManager backoff, so this retry proceeds. + target.quota = await fetchSnapshot(target.access) } target.lastQuotaRefreshError = undefined updateStoredAccount(storage, target) + // Sync to shared QuotaManager so all consumers see the same cache. The + // refreshFallback path already cached the snapshot; re-set here so + // refreshAfter reflects this storage's check interval consistently. + if (this.quotaManager && target.quota) { + const now = this.now() + this.quotaManager.setFallback(target.id, { + quota: target.quota, + refreshAfter: now + getQuotaCheckIntervalMs(storage), + checkedAt: now, + }) + } return target } } diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 3264df338..3c2b19ec4 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -9,6 +9,7 @@ export * from './dump.ts' export * from './fast.ts' export * from './logger.ts' export * from './pkce.ts' +export * from './quota-manager.ts' export * from './quotas.ts' export * from './relay.ts' export * from './routing.ts' diff --git a/packages/core/src/quota-manager.ts b/packages/core/src/quota-manager.ts new file mode 100644 index 000000000..a281e8e40 --- /dev/null +++ b/packages/core/src/quota-manager.ts @@ -0,0 +1,399 @@ +/** + * Unified quota cache and API gateway. + * + * Single source of truth for main + fallback quota state. All consumers + * share one QuotaManager instance so they see the same in-memory cache. + * Handles deduplication, rate-limiting (429 backoff), and staleness. + */ + +import type { + AccountOperationError, + AccountStorage, + OAuthAccount, + OAuthQuotaSnapshot, +} from './accounts.ts' +import { + acquireRefreshFileLock, + buildQuotaOperationError, + fetchOAuthQuotaSnapshot, + getPersistedMainQuota, + getQuotaCheckIntervalMs, + getQuotaNextRefreshAt, + getQuotaRefreshEveryNRequests, + quotaBackoffActive, +} from './accounts.ts' + +// Capture real setTimeout before tests can mock globalThis.setTimeout +const nativeSetTimeout = globalThis.setTimeout + +// --------------------------------------------------------------------------- +// Types +// --------------------------------------------------------------------------- + +export type QuotaEntry = { + quota: OAuthQuotaSnapshot + refreshAfter: number // Unix ms — earliest next refresh + checkedAt: number // when snapshot was fetched +} + +export type QuotaManagerOptions = { + storage: AccountStorage | null + fetchImpl?: typeof fetch + now?: () => number + onMainQuotaFetched?: (quota: OAuthQuotaSnapshot, checkedAt: number) => void + onApiError?: (error: AccountOperationError) => void +} + +// --------------------------------------------------------------------------- +// Class +// --------------------------------------------------------------------------- + +export class QuotaManager { + // --- State --- + private main: QuotaEntry | null = null + private mainAccessToken: string | null = null + private fallbacks = new Map() + + // --- Inflight deduplication --- + private inflightMain: Promise | null = null + private inflightFallbacks = new Map>() + + // --- Rate-limiting --- + private lastApiError: AccountOperationError | undefined = undefined + + // --- Serial API gate (prevents concurrent quota API calls) --- + private apiGate: Promise = Promise.resolve() + private lastApiCallAt = 0 + + // --- Config --- + private storage: AccountStorage | null + private readonly fetchImpl: typeof fetch + private readonly now: () => number + private readonly onMainQuotaFetched: QuotaManagerOptions['onMainQuotaFetched'] + private readonly onApiError: QuotaManagerOptions['onApiError'] + + constructor(opts: QuotaManagerOptions) { + this.storage = opts.storage + this.fetchImpl = opts.fetchImpl ?? fetch + this.now = opts.now ?? Date.now + this.onMainQuotaFetched = opts.onMainQuotaFetched + this.onApiError = opts.onApiError + + // Seed main quota from persisted storage + const persisted = getPersistedMainQuota(opts.storage) + if (persisted) { + this.main = { + quota: persisted.quota, + refreshAfter: getQuotaNextRefreshAt( + persisted.quota, + opts.storage, + persisted.checkedAt, + ), + checkedAt: persisted.checkedAt, + } + } + + // Seed backoff state from persisted storage + const persistedError = opts.storage?.quota?.mainLastQuotaApiError + if (persistedError && quotaBackoffActive(persistedError, this.now())) { + this.lastApiError = persistedError + } + } + + // ========================================================================= + // Get (synchronous, from cache) + // ========================================================================= + + getMain(): QuotaEntry | null { + return this.main + } + + getFallback(accountId: string): QuotaEntry | null { + return this.fallbacks.get(accountId) ?? null + } + + getAllFallbacks(): Map { + return this.fallbacks + } + + // ========================================================================= + // Set (manual inject — seeding from persisted account.quota on boot) + // ========================================================================= + + setMain(accessToken: string, entry: QuotaEntry): void { + this.mainAccessToken = accessToken + this.main = entry + } + + setFallback(accountId: string, entry: QuotaEntry): void { + this.fallbacks.set(accountId, entry) + } + + // ========================================================================= + // Refresh (async, deduplicated, rate-limited) + // ========================================================================= + + async refreshMain(accessToken: string): Promise { + // If token changed, invalidate cache + if (this.mainAccessToken && this.mainAccessToken !== accessToken) { + this.main = null + this.mainAccessToken = null + } + + // Deduplicate — return in-flight promise if already fetching + if (this.inflightMain) return this.inflightMain + + // Rate-limit — if API recently 429'd, return stale or throw + if (this.isBackedOff()) { + if (this.main) return this.main.quota + throw new Error('Quota API rate-limited — try again later') + } + + this.inflightMain = this._fetchMain(accessToken) + return this.inflightMain + } + + async refreshFallback( + accountId: string, + accessToken: string, + ): Promise { + // Deduplicate + const inflight = this.inflightFallbacks.get(accountId) + if (inflight) return inflight + + // Rate-limit + if (this.isBackedOff()) { + const cached = this.fallbacks.get(accountId) + if (cached) return cached.quota + throw new Error('Quota API rate-limited — try again later') + } + + const promise = this._fetchFallback(accountId, accessToken) + this.inflightFallbacks.set(accountId, promise) + return promise + } + + async refreshAllFallbacks(accounts: OAuthAccount[]): Promise { + const now = this.now() + + for (const account of accounts) { + if (account.enabled === false) continue + if (!account.access) continue + + const cached = this.fallbacks.get(account.id) + if (cached && now < cached.refreshAfter) continue + + try { + await this.refreshFallback(account.id, account.access) + } catch { + // Best-effort — keep stale cache entry if fetch fails + } + } + } + + /** + * Fire-and-forget refresh. Does not await, swallows errors. + */ + refreshMainInBackground(accessToken: string): void { + if (this.inflightMain) return + if (this.isBackedOff()) return + void this.refreshMain(accessToken).catch(() => {}) + } + + // ========================================================================= + // Staleness queries + // ========================================================================= + + isMainStale(): boolean { + if (!this.main) return true + return this.now() >= this.main.refreshAfter + } + + isFallbackStale(accountId: string): boolean { + const entry = this.fallbacks.get(accountId) + if (!entry) return true + return this.now() >= entry.refreshAfter + } + + shouldRefreshOnRequestCount(requestCount: number): boolean { + const everyN = getQuotaRefreshEveryNRequests(this.storage) + if (everyN <= 0) return false + return requestCount > 0 && requestCount % everyN === 0 + } + + /** + * Combined check: should a refresh happen right now? + * True if main is stale by time OR triggered by request count. + */ + needsRefresh(requestCount: number): boolean { + return this.isMainStale() || this.shouldRefreshOnRequestCount(requestCount) + } + + // ========================================================================= + // Config + // ========================================================================= + + updateStorage(storage: AccountStorage | null): void { + this.storage = storage + } + + /** + * Seed fallback cache entries from persisted account.quota data. + * Only seeds accounts that don't already have a cache entry. + * Prevents unnecessary API calls when persisted quota is still fresh. + */ + seedFallbacksFromAccounts(accounts: OAuthAccount[]): void { + const checkInterval = getQuotaCheckIntervalMs(this.storage) + for (const account of accounts) { + if (account.enabled === false) continue + if (this.fallbacks.has(account.id)) continue + if (!account.quota) continue + const checkedAt = Math.max( + account.quota.five_hour?.checkedAt ?? 0, + account.quota.seven_day?.checkedAt ?? 0, + ) + if (checkedAt <= 0) continue + this.fallbacks.set(account.id, { + quota: account.quota, + refreshAfter: checkedAt + checkInterval, + checkedAt, + }) + } + } + + /** + * Whether the API is currently in backoff due to a recent error. + */ + isBackedOff(): boolean { + return quotaBackoffActive(this.lastApiError, this.now()) + } + + getLastApiError(): AccountOperationError | undefined { + return this.lastApiError + } + + // ========================================================================= + // Private + // ========================================================================= + + /** Minimum gap between consecutive quota API calls (ms). */ + private static readonly API_CALL_GAP_MS = 1_000 + + /** + * Serialize API calls through a shared gate so only one + * quota API request runs at a time, with a minimum gap + * between calls. Prevents concurrent and rapid-fire calls + * from triggering Anthropic's rate limits. + */ + private _enqueueApiFetch(fn: () => Promise): Promise { + const gatedFn = async (): Promise => { + // Wait until minimum gap since last API call + const elapsed = this.now() - this.lastApiCallAt + if (elapsed < QuotaManager.API_CALL_GAP_MS) { + await new Promise((r) => { + const id = nativeSetTimeout(r, QuotaManager.API_CALL_GAP_MS - elapsed) + if (typeof id === 'object' && 'unref' in id) id.unref() + }) + } + this.lastApiCallAt = this.now() + return fn() + } + const queued = this.apiGate.then(gatedFn, gatedFn) + this.apiGate = queued.catch(() => {}) + return queued + } + + private async _fetchMain(accessToken: string): Promise { + return this._enqueueApiFetch(async () => { + try { + // Re-check backoff inside gate — may have been set by + // a preceding queued call while we waited + if (this.isBackedOff()) { + if (this.main) return this.main.quota + throw new Error('Quota API rate-limited — try again later') + } + const fileLock = await acquireRefreshFileLock({ + name: 'opencode-main-quota-refresh', + ttlMs: 30_000, + }) + if (!fileLock) { + if (this.main) return this.main.quota + throw new Error('Quota refresh is already in progress') + } + try { + const quota = await fetchOAuthQuotaSnapshot({ + accessToken, + fetchImpl: this.fetchImpl, + now: this.now, + }) + const now = this.now() + this.mainAccessToken = accessToken + this.main = { + quota, + refreshAfter: getQuotaNextRefreshAt(quota, this.storage, now), + checkedAt: now, + } + this.lastApiError = undefined + this.onMainQuotaFetched?.(quota, now) + return quota + } catch (error) { + this._handleFetchError(error) + throw error + } finally { + await fileLock.release() + } + } finally { + this.inflightMain = null + } + }) + } + + private async _fetchFallback( + accountId: string, + accessToken: string, + ): Promise { + return this._enqueueApiFetch(async () => { + try { + // Re-check backoff inside gate + if (this.isBackedOff()) { + const cached = this.fallbacks.get(accountId) + if (cached) return cached.quota + throw new Error('Quota API rate-limited — try again later') + } + const quota = await fetchOAuthQuotaSnapshot({ + accessToken, + fetchImpl: this.fetchImpl, + now: this.now, + }) + const now = this.now() + this.fallbacks.set(accountId, { + quota, + refreshAfter: now + getQuotaCheckIntervalMs(this.storage), + checkedAt: now, + }) + this.lastApiError = undefined + return quota + } catch (error) { + this._handleFetchError(error) + throw error + } finally { + this.inflightFallbacks.delete(accountId) + } + }) + } + + private _handleFetchError(error: unknown): void { + // A 401 is an auth/token problem, not a rate limit. The caller refreshes + // the token and retries; backing off the quota API here would block that + // retry — and every other account — for minutes. Surface it without + // recording backoff state. + const message = error instanceof Error ? error.message : String(error) + if (/quota check failed: 401\b/.test(message)) return + this.lastApiError = buildQuotaOperationError({ + error, + now: this.now(), + previous: this.lastApiError, + }) + this.onApiError?.(this.lastApiError) + } +} diff --git a/packages/opencode/src/index.ts b/packages/opencode/src/index.ts index 30023d32a..5429eaec9 100644 --- a/packages/opencode/src/index.ts +++ b/packages/opencode/src/index.ts @@ -23,13 +23,11 @@ import { executeFastModeCommand, executeRoutingCommand, FallbackAccountManager, - fetchOAuthQuotaSnapshot, formatRefreshBackoffMessage, + getAccountStoragePath, getCache1hMode, getCache1hPersistentMode, getCacheKeepWindow, - getQuotaCheckIntervalMs, - getQuotaNextRefreshAt, getRelayConfig, getRoutingMode, hashRefreshToken, @@ -52,6 +50,7 @@ import { parseFastModeCommandAction, parseRoutingCommandAction, type QuotaAccountSummary, + QuotaManager, quotaSnapshotPassesPolicy, type RelayConfig, refreshBackoffActive, @@ -92,12 +91,6 @@ const MIN_MAIN_REFRESH_BEFORE_EXPIRY_MINUTES = 240 const DEFAULT_MAIN_REFRESH_BEFORE_EXPIRY_MINUTES = MIN_MAIN_REFRESH_BEFORE_EXPIRY_MINUTES -type MainQuotaCache = { - accessToken: string - refreshAfter: number - quota: OAuthQuotaSnapshot -} - type NotificationRequest = { path: { id: string } body: { @@ -274,11 +267,50 @@ function appendParallelToolPrompt(system: string[]) { export const AnthropicAuthPlugin: Plugin = async (ctx) => { startEventLoopLagMonitor() const { client } = ctx - const fallbackManager = new FallbackAccountManager() + const accountStoragePath = getAccountStoragePath() + const initialStorage = await loadAccounts(accountStoragePath) + const quotaManager = new QuotaManager({ + storage: initialStorage, + onMainQuotaFetched: async (quota, checkedAt) => { + try { + const storage = (await loadAccounts(accountStoragePath)) ?? { + version: 1 as const, + accounts: [], + } + storage.quota = storage.quota ?? {} + storage.quota.mainQuota = quota + storage.quota.mainQuotaCheckedAt = checkedAt + storage.quota.mainLastQuotaApiError = undefined + await saveAccounts(storage, accountStoragePath) + } catch (error) { + log('[quota] failed to persist main quota', { + error: error instanceof Error ? error.message : String(error), + }) + } + }, + onApiError: async (error) => { + try { + const storage = (await loadAccounts(accountStoragePath)) ?? { + version: 1 as const, + accounts: [], + } + storage.quota = storage.quota ?? {} + storage.quota.mainLastQuotaApiError = error + await saveAccounts(storage, accountStoragePath) + } catch (e) { + log('[quota] failed to persist backoff state', { + error: e instanceof Error ? e.message : String(e), + }) + } + }, + }) + const fallbackManager = new FallbackAccountManager({ + quotaManager, + }) fallbackManager.startBackgroundRefresh() let latestRefreshMainAccessToken: (() => Promise) | null = null const cacheKeepManager = new CacheKeepManager({ - loadStorage: () => loadAccounts(), + loadStorage: () => loadAccounts(accountStoragePath), prepareHeaders: async (headers, target) => { if (!latestGetAuth) return headers const auth = await latestGetAuth() @@ -316,14 +348,13 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { }, log, }) - const initialCache1hStorage = await loadAccounts() - const relayConfig: RelayConfig | null = getRelayConfig(initialCache1hStorage) + const relayConfig: RelayConfig | null = getRelayConfig(initialStorage) setCache1hState({ - enabled: isCache1hPersistentlyEnabled(initialCache1hStorage), - mode: getCache1hPersistentMode(initialCache1hStorage), + enabled: isCache1hPersistentlyEnabled(initialStorage), + mode: getCache1hPersistentMode(initialStorage), }) - setDumpEnabled(isDumpPersistentlyEnabled(initialCache1hStorage)) - setFastModeEnabled(isFastModePersistentlyEnabled(initialCache1hStorage)) + setDumpEnabled(isDumpPersistentlyEnabled(initialStorage)) + setFastModeEnabled(isFastModePersistentlyEnabled(initialStorage)) let latestGetAuth: | (() => Promise<{ type: string @@ -333,6 +364,9 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { }>) | null = null let mainBackgroundRefreshTimer: ReturnType | null = null + // Per-process counter of replayable model requests. Drives the every-N + // quota refresh cadence (quota.refreshEveryNRequests) for the active route. + let sessionRequestCount = 0 function mainRefreshBeforeExpiryMs( storage: Awaited>, @@ -351,7 +385,7 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { async function clearStaleMainRefreshError(refreshToken?: string) { if (!refreshToken) return - const storage = await loadAccounts() + const storage = await loadAccounts(accountStoragePath) const error = storage?.refresh?.mainLastRefreshError if (!storage?.refresh || !error?.tokenHash) return const tokenHash = hashRefreshToken(refreshToken) @@ -370,7 +404,7 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { return } storage.refresh.mainLastRefreshError = undefined - await saveAccounts(storage) + await saveAccounts(storage, accountStoragePath) log( '[refresh] opencode main oauth cleared stale backoff after token rotation', { @@ -387,10 +421,14 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { try { const auth = await latestGetAuth() if (auth.type === 'oauth' && auth.access) { + // /claude-quota is a manual action: force a real fetch instead of + // returning the cache. refreshMain still respects 429 backoff — it + // returns the last cached snapshot when the API is backed off. + const quota = await quotaManager.refreshMain(auth.access) accounts.push({ name: 'OpenCode anthropic', role: 'main', - quota: await fetchOAuthQuotaSnapshot({ accessToken: auth.access }), + quota, }) } else if (auth.type === 'oauth') { accounts.push({ @@ -401,22 +439,25 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { }) } } catch (error) { + const msg = error instanceof Error ? error.message : String(error) accounts.push({ name: 'OpenCode anthropic', role: 'main', - error: error instanceof Error ? error.message : String(error), + error: msg.includes('429') + ? 'Usage API rate limited — try again in a moment' + : msg, }) } } + // Force a real fallback refresh and PERSIST it to anthropic-auth.json. + // refreshQuotaForAllAccounts({ force: true }) bypasses the staleness skip, + // routes through the unified QuotaManager path (429 backoff respected + // per-account), saves refreshed snapshots, and clears stale errors. const { storage, errors } = - await fallbackManager.refreshQuotaForAllAccounts() - accounts.push( - ...buildFallbackQuotaSummaries( - storage, - new Map(errors.map((error) => [error.accountId, error.message])), - ), - ) + await fallbackManager.refreshQuotaForAllAccounts({ force: true }) + const errorMap = new Map(errors.map((e) => [e.accountId, e.message])) + accounts.push(...buildFallbackQuotaSummaries(storage, errorMap)) if (!latestGetAuth) { accounts.unshift({ @@ -450,7 +491,7 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { }) } - const storage = await loadAccounts() + const storage = await loadAccounts(accountStoragePath) const enabled = isCache1hPersistentlyEnabled(storage) const mode = getCache1hPersistentMode(storage) setCache1hState({ enabled, mode }) @@ -459,7 +500,7 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { async function executePersistentCacheKeepCommand(argumentsText: string) { const action = parseCacheKeepCommandAction(argumentsText) - let storage = await loadAccounts() + let storage = await loadAccounts(accountStoragePath) if (action.type === 'window') { storage = await setCacheKeepPersistentWindow( action.startHour, @@ -490,7 +531,7 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { return executeDumpCommand({ argumentsText, enabled }) } - const storage = await loadAccounts() + const storage = await loadAccounts(accountStoragePath) const enabled = isDumpPersistentlyEnabled(storage) setDumpEnabled(enabled) return executeDumpCommand({ argumentsText, enabled }) @@ -505,7 +546,7 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { return executeFastModeCommand({ argumentsText, enabled }) } - const storage = await loadAccounts() + const storage = await loadAccounts(accountStoragePath) const enabled = isFastModePersistentlyEnabled(storage) setFastModeEnabled(enabled) return executeFastModeCommand({ argumentsText, enabled }) @@ -659,9 +700,6 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { // Shared inflight refresh promise — prevents concurrent token refreshes // from racing against each other (and causing 401 cascades with token rotation) let refreshPromise: Promise | null = null - let mainQuotaCache: MainQuotaCache | null = null - let mainQuotaRefreshPromise: Promise | null = null - let mainQuotaRetryAfter = 0 async function refreshMainAccessToken() { if (!refreshPromise) { @@ -675,14 +713,16 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { async function updateMainRefreshState( update: (storage: AccountStorage) => void, ) { - const storage: AccountStorage = (await loadAccounts()) ?? { + const storage: AccountStorage = (await loadAccounts( + accountStoragePath, + )) ?? { version: 1, main: { type: 'opencode', provider: 'anthropic' }, accounts: [], } storage.refresh = storage.refresh ?? {} update(storage) - await saveAccounts(storage) + await saveAccounts(storage, accountStoragePath) } async function waitForConcurrentMainRefresh(previous: { @@ -743,7 +783,7 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { ) } - const storage = await loadAccounts() + const storage = await loadAccounts(accountStoragePath) const refreshTokenHash = hashRefreshToken(freshAuth.refresh) const mainError = storage?.refresh?.mainLastRefreshError log('[refresh] opencode main oauth refresh check', { @@ -830,7 +870,7 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { nextStorage.refresh.mainRefreshLeaseTokenHash = refreshTokenHash }) - const latestLease = await loadAccounts() + const latestLease = await loadAccounts(accountStoragePath) log( '[refresh] opencode main oauth refresh lease acquired', { @@ -976,7 +1016,7 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { const run = async () => { try { - const storage = await loadAccounts() + const storage = await loadAccounts(accountStoragePath) if (!mainRefreshEnabled(storage)) return const latestAuth = await getAuth() if (latestAuth.type !== 'oauth') return @@ -1226,7 +1266,7 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { isCache1hEnabled() && getCache1hMode() === 'hybrid' ) { - const storage = await loadAccounts() + const storage = await loadAccounts(accountStoragePath) const tracked = await cacheKeepManager.track({ sessionId: relayAffinity, url: rewritten.url?.toString() ?? rewritten.input.toString(), @@ -1270,54 +1310,6 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { return response } - async function refreshMainQuotaCache( - accessToken: string, - storage: Awaited>, - ) { - const now = Date.now() - const quota = await fetchOAuthQuotaSnapshot({ accessToken }) - mainQuotaCache = { - accessToken, - refreshAfter: getQuotaNextRefreshAt(quota, storage, now), - quota, - } - return quota - } - - function refreshMainQuotaCacheInBackground( - accessToken: string, - storage: Awaited>, - ) { - const now = Date.now() - if (mainQuotaRefreshPromise || now < mainQuotaRetryAfter) return - mainQuotaRefreshPromise = refreshMainQuotaCache( - accessToken, - storage, - ) - .catch((error) => { - mainQuotaRetryAfter = now + getQuotaCheckIntervalMs(storage) - throw error - }) - .finally(() => { - mainQuotaRefreshPromise = null - }) - void mainQuotaRefreshPromise.catch(() => {}) - } - - async function getMainQuotaForRouting( - accessToken: string, - storage: Awaited>, - ) { - const now = Date.now() - if (mainQuotaCache?.accessToken !== accessToken) { - return await refreshMainQuotaCache(accessToken, storage) - } - if (now >= mainQuotaCache.refreshAfter) { - refreshMainQuotaCacheInBackground(accessToken, storage) - } - return mainQuotaCache.quota - } - async function tryUsableFallbackAccounts( input: string | URL | Request, init: RequestInit | undefined, @@ -1358,6 +1350,17 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { } if (!fallbackAgain) { await fallbackManager.markUsed(account) + // Active-route every-N refresh: this fallback just served the + // request, so keep its quota fresh on the same cadence as main. + // Non-blocking; only the served account, never idle fallbacks. + if ( + access && + quotaManager.shouldRefreshOnRequestCount(sessionRequestCount) + ) { + void quotaManager + .refreshFallback(account.id, access) + .catch(() => {}) + } return response } if (index < accounts.length - 1 || !returnLastOnExhausted) { @@ -1381,7 +1384,8 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { if (!isReplayableRequest(input, init?.body)) return mainResponse const loadStart = nowMs() - const storage = existingStorage ?? (await loadAccounts()) + const storage = + existingStorage ?? (await loadAccounts(accountStoragePath)) trace?.mark('fallback_load_storage', { ms: roundMs(nowMs() - loadStart), cached: !!existingStorage, @@ -1451,6 +1455,8 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { const loadStart = nowMs() const storage = await loadAccounts() trace.mark('load_storage', { ms: roundMs(nowMs() - loadStart) }) + quotaManager.updateStorage(storage) + quotaManager.seedFallbacksFromAccounts(storage?.accounts ?? []) const replayableRequest = isReplayableRequest(input, init?.body) let preselectedFallbackAccounts: | Awaited< @@ -1549,18 +1555,24 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { if (replayableRequest && mainQuotaRoutingEnabled(storage)) { try { const quotaStart = nowMs() - const mainQuota = await getMainQuotaForRouting( - auth.access, - storage, - ) + sessionRequestCount++ + // Use QuotaManager: get cached or eagerly refresh if first time + let routingQuota = quotaManager.getMain()?.quota + if (!routingQuota) { + routingQuota = await quotaManager.refreshMain(auth.access) + } else if (quotaManager.needsRefresh(sessionRequestCount)) { + // Stale OR every-N request boundary — background refresh, + // return current snapshot to avoid blocking. + void quotaManager.refreshMain(auth.access).catch(() => {}) + } trace.mark('main_quota_for_routing', { ms: roundMs(nowMs() - quotaStart), - passes: quotaSnapshotPassesPolicy(mainQuota, storage), + passes: quotaSnapshotPassesPolicy(routingQuota, storage), }) - if (!quotaSnapshotPassesPolicy(mainQuota, storage)) { + if (!quotaSnapshotPassesPolicy(routingQuota, storage)) { const fallbackStart = nowMs() preselectedFallbackAccounts = - await fallbackManager.getUsableFallbackAccounts() + await fallbackManager.getUsableFallbackAccounts(storage) trace.mark('preselect_fallback_accounts', { ms: roundMs(nowMs() - fallbackStart), accounts: preselectedFallbackAccounts.length, @@ -1588,6 +1600,7 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { // Main quota checks should optimize routing, not break requests. } } + const mainResponse = await sendWithAccessToken( input, init, diff --git a/packages/opencode/src/tests/accounts.test.ts b/packages/opencode/src/tests/accounts.test.ts index 327680826..6e3c0a80d 100644 --- a/packages/opencode/src/tests/accounts.test.ts +++ b/packages/opencode/src/tests/accounts.test.ts @@ -935,6 +935,47 @@ describe('FallbackAccountManager', () => { 'Claude OAuth refresh failed: 400 — invalid_grant', ) }) + + test('force refreshes fresh accounts and persists the new quota', async () => { + const storage = baseStorage() + storage.accounts.push({ + id: 'fresh-but-forced', + type: 'oauth', + access: 'access-token', + refresh: 'refresh-token', + expires: 20_000_000, + // Fresh quota — would normally be skipped by the staleness gate. + quota: { + five_hour: { usedPercent: 10, remainingPercent: 90, checkedAt: 1_900 }, + seven_day: { usedPercent: 20, remainingPercent: 80, checkedAt: 1_900 }, + }, + }) + await saveAccounts(storage) + + const fetchImpl = mock(() => + Promise.resolve( + new Response( + JSON.stringify({ + five_hour: { utilization: 70 }, + seven_day: { utilization: 30 }, + }), + { status: 200 }, + ), + ), + ) as unknown as typeof fetch + + const manager = new FallbackAccountManager({ fetchImpl, now: () => 2_000 }) + + // force: true bypasses the staleness skip and fetches anyway. + const result = await manager.refreshQuotaForAllAccounts({ force: true }) + + expect(result.errors).toEqual([]) + expect(fetchImpl).toHaveBeenCalled() + // Refreshed numbers are PERSISTED to disk (regression guard for #2). + const saved = await loadAccounts() + expect(saved?.accounts[0]?.quota?.five_hour?.remainingPercent).toBe(30) + expect(saved?.accounts[0]?.quota?.seven_day?.remainingPercent).toBe(70) + }) }) describe('buildRefreshOperationError', () => { diff --git a/packages/opencode/src/tests/index.test.ts b/packages/opencode/src/tests/index.test.ts index 9eee431c3..66013242d 100644 --- a/packages/opencode/src/tests/index.test.ts +++ b/packages/opencode/src/tests/index.test.ts @@ -334,6 +334,18 @@ describe('auth.loader', () => { let capturedBody: string | undefined globalThis.fetch = mock((input: any, init: any) => { + const url = extractUrl(input) + if (url.includes('/api/oauth/usage')) { + return Promise.resolve( + new Response( + JSON.stringify({ + five_hour: { utilization: 0 }, + seven_day: { utilization: 0 }, + }), + { status: 200 }, + ), + ) + } capturedHeaders = init?.headers capturedBody = init?.body return Promise.resolve(new Response(null, { status: 200 })) @@ -402,7 +414,19 @@ describe('auth.loader', () => { let capturedBody: string | undefined let capturedHeaders: Headers | undefined globalThis.fetch = mock((input: any, init: any) => { - capturedUrl = extractUrl(input) + const url = extractUrl(input) + if (url.includes('/api/oauth/usage')) { + return Promise.resolve( + new Response( + JSON.stringify({ + five_hour: { utilization: 0 }, + seven_day: { utilization: 0 }, + }), + { status: 200 }, + ), + ) + } + capturedUrl = url capturedBody = init?.body capturedHeaders = new Headers(init?.headers) return Promise.resolve( @@ -916,7 +940,19 @@ describe('auth.loader', () => { let capturedHeaders: Headers | undefined let capturedBody: string | undefined - globalThis.fetch = mock((_input: any, init: any) => { + globalThis.fetch = mock((input: any, init: any) => { + const url = extractUrl(input) + if (url.includes('/api/oauth/usage')) { + return Promise.resolve( + new Response( + JSON.stringify({ + five_hour: { utilization: 0 }, + seven_day: { utilization: 0 }, + }), + { status: 200 }, + ), + ) + } capturedHeaders = init?.headers capturedBody = init?.body return Promise.resolve(new Response(null, { status: 200 })) @@ -958,7 +994,19 @@ describe('auth.loader', () => { let capturedHeaders: Headers | undefined let capturedBody: string | undefined - globalThis.fetch = mock((_input: any, init: any) => { + globalThis.fetch = mock((input: any, init: any) => { + const url = extractUrl(input) + if (url.includes('/api/oauth/usage')) { + return Promise.resolve( + new Response( + JSON.stringify({ + five_hour: { utilization: 0 }, + seven_day: { utilization: 0 }, + }), + { status: 200 }, + ), + ) + } capturedHeaders = init?.headers capturedBody = init?.body return Promise.resolve(new Response(null, { status: 200 })) @@ -995,7 +1043,19 @@ describe('auth.loader', () => { let capturedBody: string | undefined const mockClient = createMockClient() - globalThis.fetch = mock((_input: any, init: any) => { + globalThis.fetch = mock((input: any, init: any) => { + const url = extractUrl(input) + if (url.includes('/api/oauth/usage')) { + return Promise.resolve( + new Response( + JSON.stringify({ + five_hour: { utilization: 0 }, + seven_day: { utilization: 0 }, + }), + { status: 200 }, + ), + ) + } capturedBody = init?.body return Promise.resolve(new Response(null, { status: 200 })) }) as unknown as typeof fetch @@ -1050,7 +1110,19 @@ describe('auth.loader', () => { const mockClient = createMockClient() globalThis.fetch = mock( - (_input: string | URL | Request, init?: RequestInit) => { + (input: string | URL | Request, init?: RequestInit) => { + const url = extractUrl(input) + if (url.includes('/api/oauth/usage')) { + return Promise.resolve( + new Response( + JSON.stringify({ + five_hour: { utilization: 0 }, + seven_day: { utilization: 0 }, + }), + { status: 200 }, + ), + ) + } capturedBody = String(init?.body) capturedHeaders = new Headers(init?.headers) return Promise.resolve(new Response(null, { status: 200 })) @@ -1100,7 +1172,19 @@ describe('auth.loader', () => { const mockClient = createMockClient() globalThis.fetch = mock( - (_input: string | URL | Request, init?: RequestInit) => { + (input: string | URL | Request, init?: RequestInit) => { + const url = extractUrl(input) + if (url.includes('/api/oauth/usage')) { + return Promise.resolve( + new Response( + JSON.stringify({ + five_hour: { utilization: 0 }, + seven_day: { utilization: 0 }, + }), + { status: 200 }, + ), + ) + } capturedBody = String(init?.body) return Promise.resolve(new Response(null, { status: 200 })) }, @@ -1780,7 +1864,19 @@ describe('auth.loader', () => { let capturedUrl: string | undefined globalThis.fetch = mock((input: any) => { - capturedUrl = extractUrl(input) + const url = extractUrl(input) + if (url.includes('/api/oauth/usage')) { + return Promise.resolve( + new Response( + JSON.stringify({ + five_hour: { utilization: 0 }, + seven_day: { utilization: 0 }, + }), + { status: 200 }, + ), + ) + } + capturedUrl = url return Promise.resolve(new Response(null, { status: 200 })) }) as unknown as typeof fetch @@ -2144,6 +2240,8 @@ describe('auth.loader', () => { ]) expect(second).toBe('message-2') + // Background quota refresh involves file-lock I/O; wait for it to fire. + await new Promise((r) => setTimeout(r, 50)) expect(quotaCalls).toBe(2) expect(messageCalls).toBe(2) } finally { diff --git a/packages/opencode/src/tests/quota-manager.test.ts b/packages/opencode/src/tests/quota-manager.test.ts new file mode 100644 index 000000000..a586fdab5 --- /dev/null +++ b/packages/opencode/src/tests/quota-manager.test.ts @@ -0,0 +1,324 @@ +import { afterEach, beforeEach, describe, expect, mock, test } from 'bun:test' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { QuotaManager } from '@cortexkit/anthropic-auth-core' + +function makeQuotaResponse(now: number) { + return new Response( + JSON.stringify({ + five_hour: { + utilization: 25, + resets_at: new Date(now + 3600_000).toISOString(), + }, + seven_day: { + utilization: 50, + }, + }), + { status: 200 }, + ) +} + +describe('QuotaManager', () => { + let now: number + let tempDir: string + + beforeEach(async () => { + now = 1_000_000 + tempDir = await mkdtemp(join(tmpdir(), 'qm-test-')) + process.env.OPENCODE_ANTHROPIC_AUTH_FILE = join( + tempDir, + 'anthropic-auth.json', + ) + }) + + afterEach(async () => { + delete process.env.OPENCODE_ANTHROPIC_AUTH_FILE + await rm(tempDir, { recursive: true, force: true }) + }) + + function createQM(fetchImpl?: typeof fetch) { + return new QuotaManager({ + storage: null, + fetchImpl, + now: () => now, + }) + } + + describe('backoff', () => { + test('first 429 backs off for 60s', async () => { + const fetchMock = mock(() => + Promise.resolve(new Response('rate limited', { status: 429 })), + ) as unknown as typeof fetch + const qm = createQM(fetchMock) + + try { + await qm.refreshMain('token') + } catch {} + + expect(qm.isBackedOff()).toBe(true) + now += 59_000 + expect(qm.isBackedOff()).toBe(true) + now += 2_000 + expect(qm.isBackedOff()).toBe(false) + }) + + test('401 does not arm backoff (auth error, caller retries)', async () => { + // A 401 means the access token expired — the caller refreshes the token + // and retries. Backing off here would block that retry and every other + // account, so the quota API must NOT enter backoff on 401. + const fetchMock = mock(() => + Promise.resolve(new Response('unauthorized', { status: 401 })), + ) as unknown as typeof fetch + const qm = createQM(fetchMock) + + try { + await qm.refreshMain('token') + } catch {} + + expect(qm.isBackedOff()).toBe(false) + expect(qm.getLastApiError()).toBeUndefined() + }) + + test('repeated 429s escalate backoff exponentially', async () => { + const fetchMock = mock(() => + Promise.resolve(new Response('rate limited', { status: 429 })), + ) as unknown as typeof fetch + const qm = createQM(fetchMock) + + // First failure: 60s + try { + await qm.refreshMain('token') + } catch {} + expect(qm.isBackedOff()).toBe(true) + + now += 61_000 + expect(qm.isBackedOff()).toBe(false) + + // Second failure: 120s + try { + await qm.refreshMain('token') + } catch {} + now += 119_000 + expect(qm.isBackedOff()).toBe(true) + now += 2_000 + expect(qm.isBackedOff()).toBe(false) + }) + + test('backoff caps at 15 minutes', async () => { + const fetchMock = mock(() => + Promise.resolve(new Response('rate limited', { status: 429 })), + ) as unknown as typeof fetch + const qm = createQM(fetchMock) + + // Trigger 8 failures to exceed cap + for (let i = 0; i < 8; i++) { + try { + await qm.refreshMain('token') + } catch {} + now += 16 * 60_000 + } + + try { + await qm.refreshMain('token') + } catch {} + now += 14 * 60_000 + expect(qm.isBackedOff()).toBe(true) + now += 2 * 60_000 + expect(qm.isBackedOff()).toBe(false) + }) + + test('successful fetch resets backoff', async () => { + let failNext = true + const fetchMock = mock(() => { + if (failNext) { + return Promise.resolve(new Response('rate limited', { status: 429 })) + } + return Promise.resolve(makeQuotaResponse(now)) + }) as unknown as typeof fetch + const qm = createQM(fetchMock) + + try { + await qm.refreshMain('token') + } catch {} + expect(qm.isBackedOff()).toBe(true) + + now += 61_000 + failNext = false + await qm.refreshMain('token') + expect(qm.isBackedOff()).toBe(false) + + // Next failure starts from 60s again (not escalated) + failNext = true + now += 1_100 + try { + await qm.refreshMain('token') + } catch {} + now += 59_000 + expect(qm.isBackedOff()).toBe(true) + now += 2_000 + expect(qm.isBackedOff()).toBe(false) + }) + + test('getLastApiError exposes backoff state', async () => { + const fetchMock = mock(() => + Promise.resolve(new Response('rate limited', { status: 429 })), + ) as unknown as typeof fetch + const qm = createQM(fetchMock) + + expect(qm.getLastApiError()).toBeUndefined() + + try { + await qm.refreshMain('token') + } catch {} + + const err = qm.getLastApiError() + expect(err).toBeDefined() + expect(err!.retryCount).toBe(1) + expect(err!.nextRetryAt).toBeGreaterThan(now) + }) + + test('500 errors also trigger backoff', async () => { + const fetchMock = mock(() => + Promise.resolve(new Response('internal error', { status: 500 })), + ) as unknown as typeof fetch + const qm = createQM(fetchMock) + + try { + await qm.refreshMain('token') + } catch {} + expect(qm.isBackedOff()).toBe(true) + }) + + test('returns cached quota during backoff', async () => { + let failNext = false + const fetchMock = mock(() => { + if (failNext) { + return Promise.resolve(new Response('rate limited', { status: 429 })) + } + return Promise.resolve(makeQuotaResponse(now)) + }) as unknown as typeof fetch + const qm = createQM(fetchMock) + + const first = await qm.refreshMain('token') + expect(first).toBeDefined() + + failNext = true + now += 1_100 + try { + await qm.refreshMain('token') + } catch {} + + const cached = qm.getMain() + expect(cached).not.toBeNull() + }) + }) + + describe('persistence', () => { + test('seeds main quota from persisted storage', () => { + const quota = { + quotas: [], + expires: new Date(2_000_000).toISOString(), + } + const qm = new QuotaManager({ + storage: { + version: 1, + accounts: [], + quota: { + mainQuota: quota as any, + mainQuotaCheckedAt: 900_000, + }, + }, + now: () => 1_000_000, + }) + + const main = qm.getMain() + expect(main).not.toBeNull() + expect(main!.checkedAt).toBe(900_000) + }) + + test('calls onMainQuotaFetched after successful fetch', async () => { + let callbackQuota: any = null + const fetchMock = mock(() => + Promise.resolve(makeQuotaResponse(now)), + ) as unknown as typeof fetch + + const qm = new QuotaManager({ + storage: null, + fetchImpl: fetchMock, + now: () => now, + onMainQuotaFetched: (quota, checkedAt) => { + callbackQuota = { quota, checkedAt } + }, + }) + + await qm.refreshMain('token') + expect(callbackQuota).not.toBeNull() + expect(callbackQuota.checkedAt).toBe(now) + }) + + test('seeds backoff state from persisted storage', () => { + const qm = new QuotaManager({ + storage: { + version: 1, + accounts: [], + quota: { + mainLastQuotaApiError: { + message: 'Claude quota check failed: 429 — rate limited', + checkedAt: now - 30_000, + nextRetryAt: now + 30_000, + retryCount: 1, + }, + }, + }, + now: () => now, + }) + + expect(qm.isBackedOff()).toBe(true) + }) + + test('ignores expired persisted backoff', () => { + const qm = new QuotaManager({ + storage: { + version: 1, + accounts: [], + quota: { + mainLastQuotaApiError: { + message: 'old error', + checkedAt: now - 120_000, + nextRetryAt: now - 60_000, + retryCount: 1, + }, + }, + }, + now: () => now, + }) + + expect(qm.isBackedOff()).toBe(false) + }) + + test('calls onApiError callback on failure', async () => { + let errorCallback: any = null + const fetchMock = mock(() => + Promise.resolve(new Response('rate limited', { status: 429 })), + ) as unknown as typeof fetch + + const qm = new QuotaManager({ + storage: null, + fetchImpl: fetchMock, + now: () => now, + onApiError: (error) => { + errorCallback = error + }, + }) + + try { + await qm.refreshMain('token') + } catch {} + + expect(errorCallback).not.toBeNull() + expect(errorCallback.retryCount).toBe(1) + expect(errorCallback.nextRetryAt).toBeGreaterThan(now) + }) + }) +}) From c4e76b68c680b9ab88fef0bf7f4d8d230ef5ae4b Mon Sep 17 00:00:00 2001 From: iceteaSA <171169159+iceteaSA@users.noreply.github.com> Date: Mon, 1 Jun 2026 13:45:14 +0200 Subject: [PATCH 2/5] fix(quota): bind persisted main seed to token, count all replayable requests Two AI-review fixes: - Token-bind the persisted main quota seed. The constructor seeded this.main from storage without recording which account produced it, so refreshMain's token-change guard had no baseline and a different account's stale quota could be served during backoff. Persist a token fingerprint (truncated SHA-256) with the main quota, seed it alongside this.main, and invalidate the seed in refreshMain before the backoff short-circuit when the live token's fingerprint differs. - Increment sessionRequestCount for every replayable request instead of only inside the mainQuotaRoutingEnabled branch, so the every-N refresh cadence (and the active-route fallback refresh) advances on all routing paths. Adds regression tests for seed drop/keep on token change during backoff. --- packages/core/src/accounts.ts | 12 +++- packages/core/src/quota-manager.ts | 39 ++++++++--- packages/opencode/src/index.ts | 8 ++- .../opencode/src/tests/quota-manager.test.ts | 67 ++++++++++++++++++- 4 files changed, 111 insertions(+), 15 deletions(-) diff --git a/packages/core/src/accounts.ts b/packages/core/src/accounts.ts index f6d89c848..3e8d4ddc9 100644 --- a/packages/core/src/accounts.ts +++ b/packages/core/src/accounts.ts @@ -80,6 +80,9 @@ export type AccountStorage = { failClosedOnUnknownQuota?: boolean mainQuota?: OAuthQuotaSnapshot mainQuotaCheckedAt?: number + // Fingerprint of the access token that produced mainQuota. Used to avoid + // seeding a different account's persisted quota after a main-account switch. + mainQuotaToken?: string mainLastQuotaApiError?: AccountOperationError } claudeCache?: { @@ -727,14 +730,17 @@ export function getQuotaCheckIntervalMs(storage: AccountStorage | null) { return Math.max(1, minutes) * 60_000 } -export function getPersistedMainQuota( - storage: AccountStorage | null, -): { quota: OAuthQuotaSnapshot; checkedAt: number } | null { +export function getPersistedMainQuota(storage: AccountStorage | null): { + quota: OAuthQuotaSnapshot + checkedAt: number + tokenFingerprint?: string +} | null { if (!storage?.quota?.mainQuota || !storage.quota.mainQuotaCheckedAt) return null return { quota: storage.quota.mainQuota, checkedAt: storage.quota.mainQuotaCheckedAt, + tokenFingerprint: storage.quota.mainQuotaToken, } } diff --git a/packages/core/src/quota-manager.ts b/packages/core/src/quota-manager.ts index a281e8e40..3e30f1d5e 100644 --- a/packages/core/src/quota-manager.ts +++ b/packages/core/src/quota-manager.ts @@ -6,6 +6,7 @@ * Handles deduplication, rate-limiting (429 backoff), and staleness. */ +import { createHash } from 'node:crypto' import type { AccountOperationError, AccountStorage, @@ -26,6 +27,15 @@ import { // Capture real setTimeout before tests can mock globalThis.setTimeout const nativeSetTimeout = globalThis.setTimeout +/** + * Stable, non-reversible fingerprint of an access token. Used to detect a + * main-account switch so a different account's persisted/cached quota is never + * reused. Not a secret — a truncated SHA-256, safe to persist alongside quota. + */ +export function tokenFingerprint(token: string): string { + return createHash('sha256').update(token).digest('hex').slice(0, 16) +} + // --------------------------------------------------------------------------- // Types // --------------------------------------------------------------------------- @@ -40,7 +50,11 @@ export type QuotaManagerOptions = { storage: AccountStorage | null fetchImpl?: typeof fetch now?: () => number - onMainQuotaFetched?: (quota: OAuthQuotaSnapshot, checkedAt: number) => void + onMainQuotaFetched?: ( + quota: OAuthQuotaSnapshot, + checkedAt: number, + tokenFingerprint: string, + ) => void onApiError?: (error: AccountOperationError) => void } @@ -51,7 +65,7 @@ export type QuotaManagerOptions = { export class QuotaManager { // --- State --- private main: QuotaEntry | null = null - private mainAccessToken: string | null = null + private mainTokenFp: string | null = null private fallbacks = new Map() // --- Inflight deduplication --- @@ -79,7 +93,10 @@ export class QuotaManager { this.onMainQuotaFetched = opts.onMainQuotaFetched this.onApiError = opts.onApiError - // Seed main quota from persisted storage + // Seed main quota from persisted storage, bound to the token fingerprint + // that produced it. refreshMain() drops this seed if the live token's + // fingerprint differs (main-account switch), preventing stale wrong-account + // quota from being served during backoff. const persisted = getPersistedMainQuota(opts.storage) if (persisted) { this.main = { @@ -91,6 +108,7 @@ export class QuotaManager { ), checkedAt: persisted.checkedAt, } + this.mainTokenFp = persisted.tokenFingerprint ?? null } // Seed backoff state from persisted storage @@ -121,7 +139,7 @@ export class QuotaManager { // ========================================================================= setMain(accessToken: string, entry: QuotaEntry): void { - this.mainAccessToken = accessToken + this.mainTokenFp = tokenFingerprint(accessToken) this.main = entry } @@ -134,10 +152,13 @@ export class QuotaManager { // ========================================================================= async refreshMain(accessToken: string): Promise { - // If token changed, invalidate cache - if (this.mainAccessToken && this.mainAccessToken !== accessToken) { + // If the main account/token changed, invalidate the cache (including a + // persisted seed) BEFORE the backoff short-circuit so a different account's + // stale quota is never returned while the quota API is backed off. + const fp = tokenFingerprint(accessToken) + if (this.mainTokenFp && this.mainTokenFp !== fp) { this.main = null - this.mainAccessToken = null + this.mainTokenFp = null } // Deduplicate — return in-flight promise if already fetching @@ -327,14 +348,14 @@ export class QuotaManager { now: this.now, }) const now = this.now() - this.mainAccessToken = accessToken + this.mainTokenFp = tokenFingerprint(accessToken) this.main = { quota, refreshAfter: getQuotaNextRefreshAt(quota, this.storage, now), checkedAt: now, } this.lastApiError = undefined - this.onMainQuotaFetched?.(quota, now) + this.onMainQuotaFetched?.(quota, now, this.mainTokenFp) return quota } catch (error) { this._handleFetchError(error) diff --git a/packages/opencode/src/index.ts b/packages/opencode/src/index.ts index 5429eaec9..ada05c3d2 100644 --- a/packages/opencode/src/index.ts +++ b/packages/opencode/src/index.ts @@ -271,7 +271,7 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { const initialStorage = await loadAccounts(accountStoragePath) const quotaManager = new QuotaManager({ storage: initialStorage, - onMainQuotaFetched: async (quota, checkedAt) => { + onMainQuotaFetched: async (quota, checkedAt, tokenFingerprint) => { try { const storage = (await loadAccounts(accountStoragePath)) ?? { version: 1 as const, @@ -280,6 +280,7 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { storage.quota = storage.quota ?? {} storage.quota.mainQuota = quota storage.quota.mainQuotaCheckedAt = checkedAt + storage.quota.mainQuotaToken = tokenFingerprint storage.quota.mainLastQuotaApiError = undefined await saveAccounts(storage, accountStoragePath) } catch (error) { @@ -1552,10 +1553,13 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { trace.done('missing_access_error') throw new Error('OAuth access token is missing after refresh') } + // Count every replayable request so the every-N refresh cadence + // (quota.refreshEveryNRequests) advances on all routing paths, + // not only when main-quota routing is enabled. + if (replayableRequest) sessionRequestCount++ if (replayableRequest && mainQuotaRoutingEnabled(storage)) { try { const quotaStart = nowMs() - sessionRequestCount++ // Use QuotaManager: get cached or eagerly refresh if first time let routingQuota = quotaManager.getMain()?.quota if (!routingQuota) { diff --git a/packages/opencode/src/tests/quota-manager.test.ts b/packages/opencode/src/tests/quota-manager.test.ts index a586fdab5..c1cff9958 100644 --- a/packages/opencode/src/tests/quota-manager.test.ts +++ b/packages/opencode/src/tests/quota-manager.test.ts @@ -2,7 +2,7 @@ import { afterEach, beforeEach, describe, expect, mock, test } from 'bun:test' import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' -import { QuotaManager } from '@cortexkit/anthropic-auth-core' +import { QuotaManager, tokenFingerprint } from '@cortexkit/anthropic-auth-core' function makeQuotaResponse(now: number) { return new Response( @@ -237,6 +237,71 @@ describe('QuotaManager', () => { expect(main!.checkedAt).toBe(900_000) }) + test('drops persisted main seed during backoff when the token changed', async () => { + // Regression: a persisted seed must be bound to the account that produced + // it. After a main-account switch (different access token), the seed must + // not be served — even while the quota API is backed off. + const quota = { + quotas: [], + expires: new Date(2_000_000).toISOString(), + } + const qm = new QuotaManager({ + storage: { + version: 1, + accounts: [], + quota: { + mainQuota: quota as any, + mainQuotaCheckedAt: 900_000, + mainQuotaToken: 'old-account-fingerprint', + mainLastQuotaApiError: { + message: 'Claude quota check failed: 429 — rate limited', + checkedAt: 999_000, + nextRetryAt: 1_030_000, + retryCount: 1, + }, + }, + }, + now: () => 1_000_000, + }) + + expect(qm.getMain()).not.toBeNull() + expect(qm.isBackedOff()).toBe(true) + + // Different account token: seed is invalidated rather than returned. + await expect(qm.refreshMain('different-account-token')).rejects.toThrow() + expect(qm.getMain()).toBeNull() + }) + + test('keeps persisted main seed during backoff when the token matches', async () => { + const quota = { + quotas: [], + expires: new Date(2_000_000).toISOString(), + } + const qm = new QuotaManager({ + storage: { + version: 1, + accounts: [], + quota: { + mainQuota: quota as any, + mainQuotaCheckedAt: 900_000, + mainQuotaToken: tokenFingerprint('same-account-token'), + mainLastQuotaApiError: { + message: 'Claude quota check failed: 429 — rate limited', + checkedAt: 999_000, + nextRetryAt: 1_030_000, + retryCount: 1, + }, + }, + }, + now: () => 1_000_000, + }) + + // Same account token: backed off, so the seed is returned (not refetched). + const result = await qm.refreshMain('same-account-token') + expect(result).toEqual(quota as any) + expect(qm.getMain()).not.toBeNull() + }) + test('calls onMainQuotaFetched after successful fetch', async () => { let callbackQuota: any = null const fetchMock = mock(() => From 53dc5a1a9edaf1d9debe2090480351c6e1e3631a Mon Sep 17 00:00:00 2001 From: iceteaSA <171169159+iceteaSA@users.noreply.github.com> Date: Mon, 1 Jun 2026 13:49:00 +0200 Subject: [PATCH 3/5] fix(quota): scope quota-API backoff per route (main vs each fallback) A single shared lastApiError meant a fallback account's quota 429 backed off the main account and persisted as the main quota error: _fetchFallback called the shared error handler, which set lastApiError and invoked onApiError (which the OpenCode plugin persists as storage.quota.mainLastQuotaApiError); refreshMain then short-circuited via isBackedOff() and threw without ever trying the main token. Scope backoff state by route: - mainLastApiError drives isBackedOff()/getLastApiError() and refreshMain; only main failures arm it and call onApiError. - fallbackApiErrors is a per-account map driving the new isFallbackBackedOff(); fallback failures arm only that account's backoff and never touch main state or onApiError (the per-account error is already recorded via the account's lastQuotaRefreshError). Adds regression tests: fallback 429 does not back off main or fire onApiError, and main 429 does not back off a fallback. --- packages/core/src/quota-manager.ts | 78 +++++++++++++------ .../opencode/src/tests/quota-manager.test.ts | 45 +++++++++++ 2 files changed, 100 insertions(+), 23 deletions(-) diff --git a/packages/core/src/quota-manager.ts b/packages/core/src/quota-manager.ts index 3e30f1d5e..f02ad3b39 100644 --- a/packages/core/src/quota-manager.ts +++ b/packages/core/src/quota-manager.ts @@ -72,8 +72,10 @@ export class QuotaManager { private inflightMain: Promise | null = null private inflightFallbacks = new Map>() - // --- Rate-limiting --- - private lastApiError: AccountOperationError | undefined = undefined + // --- Rate-limiting (scoped per route so a fallback 429 never backs off the + // main account or vice versa) --- + private mainLastApiError: AccountOperationError | undefined = undefined + private fallbackApiErrors = new Map() // --- Serial API gate (prevents concurrent quota API calls) --- private apiGate: Promise = Promise.resolve() @@ -114,7 +116,7 @@ export class QuotaManager { // Seed backoff state from persisted storage const persistedError = opts.storage?.quota?.mainLastQuotaApiError if (persistedError && quotaBackoffActive(persistedError, this.now())) { - this.lastApiError = persistedError + this.mainLastApiError = persistedError } } @@ -182,8 +184,8 @@ export class QuotaManager { const inflight = this.inflightFallbacks.get(accountId) if (inflight) return inflight - // Rate-limit - if (this.isBackedOff()) { + // Rate-limit — scoped to THIS fallback account only + if (this.isFallbackBackedOff(accountId)) { const cached = this.fallbacks.get(accountId) if (cached) return cached.quota throw new Error('Quota API rate-limited — try again later') @@ -283,14 +285,22 @@ export class QuotaManager { } /** - * Whether the API is currently in backoff due to a recent error. + * Whether the MAIN quota API is currently in backoff. Scoped to the main + * account — a fallback account's 429 never reports here. */ isBackedOff(): boolean { - return quotaBackoffActive(this.lastApiError, this.now()) + return quotaBackoffActive(this.mainLastApiError, this.now()) + } + + /** + * Whether a specific fallback account's quota API is in backoff. + */ + isFallbackBackedOff(accountId: string): boolean { + return quotaBackoffActive(this.fallbackApiErrors.get(accountId), this.now()) } getLastApiError(): AccountOperationError | undefined { - return this.lastApiError + return this.mainLastApiError } // ========================================================================= @@ -354,11 +364,11 @@ export class QuotaManager { refreshAfter: getQuotaNextRefreshAt(quota, this.storage, now), checkedAt: now, } - this.lastApiError = undefined + this.mainLastApiError = undefined this.onMainQuotaFetched?.(quota, now, this.mainTokenFp) return quota } catch (error) { - this._handleFetchError(error) + this._handleMainFetchError(error) throw error } finally { await fileLock.release() @@ -375,8 +385,8 @@ export class QuotaManager { ): Promise { return this._enqueueApiFetch(async () => { try { - // Re-check backoff inside gate - if (this.isBackedOff()) { + // Re-check backoff inside gate — scoped to this fallback account + if (this.isFallbackBackedOff(accountId)) { const cached = this.fallbacks.get(accountId) if (cached) return cached.quota throw new Error('Quota API rate-limited — try again later') @@ -392,10 +402,10 @@ export class QuotaManager { refreshAfter: now + getQuotaCheckIntervalMs(this.storage), checkedAt: now, }) - this.lastApiError = undefined + this.fallbackApiErrors.delete(accountId) return quota } catch (error) { - this._handleFetchError(error) + this._handleFallbackFetchError(accountId, error) throw error } finally { this.inflightFallbacks.delete(accountId) @@ -403,18 +413,40 @@ export class QuotaManager { }) } - private _handleFetchError(error: unknown): void { - // A 401 is an auth/token problem, not a rate limit. The caller refreshes - // the token and retries; backing off the quota API here would block that - // retry — and every other account — for minutes. Surface it without - // recording backoff state. + // A 401 is an auth/token problem, not a rate limit. The caller refreshes the + // token and retries; backing off the quota API here would block that retry, + // so surface 401s without recording backoff state. + private static isAuthError(error: unknown): boolean { const message = error instanceof Error ? error.message : String(error) - if (/quota check failed: 401\b/.test(message)) return - this.lastApiError = buildQuotaOperationError({ + return /quota check failed: 401\b/.test(message) + } + + /** Main quota failure: arms main-only backoff and persists via onApiError. */ + private _handleMainFetchError(error: unknown): void { + if (QuotaManager.isAuthError(error)) return + this.mainLastApiError = buildQuotaOperationError({ error, now: this.now(), - previous: this.lastApiError, + previous: this.mainLastApiError, }) - this.onApiError?.(this.lastApiError) + this.onApiError?.(this.mainLastApiError) + } + + /** + * Fallback quota failure: arms backoff for THIS account only. Never touches + * main backoff state and never calls onApiError (which persists the main + * quota error) — the per-account error is recorded by the caller via the + * account's lastQuotaRefreshError. + */ + private _handleFallbackFetchError(accountId: string, error: unknown): void { + if (QuotaManager.isAuthError(error)) return + this.fallbackApiErrors.set( + accountId, + buildQuotaOperationError({ + error, + now: this.now(), + previous: this.fallbackApiErrors.get(accountId), + }), + ) } } diff --git a/packages/opencode/src/tests/quota-manager.test.ts b/packages/opencode/src/tests/quota-manager.test.ts index c1cff9958..58edfa61c 100644 --- a/packages/opencode/src/tests/quota-manager.test.ts +++ b/packages/opencode/src/tests/quota-manager.test.ts @@ -63,6 +63,51 @@ describe('QuotaManager', () => { expect(qm.isBackedOff()).toBe(false) }) + test('fallback 429 does NOT back off main or fire onApiError', async () => { + // Regression: backoff is scoped per route. A fallback account's quota + // 429 must not suppress main quota checks nor persist as the main quota + // API error (onApiError -> mainLastQuotaApiError). + let apiErrorCount = 0 + const fetchMock = mock(() => + Promise.resolve(new Response('rate limited', { status: 429 })), + ) as unknown as typeof fetch + const qm = new QuotaManager({ + storage: null, + fetchImpl: fetchMock, + now: () => now, + onApiError: () => { + apiErrorCount++ + }, + }) + + try { + await qm.refreshFallback('fallback-1', 'fallback-token') + } catch {} + + // Fallback account is backed off; main is not. + expect(qm.isFallbackBackedOff('fallback-1')).toBe(true) + expect(qm.isBackedOff()).toBe(false) + // onApiError (persists mainLastQuotaApiError) must NOT have fired. + expect(apiErrorCount).toBe(0) + expect(qm.getLastApiError()).toBeUndefined() + // A different fallback is unaffected. + expect(qm.isFallbackBackedOff('fallback-2')).toBe(false) + }) + + test('main 429 does NOT back off a fallback account', async () => { + const fetchMock = mock(() => + Promise.resolve(new Response('rate limited', { status: 429 })), + ) as unknown as typeof fetch + const qm = createQM(fetchMock) + + try { + await qm.refreshMain('main-token') + } catch {} + + expect(qm.isBackedOff()).toBe(true) + expect(qm.isFallbackBackedOff('fallback-1')).toBe(false) + }) + test('401 does not arm backoff (auth error, caller retries)', async () => { // A 401 means the access token expired — the caller refreshes the token // and retries. Backing off here would block that retry and every other From 9c8f23a71bd62ef8bca8472695f4f63f043ad2db Mon Sep 17 00:00:00 2001 From: iceteaSA <171169159+iceteaSA@users.noreply.github.com> Date: Mon, 1 Jun 2026 14:28:12 +0200 Subject: [PATCH 4/5] fix(quota): token-aware getMain read and count fallback-first requests Two request-flow bugs from review: - getMain() bypassed the token-fingerprint check. The routing path reads quotaManager.getMain()?.quota before refreshMain, so a fresh cached snapshot from a previous main account could be used for a new access token. getMain now accepts the live access token and drops a cached entry bound to a different token (main-account switch); the routing path passes auth.access. - sessionRequestCount was incremented after the fallback-first early return, so a successful fallback-first request never advanced the every-N cadence and the served fallback's active-route refresh never fired. The counter now increments right after replayableRequest is determined, before any routing. Adds regression tests: getMain drops a cross-token cached entry, and a successful fallback-first request (refreshEveryNRequests=1) refreshes the served fallback. --- packages/core/src/quota-manager.ts | 18 +++++- packages/opencode/src/index.ts | 15 +++-- packages/opencode/src/tests/index.test.ts | 59 +++++++++++++++++++ .../opencode/src/tests/quota-manager.test.ts | 30 ++++++++++ 4 files changed, 115 insertions(+), 7 deletions(-) diff --git a/packages/core/src/quota-manager.ts b/packages/core/src/quota-manager.ts index f02ad3b39..52ac1bac9 100644 --- a/packages/core/src/quota-manager.ts +++ b/packages/core/src/quota-manager.ts @@ -124,7 +124,23 @@ export class QuotaManager { // Get (synchronous, from cache) // ========================================================================= - getMain(): QuotaEntry | null { + /** + * Cached main quota entry. Pass the live access token to enforce token + * binding: if the cached entry was produced by a different token (main + * account switched), it is dropped and null is returned so the caller + * refetches for the current account. Called without a token (e.g. for + * display) it returns whatever is cached. + */ + getMain(accessToken?: string): QuotaEntry | null { + if ( + accessToken && + this.main && + this.mainTokenFp && + this.mainTokenFp !== tokenFingerprint(accessToken) + ) { + this.main = null + this.mainTokenFp = null + } return this.main } diff --git a/packages/opencode/src/index.ts b/packages/opencode/src/index.ts index ada05c3d2..1a523be91 100644 --- a/packages/opencode/src/index.ts +++ b/packages/opencode/src/index.ts @@ -1459,6 +1459,11 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { quotaManager.updateStorage(storage) quotaManager.seedFallbacksFromAccounts(storage?.accounts ?? []) const replayableRequest = isReplayableRequest(input, init?.body) + // Count every replayable request up front — before the + // fallback-first early return — so the every-N refresh cadence + // (quota.refreshEveryNRequests) advances for main and the active + // fallback route on all paths, including successful fallback-first. + if (replayableRequest) sessionRequestCount++ let preselectedFallbackAccounts: | Awaited< ReturnType< @@ -1553,15 +1558,13 @@ export const AnthropicAuthPlugin: Plugin = async (ctx) => { trace.done('missing_access_error') throw new Error('OAuth access token is missing after refresh') } - // Count every replayable request so the every-N refresh cadence - // (quota.refreshEveryNRequests) advances on all routing paths, - // not only when main-quota routing is enabled. - if (replayableRequest) sessionRequestCount++ if (replayableRequest && mainQuotaRoutingEnabled(storage)) { try { const quotaStart = nowMs() - // Use QuotaManager: get cached or eagerly refresh if first time - let routingQuota = quotaManager.getMain()?.quota + // Token-aware read: getMain(auth.access) drops a cached entry + // bound to a different access token (main-account switch) so + // routing never uses the previous account's quota. + let routingQuota = quotaManager.getMain(auth.access)?.quota if (!routingQuota) { routingQuota = await quotaManager.refreshMain(auth.access) } else if (quotaManager.needsRefresh(sessionRequestCount)) { diff --git a/packages/opencode/src/tests/index.test.ts b/packages/opencode/src/tests/index.test.ts index 66013242d..e5ff6a261 100644 --- a/packages/opencode/src/tests/index.test.ts +++ b/packages/opencode/src/tests/index.test.ts @@ -1979,6 +1979,65 @@ describe('auth.loader', () => { expect(authorizations).toEqual(['Bearer fallback-access']) }) + test('successful fallback-first request advances the every-N counter and refreshes the served fallback', async () => { + // Regression: the request counter must increment before the fallback-first + // early return, so a served fallback's active-route every-N refresh fires. + await useTempAccountFile( + createFallbackStorage({ + routing: { mode: 'fallback-first' }, + quota: { + enabled: true, + checkIntervalMinutes: 5, + minimumRemaining: { five_hour: 10, seven_day: 20 }, + refreshEveryNRequests: 1, + }, + }), + ) + const usageTokens: string[] = [] + globalThis.fetch = mock((input: any, init: any) => { + const url = extractUrl(input) + if (url.includes('/api/oauth/usage')) { + usageTokens.push(new Headers(init?.headers).get('authorization') ?? '') + return Promise.resolve( + new Response( + JSON.stringify({ + five_hour: { utilization: 20 }, + seven_day: { utilization: 20 }, + }), + { status: 200 }, + ), + ) + } + return Promise.resolve(new Response('fallback ok', { status: 200 })) + }) as unknown as typeof fetch + + const plugin = await getPlugin() + const result = await plugin.auth.loader( + () => + Promise.resolve({ + type: 'oauth', + access: 'main-access', + refresh: 'main-refresh', + expires: Date.now() + 100000, + }), + { models: {} }, + ) + + const response = await result.fetch(MESSAGES_URL, EMPTY_POST) + expect(response.status).toBe(200) + expect(await response.text()).toBe('fallback ok') + + // The active-route refresh is fire-and-forget; wait for it to land. + for ( + let i = 0; + i < 50 && !usageTokens.includes('Bearer fallback-access'); + i++ + ) { + await new Promise((r) => setTimeout(r, 10)) + } + expect(usageTokens).toContain('Bearer fallback-access') + }) + test('fallback-first routing does not refresh expired main oauth when fallback succeeds', async () => { await useTempAccountFile( createFallbackStorage({ diff --git a/packages/opencode/src/tests/quota-manager.test.ts b/packages/opencode/src/tests/quota-manager.test.ts index 58edfa61c..5dd3c045c 100644 --- a/packages/opencode/src/tests/quota-manager.test.ts +++ b/packages/opencode/src/tests/quota-manager.test.ts @@ -282,6 +282,36 @@ describe('QuotaManager', () => { expect(main!.checkedAt).toBe(900_000) }) + test('getMain(accessToken) drops a cached entry bound to a different token', () => { + // Regression: the request path reads getMain() before refreshMain, so the + // token-binding check must also apply on the cached read — otherwise a + // stale previous-account quota is used for a new access token. + const quota = { + quotas: [], + expires: new Date(2_000_000).toISOString(), + } + const qm = new QuotaManager({ + storage: { + version: 1, + accounts: [], + quota: { + mainQuota: quota as any, + mainQuotaCheckedAt: 900_000, + mainQuotaToken: tokenFingerprint('old-main-token'), + }, + }, + now: () => 1_000_000, + }) + + // Tokenless read returns the cached entry (e.g. display paths). + expect(qm.getMain()).not.toBeNull() + // Matching token keeps the entry. + expect(qm.getMain('old-main-token')).not.toBeNull() + // Different token (account switch) drops the entry and returns null. + expect(qm.getMain('new-main-token')).toBeNull() + expect(qm.getMain()).toBeNull() + }) + test('drops persisted main seed during backoff when the token changed', async () => { // Regression: a persisted seed must be bound to the account that produced // it. After a main-account switch (different access token), the seed must From b87233074424736b97388d783f9bd48065ef5936 Mon Sep 17 00:00:00 2001 From: iceteaSA <171169159+iceteaSA@users.noreply.github.com> Date: Mon, 1 Jun 2026 14:58:52 +0200 Subject: [PATCH 5/5] fix(quota): one source of truth for fallback selection (QM cache) Fallback selection used the QuotaManager cache for staleness but evaluated quota policy from the storage account.quota, and the active-route every-N refresh updates only the cache. So a fresh cache entry could suppress a refetch while being ignored by the routing decision (route from stale storage quota), and a same-id re-login could skip the fetch then fail closed on missing account.quota. - getUsableFallbackAccounts now evaluates policy via quotaPolicyAccount(), which overlays the QuotaManager-cached quota (the same source as the staleness check). refreshAccountQuota already updates both, so refreshed and cached decisions stay consistent. - Token-bind fallback cache entries: getFallback/isFallbackStale accept the live access token and drop an entry produced by a different token (re-login); _fetchFallback records the fingerprint. getUsableFallbackAccounts passes next.access so a credential change invalidates the stale entry and refetches. Adds regression tests for both repros. --- packages/core/src/accounts.ts | 25 ++++++- packages/core/src/quota-manager.ts | 25 ++++++- packages/opencode/src/tests/accounts.test.ts | 71 ++++++++++++++++++++ 3 files changed, 116 insertions(+), 5 deletions(-) diff --git a/packages/core/src/accounts.ts b/packages/core/src/accounts.ts index 3e8d4ddc9..65aa0476d 100644 --- a/packages/core/src/accounts.ts +++ b/packages/core/src/accounts.ts @@ -1065,7 +1065,7 @@ export class FallbackAccountManager { } this.seedFallbackQuota(next, storage) const stale = this.quotaManager - ? this.quotaManager.isFallbackStale(next.id) + ? this.quotaManager.isFallbackStale(next.id, next.access) : quotaIsStale(next, storage, this.now()) // Skip the request-time refresh when this account's quota API is // backed off (recent 429/5xx). Hitting it again would extend the @@ -1078,7 +1078,13 @@ export class FallbackAccountManager { next = await this.refreshAccountQuota(next, storage) changed = true } - if (this.accountPassesQuotaPolicy(next, storage)) usable.push(next) + // Single source of truth: evaluate quota policy from the unified + // QuotaManager cache (the same source as the staleness check above) so + // an active-route refresh that updated only the cache is not ignored. + if ( + this.accountPassesQuotaPolicy(this.quotaPolicyAccount(next), storage) + ) + usable.push(next) } catch (error) { if ( canUseCachedQuotaAfterRefreshError( @@ -1124,6 +1130,21 @@ export class FallbackAccountManager { return quotaSnapshotPassesPolicy(account.quota, storage) } + /** + * Return the account with its quota overlaid from the unified QuotaManager + * cache (token-bound) when available, so quota-policy decisions use the same + * source of truth as the staleness check. Falls back to the stored + * account.quota when no manager is wired or the cache has no entry. + */ + private quotaPolicyAccount(account: OAuthAccount): OAuthAccount { + if (!this.quotaManager) return account + const cached = this.quotaManager.getFallback( + account.id, + account.access, + )?.quota + return cached ? { ...account, quota: cached } : account + } + async refreshDueAccounts() { const storage = await this.load() if (!storage || !refreshEnabled(storage)) return diff --git a/packages/core/src/quota-manager.ts b/packages/core/src/quota-manager.ts index 52ac1bac9..099efbbc1 100644 --- a/packages/core/src/quota-manager.ts +++ b/packages/core/src/quota-manager.ts @@ -67,6 +67,10 @@ export class QuotaManager { private main: QuotaEntry | null = null private mainTokenFp: string | null = null private fallbacks = new Map() + // Fingerprint of the access token that produced each fallback cache entry, so + // a re-login (credential change) for the same account id invalidates the + // stale entry instead of being treated as fresh. + private fallbackTokenFps = new Map() // --- Inflight deduplication --- private inflightMain: Promise | null = null @@ -144,7 +148,20 @@ export class QuotaManager { return this.main } - getFallback(accountId: string): QuotaEntry | null { + /** + * Cached fallback quota entry. Pass the live access token to enforce token + * binding: if the entry was produced by a different token (account re-login), + * it is dropped and null is returned so the caller refetches. + */ + getFallback(accountId: string, accessToken?: string): QuotaEntry | null { + if (accessToken) { + const fp = this.fallbackTokenFps.get(accountId) + if (fp && fp !== tokenFingerprint(accessToken)) { + this.fallbacks.delete(accountId) + this.fallbackTokenFps.delete(accountId) + return null + } + } return this.fallbacks.get(accountId) ?? null } @@ -248,8 +265,9 @@ export class QuotaManager { return this.now() >= this.main.refreshAfter } - isFallbackStale(accountId: string): boolean { - const entry = this.fallbacks.get(accountId) + isFallbackStale(accountId: string, accessToken?: string): boolean { + // Token-aware: a credential change invalidates the entry (treated as stale). + const entry = this.getFallback(accountId, accessToken) if (!entry) return true return this.now() >= entry.refreshAfter } @@ -418,6 +436,7 @@ export class QuotaManager { refreshAfter: now + getQuotaCheckIntervalMs(this.storage), checkedAt: now, }) + this.fallbackTokenFps.set(accountId, tokenFingerprint(accessToken)) this.fallbackApiErrors.delete(accountId) return quota } catch (error) { diff --git a/packages/opencode/src/tests/accounts.test.ts b/packages/opencode/src/tests/accounts.test.ts index 6e3c0a80d..21c9adabb 100644 --- a/packages/opencode/src/tests/accounts.test.ts +++ b/packages/opencode/src/tests/accounts.test.ts @@ -12,6 +12,7 @@ import { getCache1hPersistentMode, isFastModePersistentlyEnabled, loadAccounts, + QuotaManager, saveAccounts, setCache1hPersistentEnabled, setCache1hPersistentMode, @@ -976,6 +977,76 @@ describe('FallbackAccountManager', () => { expect(saved?.accounts[0]?.quota?.five_hour?.remainingPercent).toBe(30) expect(saved?.accounts[0]?.quota?.seven_day?.remainingPercent).toBe(70) }) + + test('fallback policy uses the QuotaManager cache, not stale storage quota', async () => { + // Regression: an active-route refresh updates only the QM cache. Selection + // must evaluate quota policy from that cache (single source of truth), not + // the older storage account.quota, or it routes to an exhausted account. + const storage = baseStorage() + storage.accounts.push({ + id: 'fallback-1', + type: 'oauth', + access: 'fallback-access', + refresh: 'fallback-refresh', + expires: Date.now() + 5 * 60 * 60_000, + // Storage says PASSING (100% remaining). + quota: { + five_hour: { usedPercent: 0, remainingPercent: 100, checkedAt: 900 }, + seven_day: { usedPercent: 0, remainingPercent: 100, checkedAt: 900 }, + }, + }) + await saveAccounts(storage) + + const fetchImpl = mock(() => { + throw new Error('should not fetch — QM entry is fresh') + }) as unknown as typeof fetch + const qm = new QuotaManager({ storage, fetchImpl, now: () => 1_000 }) + // Active-route refresh left a FRESH but EXHAUSTED entry in the QM cache. + qm.setFallback('fallback-1', { + quota: { + five_hour: { usedPercent: 100, remainingPercent: 0, checkedAt: 1_000 }, + seven_day: { usedPercent: 100, remainingPercent: 0, checkedAt: 1_000 }, + }, + refreshAfter: 1_000 + 10 * 60_000, + checkedAt: 1_000, + }) + + const manager = new FallbackAccountManager({ + fetchImpl, + now: () => 1_000, + quotaManager: qm, + }) + const accounts = await manager.getUsableFallbackAccounts(storage) + // QM cache is exhausted → account must NOT be usable (was usable when policy + // read the stale storage quota). + expect(accounts.map((a) => a.id)).not.toContain('fallback-1') + }) + + test('re-login (token change) invalidates a fresh fallback cache entry', async () => { + // Regression: a same-id re-login changes the access token. The token-bound + // fallback cache entry from the old token must be treated as stale so the + // new credentials trigger a refetch instead of reusing the old quota. + const fetchImpl = mock(() => + Promise.resolve( + new Response( + JSON.stringify({ + five_hour: { utilization: 10 }, + seven_day: { utilization: 10 }, + }), + { status: 200 }, + ), + ), + ) as unknown as typeof fetch + const qm = new QuotaManager({ storage: null, fetchImpl, now: () => 2_000 }) + + // Cache populated by the OLD token (binds its fingerprint), entry is fresh. + await qm.refreshFallback('fallback-1', 'old-access') + expect(qm.isFallbackStale('fallback-1', 'old-access')).toBe(false) + + // Same account id, NEW token (re-login): entry is invalidated → stale. + expect(qm.isFallbackStale('fallback-1', 'new-access')).toBe(true) + expect(qm.getFallback('fallback-1', 'new-access')).toBeNull() + }) }) describe('buildRefreshOperationError', () => {