From f06561264c31924910eba4638bd95c159bcb31d2 Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Tue, 27 Oct 2026 23:09:14 +0900 Subject: [PATCH 1/6] fix(anthropic): preserve and persist OAuth session affinity --- .../src/content/docs/fr/guides/claude-code.md | 2 +- .../src/content/docs/guides/claude-code.md | 2 +- .../src/content/docs/tr/guides/claude-code.md | 3 +- .../content/docs/zh-tw/guides/claude-code.md | 2 +- scripts/test-layout/layout-adapters.json | 4 +- src/oauth/anthropic-routing.ts | 47 +++--- src/oauth/anthropic-session-affinity-store.ts | 156 ++++++++++++++++++ structure/providers/anthropic-account-pool.md | 17 ++ .../anthropic-affinity-restart.test.ts | 85 ++++++++++ .../anthropic-affinity-store.test.ts | 116 +++++++++++++ tests/fixtures/test-layout-expected.json | 1 + 11 files changed, 405 insertions(+), 30 deletions(-) create mode 100644 src/oauth/anthropic-session-affinity-store.ts create mode 100644 tests/adapters/anthropic/anthropic-affinity-restart.test.ts create mode 100644 tests/adapters/anthropic/anthropic-affinity-store.test.ts 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 0405b2807d9..7ca1807e148 100644 --- a/docs-site/src/content/docs/fr/guides/claude-code.md +++ b/docs-site/src/content/docs/fr/guides/claude-code.md @@ -35,7 +35,7 @@ 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. +- 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 4a815b3c6d1..e8e8f8ce6e7 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 3b6b953d344..c161eecff08 100644 --- a/docs-site/src/content/docs/tr/guides/claude-code.md +++ b/docs-site/src/content/docs/tr/guides/claude-code.md @@ -39,8 +39,7 @@ 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). +- 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 dee9b6374a1..0e64c169f06 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 @@ -30,7 +30,7 @@ Anthropic 未認可自動帳號池；同一組織的帳號可能共用配額（ - 上游 **429** 會讓該帳號冷卻（有 `Retry-After` 時使用它，否則用預設 backoff）、清除其 affinity， 並可能在同一個請求內輪換到另一個合格帳號（有上限）。 -- Affinity 是**程序本機**的（proxy 重啟後就會遺失）。 +- 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 08f0ab47160..3f542d08451 100644 --- a/scripts/test-layout/layout-adapters.json +++ b/scripts/test-layout/layout-adapters.json @@ -5,5 +5,7 @@ "responses-sidecar-usage.test.ts": "responses", "google-vertex-key-project.test.ts": "adapters/google", "provider-quota-abliteration.test.ts": "providers", - "command-code-plan-denial.test.ts": "providers" + "command-code-plan-denial.test.ts": "providers", + "anthropic-affinity-restart.test.ts": "adapters/anthropic", + "anthropic-affinity-store.test.ts": "adapters/anthropic" } diff --git a/src/oauth/anthropic-routing.ts b/src/oauth/anthropic-routing.ts index c9de9371896..1cafbf33922 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, MAX_AFFINITY_ENTRIES } 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 : ""; @@ -372,7 +366,8 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting // 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; @@ -1013,7 +1008,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))) { if (manualPreference.accountId !== set.activeAccountId || manualPreference.revision !== set.selectionRevision) { manualPreference = null; } else { @@ -1026,7 +1021,7 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting } } - const key = normalizeAffinityComponent(sessionKey); + const key = affinityKey(sessionKey); if (key) { const affined = sessionAffinity.get(key); if (affined && now - affined.lastUsedAt <= AFFINITY_IDLE_TTL_MS) { @@ -1034,6 +1029,7 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting 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 }); 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,7 +1130,7 @@ 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); @@ -1309,16 +1305,14 @@ 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 temporary alternate is never evidence against an established account. Shared + // quota/entitlement refusals and operator resets explicitly remove the binding. + const preserveBinding = bound && Date.now() - bound.lastUsedAt <= AFFINITY_IDLE_TTL_MS + && !isCooled(bound.accountId, Date.now()) && isPoolCredentialUsable(bound.accountId, Date.now()); + if (preserveBinding) 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 +1391,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; + for (const [key] of sessionAffinity) sessionAffinity.get(key); // Persist invalidated pauses/reauthentication. + } // 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,7 +1407,9 @@ 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; + if (event.provider !== PROVIDER || event.kind !== "oauth") return; + for (const [key] of sessionAffinity) sessionAffinity.get(key); + if (!manualPreference) return; const current = captureOAuthAccountSelection(PROVIDER); if (current?.accountId !== manualPreference.accountId || current.revision !== manualPreference.revision) manualPreference = 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..4f72147220f --- /dev/null +++ b/src/oauth/anthropic-session-affinity-store.ts @@ -0,0 +1,156 @@ +/** Private, bounded Anthropic affinity snapshots. No session or credential bytes reach disk. */ +import { createHash } from "node:crypto"; +import { closeSync, fstatSync, openSync, readSync } from "node:fs"; +import { join } from "node:path"; +import { atomicWriteFile } 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 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 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 = atomicWriteFile) { owners.add(this); } + private fence(accountId: string): string | undefined { + const account = getAccountSet(this.instance)?.accounts.find(row => row.id === accountId); + // Legacy accounts without a persistent login identity remain memory-only until next login. + return account && !account.paused && !account.needsReauth && account.loginId + ? digest("opencodex/anthropic-affinity/login/v1", account.loginId) : undefined; + } + private valid(entry: Entry): boolean { + const account = getAccountSet(this.instance)?.accounts.find(row => row.id === entry.accountId); + return !!account && !account.paused && !account.needsReauth && this.fence(entry.accountId) === entry.loginFence; + } + private current(): 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(); }); + let descriptor: number | undefined; + try { + descriptor = openSync(anthropicAffinityPath(home, this.instance), "r"); + const stat = fstatSync(descriptor); + if (!stat.isFile() || stat.size > MAX_AFFINITY_FILE_BYTES) return scope; + const bytes = Buffer.alloc(stat.size + 1); + const count = readSync(descriptor, bytes, 0, bytes.length, 0); + if (count !== stat.size) return scope; + const snapshot = JSON.parse(bytes.subarray(0, count).toString("utf8")); + if (!snapshot || snapshot.version !== 1 || !Array.isArray(snapshot.entries)) return scope; + 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)) 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.prune(scope, now); + 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"); + } + } finally { if (descriptor !== undefined) closeSync(descriptor); } + return scope; + } + private prune(scope: Scope, now: number): void { + for (const [key, row] of scope.rows) { + if (row.lastUsedAt > now || now - row.lastUsedAt > AFFINITY_IDLE_TTL_MS || !this.valid(row)) scope.rows.delete(key); + } + 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.flush(); }, TOUCH_DELAY_MS); + scope.timer.unref(); + } else { void this.flush(); } + } + private flushSync(scope: Scope): void { + if (!scope.active || !scope.dirty || getConfigDir() !== scope.home) return; + try { + this.prune(scope, Date.now()); + 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?.active) 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.flush(); + })); + 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): Entry | undefined { + const scope = this.current(); + const row = scope.rows.get(key); + if (row && !this.valid(row)) { scope.rows.delete(key); this.changed(scope); return undefined; } + return row; + } + set(key: string, value: Entry): void { + if (!/^[a-f0-9]{64}$/.test(key) || !Number.isFinite(value.lastUsedAt)) return; + const scope = this.current(); + const previous = scope.rows.get(key); + scope.rows.set(key, { accountId: value.accountId, lastUsedAt: value.lastUsedAt, loginFence: this.fence(value.accountId) }); + this.prune(scope, value.lastUsedAt); + this.changed(scope, previous?.accountId === value.accountId && previous.loginFence === this.fence(value.accountId)); + } + 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..435afd4740e 100644 --- a/structure/providers/anthropic-account-pool.md +++ b/structure/providers/anthropic-account-pool.md @@ -470,6 +470,23 @@ 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 even if the admission pause +expires before commit, and refresh its idle timestamp. Shared-quota and entitlement refusal invalidation remains +explicit. Structural writes use the config ACL settlement barrier and private atomic publisher; 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..0d2b229bcfd --- /dev/null +++ b/tests/adapters/anthropic/anthropic-affinity-restart.test.ts @@ -0,0 +1,85 @@ +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"; + +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(); +}); +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); +}); 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..04c3d2873df --- /dev/null +++ b/tests/adapters/anthropic/anthropic-affinity-store.test.ts @@ -0,0 +1,116 @@ +import { afterEach, beforeEach, expect, test, spyOn } from "bun:test"; +import { mkdtempSync, readFileSync, 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 } from "../../../src/config/atomic-write"; +import { runOptionalShutdownHooks } from "../../../src/lib/optional-shutdown-hooks"; +import { settleBackgroundWork } from "../../../src/lib/background-work"; + +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" | "anthropic-team" = "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("anthropic-team").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); +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index d1f3dad7a49..8f9678ef3da 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -241,6 +241,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", From 3958f184a537fd48fe6828905c4044b94f945781 Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Wed, 28 Oct 2026 00:00:42 +0900 Subject: [PATCH 2/6] fix(anthropic): preserve affinity only for detours, not for exhausted or replaced accounts Restore the commit-time rule that rebinds unless the bound account was excluded by a model route or detoured around by a transient pause or family rejection, and extend the pause test with a bounded five-minute detour grace so a pause expiring during the alternate's awaits does not rebind (#7813). Measured quota exhaustion and plain recovery candidates rebind again. --- src/oauth/anthropic-rate-limit-policy.ts | 28 ++++++++++++++++++++++-- src/oauth/anthropic-routing.ts | 19 +++++++++++----- 2 files changed, 39 insertions(+), 8 deletions(-) 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 1cafbf33922..869b701c28c 100644 --- a/src/oauth/anthropic-routing.ts +++ b/src/oauth/anthropic-routing.ts @@ -360,7 +360,7 @@ 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. @@ -1307,11 +1307,18 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting } const key = affinityKey(options.sessionKey); const bound = key ? sessionAffinity.get(key) : undefined; - // A temporary alternate is never evidence against an established account. Shared - // quota/entitlement refusals and operator resets explicitly remove the binding. - const preserveBinding = bound && Date.now() - bound.lastUsedAt <= AFFINITY_IDLE_TTL_MS - && !isCooled(bound.accountId, Date.now()) && isPoolCredentialUsable(bound.accountId, Date.now()); - if (preserveBinding) sessionAffinity.set(key, { ...bound, lastUsedAt: Date.now() }); + 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); + // 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. + const preservePausedAffinity = bound && bound.accountId !== accountId + && (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 From dbaa5ed11d75e93cd8c2013ad79ab830ee7f8834 Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Wed, 28 Oct 2026 00:16:47 +0900 Subject: [PATCH 3/6] fix(anthropic): bound affinity validation and reject linked snapshots --- .../src/content/docs/fr/guides/claude-code.md | 2 +- .../src/content/docs/tr/guides/claude-code.md | 4 +- .../content/docs/zh-tw/guides/claude-code.md | 4 +- src/oauth/anthropic-routing.ts | 40 ++----- src/oauth/anthropic-session-affinity-store.ts | 101 +++++++++++------- structure/providers/anthropic-account-pool.md | 14 ++- .../anthropic-affinity-restart.test.ts | 16 +++ .../anthropic-affinity-store.test.ts | 73 ++++++++++++- .../active-registry-admission.test.ts | 23 +++- 9 files changed, 192 insertions(+), 85 deletions(-) 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 47c9412d6ab..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,7 +34,7 @@ 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. + 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` 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 2b0a360f02f..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,8 +37,8 @@ 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ı). + ç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 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 1653be3bc06..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,8 +28,8 @@ Anthropic 未認可自動帳號池；同一組織的帳號可能共用配額（ 啟用時的營運契約： -- 上游 **429** 會讓該帳號冷卻（有 `Retry-After` 時使用它，否則用預設 backoff）、清除其 affinity， - 並可能在同一個請求內輪換到另一個合格帳號（有上限）。 +- 上游 **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`。 diff --git a/src/oauth/anthropic-routing.ts b/src/oauth/anthropic-routing.ts index 869b701c28c..ad34c78c431 100644 --- a/src/oauth/anthropic-routing.ts +++ b/src/oauth/anthropic-routing.ts @@ -46,7 +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, MAX_AFFINITY_ENTRIES } from "./anthropic-session-affinity-store"; +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 @@ -875,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); } @@ -971,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 @@ -1008,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 && !sessionAffinity.get(affinityKey(sessionKey))) { + if (manualPreference && !sessionAffinity.get(affinityKey(sessionKey), set, now)) { if (manualPreference.accountId !== set.activeAccountId || manualPreference.revision !== set.selectionRevision) { manualPreference = null; } else { @@ -1023,13 +1012,13 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting 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 }); + 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 @@ -1133,13 +1122,10 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting 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. @@ -1400,7 +1386,7 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting // Pause changes eligibility, not health. Do not reset cooldowns or cancel sent turns. function onPause(): void { quorumCache = null; - for (const [key] of sessionAffinity) sessionAffinity.get(key); // Persist invalidated pauses/reauthentication. + 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; @@ -1415,11 +1401,11 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting // Threshold mutations rebase it first, before this generic notification runs. function onSelection(event: Parameters[0]>[0]): void { if (event.provider !== PROVIDER || event.kind !== "oauth") return; - for (const [key] of sessionAffinity) sessionAffinity.get(key); + const snapshot = getAccountSet(PROVIDER); + sessionAffinity.invalidate(snapshot); if (!manualPreference) return; - const current = captureOAuthAccountSelection(PROVIDER); - if (current?.accountId !== manualPreference.accountId - || current.revision !== manualPreference.revision) manualPreference = null; + if (snapshot?.activeAccountId !== manualPreference.accountId + || snapshot.selectionRevision !== manualPreference.revision) manualPreference = null; } @@ -1433,11 +1419,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 index 4f72147220f..299d06d3249 100644 --- a/src/oauth/anthropic-session-affinity-store.ts +++ b/src/oauth/anthropic-session-affinity-store.ts @@ -1,12 +1,12 @@ /** Private, bounded Anthropic affinity snapshots. No session or credential bytes reach disk. */ import { createHash } from "node:crypto"; -import { closeSync, fstatSync, openSync, readSync } from "node:fs"; import { join } from "node:path"; -import { atomicWriteFile } from "../config/atomic-write"; +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; @@ -14,6 +14,8 @@ 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(); @@ -28,18 +30,20 @@ function warn(): void { console.warn("[anthropic-affinity] snapshot persistence export class AnthropicSessionAffinityStore implements Iterable<[string, Entry]> { private scope?: Scope; - constructor(private readonly instance: AnthropicInstanceId, private readonly write = atomicWriteFile) { owners.add(this); } - private fence(accountId: string): string | undefined { - const account = getAccountSet(this.instance)?.accounts.find(row => row.id === accountId); - // Legacy accounts without a persistent login identity remain memory-only until next login. - return account && !account.paused && !account.needsReauth && account.loginId - ? digest("opencodex/anthropic-affinity/login/v1", account.loginId) : undefined; + 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): boolean { - const account = getAccountSet(this.instance)?.accounts.find(row => row.id === entry.accountId); - return !!account && !account.paused && !account.needsReauth && this.fence(entry.accountId) === entry.loginFence; + private valid(entry: Entry, fences: Fences): boolean { + return fences.has(entry.accountId) && fences.get(entry.accountId) === entry.loginFence; } - private current(): Scope { + private current(fences?: Fences): Scope { const home = getConfigDir(); if (this.scope?.home === home) return this.scope; this.resetMemory(); @@ -47,41 +51,34 @@ export class AnthropicSessionAffinityStore implements Iterable<[string, Entry]> this.scope = scope; // Starts synchronously; retained ACL children are joined by the tracked writer. registerOptionalShutdownHook(`anthropic-affinity:${this.instance}`, () => { void this.flush(); }); - let descriptor: number | undefined; try { - descriptor = openSync(anthropicAffinityPath(home, this.instance), "r"); - const stat = fstatSync(descriptor); - if (!stat.isFile() || stat.size > MAX_AFFINITY_FILE_BYTES) return scope; - const bytes = Buffer.alloc(stat.size + 1); - const count = readSync(descriptor, bytes, 0, bytes.length, 0); - if (count !== stat.size) return scope; - const snapshot = JSON.parse(bytes.subarray(0, count).toString("utf8")); + 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)) continue; + || 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.prune(scope, now); + 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"); } - } finally { if (descriptor !== undefined) closeSync(descriptor); } + } return scope; } - private prune(scope: Scope, now: number): void { - for (const [key, row] of scope.rows) { - if (row.lastUsedAt > now || now - row.lastUsedAt > AFFINITY_IDLE_TTL_MS || !this.valid(row)) scope.rows.delete(key); - } + 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); } @@ -89,14 +86,14 @@ export class AnthropicSessionAffinityStore implements Iterable<[string, Entry]> scope.dirty = true; scope.revision++; if (touch) { - scope.timer ??= setTimeout(() => { scope.timer = undefined; void this.flush(); }, TOUCH_DELAY_MS); + scope.timer ??= setTimeout(() => { scope.timer = undefined; void this.flushScope(scope); }, TOUCH_DELAY_MS); scope.timer.unref(); - } else { void this.flush(); } + } else { void this.flushScope(scope); } } private flushSync(scope: Scope): void { - if (!scope.active || !scope.dirty || getConfigDir() !== scope.home) return; + if (!scope.active || !scope.dirty || this.scope !== scope) return; + if (getConfigDir() !== scope.home) { this.resetMemory(); return; } try { - this.prune(scope, Date.now()); 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; @@ -104,7 +101,11 @@ export class AnthropicSessionAffinityStore implements Iterable<[string, Entry]> } 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; @@ -115,7 +116,7 @@ export class AnthropicSessionAffinityStore implements Iterable<[string, Entry]> }).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.flush(); + if (scope.active && scope.dirty && scope.revision !== attemptedRevision) return this.flushScope(scope); })); return scope.pending; } @@ -127,19 +128,39 @@ export class AnthropicSessionAffinityStore implements Iterable<[string, Entry]> this.scope = undefined; } get size(): number { return this.current().rows.size; } - get(key: string): Entry | undefined { - const scope = this.current(); + 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)) { scope.rows.delete(key); this.changed(scope); return undefined; } + 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): void { + set(key: string, value: Entry, snapshot: AccountSnapshot = getAccountSet(this.instance)): void { if (!/^[a-f0-9]{64}$/.test(key) || !Number.isFinite(value.lastUsedAt)) return; - const scope = this.current(); + 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); - scope.rows.set(key, { accountId: value.accountId, lastUsedAt: value.lastUsedAt, loginFence: this.fence(value.accountId) }); - this.prune(scope, value.lastUsedAt); - this.changed(scope, previous?.accountId === value.accountId && previous.loginFence === this.fence(value.accountId)); + 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(); diff --git a/structure/providers/anthropic-account-pool.md b/structure/providers/anthropic-account-pool.md index 435afd4740e..da57bf03579 100644 --- a/structure/providers/anthropic-account-pool.md +++ b/structure/providers/anthropic-account-pool.md @@ -480,9 +480,17 @@ removal/re-add, pause and reauthentication invalidate it. Invalid, oversized, fu 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 even if the admission pause -expires before commit, and refresh its idle timestamp. Shared-quota and entitlement refusal invalidation remains -explicit. Structural writes use the config ACL settlement barrier and private atomic publisher; timestamp-only +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. diff --git a/tests/adapters/anthropic/anthropic-affinity-restart.test.ts b/tests/adapters/anthropic/anthropic-affinity-restart.test.ts index 0d2b229bcfd..8df67de6c41 100644 --- a/tests/adapters/anthropic/anthropic-affinity-restart.test.ts +++ b/tests/adapters/anthropic/anthropic-affinity-restart.test.ts @@ -7,6 +7,7 @@ import { captureOAuthAccountSelection, getAccountSet, saveCredential, setActiveA 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; @@ -24,6 +25,7 @@ beforeEach(() => { home = mkdtempSync(join(tmpdir(), "ocx-affinity-restart-")); process.env.OPENCODEX_HOME = home; clearAnthropicAccountPoolState(); + clearAccountQuotaCache(); }); afterEach(async () => { clock?.mockRestore(); clock = undefined; @@ -83,3 +85,17 @@ test("shared-quota rejection rebinds even after a short Retry-After expires", as await promoteAnthropicActiveAccount(b, captureOAuthAccountSelection("anthropic"), { config, sessionKey: session }); expect(resolveAnthropicAccountForSession(session, config).accountId).toBe(b); }); +test.each(["shared five-hour", "model-specific"] as const)("measured %s exhaustion commits a replacement binding", async kind => { + const { a, b } = await seed(); + const model = "claude-opus-5-5"; + 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 index 04c3d2873df..6e967f3daeb 100644 --- a/tests/adapters/anthropic/anthropic-affinity-store.test.ts +++ b/tests/adapters/anthropic/anthropic-affinity-store.test.ts @@ -1,5 +1,5 @@ import { afterEach, beforeEach, expect, test, spyOn } from "bun:test"; -import { mkdtempSync, readFileSync, writeFileSync } from "node:fs"; +import { existsSync, lstatSync, mkdtempSync, readFileSync, symlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { removeTreeWithRetry } from "../../helpers/remove-tree"; @@ -9,6 +9,7 @@ import { AnthropicSessionAffinityStore, anthropicAffinityDigest, anthropicAffini import { atomicWriteFile } 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; @@ -18,7 +19,7 @@ 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" | "anthropic-team" = "anthropic", write?: typeof atomicWriteFile) { +function owner(instance: "anthropic" | "anthropic2" = "anthropic", write?: typeof atomicWriteFile) { const store = new AnthropicSessionAffinityStore(instance, write); stores.push(store); return store; } beforeEach(async () => { @@ -66,7 +67,7 @@ test.each(["{bad json", JSON.stringify({ version: 2, entries: [] }), JSON.string }); test("provider instance and config home are isolated", async () => { const { store } = await snapshot(); - expect(owner("anthropic-team").get(key)).toBeUndefined(); + 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); } @@ -114,3 +115,69 @@ test("timestamp touches coalesce and shutdown flushes the latest use", async () 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); expect(reads).toHaveBeenCalledTimes(1); + reads.mockClear(); + const touched = "0".repeat(64); + store.set(touched, { accountId: id, lastUsedAt: Date.now() }); + expect(reads).toHaveBeenCalledTimes(1); + reads.mockClear(); expect(store.get(touched)?.accountId).toBe(id); + expect(reads).toHaveBeenCalledTimes(1); + reads.mockClear(); await store.flush(); expect(reads).not.toHaveBeenCalled(); + 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. + const { unlinkSync } = await import("node:fs"); 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 () => { + const store = owner(); + 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); + process.env.OPENCODEX_HOME = home; + expect(store.get(key)).toBeUndefined(); + } finally { store.resetMemory(); process.env.OPENCODEX_HOME = home; removeTreeWithRetry(other); } +}); 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); + } }); }); From 8f18f604ca6be5f9287918a524e17492fc19e7d4 Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Wed, 28 Oct 2026 00:30:41 +0900 Subject: [PATCH 4/6] test(anthropic): restore sidecar mocks and pin affinity boundary regressions --- structure/providers/anthropic-account-pool.md | 3 +- .../anthropic-affinity-store.test.ts | 29 +++++++++++++++---- ...anthropic-sidecar-account-failover.test.ts | 10 ++++++- 3 files changed, 33 insertions(+), 9 deletions(-) diff --git a/structure/providers/anthropic-account-pool.md b/structure/providers/anthropic-account-pool.md index da57bf03579..76bcf5cca20 100644 --- a/structure/providers/anthropic-account-pool.md +++ b/structure/providers/anthropic-account-pool.md @@ -490,8 +490,7 @@ touched row; ordinary publication does not reread auth or revalidate every bindi 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 +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. diff --git a/tests/adapters/anthropic/anthropic-affinity-store.test.ts b/tests/adapters/anthropic/anthropic-affinity-store.test.ts index 6e967f3daeb..0ba6229fb4b 100644 --- a/tests/adapters/anthropic/anthropic-affinity-store.test.ts +++ b/tests/adapters/anthropic/anthropic-affinity-store.test.ts @@ -1,5 +1,5 @@ import { afterEach, beforeEach, expect, test, spyOn } from "bun:test"; -import { existsSync, lstatSync, mkdtempSync, readFileSync, symlinkSync, writeFileSync } from "node:fs"; +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"; @@ -124,14 +124,15 @@ test("2000 bindings use one auth snapshot per hydration, touch and invalidation" const accounts = getAccountSet("anthropic"); const reads = spyOn(authStore, "getAccountSet").mockImplementation(() => accounts); try { - expect(store.size).toBe(2000); expect(reads).toHaveBeenCalledTimes(1); + 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() }); - expect(reads).toHaveBeenCalledTimes(1); + const setReads = reads.mock.calls.length; reads.mockClear(); expect(store.get(touched)?.accountId).toBe(id); - expect(reads).toHaveBeenCalledTimes(1); - reads.mockClear(); await store.flush(); expect(reads).not.toHaveBeenCalled(); + 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); @@ -154,7 +155,7 @@ for (const dangling of [false, true]) test.skipIf(linkUnavailable)(`final ${dang const warning = spyOn(console, "warn").mockImplementation(() => {}); try { // Replace the fixture's own snapshot with a link to a target outside the config home. - const { unlinkSync } = await import("node:fs"); unlinkSync(path()); + unlinkSync(path()); if (!dangling) writeFileSync(target, JSON.stringify(data)); symlinkSync(target, path(), "file"); expect(store.size).toBe(0); expect(warning).toHaveBeenCalled(); @@ -181,3 +182,19 @@ test("pending writes and shutdown cannot publish a previous home's rows", async expect(store.get(key)).toBeUndefined(); } 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 () => { From b7115c894750f4c38c631d4015bd7f85775d45b0 Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Wed, 28 Oct 2026 00:53:21 +0900 Subject: [PATCH 5/6] fix(anthropic): let measured exhaustion outrank a recent throttle at affinity commit A short throttle that already expired kept the detour grace marker, so committing a replacement for a measured-exhausted bound account still preserved the old binding. Measured exhaustion of the bound account now always rebinds; regressions cover shared and model-specific exhaustion after an expired throttle. --- src/oauth/anthropic-routing.ts | 4 +++- .../anthropic/anthropic-affinity-restart.test.ts | 11 ++++++++++- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/src/oauth/anthropic-routing.ts b/src/oauth/anthropic-routing.ts index ad34c78c431..4fe1120d905 100644 --- a/src/oauth/anthropic-routing.ts +++ b/src/oauth/anthropic-routing.ts @@ -1300,8 +1300,10 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting && !routeCandidates(eligibleAtCommit, options.routeDecision).includes(bound.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. + // 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() }); diff --git a/tests/adapters/anthropic/anthropic-affinity-restart.test.ts b/tests/adapters/anthropic/anthropic-affinity-restart.test.ts index 8df67de6c41..89744c72920 100644 --- a/tests/adapters/anthropic/anthropic-affinity-restart.test.ts +++ b/tests/adapters/anthropic/anthropic-affinity-restart.test.ts @@ -85,9 +85,18 @@ test("shared-quota rejection rebinds even after a short Retry-After expires", as await promoteAnthropicActiveAccount(b, captureOAuthAccountSelection("anthropic"), { config, sessionKey: session }); expect(resolveAnthropicAccountForSession(session, config).accountId).toBe(b); }); -test.each(["shared five-hour", "model-specific"] as const)("measured %s exhaustion commits a replacement binding", async kind => { +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 }); From fa692c956917ac6f93770f3ef0750267d760ab98 Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Wed, 28 Oct 2026 03:39:10 +0900 Subject: [PATCH 6/6] test(anthropic): assert the home-isolation invariant, not write timing On POSIX the affinity snapshot is written as soon as it changes, so the original home could hold the row before the home switch and the old final assertion failed in CI. Record every publication target instead and assert that rows bound to a home are only ever written under that home. --- .../anthropic/anthropic-affinity-store.test.ts | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/tests/adapters/anthropic/anthropic-affinity-store.test.ts b/tests/adapters/anthropic/anthropic-affinity-store.test.ts index 0ba6229fb4b..ead6f20538d 100644 --- a/tests/adapters/anthropic/anthropic-affinity-store.test.ts +++ b/tests/adapters/anthropic/anthropic-affinity-store.test.ts @@ -6,7 +6,7 @@ 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 } from "../../../src/config/atomic-write"; +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"; @@ -169,7 +169,14 @@ for (const dangling of [false, true]) test.skipIf(linkUnavailable)(`final ${dang }); test("pending writes and shutdown cannot publish a previous home's rows", async () => { - const store = owner(); + // 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() }); @@ -178,8 +185,12 @@ test("pending writes and shutdown cannot publish a previous home's rows", async 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; - expect(store.get(key)).toBeUndefined(); + // 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); } });