diff --git a/docs-site/src/content/docs/fr/guides/claude-code.md b/docs-site/src/content/docs/fr/guides/claude-code.md index b9537a44e79..0b167f7c9d7 100644 --- a/docs-site/src/content/docs/fr/guides/claude-code.md +++ b/docs-site/src/content/docs/fr/guides/claude-code.md @@ -34,8 +34,8 @@ consultez les conditions actuelles d'Anthropic. Comportement lorsque cette option est activée : - Un **429** en amont place le compte en temporisation selon `Retry-After` lorsqu'il est présent, ou selon un délai de repli, - efface ses affinités et peut faire basculer la requête vers un autre compte admissible, dans les limites prévues. -- L'affinité est **locale au processus** et disparaît au redémarrage du proxy. + et peut faire basculer la requête vers un autre compte admissible, dans les limites prévues. Seul un refus de quota partagé efface ses affinités ; une limitation temporaire les conserve. +- L'affinité survit au redémarrage dans le même répertoire de configuration et la même instance fournisseur (24 heures d'inactivité, 2 000 liaisons au maximum). Le fichier privé ne contient que des empreintes de session et d'identité de connexion, jamais de clé de session brute ni de jeton. Les anciens comptes sans identité persistante restent locaux au processus jusqu'à leur prochaine connexion. Le renouvellement du jeton conserve la liaison ; une nouvelle connexion, la suppression, la pause, une réauthentification requise ou une sélection manuelle pendant l'exécution l'invalident. Un détour temporaire dû à une limitation de débit conserve la liaison initiale. - Les erreurs de renouvellement du jeton conservent la règle `needsReauth`. Un 403 confirmé lié à un abonnement ou à la facturation du compte peut déclencher un basculement avant la sortie, avec une temporisation selon `Retry-After` ou de dix minutes. Un refus d’autorisation ordinaire reste terminal. Voir la [reprise des comptes](/guides/claude-code/). - Si chaque compte éligible est en temporisation, le proxy renvoie **429** (et non 401) avec `Retry-After` lorsqu'il est connu. diff --git a/docs-site/src/content/docs/guides/claude-code.md b/docs-site/src/content/docs/guides/claude-code.md index efe3336f3e4..f3464fcb0b2 100644 --- a/docs-site/src/content/docs/guides/claude-code.md +++ b/docs-site/src/content/docs/guides/claude-code.md @@ -109,7 +109,7 @@ Operational contract when enabled: not postpone usage probes or clear a failed usage probe's unavailable status. Measurements whose known reset time has passed are discarded as unknown, including retained model-specific windows. Values without a known reset are preserved; missing data is never reported as zero usage. -- Affinity is **process-local** (lost on proxy restart). +- Affinity survives proxy restart within the same config home and provider instance (24-hour idle lifetime, newest 2,000 bindings). The private snapshot stores session digests and login fences, never raw session keys or tokens. Legacy accounts without a persistent login identity remain process-local until their next login. Token refresh preserves affinity; replacement login, account removal, pause/reauthentication and explicit runtime account selection invalidate it. A transient throttle detour preserves the original binding even when its pause expires before the retry commits. - A complete, structured **403** account-entitlement or billing refusal can rotate before output. Recognized cases include no Claude Code access, an expired/inactive subscription, and an insufficient Anthropic credit balance. The refused account loses its affinities and diff --git a/docs-site/src/content/docs/tr/guides/claude-code.md b/docs-site/src/content/docs/tr/guides/claude-code.md index bc01fe07cef..bc4d17a2ec3 100644 --- a/docs-site/src/content/docs/tr/guides/claude-code.md +++ b/docs-site/src/content/docs/tr/guides/claude-code.md @@ -37,10 +37,9 @@ tavsiye değildir; Anthropic'in güncel koşullarını kontrol edin. Etkinleştirildiğinde operasyonel sözleşme: - Yukarı akıştan gelen **429**, varsa `Retry-After` (yoksa varsayılan bir geri - çekilme) kullanarak o hesabı soğutur, bağlılıklarını temizler ve aynı istek - içinde uygun başka bir hesaba dönebilir (sınırlı). -- Bağlılık **işleme özeldir (process-local)** (proxy yeniden başlatıldığında - kaybolur). + çekilme) kullanarak o hesabı soğutur ve aynı istek içinde uygun başka bir hesaba + dönebilir (sınırlı). Yalnızca ortak kota reddi bağlılıkları temizler; geçici hız sınırlaması bunları korur. +- Bağlılık aynı yapılandırma dizini ve sağlayıcı örneğinde yeniden başlatma sonrasında korunur (24 saat boşta kalma, en yeni 2.000 bağ). Özel dosya yalnızca oturum ve giriş kimliği özetlerini saklar; ham oturum anahtarı veya token içermez. Kalıcı giriş kimliği olmayan eski hesaplar yeniden girişe kadar işleme özeldir. Token yenileme bağı korur; yeniden giriş, kaldırma, duraklatma, yeniden kimlik doğrulama gereksinimi veya çalışma sırasında elle hesap seçimi bağı geçersiz kılar. Geçici hız sınırlaması nedeniyle başka hesaba yönelme özgün bağı değiştirmez. - Token yenileme hataları mevcut `needsReauth` kuralını korur. Doğrulanmış abonelik veya hesap ödeme 403 hatası çıktı başlamadan hesap değiştirebilir; soğuma `Retry-After` veya varsayılan on dakikadır. Genel izin reddi terminal kalır. Bkz. [hesap kurtarma](/guides/claude-code/). - Uygun tüm hesaplar soğutuluyorsa, proxy bilindiğinde `Retry-After` ile birlikte **429** (401 değil) döndürür. diff --git a/docs-site/src/content/docs/zh-tw/guides/claude-code.md b/docs-site/src/content/docs/zh-tw/guides/claude-code.md index 2e392ba1bf3..8b4e9ec42fd 100644 --- a/docs-site/src/content/docs/zh-tw/guides/claude-code.md +++ b/docs-site/src/content/docs/zh-tw/guides/claude-code.md @@ -28,9 +28,9 @@ Anthropic 未認可自動帳號池；同一組織的帳號可能共用配額（ 啟用時的營運契約： -- 上游 **429** 會讓該帳號冷卻（有 `Retry-After` 時使用它，否則用預設 backoff）、清除其 affinity， - 並可能在同一個請求內輪換到另一個合格帳號（有上限）。 -- Affinity 是**程序本機**的（proxy 重啟後就會遺失）。 +- 上游 **429** 會讓該帳號冷卻（有 `Retry-After` 時使用它，否則用預設 backoff）， + 並可能在同一個請求內輪換到另一個合格帳號（有上限）。只有共享配額拒絕會清除 affinity；暫時限流會保留綁定。 +- Affinity 可在相同設定目錄及提供者實例中跨 proxy 重啟保留（閒置 24 小時，最多 2,000 筆）。快照只存工作階段摘要及登入識別圍欄，不存原始工作階段金鑰或 token。沒有持久登入識別的舊帳號在再次登入前僅保留程序內綁定；一般 token 更新保留綁定，重新登入、移除、暫停、需要重新驗證或執行期間手動選擇帳號則使其失效。暫時限流的替代帳號不會覆寫原綁定。 - Token 更新失敗保留既有 `needsReauth` 規則。明確的訂閱或帳號計費 403 可在輸出前切換帳號，冷卻遵循 `Retry-After` 或預設十分鐘；一般權限拒絕不切換。詳見[帳號復原](/guides/claude-code/)。 - 如果每個合格帳號都在冷卻，proxy 會回傳 **429**（不是 401），並在已知時附上 `Retry-After`。 - 復原（包括 429 容錯移轉）會使用 `quotaWindow` 為合格的替代帳號排序，且不改變現有的冷卻或 diff --git a/scripts/test-layout/layout-adapters.json b/scripts/test-layout/layout-adapters.json index 49d42e432b1..b52b3fd193d 100644 --- a/scripts/test-layout/layout-adapters.json +++ b/scripts/test-layout/layout-adapters.json @@ -11,5 +11,7 @@ "provider-quota-abliteration.test.ts": "providers", "claude-inbound-tool-loading.test.ts": "claude-integration", "command-code-plan-denial.test.ts": "providers", - "claude-tool-loading-destinations.test.ts": "claude-integration" + "claude-tool-loading-destinations.test.ts": "claude-integration", + "anthropic-affinity-restart.test.ts": "adapters/anthropic", + "anthropic-affinity-store.test.ts": "adapters/anthropic" } diff --git a/src/oauth/anthropic-rate-limit-policy.ts b/src/oauth/anthropic-rate-limit-policy.ts index 5efbf3c9a2e..d191e4139af 100644 --- a/src/oauth/anthropic-rate-limit-policy.ts +++ b/src/oauth/anthropic-rate-limit-policy.ts @@ -6,6 +6,12 @@ export type AnthropicRateLimitKind = "shared-quota" | "family-quota" | "transien export type RateLimitHeaders = Pick; export const ANTHROPIC_SHORT_RETRY_MS = 100; export const ANTHROPIC_MAX_INLINE_THROTTLE_MS = 1_000; +/** + * How long a transient pause keeps marking its account as only detoured from (#7813). A detour + * awaits the alternate's credential and selection commit after the pause is set; a pause that + * expires during those awaits must not turn the detour into a permanent rebinding. + */ +export const ANTHROPIC_DETOUR_GRACE_MS = 5 * 60_000; export function anthropicRetryAfterMs(value: string | null | undefined, now: number): number | undefined { const text = value?.trim(); @@ -27,6 +33,9 @@ export function classifyAnthropic429(headers: RateLimitHeaders, now = Date.now() function createAnthropicRatePolicy(instance: AnthropicInstanceId) { let pauses: Map | undefined; + // Accounts paused within the detour grace window, by credential generation. Bounded by the + // stored account set (one row per account) and pruned with the pauses. + let recent: Map | undefined; function pauseAnthropicRateAdmission(accountId: string, until: number): void { const credential = getAccountCredential(instance, accountId); if (!credential) return; @@ -36,6 +45,9 @@ function createAnthropicRatePolicy(instance: AnthropicInstanceId) { for (const [id, pause] of rows) if (pause.until <= now) rows.delete(id); const previous = rows.get(accountId); rows.set(accountId, { until: Math.max(previous?.generation === generation ? previous.until : 0, until), generation }); + const marks = recent ??= new Map(); + for (const [id, mark] of marks) if (now - mark.at > ANTHROPIC_DETOUR_GRACE_MS) marks.delete(id); + marks.set(accountId, { at: now, generation }); } function anthropicRatePauseUntil(accountId: string, now = Date.now()): number | undefined { const pause = pauses?.get(accountId); @@ -46,21 +58,33 @@ function createAnthropicRatePolicy(instance: AnthropicInstanceId) { } return pause.until; } - function clearAnthropicRatePauses(): void { pauses = undefined; } + /** True while the account's latest transient pause is younger than the detour grace window. */ + function anthropicRatePausedRecently(accountId: string, now = Date.now()): boolean { + const mark = recent?.get(accountId); + if (!mark) return false; + const credential = getAccountCredential(instance, accountId); + if (now - mark.at > ANTHROPIC_DETOUR_GRACE_MS || !credential || credentialGeneration(credential) !== mark.generation) { + recent!.delete(accountId); return false; + } + return true; + } + function clearAnthropicRatePauses(): void { pauses = undefined; recent = undefined; } function reconcileAnthropicRatePauses(context: GenerationContext): number { let removed = 0; for (const id of pauses?.keys() ?? []) { if (context.oauthAccountKeys.has(`${instance}\0${id}`)) continue; pauses!.delete(id); removed++; } + for (const id of recent?.keys() ?? []) if (!context.oauthAccountKeys.has(`${instance}\0${id}`)) recent!.delete(id); return removed; } function sweepExpiredAnthropicRatePauses(now = Date.now()): number { let removed = 0; for (const [id, pause] of pauses ?? []) if (pause.until <= now) { pauses!.delete(id); removed++; } + for (const [id, mark] of recent ?? []) if (now - mark.at > ANTHROPIC_DETOUR_GRACE_MS) recent!.delete(id); return removed; } - return Object.freeze({ instance, pauseAnthropicRateAdmission, anthropicRatePauseUntil, + return Object.freeze({ instance, pauseAnthropicRateAdmission, anthropicRatePauseUntil, anthropicRatePausedRecently, clearAnthropicRatePauses, reconcileAnthropicRatePauses, sweepExpiredAnthropicRatePauses }); } export type AnthropicRatePolicy = ReturnType; diff --git a/src/oauth/anthropic-routing.ts b/src/oauth/anthropic-routing.ts index c9de9371896..4fe1120d905 100644 --- a/src/oauth/anthropic-routing.ts +++ b/src/oauth/anthropic-routing.ts @@ -12,7 +12,7 @@ import { clearAllAnthropicFamilyQuota, anthropicModelQuotaFor, anthropicModelPer * Intentionally narrower than the Codex pool: no mid-session quota rotation, * soft-avoid ladders, or probe leases. Anthropic OAuth is ToS-sensitive. * - * Affinity is process-local (lost on restart). Cooldown uses Retry-After when present, else + * Affinity survives restart in a private digest-only snapshot. Cooldown uses Retry-After when present, else * the reset time of whichever rate-limit window upstream reports as rejected, else a default * backoff. Classified entitlement/billing 403s use a ten-minute default cooldown. * Token refresh failures retain the existing store needsReauth policy. @@ -46,6 +46,7 @@ import { configuredAnthropicInstance, isAnthropicInstanceId, type AnthropicInsta import { resolveAnthropicAccountPoolConfig } from "./anthropic-pool-config"; import { anthropicPoolKey } from "./pool-kernel"; import type { GenerationContext } from "../lib/state-store-sweeper"; +import { AnthropicSessionAffinityStore, anthropicAffinityDigest, AFFINITY_IDLE_TTL_MS } from "./anthropic-session-affinity-store"; /** * The read side of a `Headers` object, so a caller can pass the live upstream response's @@ -57,8 +58,6 @@ export type AnthropicRateLimitHeaders = Pick; /** Backoff only when upstream supplies no usable deadline. */ const DEFAULT_COOLDOWN_MS = 60_000; const ACCOUNT_REFUSAL_COOLDOWN_MS = 10 * 60_000; -const AFFINITY_IDLE_TTL_MS = 24 * 60 * 60_000; -const MAX_AFFINITY_ENTRIES = 2_000; const MAX_AFFINITY_COMPONENT_BYTES = 512; const UNKNOWN_USAGE_SCORE = 100; const DEFAULT_AUTO_SWITCH_THRESHOLD = 80; @@ -132,11 +131,6 @@ interface AnthropicAccountHold { resetAt?: number; } -interface AffinityEntry { - accountId: string; - lastUsedAt: number; -} - function normalizeAffinityComponent(value: string | null | undefined): string { const normalized = value?.trim() ?? ""; return normalized && retainedUtf8Bytes(normalized) <= MAX_AFFINITY_COMPONENT_BYTES ? normalized : ""; @@ -366,13 +360,14 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting const PROVIDER = instance; const poolKey = anthropicPoolKey(instance); const { anthropicFamilyRejected, anthropicFamilyRetryAt, anthropicFamilyHoldState, anthropicModelExhausted, clearAnthropicFamilyQuota } = anthropicModelQuotaFor(instance); - const { anthropicRatePauseUntil, clearAnthropicRatePauses, pauseAnthropicRateAdmission } = anthropicRatePolicyFor(instance); + const { anthropicRatePauseUntil, anthropicRatePausedRecently, clearAnthropicRatePauses, pauseAnthropicRateAdmission } = anthropicRatePolicyFor(instance); const { anthropicCooldownGeneration, clearAnthropicCooldownGenerations, noteAnthropicCooldownMutation } = anthropicCooldownRecoveryFor(instance); // Request-owned config authorizes selection/commit. Namespace-only credential and roster // methods carry no ambient disk authority; physical consumers recheck their captured config/target. function admitted(config: OcxConfig): boolean { return configuredAnthropicInstance(config, instance) === instance; } const upstreamHealth = new RoutingMap(); - const sessionAffinity = new RoutingMap(); + const sessionAffinity = new AnthropicSessionAffinityStore(instance); + const affinityKey = (key: string | null | undefined) => anthropicAffinityDigest(instance, normalizeAffinityComponent(key)); // Undefined means this runtime has not admitted a selection yet; null means consumed. // The startup baseline comes from the authoritative store, never a second persisted pin. let manualPreference: OAuthAccountSelection | null | undefined; @@ -532,7 +527,7 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting clearAnthropicRatePauses(); clearAnthropicFamilyQuota(); clearAnthropicCooldownGenerations(); - sessionAffinity.clear(); + sessionAffinity.resetMemory(); manualPreference = undefined; manualSelectionGeneration++; quorumCache = null; @@ -880,16 +875,6 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting return pickLowestUsage(config, excludeId, now, decision, model, excludedAccountIds); } - function pruneExpiredAffinity(now: number): void { - for (const [key, entry] of sessionAffinity) { - if (now - entry.lastUsedAt > AFFINITY_IDLE_TTL_MS) sessionAffinity.delete(key); - } - if (sessionAffinity.size <= MAX_AFFINITY_ENTRIES) return; - const sorted = [...sessionAffinity.entries()].sort((a, b) => a[1].lastUsedAt - b[1].lastUsedAt); - const drop = sessionAffinity.size - MAX_AFFINITY_ENTRIES; - for (let i = 0; i < drop; i++) sessionAffinity.delete(sorted[i]![0]); - } - function stickyLimitForPool(config: OcxConfig): number { return normalizeAccountPoolStickyLimit(anthropicAccountPoolConfig(config).stickyLimit); } @@ -976,7 +961,6 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting model?: string, ): AnthropicAccountSelection { if (!admitted(config)) return { accountId: null, reason: "none", routePosition: decision?.position }; - pruneExpiredAffinity(now); const set = getAccountSet(PROVIDER); if (!set || set.accounts.length === 0) return { accountId: null, reason: "none", ...(decision ? { routePosition: decision.position } : {}) }; const scoped = decision && !decision.fallback @@ -1013,7 +997,7 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting // A manual choice is a one-dispatch preference, not a lower-priority quota hint. // Consume it only after admission commits, so a failed token lookup cannot spend it. - if (manualPreference) { + if (manualPreference && !sessionAffinity.get(affinityKey(sessionKey), set, now)) { if (manualPreference.accountId !== set.activeAccountId || manualPreference.revision !== set.selectionRevision) { manualPreference = null; } else { @@ -1026,14 +1010,15 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting } } - const key = normalizeAffinityComponent(sessionKey); + const key = affinityKey(sessionKey); if (key) { - const affined = sessionAffinity.get(key); + const affined = sessionAffinity.get(key, set, now); if (affined && now - affined.lastUsedAt <= AFFINITY_IDLE_TTL_MS) { const stillThere = set.accounts.some(a => a.id === affined.accountId && a.paused !== true && a.needsReauth !== true); const stillUsable = stillThere && !isCooled(affined.accountId, now) && isPoolCredentialUsable(affined.accountId, now); if (stillUsable && eligible.includes(affined.accountId) && (!model || !anthropicModelExhausted(affined.accountId, model))) { + sessionAffinity.set(key, { ...affined, lastUsedAt: now }, set); return { accountId: affined.accountId, reason: "affinity", routePosition: decision?.position }; } // A model route may exclude a healthy binding only for this request. Keep it for @@ -1134,16 +1119,13 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting accountId: string, now = Date.now(), ): void { - const key = normalizeAffinityComponent(sessionKey); + const key = affinityKey(sessionKey); if (!key || !normalizeAffinityComponent(accountId)) return; sessionAffinity.set(key, { accountId, lastUsedAt: now }); - pruneExpiredAffinity(now); } function clearAnthropicSessionAffinityForAccount(accountId: string): void { - for (const [key, entry] of sessionAffinity) { - if (entry.accountId === accountId) sessionAffinity.delete(key); - } + sessionAffinity.invalidate(getAccountSet(PROVIDER), id => id !== accountId); // The roster just lost or changed a member. This is the account-removal path, so the next // activation question must re-read rather than answer from a count taken while the account // was still present -- otherwise a delete leaves a stale quorum for the length of the TTL. @@ -1309,16 +1291,23 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting if (picked !== accountId) seedPoolRotationAccount(poolKey, accountId); notePoolRotationSuccess(poolKey, accountId, limit); } - const key = normalizeAffinityComponent(options.sessionKey); + const key = affinityKey(options.sessionKey); const bound = key ? sessionAffinity.get(key) : undefined; const eligibleAtCommit = options.routeDecision && bound ? getEligibleAnthropicAccounts() : []; const preserveExcludedAffinity = options.routeDecision && bound && bound.accountId !== accountId && Date.now() - bound.lastUsedAt <= AFFINITY_IDLE_TTL_MS && eligibleAtCommit.includes(bound.accountId) && !routeCandidates(eligibleAtCommit, options.routeDecision).includes(bound.accountId); - const preservePausedAffinity = bound && (anthropicRatePauseUntil(bound.accountId) !== undefined - || anthropicFamilyRejected(bound.accountId, options.model)); - if (!preserveExcludedAffinity && !preservePausedAffinity) bindAnthropicSessionAffinity(options.sessionKey, accountId); + // A transient detour must not become a rebinding just because the pause expired while + // the alternate's credential and this commit were awaited (#7813). Shared-quota and + // entitlement refusals clear the binding explicitly before any alternate is committed, + // and measured exhaustion of the bound account always outranks a recent throttle. + const preservePausedAffinity = bound && bound.accountId !== accountId + && !(options.model && anthropicModelExhausted(bound.accountId, options.model)) + && (anthropicRatePauseUntil(bound.accountId) !== undefined || anthropicRatePausedRecently(bound.accountId) + || anthropicFamilyRejected(bound.accountId, options.model)); + if (key && bound && (preserveExcludedAffinity || preservePausedAffinity)) sessionAffinity.set(key, { ...bound, lastUsedAt: Date.now() }); + else bindAnthropicSessionAffinity(options.sessionKey, accountId); } if (manualPreference === undefined || (manualPreference?.accountId === expectedSelection.accountId && manualPreference.revision === expectedSelection.revision)) manualPreference = null; @@ -1397,7 +1386,10 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting } // Pause changes eligibility, not health. Do not reset cooldowns or cancel sent turns. - function onPause(): void { quorumCache = null; } + function onPause(): void { + quorumCache = null; + sessionAffinity.invalidate(); + } // A threshold write must fence in-flight automatic proposals, but it does not // revoke an operator's one-dispatch choice. Rebase only that still-owned choice; // an intervening account change clears it, so an ABA selection is not resurrected. @@ -1410,10 +1402,12 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting // Any non-policy selection generation supersedes the pending one-shot choice. // Threshold mutations rebase it first, before this generic notification runs. function onSelection(event: Parameters[0]>[0]): void { - if (event.provider !== PROVIDER || event.kind !== "oauth" || !manualPreference) return; - const current = captureOAuthAccountSelection(PROVIDER); - if (current?.accountId !== manualPreference.accountId - || current.revision !== manualPreference.revision) manualPreference = null; + if (event.provider !== PROVIDER || event.kind !== "oauth") return; + const snapshot = getAccountSet(PROVIDER); + sessionAffinity.invalidate(snapshot); + if (!manualPreference) return; + if (snapshot?.activeAccountId !== manualPreference.accountId + || snapshot.selectionRevision !== manualPreference.revision) manualPreference = null; } @@ -1427,11 +1421,7 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting upstreamHealth.delete(id); removed++; } - for (const [key, entry] of sessionAffinity) { - if (valid(entry.accountId)) continue; - sessionAffinity.delete(key); - removed++; - } + removed += sessionAffinity.invalidate(getAccountSet(PROVIDER), valid); if (manualPreference && !valid(manualPreference.accountId)) { manualPreference = null; removed++; } if (removed > 0) manualSelectionGeneration++; quorumCache = null; diff --git a/src/oauth/anthropic-session-affinity-store.ts b/src/oauth/anthropic-session-affinity-store.ts new file mode 100644 index 00000000000..299d06d3249 --- /dev/null +++ b/src/oauth/anthropic-session-affinity-store.ts @@ -0,0 +1,177 @@ +/** Private, bounded Anthropic affinity snapshots. No session or credential bytes reach disk. */ +import { createHash } from "node:crypto"; +import { join } from "node:path"; +import { atomicWriteFileNoFollow } from "../config/atomic-write"; +import { getConfigDir, withConfigDirHardeningSettled } from "../config/paths"; +import { registerOptionalShutdownHook } from "../lib/optional-shutdown-hooks"; +import { trackBackgroundWork } from "../lib/background-work"; +import { getAccountSet } from "./store"; +import { readBoundedRegularFileSync } from "../lib/bounded-file-read"; +import type { AnthropicInstanceId } from "../providers/anthropic-instance-id"; + +export const AFFINITY_IDLE_TTL_MS = 24 * 60 * 60_000; +export const MAX_AFFINITY_ENTRIES = 2_000; +export const MAX_AFFINITY_FILE_BYTES = 1_048_576; +const TOUCH_DELAY_MS = 1_000; +type Entry = { accountId: string; lastUsedAt: number; loginFence?: string }; +type AccountSnapshot = ReturnType; +type Fences = ReadonlyMap; +type Scope = { home: string; rows: Map; dirty: boolean; active: boolean; revision: number; + timer?: ReturnType; pending?: Promise }; +const owners = new Set(); +const digest = (domain: string, value: string) => createHash("sha256").update(`${domain}\0${value}`).digest("hex"); +export function anthropicAffinityDigest(instance: AnthropicInstanceId, normalizedKey: string): string { + return normalizedKey ? digest(`opencodex/anthropic-affinity/session/v1/${instance}`, normalizedKey) : ""; +} +export function anthropicAffinityPath(home: string, instance: AnthropicInstanceId): string { + return join(home, `anthropic-affinity-${digest("opencodex/anthropic-affinity/instance/v1", instance)}.json`); +} +function warn(): void { console.warn("[anthropic-affinity] snapshot persistence failed; using runtime affinity"); } + +export class AnthropicSessionAffinityStore implements Iterable<[string, Entry]> { + private scope?: Scope; + constructor(private readonly instance: AnthropicInstanceId, private readonly write = atomicWriteFileNoFollow) { owners.add(this); } + private fences(snapshot: AccountSnapshot = getAccountSet(this.instance)): Fences { + const fences = new Map(); + for (const account of snapshot?.accounts ?? []) { + if (account.paused || account.needsReauth) continue; + // Legacy accounts without a persistent login identity remain memory-only until next login. + fences.set(account.id, account.loginId ? digest("opencodex/anthropic-affinity/login/v1", account.loginId) : undefined); + } + return fences; + } + private valid(entry: Entry, fences: Fences): boolean { + return fences.has(entry.accountId) && fences.get(entry.accountId) === entry.loginFence; + } + private current(fences?: Fences): Scope { + const home = getConfigDir(); + if (this.scope?.home === home) return this.scope; + this.resetMemory(); + const scope: Scope = { home, rows: new Map(), dirty: false, active: true, revision: 0 }; + this.scope = scope; + // Starts synchronously; retained ACL children are joined by the tracked writer. + registerOptionalShutdownHook(`anthropic-affinity:${this.instance}`, () => { void this.flush(); }); + try { + const bytes = readBoundedRegularFileSync(anthropicAffinityPath(home, this.instance), MAX_AFFINITY_FILE_BYTES); + const snapshot = JSON.parse(bytes.toString("utf8")); + if (!snapshot || snapshot.version !== 1 || !Array.isArray(snapshot.entries)) return scope; + fences ??= this.fences(); + const now = Date.now(); + for (const row of snapshot.entries) { + if (!row || typeof row.sessionDigest !== "string" || !/^[a-f0-9]{64}$/.test(row.sessionDigest) + || typeof row.accountId !== "string" || row.accountId.length > 512 + || typeof row.loginFence !== "string" || !/^[a-f0-9]{64}$/.test(row.loginFence) + || !Number.isFinite(row.lastUsedAt) || row.lastUsedAt > now + || now - row.lastUsedAt > AFFINITY_IDLE_TTL_MS || !this.valid(row, fences)) continue; + const previous = scope.rows.get(row.sessionDigest); + if (!previous || previous.lastUsedAt < row.lastUsedAt) scope.rows.set(row.sessionDigest, { + accountId: row.accountId, lastUsedAt: row.lastUsedAt, loginFence: row.loginFence, + }); + } + this.trim(scope); + if (scope.rows.size !== snapshot.entries.length) this.changed(scope); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") { + console.warn("[anthropic-affinity] snapshot unavailable or invalid; using runtime affinity"); + } + } + return scope; + } + private trim(scope: Scope): void { + if (scope.rows.size <= MAX_AFFINITY_ENTRIES) return; + const ordered = [...scope.rows].sort((a, b) => b[1].lastUsedAt - a[1].lastUsedAt); + for (const [key] of ordered.slice(MAX_AFFINITY_ENTRIES)) scope.rows.delete(key); + } + private changed(scope: Scope, touch = false): void { + scope.dirty = true; + scope.revision++; + if (touch) { + scope.timer ??= setTimeout(() => { scope.timer = undefined; void this.flushScope(scope); }, TOUCH_DELAY_MS); + scope.timer.unref(); + } else { void this.flushScope(scope); } + } + private flushSync(scope: Scope): void { + if (!scope.active || !scope.dirty || this.scope !== scope) return; + if (getConfigDir() !== scope.home) { this.resetMemory(); return; } + try { + const entries = [...scope.rows].filter(([, row]) => row.loginFence).map(([sessionDigest, row]) => ({ sessionDigest, ...row })); + this.write(anthropicAffinityPath(scope.home, this.instance), JSON.stringify({ version: 1, entries })); + scope.dirty = false; + } catch { warn(); } + } + async flush(): Promise { + const scope = this.scope; + if (scope) await this.flushScope(scope); + } + private async flushScope(scope: Scope): Promise { + if (!scope?.active) return; + if (getConfigDir() !== scope.home) { if (this.scope === scope) this.resetMemory(); return; } + if (scope.timer) { clearTimeout(scope.timer); scope.timer = undefined; } + if (scope.pending) return scope.pending; + if (!scope.dirty) return; + let attemptedRevision = scope.revision; + scope.pending = trackBackgroundWork("anthropic-affinity-write", withConfigDirHardeningSettled(scope.home, () => { + attemptedRevision = scope.revision; + this.flushSync(scope); + }).catch(() => { warn(); }).finally(() => { + scope.pending = undefined; + // A mutation after publication but before settlement still needs its own write. + if (scope.active && scope.dirty && scope.revision !== attemptedRevision) return this.flushScope(scope); + })); + return scope.pending; + } + resetMemory(): void { + if (this.scope) { + this.scope.active = false; + if (this.scope.timer) clearTimeout(this.scope.timer); + } + this.scope = undefined; + } + get size(): number { return this.current().rows.size; } + get(key: string, snapshot: AccountSnapshot = getAccountSet(this.instance), now = Date.now()): Entry | undefined { + const fences = this.fences(snapshot); + const scope = this.current(fences); + const row = scope.rows.get(key); + if (row && (!this.valid(row, fences) || now - row.lastUsedAt > AFFINITY_IDLE_TTL_MS)) { + scope.rows.delete(key); this.changed(scope); return undefined; + } + return row; + } + set(key: string, value: Entry, snapshot: AccountSnapshot = getAccountSet(this.instance)): void { + if (!/^[a-f0-9]{64}$/.test(key) || !Number.isFinite(value.lastUsedAt)) return; + const fences = this.fences(snapshot); + const scope = this.current(fences); + if (!fences.has(value.accountId)) { this.delete(key); return; } + const previous = scope.rows.get(key); + const loginFence = fences.get(value.accountId); + scope.rows.set(key, { accountId: value.accountId, lastUsedAt: value.lastUsedAt, loginFence }); + this.trim(scope); + this.changed(scope, previous?.accountId === value.accountId && previous.loginFence === loginFence); + } + /** Roster mutations validate every binding once, against one captured auth generation. */ + invalidate(snapshot: AccountSnapshot = getAccountSet(this.instance), keep: (accountId: string) => boolean = () => true): number { + const fences = this.fences(snapshot); + const scope = this.current(fences); + const now = Date.now(); + let removed = 0; + for (const [key, row] of scope.rows) { + if (this.valid(row, fences) && keep(row.accountId) + && row.lastUsedAt <= now && now - row.lastUsedAt <= AFFINITY_IDLE_TTL_MS) continue; + scope.rows.delete(key); removed++; + } + if (removed) this.changed(scope); + return removed; + } + delete(key: string): boolean { + const scope = this.current(); + const deleted = scope.rows.delete(key); + if (deleted) this.changed(scope); + return deleted; + } + clear(): void { const scope = this.current(); scope.rows.clear(); this.changed(scope); } + *entries(): IterableIterator<[string, Entry]> { yield* this.current().rows; } + [Symbol.iterator](): IterableIterator<[string, Entry]> { return this.entries(); } +} +export async function flushAnthropicSessionAffinity(): Promise { + for (const owner of owners) await owner.flush(); +} diff --git a/structure/providers/anthropic-account-pool.md b/structure/providers/anthropic-account-pool.md index adc84b66ecb..76bcf5cca20 100644 --- a/structure/providers/anthropic-account-pool.md +++ b/structure/providers/anthropic-account-pool.md @@ -470,6 +470,30 @@ than a promise that a route is eligible; explicit rollout opt-outs still apply. The shared conversation-key precedence is implemented by `src/oauth/session-affinity-key.ts` and re-exported as `anthropicSessionKeyFromParts`; Anthropic routing semantics remain owned here. [Generic OAuth affinity](generic-account-pool.md#conversation-affinity) uses independent request-local receipts. +### Durable session affinity + +`src/oauth/anthropic-session-affinity-store.ts` lazily hydrates a versioned `anthropic-affinity-.json` +under the current config home before affinity selection. Each row contains only a domain-separated SHA-256 +session digest, opaque account ID, last-use timestamp and hashed persistent login identity. Legacy accounts +without `loginId` retain runtime-only affinity. Token refresh preserves the fence; replacement login, +removal/re-add, pause and reauthentication invalidate it. Invalid, oversized, future and expired rows are ignored; +the newest 2,000 entries survive, with a 24-hour idle lifetime. A restored binding precedes the startup active +preference; an explicit runtime selection clears bindings and persists the reset. + +Temporary throttle and model-route/family detours retain a valid original binding and refresh its idle timestamp. +Commit preserves a binding excluded by the route, an active admission pause or family rejection, or a transient +pause recorded within `ANTHROPIC_DETOUR_GRACE_MS` (five minutes), even if that pause expired during preparation. +Measured shared or model-specific exhaustion rebinds to the committed replacement; shared-quota and entitlement +refusals explicitly invalidate bindings. Hydration and pause, selection, removal and reconciliation events validate +rows against one auth snapshot, deriving each account's login fence once. Hot reads and writes validate only the +touched row; ordinary publication does not reread auth or revalidate every binding. Structural writes use the config +ACL settlement barrier and private no-follow atomic publisher; bounded hydration rejects final-entry symlinks, +junctions and non-regular files with identifier-free warnings. Aliases above the config root remain supported. +Memory and pending writes belong to the resolved config home; reset or an observed home change retires that scope. +Timestamp-only updates coalesce for at most one second, with a best-effort shutdown flush. Failures emit identifier-free warnings +and preserve the previous snapshot. `tests/adapters/anthropic/anthropic-affinity-restart.test.ts` and +`tests/adapters/anthropic/anthropic-affinity-store.test.ts` exercise routing and persistence boundaries. + ## JEV advisory publication While a live JEV combo enables quota signals, auth-store load/persist and accepted quota producer diff --git a/tests/adapters/anthropic/anthropic-affinity-restart.test.ts b/tests/adapters/anthropic/anthropic-affinity-restart.test.ts new file mode 100644 index 00000000000..89744c72920 --- /dev/null +++ b/tests/adapters/anthropic/anthropic-affinity-restart.test.ts @@ -0,0 +1,110 @@ +import { afterEach, beforeEach, expect, test, spyOn } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { removeTreeWithRetry } from "../../helpers/remove-tree"; +import { captureOAuthAccountSelection, getAccountSet, saveCredential, setActiveAccount } from "../../../src/oauth/store"; +import { bindAnthropicSessionAffinity, clearAnthropicAccountPoolState, resolveAnthropicAccountForSession, + promoteAnthropicActiveAccount, recordAnthropicAccount429, resetAnthropicRoutingForManualSelection } from "../../../src/oauth/anthropic-routing"; +import type { OcxConfig } from "../../../src/types"; +import { clearAccountQuotaCache, setCachedProviderAccountQuotaForTests } from "../../../src/providers/quota"; + +const originalHome = process.env.OPENCODEX_HOME; +let home: string; +let clock: ReturnType | undefined; +const config = { providers: { anthropic: { adapter: "anthropic", authMode: "oauth" } }, + anthropicAccountPool: { enabled: true, autoSwitchThreshold: 0 } } as unknown as OcxConfig; +const session = "private-session-L7-never-persist-raw"; +async function flush() { + // This owner exists with the implementation; keeping the regression runnable on dev + // proves the behavioral failure rather than a missing-export failure. + const store = await import("../../../src/oauth/anthropic-session-affinity-store").catch(() => null); + await store?.flushAnthropicSessionAffinity(); +} +beforeEach(() => { + home = mkdtempSync(join(tmpdir(), "ocx-affinity-restart-")); + process.env.OPENCODEX_HOME = home; + clearAnthropicAccountPoolState(); + clearAccountQuotaCache(); +}); +afterEach(async () => { + clock?.mockRestore(); clock = undefined; + await flush(); + clearAnthropicAccountPoolState(); + if (originalHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = originalHome; + removeTreeWithRetry(home); +}); +async function seed() { + for (const id of ["A", "B"]) await saveCredential("anthropic", { + access: `fixture-access-${id}`, refresh: `fixture-refresh-${id}`, expires: Date.now() + 3_600_000, accountId: id, + }); + const [a, b] = getAccountSet("anthropic")!.accounts; + await setActiveAccount("anthropic", a!.id); + await promoteAnthropicActiveAccount(a!.id, captureOAuthAccountSelection("anthropic"), { config }); + bindAnthropicSessionAffinity(session, a!.id); + return { a: a!.id, b: b!.id }; +} +test("transient detour commit preserves binding after pause expires", async () => { + const { a, b } = await seed(); + let now = Date.now(); clock = spyOn(Date, "now").mockImplementation(() => now); + recordAnthropicAccount429(config, a, "0.01", now); + expect(resolveAnthropicAccountForSession(session, config).accountId).toBe(b); + now += 20; + await promoteAnthropicActiveAccount(b, captureOAuthAccountSelection("anthropic"), { config, sessionKey: session }); + expect(resolveAnthropicAccountForSession(session, config).accountId).toBe(a); +}); +test("restored binding precedes startup active preference", async () => { + const { a, b } = await seed(); + await flush(); + await setActiveAccount("anthropic", b); + clearAnthropicAccountPoolState(); + expect(resolveAnthropicAccountForSession(session, config).accountId).toBe(a); +}); +test("runtime manual selection overrides binding", async () => { + const { b } = await seed(); + await setActiveAccount("anthropic", b); + resetAnthropicRoutingForManualSelection(b); + expect(resolveAnthropicAccountForSession(session, config).accountId).toBe(b); +}); +test("normal transient detour returns to the original and refreshes idle lifetime", async () => { + const { a, b } = await seed(); + let now = Date.now(); clock = spyOn(Date, "now").mockImplementation(() => now); + bindAnthropicSessionAffinity(session, a, now - 24 * 60 * 60_000 + 5); + recordAnthropicAccount429(config, a, "0.1", now); + expect(resolveAnthropicAccountForSession(session, config).accountId).toBe(b); + await promoteAnthropicActiveAccount(b, captureOAuthAccountSelection("anthropic"), { config, sessionKey: session }); + now += 200; + expect(resolveAnthropicAccountForSession(session, config).accountId).toBe(a); +}); +test("shared-quota rejection rebinds even after a short Retry-After expires", async () => { + const { a, b } = await seed(); + let now = Date.now(); clock = spyOn(Date, "now").mockImplementation(() => now); + recordAnthropicAccount429(config, a, "0.01", now, new Headers({ "anthropic-ratelimit-unified-5h-status": "rejected" })); + now += 20; + await promoteAnthropicActiveAccount(b, captureOAuthAccountSelection("anthropic"), { config, sessionKey: session }); + expect(resolveAnthropicAccountForSession(session, config).accountId).toBe(b); +}); +test.each([ + ["shared five-hour", false], ["model-specific", false], ["shared five-hour", true], ["model-specific", true], +] as const)("measured %s exhaustion commits a replacement binding (earlier expired throttle: %p)", async (kind, throttled) => { + const { a, b } = await seed(); + const model = "claude-opus-5-5"; + if (throttled) { + // A short throttle that has already expired must not outrank the measured exhaustion below. + const start = Date.now(); + let now = start; clock = spyOn(Date, "now").mockImplementation(() => now); + recordAnthropicAccount429(config, a, "0.01", now); + now = start + 50; + } + setCachedProviderAccountQuotaForTests("anthropic", a, kind === "shared five-hour" + ? { fiveHourPercent: 100 } : { customWindows: [{ label: "Opus", scope: "model", percent: 100 }] }); + setCachedProviderAccountQuotaForTests("anthropic", b, { fiveHourPercent: 10 }); + const selected = resolveAnthropicAccountForSession(session, config, Date.now(), null, model); + expect(selected.accountId).toBe(b); + expect(await promoteAnthropicActiveAccount(b, captureOAuthAccountSelection("anthropic"), { + config, sessionKey: session, model, reason: selected.reason, + })).not.toBeNull(); + setCachedProviderAccountQuotaForTests("anthropic", a, { fiveHourPercent: 0 }); + expect(resolveAnthropicAccountForSession(session, config, Date.now(), null, model)).toMatchObject({ accountId: b, reason: "affinity" }); +}); diff --git a/tests/adapters/anthropic/anthropic-affinity-store.test.ts b/tests/adapters/anthropic/anthropic-affinity-store.test.ts new file mode 100644 index 00000000000..ead6f20538d --- /dev/null +++ b/tests/adapters/anthropic/anthropic-affinity-store.test.ts @@ -0,0 +1,211 @@ +import { afterEach, beforeEach, expect, test, spyOn } from "bun:test"; +import { existsSync, lstatSync, mkdtempSync, readFileSync, symlinkSync, unlinkSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { removeTreeWithRetry } from "../../helpers/remove-tree"; +import { getAccountSet, removeAccount, saveCredential, saveAccountCredential, markAccountNeedsReauth } from "../../../src/oauth/store"; +import { AnthropicSessionAffinityStore, anthropicAffinityDigest, anthropicAffinityPath, + AFFINITY_IDLE_TTL_MS, MAX_AFFINITY_FILE_BYTES } from "../../../src/oauth/anthropic-session-affinity-store"; +import { atomicWriteFile, atomicWriteFileNoFollow } from "../../../src/config/atomic-write"; +import { runOptionalShutdownHooks } from "../../../src/lib/optional-shutdown-hooks"; +import { settleBackgroundWork } from "../../../src/lib/background-work"; +import * as authStore from "../../../src/oauth/store"; + +const originalHome = process.env.OPENCODEX_HOME; +let home: string; +let id: string; +let stores: AnthropicSessionAffinityStore[]; +const secret = "never-store-this-session-or-bearer"; +const key = anthropicAffinityDigest("anthropic", secret); +const path = () => anthropicAffinityPath(home, "anthropic"); +const credential = () => ({ access: "fixture-only-bearer", refresh: "fixture-only-refresh", expires: Date.now() + 3_600_000, accountId: "fixture-account" }); +function owner(instance: "anthropic" | "anthropic2" = "anthropic", write?: typeof atomicWriteFile) { + const store = new AnthropicSessionAffinityStore(instance, write); stores.push(store); return store; +} +beforeEach(async () => { + home = mkdtempSync(join(tmpdir(), "ocx-affinity-store-")); process.env.OPENCODEX_HOME = home; stores = []; + await saveCredential("anthropic", credential()); id = getAccountSet("anthropic")!.accounts[0]!.id; +}); +afterEach(async () => { + for (const store of stores) { await store.flush(); store.resetMemory(); } + if (originalHome === undefined) delete process.env.OPENCODEX_HOME; else process.env.OPENCODEX_HOME = originalHome; + removeTreeWithRetry(home); +}); +async function snapshot() { + const store = owner(); store.set(key, { accountId: id, lastUsedAt: Date.now() }); await store.flush(); + return { store, data: JSON.parse(readFileSync(path(), "utf8")) }; +} +test("snapshot contains only digests, opaque account and timestamp", async () => { + const { store } = await snapshot(); const bytes = readFileSync(path(), "utf8"); + for (const value of [secret, "fixture-only-bearer", "fixture-only-refresh", "fixture-account"]) expect(bytes).not.toContain(value); + const row = JSON.parse(bytes).entries[0]; + expect(Object.keys(row).sort()).toEqual(["accountId", "lastUsedAt", "loginFence", "sessionDigest"]); + expect(row.sessionDigest).toBe(key); expect(row.loginFence).toMatch(/^[a-f0-9]{64}$/); + store.resetMemory(); expect(store.get(key)?.accountId).toBe(id); +}); +test("back-to-back structural mutations are both persisted and raw keys are refused", async () => { + const store = owner(); const other = anthropicAffinityDigest("anthropic", "second"); + store.set(key, { accountId: id, lastUsedAt: Date.now() }); + store.set(other, { accountId: id, lastUsedAt: Date.now() }); + store.set(secret, { accountId: id, lastUsedAt: Date.now() }); + await store.flush(); store.resetMemory(); + expect(store.size).toBe(2); expect(store.get(other)?.accountId).toBe(id); + expect(readFileSync(path(), "utf8")).not.toContain(secret); +}); +test("load rejects expired and future rows and keeps the newest 2000", async () => { + const { store, data } = await snapshot(); const now = Date.now(); const base = data.entries[0]; + data.entries = Array.from({ length: 2002 }, (_, i) => ({ ...base, + sessionDigest: i.toString(16).padStart(64, "0"), lastUsedAt: now - i })); + data.entries.push({ ...base, sessionDigest: "a".repeat(64), lastUsedAt: now - AFFINITY_IDLE_TTL_MS - 1 }, + { ...base, sessionDigest: "b".repeat(64), lastUsedAt: now + 60_000 }); + writeFileSync(path(), JSON.stringify(data)); store.resetMemory(); + expect(store.size).toBe(2000); expect(store.get((2001).toString(16).padStart(64, "0"))).toBeUndefined(); + expect(store.get("a".repeat(64))).toBeUndefined(); expect(store.get("b".repeat(64))).toBeUndefined(); +}); +test.each(["{bad json", JSON.stringify({ version: 2, entries: [] }), JSON.stringify({ version: 1, entries: [{}] }), " ".repeat(MAX_AFFINITY_FILE_BYTES + 1)])("invalid or oversized snapshot is ignored (%#)", bytes => { + writeFileSync(path(), bytes); expect(owner().size).toBe(0); +}); +test("provider instance and config home are isolated", async () => { + const { store } = await snapshot(); + expect(owner("anthropic2").get(key)).toBeUndefined(); + const other = mkdtempSync(join(tmpdir(), "ocx-affinity-other-")); + try { process.env.OPENCODEX_HOME = other; expect(store.get(key)).toBeUndefined(); } + finally { store.resetMemory(); process.env.OPENCODEX_HOME = home; removeTreeWithRetry(other); } + expect(store.get(key)?.accountId).toBe(id); +}); +test("token refresh keeps the fence while replacement login invalidates", async () => { + const { store } = await snapshot(); + await saveAccountCredential("anthropic", id, { ...credential(), access: "fixture-refreshed-token" }); + store.resetMemory(); expect(store.get(key)?.accountId).toBe(id); + await saveCredential("anthropic", credential()); + store.resetMemory(); expect(store.get(key)).toBeUndefined(); +}); +test("removal and re-add cannot resurrect the old snapshot", async () => { + const { store } = await snapshot(); const old = readFileSync(path(), "utf8"); + await removeAccount("anthropic", id); await saveCredential("anthropic", credential()); + writeFileSync(path(), old); store.resetMemory(); expect(store.get(key)).toBeUndefined(); +}); +test("reauthentication invalidation is persisted before the flag clears", async () => { + const { store } = await snapshot(); + await markAccountNeedsReauth("anthropic", id, true); + store.resetMemory(); expect(store.get(key)).toBeUndefined(); await store.flush(); + await markAccountNeedsReauth("anthropic", id, false); + store.resetMemory(); expect(store.get(key)).toBeUndefined(); + expect(JSON.parse(readFileSync(path(), "utf8")).entries).toEqual([]); +}); +test("failed atomic publication retains previous bytes and reports no identifiers", async () => { + await snapshot(); const before = readFileSync(path(), "utf8"); + const warning = spyOn(console, "warn").mockImplementation(() => {}); + try { + const store = owner("anthropic", (p, content) => atomicWriteFile(p, content, undefined, { + beforeRename: () => { throw new Error("fixture publication failure"); }, + })); + store.set(anthropicAffinityDigest("anthropic", "second"), { accountId: id, lastUsedAt: Date.now() }); + await store.flush(); expect(readFileSync(path(), "utf8")).toBe(before); + expect(warning).toHaveBeenCalled(); expect(JSON.stringify(warning.mock.calls)).not.toContain(secret); + store.resetMemory(); + } finally { warning.mockRestore(); } +}); +test("timestamp touches coalesce and shutdown flushes the latest use", async () => { + let writes = 0; const store = owner("anthropic", (...args) => { writes++; atomicWriteFile(...args); }); + const now = Date.now(); store.set(key, { accountId: id, lastUsedAt: now - 100 }); await store.flush(); + store.set(key, { accountId: id, lastUsedAt: now - 50 }); store.set(key, { accountId: id, lastUsedAt: now }); + expect(writes).toBe(1); runOptionalShutdownHooks(); + expect((await settleBackgroundWork({ timeoutMs: 5_000 })).settled).toBe(true); + expect(writes).toBe(2); + expect(JSON.parse(readFileSync(path(), "utf8")).entries[0].lastUsedAt).toBe(now); +}); + +test("2000 bindings use one auth snapshot per hydration, touch and invalidation", async () => { + const { store, data } = await snapshot(); + data.entries = Array.from({ length: 2000 }, (_, i) => ({ ...data.entries[0], + sessionDigest: i.toString(16).padStart(64, "0") })); + writeFileSync(path(), JSON.stringify(data)); store.resetMemory(); + const accounts = getAccountSet("anthropic"); + const reads = spyOn(authStore, "getAccountSet").mockImplementation(() => accounts); + try { + expect(store.size).toBe(2000); const hydrationReads = reads.mock.calls.length; + reads.mockClear(); + const touched = "0".repeat(64); + store.set(touched, { accountId: id, lastUsedAt: Date.now() }); + const setReads = reads.mock.calls.length; + reads.mockClear(); expect(store.get(touched)?.accountId).toBe(id); + const getReads = reads.mock.calls.length; + reads.mockClear(); await store.flush(); + expect([hydrationReads, setReads, getReads, reads.mock.calls.length]).toEqual([1, 1, 1, 0]); + accounts!.accounts[0]!.paused = true; + reads.mockClear(); store.invalidate(accounts); + expect(reads).not.toHaveBeenCalled(); expect(store.size).toBe(0); + reads.mockClear(); store.invalidate(); expect(reads).toHaveBeenCalledTimes(1); + } finally { reads.mockRestore(); } +}); + +const linkProbe = mkdtempSync(join(tmpdir(), "ocx-affinity-link-probe-")); +let linkUnavailable = false; +try { symlinkSync(join(linkProbe, "absent"), join(linkProbe, "link"), "file"); } +catch (error) { + if (process.platform !== "win32" || (error as NodeJS.ErrnoException).code !== "EPERM") throw error; + linkUnavailable = true; + console.warn("[affinity test] skipping file symlinks: Windows EPERM (symlink privilege unavailable)"); +} finally { removeTreeWithRetry(linkProbe); } +for (const dangling of [false, true]) test.skipIf(linkUnavailable)(`final ${dangling ? "dangling" : "external-target"} symlink is neither loaded nor followed on publication`, async () => { + const { store, data } = await snapshot(); store.resetMemory(); + const external = mkdtempSync(join(tmpdir(), "ocx-affinity-external-")); + const target = join(external, "snapshot.json"); + const warning = spyOn(console, "warn").mockImplementation(() => {}); + try { + // Replace the fixture's own snapshot with a link to a target outside the config home. + unlinkSync(path()); + if (!dangling) writeFileSync(target, JSON.stringify(data)); + symlinkSync(target, path(), "file"); + expect(store.size).toBe(0); expect(warning).toHaveBeenCalled(); + expect(JSON.stringify(warning.mock.calls)).not.toContain(home); + expect(JSON.stringify(warning.mock.calls)).not.toContain(id); + store.set(key, { accountId: id, lastUsedAt: Date.now() }); await store.flush(); + expect(lstatSync(path()).isSymbolicLink()).toBe(false); + if (dangling) expect(existsSync(target)).toBe(false); + else expect(readFileSync(target, "utf8")).toBe(JSON.stringify(data)); + } finally { warning.mockRestore(); store.resetMemory(); removeTreeWithRetry(external); } +}); + +test("pending writes and shutdown cannot publish a previous home's rows", async () => { + // Every publication is recorded: the invariant is that rows bound to `home` are only ever + // written under `home`. Whether the first write settles before the switch is platform + // timing (Windows waits for ACL settlement, POSIX writes at once), so it is not asserted. + const targets: string[] = []; + const store = owner("anthropic", ((target: string, ...rest: unknown[]) => { + targets.push(target); + return (atomicWriteFileNoFollow as (...args: unknown[]) => unknown)(target, ...rest); + }) as typeof atomicWriteFile); + const other = mkdtempSync(join(tmpdir(), "ocx-affinity-switched-")); + try { + store.set(key, { accountId: id, lastUsedAt: Date.now() }); + process.env.OPENCODEX_HOME = other; + await store.flush(); + expect(existsSync(anthropicAffinityPath(other, "anthropic"))).toBe(false); + runOptionalShutdownHooks(); await settleBackgroundWork({ timeoutMs: 5_000 }); + expect(existsSync(anthropicAffinityPath(other, "anthropic"))).toBe(false); + expect(targets.every(target => target === path())).toBe(true); + process.env.OPENCODEX_HOME = home; + // Back in its own home the store never serves a row written for another one. + const restored = store.get(key); + expect(restored === undefined || restored.accountId === id).toBe(true); + expect(existsSync(anthropicAffinityPath(other, "anthropic"))).toBe(false); + } finally { store.resetMemory(); process.env.OPENCODEX_HOME = home; removeTreeWithRetry(other); } +}); + +test("a config-home directory alias preserves hydration and no-follow publication", async () => { + const { store } = await snapshot(); store.resetMemory(); + const other = mkdtempSync(join(tmpdir(), "ocx-affinity-alias-")); + const alias = join(other, "home"); + try { + symlinkSync(home, alias, process.platform === "win32" ? "junction" : "dir"); + process.env.OPENCODEX_HOME = alias; + expect(store.get(key)?.accountId).toBe(id); + const otherKey = anthropicAffinityDigest("anthropic", "aliased-session"); + store.set(otherKey, { accountId: id, lastUsedAt: Date.now() }); await store.flush(); + expect(lstatSync(alias).isSymbolicLink()).toBe(true); + process.env.OPENCODEX_HOME = home; + expect(store.get(otherKey)?.accountId).toBe(id); + } finally { store.resetMemory(); process.env.OPENCODEX_HOME = home; removeTreeWithRetry(other); } +}); diff --git a/tests/adapters/anthropic/anthropic-sidecar-account-failover.test.ts b/tests/adapters/anthropic/anthropic-sidecar-account-failover.test.ts index d80477ea2f7..f0022ccc9f6 100644 --- a/tests/adapters/anthropic/anthropic-sidecar-account-failover.test.ts +++ b/tests/adapters/anthropic/anthropic-sidecar-account-failover.test.ts @@ -24,6 +24,7 @@ let releaseSpendHome: (() => void) | undefined; let handleResponses: typeof import("../../../src/server/responses")["handleResponses"]; let observedKeys: string[] = []; let sidecarMode = false; +let restoreModuleMocks: (() => void) | undefined; function fixtureAdapter(provider: OcxProviderConfig): ProviderAdapter { return { @@ -43,7 +44,13 @@ function fixtureAdapter(provider: OcxProviderConfig): ProviderAdapter { } beforeAll(async () => { - const actualResolver = await import("../../../src/server/adapter-resolve"); + // Capture values: Bun's module namespaces are live and mock.restore() only restores spies. + const actualResolver = { ...await import("../../../src/server/adapter-resolve") }; + const actualWebSearch = { ...await import("../../../src/web-search") }; + restoreModuleMocks = () => { + mock.module("../../../src/server/adapter-resolve", () => actualResolver); + mock.module("../../../src/web-search", () => actualWebSearch); + }; const actualResolveAdapter = actualResolver.resolveAdapter; mock.module("../../../src/server/adapter-resolve", () => ({ ...actualResolver, @@ -128,6 +135,7 @@ afterAll(() => { if (previousHome === undefined) delete process.env.OPENCODEX_HOME; else process.env.OPENCODEX_HOME = previousHome; mock.restore(); + restoreModuleMocks?.(); }); test("Anthropic sidecar dispatch seam records A429 and B200 when proactive pooling is disabled", async () => { diff --git a/tests/codex-integration/active-registry-admission.test.ts b/tests/codex-integration/active-registry-admission.test.ts index e531dcc718c..f12cd002275 100644 --- a/tests/codex-integration/active-registry-admission.test.ts +++ b/tests/codex-integration/active-registry-admission.test.ts @@ -26,6 +26,8 @@ import type { OcxConfig } from "../../src/types"; import { removeTreeWithRetry } from "../helpers/remove-tree"; import { currentServerFixtureConfig, settleServerAuthFixture, startManagementServerFixture, type ManagementServerFixture } from "../helpers/server-auth-fixture"; import { SERVER_BUDGET_MS } from "../helpers/test-budget"; +import { getAccountSet, saveCredential } from "../../src/oauth/store"; +import { flushAnthropicSessionAffinity } from "../../src/oauth/anthropic-session-affinity-store"; describe("active registry admission", () => { describe("real HTTP boundaries", () => { @@ -360,11 +362,22 @@ describe("active registry admission", () => { expect(peak).toBe(1); }); - test("affinity rejects an oversized key component without colliding or changing routing", () => { - clearAnthropicAccountPoolState(); - bindAnthropicSessionAffinity("valid-session", "account-a"); - bindAnthropicSessionAffinity(`${"x".repeat(512)}-different`, "account-b"); - expect(anthropicSessionAffinitySizeForTests()).toBe(1); + test("affinity rejects an oversized key component without colliding or changing routing", async () => { + const previousHome = process.env.OPENCODEX_HOME; + const home = mkdtempSync(join(tmpdir(), "ocx-affinity-admission-")); + process.env.OPENCODEX_HOME = home; clearAnthropicAccountPoolState(); + try { + await saveCredential("anthropic", { access: "fixture-affinity-access", refresh: "fixture-affinity-refresh", + expires: Date.now() + 3_600_000, accountId: "fixture-affinity-account" }); + const accountId = getAccountSet("anthropic")!.accounts[0]!.id; + bindAnthropicSessionAffinity("valid-session", accountId); + bindAnthropicSessionAffinity(`${"x".repeat(512)}-different`, accountId); + expect(anthropicSessionAffinitySizeForTests()).toBe(1); + } finally { + await flushAnthropicSessionAffinity(); clearAnthropicAccountPoolState(); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; else process.env.OPENCODEX_HOME = previousHome; + removeTreeWithRetry(home); + } }); }); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index e87d32a01e8..07a0b2e1487 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -252,6 +252,7 @@ "always-on-429-failover.test.ts": "routing", "ambiguous-resend-composition.test.ts": "lib", "ambiguous-resend-gate.test.ts": "lib", "anthropic-client-identity.test.ts": "adapters/anthropic", "anthropic-account-metadata.test.ts": "adapters/anthropic", "anthropic-account-pool.test.ts": "adapters/anthropic", + "anthropic-affinity-restart.test.ts": "adapters/anthropic", "anthropic-affinity-store.test.ts": "adapters/anthropic", "anthropic-account-pause-outbound.test.ts": "adapters/anthropic", "anthropic-account-pause.test.ts": "adapters/anthropic", "anthropic-account-threshold.test.ts": "adapters/anthropic", "anthropic-combo-account-cooldown.test.ts": "adapters/anthropic", "cli-alias-json.test.ts": "cli",