diff --git a/README.md b/README.md index 87f8b1de..e462208e 100644 --- a/README.md +++ b/README.md @@ -332,6 +332,14 @@ In OpenCode, this includes the main Anthropic account and sidecar fallback accou Reset times are rendered as relative durations, such as `resets in 10m` or `resets in 1h 15m`. +### Quota header feed + +The optional quota header feed writes one lease file per process under `${TMPDIR:-/tmp}/opencode-anthropic-auth/quota-header-feed/`. Set `OPENCODE_ANTHROPIC_AUTH_QUOTA_FEED_DIR` to override the directory. A file contains only accounts whose response headers THAT process harvested. Consumers MUST union entries from every file inside `lease_horizon_ms`, then deduplicate by account. "Newest file wins" drops accounts seen by other processes. + +For each account, the newest entry wins for quota values. Resolve `anthropic_account_uuid` from any entry in that account's group that carries it. During a rollout, an older pre-restart publisher can write the newest entry without that field beside newer code that has it. + +Each entry always includes `anthropic_account_uuid`; an absent key identifies an older producer. Its value is provider-derived or `null`, never a local substitute. A consumer that sees `null` must count a gap, not fall back to `account_ref`; that field is store-local and never a join key. + ## Safety fallback (OpenCode) Eligible Fable 5/5.1 and Opus 5 OAuth requests try Anthropic's server-side safety fallback first. The plugin sends `fallbacks: "default"` with Anthropic's server-side fallback beta, preserves fallback conversation boundaries in OpenCode history, and reports model handoffs and restoration in the TUI sidebar or OpenCode Desktop. Follow-up requests may remain on Anthropic's selected fallback model for approximately one hour. diff --git a/packages/core/src/accounts.ts b/packages/core/src/accounts.ts index e1f55fb0..582eef1d 100644 --- a/packages/core/src/accounts.ts +++ b/packages/core/src/accounts.ts @@ -49,6 +49,7 @@ export type AccountBase = { export type OAuthAccount = AccountBase & { type: 'oauth' authLineageId?: string + anthropicAccountUuid?: string claustrumHandle?: string access?: string refresh: string @@ -556,6 +557,11 @@ function normalizeAccount(value: unknown): FallbackAccount | null { typeof value.claustrumHandle === 'string' && value.claustrumHandle.trim() ? value.claustrumHandle.trim() : undefined, + anthropicAccountUuid: + typeof value.anthropicAccountUuid === 'string' && + value.anthropicAccountUuid.trim() + ? value.anthropicAccountUuid.trim() + : undefined, access: typeof value.access === 'string' ? value.access : undefined, refresh: value.refresh, expires: typeof value.expires === 'number' ? value.expires : undefined, @@ -1197,6 +1203,7 @@ function accountRuntimeState(account: FallbackAccount) { } return objectWithDefinedEntries({ authLineageId: account.authLineageId, + anthropicAccountUuid: account.anthropicAccountUuid, claustrumHandle: account.claustrumHandle, access: account.access, refresh: account.refresh, @@ -3714,10 +3721,16 @@ export function upsertAccount( (account.label && candidate.label === account.label), ) if (index >= 0) { - storage.accounts[index] = { - ...storage.accounts[index], + const existing = storage.accounts[index] + if (!existing) return + const lineageChanged = + existing.type === 'oauth' && + account.type === 'oauth' && + existing.authLineageId !== account.authLineageId + const updated: FallbackAccount = { + ...existing, ...account, - addedAt: storage.accounts[index]?.addedAt ?? account.addedAt, + addedAt: existing.addedAt ?? account.addedAt, ...(account.type === 'oauth' && { quota: account.quota, profile: account.profile, @@ -3726,11 +3739,69 @@ export function upsertAccount( lastQuotaRefreshError: account.lastQuotaRefreshError, }), } + if (lineageChanged && updated.type === 'oauth') { + delete updated.anthropicAccountUuid + } + storage.accounts[index] = updated return } storage.accounts.push(account) } +export function persistFallbackQuotaHeaderPersistent( + input: { + accountId: string + authLineageId?: string + quota: OAuthQuotaSnapshot + anthropicAccountUuid?: string + }, + path = getAccountStoragePath(), +): Promise { + return enqueueSave(async () => { + const configLock = await acquireAccountConfigWriteLock(path) + try { + const stateLock = await acquireAccountStateWriteLock(path) + try { + const storage = await loadAccounts(path) + const account = storage?.accounts.find( + (candidate): candidate is OAuthAccount => + candidate.id === input.accountId && isOAuthAccount(candidate), + ) + if ( + !storage || + !account || + account.authLineageId !== input.authLineageId + ) { + return false + } + account.quota = { + ...input.quota, + accountIdentity: account.id, + } + if (input.anthropicAccountUuid !== undefined) { + account.anthropicAccountUuid = input.anthropicAccountUuid + } + await saveAccountStateUnlocked(storage, path, { + accounts: [input.accountId], + }) + return true + } finally { + await stateLock.release() + } + } finally { + await configLock.release() + } + }) +} + +export function fallbackAccountUuidForLineage( + account: OAuthAccount | undefined, + authLineageId?: string, +): string | null { + if (!account || account.authLineageId !== authLineageId) return null + return account.anthropicAccountUuid ?? null +} + export function removeAccount(storage: AccountStorage, id: string): boolean { const index = storage.accounts.findIndex((c) => c.id === id) if (index < 0) return false diff --git a/packages/core/src/claude-code.ts b/packages/core/src/claude-code.ts index ac093ebf..45d97ad4 100644 --- a/packages/core/src/claude-code.ts +++ b/packages/core/src/claude-code.ts @@ -188,7 +188,9 @@ export async function resolveClaudeCodeIdentity( setBounded(identityCache, accessToken, identity) } - if (!accessToken.startsWith('sk-ant-oat')) return identity + if (!accessToken.startsWith('sk-ant-oat')) { + return clearCachedAccountUuid(cacheKey, identity) + } const now = Date.now() // A slot-stable identity survives account replacement; bootstrap is the diff --git a/packages/core/src/quota-header-feed.ts b/packages/core/src/quota-header-feed.ts index 065f4085..594c41b8 100644 --- a/packages/core/src/quota-header-feed.ts +++ b/packages/core/src/quota-header-feed.ts @@ -1,11 +1,13 @@ import { randomUUID } from 'node:crypto' import { chmod, + lstat, mkdir, readdir, readFile, rename, rm, + stat, writeFile, } from 'node:fs/promises' import { tmpdir } from 'node:os' @@ -28,6 +30,13 @@ import { export const QUOTA_HEADER_FEED_SCHEMA_VERSION = 3 export const QUOTA_HEADER_FEED_LEASE_MS = 180_000 +/** + * Lease files are per process; each carries only the accounts whose response + * headers THAT process harvested. Consumers MUST union entries across all files + * inside `lease_horizon_ms`, deduplicating by account; "newest file wins" is wrong. + * `anthropic_account_uuid` is always present: null is unresolvable, while absence + * identifies an old producer. A fallback `account_ref` is store-local. + */ export type QuotaHeaderFeedIdentity = | { identity_source: 'credential_id'; credential_id: string } | { identity_source: 'account_ref'; account_ref: string } @@ -59,11 +68,14 @@ type QuotaHeaderFeedMetadata = { export type QuotaHeaderFeedEntry = QuotaHeaderFeedIdentity & QuotaHeaderFeedMetadata & { + /** Always present: null means UUID resolution failed; absence denotes an old producer. */ + anthropic_account_uuid: string | null quota: QuotaHeaderFeedQuota } export type QuotaHeaderFeedPublishEntry = QuotaHeaderFeedIdentity & QuotaHeaderFeedMetadata & { + anthropic_account_uuid: string | null quota: Omit & { fieldSources?: QuotaFieldSources } @@ -72,6 +84,7 @@ export type QuotaHeaderFeedPublishEntry = QuotaHeaderFeedIdentity & type FeedRecord = { version: typeof QUOTA_HEADER_FEED_SCHEMA_VERSION + lease_horizon_ms: number entries: Record } @@ -105,6 +118,12 @@ function validIdentity( ) return false if (!entry.quota || typeof entry.quota !== 'object') return false + if ( + !Object.hasOwn(entry, 'anthropic_account_uuid') || + (entry.anthropic_account_uuid !== null && + typeof entry.anthropic_account_uuid !== 'string') + ) + return false if (entry.identity_source === 'none') { return !('credential_id' in entry) && !('account_ref' in entry) } @@ -292,6 +311,8 @@ export class QuotaHeaderFeedRegistry { now?: () => number leaseMs?: number instanceId?: string + removeFile?: (path: string) => Promise + beforeRemoveFile?: (path: string) => Promise } = {}, ) { const instanceId = options.instanceId ?? `${process.pid}-${randomUUID()}` @@ -302,16 +323,33 @@ export class QuotaHeaderFeedRegistry { } publish(entry: QuotaHeaderFeedPublishEntry): Promise { - const { accountKey, quota, ...entryWithoutQuota } = entry - const cleanEntry = { - ...entryWithoutQuota, - quota: projectQuota(quota), - } as QuotaHeaderFeedEntry + const { accountKey, quota } = entry try { - validatePublishEntry(cleanEntry) + validatePublishEntry(entry) } catch (error) { return Promise.reject(error) } + const identity = + entry.identity_source === 'credential_id' + ? { + identity_source: 'credential_id' as const, + credential_id: entry.credential_id, + } + : entry.identity_source === 'account_ref' + ? { + identity_source: 'account_ref' as const, + account_ref: entry.account_ref, + } + : { identity_source: 'none' as const } + const cleanEntry: QuotaHeaderFeedEntry = { + ...identity, + schema_version: entry.schema_version, + provider: entry.provider, + configured_account_count: entry.configured_account_count, + observed_at_ms: entry.observed_at_ms, + anthropic_account_uuid: entry.anthropic_account_uuid, + quota: projectQuota(quota), + } if (!accountKey) return Promise.reject(new Error('Invalid quota header feed account key')) this.writeChain = this.writeChain @@ -321,6 +359,7 @@ export class QuotaHeaderFeedRegistry { this.options.directory ?? getDefaultQuotaHeaderFeedDirectory() await mkdir(directory, { recursive: true, mode: 0o700 }) await chmod(directory, 0o700) + await this.reapStaleSiblingLeases(directory) let entries: Record = {} try { const record = JSON.parse( @@ -338,7 +377,7 @@ export class QuotaHeaderFeedRegistry { try { await writeFile( tempPath, - `${JSON.stringify({ version: QUOTA_HEADER_FEED_SCHEMA_VERSION, entries })}\n`, + `${JSON.stringify({ version: QUOTA_HEADER_FEED_SCHEMA_VERSION, lease_horizon_ms: this.options.leaseMs ?? QUOTA_HEADER_FEED_LEASE_MS, entries })}\n`, { mode: 0o600 }, ) await chmod(tempPath, 0o600) @@ -350,6 +389,36 @@ export class QuotaHeaderFeedRegistry { return this.writeChain } + private async reapStaleSiblingLeases(directory: string): Promise { + const now = this.options.now?.() ?? Date.now() + const leaseMs = this.options.leaseMs ?? QUOTA_HEADER_FEED_LEASE_MS + let names: string[] + try { + names = await readdir(directory) + } catch { + return + } + await Promise.all( + names + .filter((name) => /^\d+-[0-9a-f-]+\.json$/i.test(name)) + .map(async (name) => { + const path = join(directory, name) + if (path === this.filePath) return + try { + const file = await stat(path) + if (file.mtimeMs > now || now - file.mtimeMs < leaseMs) return + await this.options.beforeRemoveFile?.(path) + const current = await lstat(path) + if (current.ino !== file.ino || current.mtimeMs !== file.mtimeMs) + return + await (this.options.removeFile ?? ((target) => rm(target)))(path) + } catch { + // A missed cleanup must not prevent this process from refreshing its lease. + } + }), + ) + } + async list(): Promise { await this.writeChain.catch(() => {}) const directory = diff --git a/packages/opencode/src/index.ts b/packages/opencode/src/index.ts index 5822c771..305bf896 100644 --- a/packages/opencode/src/index.ts +++ b/packages/opencode/src/index.ts @@ -58,6 +58,7 @@ import { executeRoutingCommand, FALLBACK_BACKGROUND_TICK_MS, FallbackAccountManager, + fallbackAccountUuidForLineage, fetchOAuthAccountProfile, formatOAuthAccountTier, formatQuotaBackoffMessage, @@ -135,6 +136,7 @@ import { parseLoggingCommandAction, parsePrimeCommandAction, parseRoutingCommandAction, + persistFallbackQuotaHeaderPersistent, primeStorageFingerprint, QUOTA_HEADER_FEED_SCHEMA_VERSION, type QuotaAccountSummary, @@ -687,14 +689,35 @@ type StickyOAuthRoute = { } type MainQuotaIdentityBinding = { - accountIdentity: string | undefined + quotaKey: string | undefined generation: number } type MainQuotaIdentityResolution = MainQuotaIdentityBinding & { + providerAccountUuid: ProviderAccountUuid | undefined stale: boolean } +declare const providerAccountUuidBrand: unique symbol + +type ProviderAccountUuid = string & { + readonly [providerAccountUuidBrand]: 'ProviderAccountUuid' +} + +type ServedQuotaHeaders = { + accountId: 'main' | string + accessToken: string + authLineageId?: string + anthropicAccountUuid?: ProviderAccountUuid | null + mainQuotaIdentity?: MainQuotaIdentityBinding +} + +function asProviderAccountUuid( + value: string | null | undefined, +): ProviderAccountUuid | undefined { + return value as ProviderAccountUuid | undefined +} + const FABLE_SWITCHED_TO_OPUS_NOTICE = 'Fable content filter detected. Switched to Opus 4.8 for a 10-response recovery window while keeping the Fable cache warm.' const FABLE_RESTORED_NOTICE = @@ -1133,29 +1156,29 @@ const anthropicAuthPlugin = async ( async function reconcileMainQuotaAccountIdentity( accessToken: string, - accountIdentity: string | undefined, + quotaKey: string | undefined, ): Promise { if ( mainQuotaIdentityAccessToken === accessToken && - mainQuotaAccountId === accountIdentity + mainQuotaAccountId === quotaKey ) { return } mainQuotaIdentityAccessToken = accessToken - mainQuotaAccountId = accountIdentity + mainQuotaAccountId = quotaKey const clearedGeneration = quotaManager.setMainQuotaAccountIdentity( - accountIdentity, + quotaKey, accessToken.startsWith('sk-ant-oat'), ) - if (accountIdentity === undefined) return + if (quotaKey === undefined) return const storage = await loadAccounts(accountStoragePath) const persistedError = storage?.quota?.mainLastQuotaApiError if ( !storage || !persistedError || - persistedError.accountIdentity === accountIdentity || + persistedError.accountIdentity === quotaKey || (!accessToken.startsWith('sk-ant-oat') && persistedError.accountIdentity === undefined) ) { @@ -1189,25 +1212,27 @@ const anthropicAuthPlugin = async ( model, mainAccountId, ) - // Adapters that do not use Claude's OAuth token format cannot be bootstrapped; - // keep the slot identity only for that compatibility path. Claude OAuth - // credentials use the oat prefix and stay unknown when bootstrap fails. - const quotaIdentity = + // Non-oat adapters retain their local slot for quota fencing, but it is + // not a provider identity and must never reach a provider-facing field. + const quotaKey = identity.accountUuid ?? (!accessToken.startsWith('sk-ant-oat') ? mainAccountId : undefined) + const providerAccountUuid = asProviderAccountUuid(identity.accountUuid) if (resolutionGeneration !== mainQuotaIdentityResolutionGeneration) { const identityChanged = mainQuotaIdentityAccessToken !== accessToken || - mainQuotaAccountId !== quotaIdentity + mainQuotaAccountId !== quotaKey return { - accountIdentity: quotaIdentity, + quotaKey, + providerAccountUuid, generation: quotaManager.getMainQuotaIdentityGeneration(), stale: identityChanged, } } - await reconcileMainQuotaAccountIdentity(accessToken, quotaIdentity) + await reconcileMainQuotaAccountIdentity(accessToken, quotaKey) return { - accountIdentity: quotaIdentity, + quotaKey, + providerAccountUuid, generation: quotaManager.getMainQuotaIdentityGeneration(), stale: false, } @@ -1410,21 +1435,17 @@ const anthropicAuthPlugin = async ( } const warnedQuotaNormalizeErrors = new Set() + const warnedStaleFallbackQuotaPersists = new Set() async function persistPushedQuota( - served: { - accountId: 'main' | string - accessToken: string - authLineageId?: string - mainQuotaIdentity?: MainQuotaIdentityBinding - }, + served: ServedQuotaHeaders, entry: QuotaEntry, ): Promise { const storage = (await loadAccounts(accountStoragePath)) ?? createEmptyStorage() if (served.accountId === 'main') { - const accountIdentity = served.mainQuotaIdentity?.accountIdentity - if (!accountIdentity || entry.quota.accountIdentity !== accountIdentity) { + const quotaKey = served.mainQuotaIdentity?.quotaKey + if (!quotaKey || entry.quota.accountIdentity !== quotaKey) { logger.trace( 'quota', 'skipped quota persistence without verified main account identity', @@ -1434,14 +1455,14 @@ const anthropicAuthPlugin = async ( storage.quota = storage.quota ?? {} storage.quota.mainQuota = { ...entry.quota, - accountIdentity, + accountIdentity: quotaKey, } storage.quota.mainQuotaCheckedAt = entry.checkedAt await saveAccountState(storage, accountStoragePath, { mainQuota: true }) const reloaded = await loadAccounts(accountStoragePath) const persistedQuota = reloaded?.quota?.mainQuota const persistedQuotaBelongsToRequest = Boolean( - persistedQuota && persistedQuota.accountIdentity === accountIdentity, + persistedQuota && persistedQuota.accountIdentity === quotaKey, ) const mergedQuota = persistedQuotaBelongsToRequest ? mergeHeaderQuotaForPersistence(persistedQuota, entry.quota) @@ -1454,23 +1475,28 @@ const anthropicAuthPlugin = async ( } : entry } - const account = storage.accounts.find( - (candidate): candidate is OAuthAccount => - candidate.id === served.accountId && isOAuthAccount(candidate), + const persisted = await persistFallbackQuotaHeaderPersistent( + { + accountId: served.accountId, + authLineageId: served.authLineageId, + quota: entry.quota, + anthropicAccountUuid: served.anthropicAccountUuid ?? undefined, + }, + accountStoragePath, ) - if (!account) return null - // A lineage mismatch is a confirmed replacement; two absent markers remain - // legacy-compatible, while one-sided absence is too ambiguous to persist. - if (account.authLineageId !== served.authLineageId) { - logger.trace( - 'quota', - 'skipped stale fallback quota persistence after lineage change', - { - accountId: served.accountId, - storedLineage: account.authLineageId, - servedLineage: served.authLineageId, - }, - ) + if (!persisted) { + const key = `${served.accountId}:${served.authLineageId ?? ''}` + if (!warnedStaleFallbackQuotaPersists.has(key)) { + warnedStaleFallbackQuotaPersists.add(key) + logger.debug( + 'quota', + 'skipped stale fallback quota persistence after lineage change', + { + accountId: served.accountId, + servedLineage: served.authLineageId, + }, + ) + } // The persistence fence is authoritative; discard the optimistic cache // write unless a newer replacement observation already won the race. if (quotaManager.getAllFallbacks().get(served.accountId) === entry) { @@ -1478,13 +1504,6 @@ const anthropicAuthPlugin = async ( } return null } - account.quota = { - ...entry.quota, - accountIdentity: account.id, - } - await saveAccountState(storage, accountStoragePath, { - accounts: [served.accountId], - }) return entry } @@ -1520,18 +1539,11 @@ const anthropicAuthPlugin = async ( } async function publishQuotaHeaderFeed( - served: { - accountId: 'main' | string - accessToken: string - mainQuotaIdentity?: MainQuotaIdentityBinding - }, + served: ServedQuotaHeaders, entry: QuotaEntry, ): Promise { if (!quotaHeaderFeedRegistry) return - if ( - served.accountId === 'main' && - !served.mainQuotaIdentity?.accountIdentity - ) { + if (served.accountId === 'main' && !served.mainQuotaIdentity?.quotaKey) { // Observed-at staleness exposes silence; an unknown key must not replace verified data. return } @@ -1563,7 +1575,7 @@ const anthropicAuthPlugin = async ( const accountKey = credentialId ?? (served.accountId === 'main' - ? (served.mainQuotaIdentity?.accountIdentity ?? 'main') + ? (served.mainQuotaIdentity?.quotaKey ?? 'main') : served.accountId) const feedEntry: QuotaHeaderFeedPublishEntry = credentialId ? { @@ -1573,17 +1585,19 @@ const anthropicAuthPlugin = async ( provider: 'anthropic', configured_account_count: configuredAccountCount, observed_at_ms: observedAtMs, + anthropic_account_uuid: served.anthropicAccountUuid ?? null, quota, accountKey, } - : served.accountId === 'main' && served.mainQuotaIdentity?.accountIdentity + : served.accountId === 'main' && served.mainQuotaIdentity?.quotaKey ? { identity_source: 'account_ref', - account_ref: served.mainQuotaIdentity.accountIdentity, + account_ref: served.mainQuotaIdentity.quotaKey, schema_version: QUOTA_HEADER_FEED_SCHEMA_VERSION, provider: 'anthropic', configured_account_count: configuredAccountCount, observed_at_ms: observedAtMs, + anthropic_account_uuid: served.anthropicAccountUuid ?? null, quota, accountKey, } @@ -1594,6 +1608,7 @@ const anthropicAuthPlugin = async ( provider: 'anthropic', configured_account_count: configuredAccountCount, observed_at_ms: observedAtMs, + anthropic_account_uuid: served.anthropicAccountUuid ?? null, quota, accountKey, } @@ -1604,6 +1619,7 @@ const anthropicAuthPlugin = async ( provider: 'anthropic', configured_account_count: configuredAccountCount, observed_at_ms: observedAtMs, + anthropic_account_uuid: served.anthropicAccountUuid ?? null, quota, accountKey, } @@ -1618,12 +1634,7 @@ const anthropicAuthPlugin = async ( function harvestQuotaHeaders( headers: Headers, - served: { - accountId: 'main' | string - accessToken: string - authLineageId?: string - mainQuotaIdentity?: MainQuotaIdentityBinding - }, + served: ServedQuotaHeaders, ): void { try { if (!isQuotaBearingHeaderFrame(headers)) { @@ -1642,7 +1653,7 @@ const anthropicAuthPlugin = async ( mainQuotaIdentity.generation ) { logger.trace('quota', 'discarded stale main quota headers', { - boundIdentity: mainQuotaIdentity?.accountIdentity, + boundIdentity: mainQuotaIdentity?.quotaKey, currentGeneration: quotaManager.getMainQuotaIdentityGeneration(), }) return @@ -1651,7 +1662,7 @@ const anthropicAuthPlugin = async ( const entry = served.accountId === 'main' ? quotaManager.pushMainFromHeaders( - mainQuotaIdentity?.accountIdentity, + mainQuotaIdentity?.quotaKey, incoming, ) : quotaManager.pushFallbackFromHeaders(served.accountId, incoming, { @@ -3077,14 +3088,13 @@ const anthropicAuthPlugin = async ( ) { if (!accessToken) return if (storage?.quota?.enabled !== true) return - const accountIdentity = - mainQuotaIdentity?.accountIdentity ?? mainQuotaAccountId - if (quotaManager.getMain(accountIdentity)) return + const quotaKey = mainQuotaIdentity?.quotaKey ?? mainQuotaAccountId + if (quotaManager.getMain(quotaKey)) return if (sidebarMainQuotaRefreshInFlight) return sidebarMainQuotaRefreshInFlight = true void quotaManager - .refreshMain(accountIdentity, accessToken, mainQuotaIdentity?.generation) + .refreshMain(quotaKey, accessToken, mainQuotaIdentity?.generation) .then(() => refreshSidebarQuota()) .catch(() => {}) .finally(() => { @@ -5399,7 +5409,7 @@ const anthropicAuthPlugin = async ( fallbackAuthLineageId?: string, fableRequest?: FableRequestContext, laneStartRequest = false, - mainQuotaIdentity?: MainQuotaIdentityBinding, + mainQuotaIdentity?: MainQuotaIdentityResolution, claustrumResolution?: ClaustrumAccessResolution, ) { const start = nowMs() @@ -5681,11 +5691,28 @@ const anthropicAuthPlugin = async ( } } - const relayConfig = getRelayConfig(await getRequestStorage()) + const requestStorageForIdentity = await getRequestStorage() + const relayConfig = getRelayConfig(requestStorageForIdentity) + const persistedFallbackAccount = + oauthAccountId === 'main' + ? undefined + : requestStorageForIdentity?.accounts.find( + (account): account is OAuthAccount => + account.id === oauthAccountId && isOAuthAccount(account), + ) + const persistedFallbackAccountUuid = fallbackAccountUuidForLineage( + persistedFallbackAccount, + fallbackAuthLineageId, + ) const served = { accountId: oauthAccountId, accessToken, authLineageId: fallbackAuthLineageId, + anthropicAccountUuid: + asProviderAccountUuid(identity.accountUuid) ?? + (oauthAccountId === 'main' + ? mainQuotaIdentity?.providerAccountUuid + : asProviderAccountUuid(persistedFallbackAccountUuid)), ...(oauthAccountId === 'main' && mainQuotaIdentity ? { mainQuotaIdentity } : {}), @@ -5778,7 +5805,7 @@ const anthropicAuthPlugin = async ( ) { if (!accessToken) return false const entry = quotaManager.getMain( - mainQuotaIdentity?.accountIdentity ?? mainQuotaAccountId, + mainQuotaIdentity?.quotaKey ?? mainQuotaAccountId, ) // A genuine response header is live routing evidence like a 429, but // it gets no exemption from the shared freshness and token gates. @@ -5796,7 +5823,7 @@ const anthropicAuthPlugin = async ( if (!accessToken) return false try { await quotaManager.refreshMain( - mainQuotaIdentity?.accountIdentity ?? mainQuotaAccountId, + mainQuotaIdentity?.quotaKey ?? mainQuotaAccountId, accessToken, mainQuotaIdentity?.generation, ) @@ -5874,8 +5901,8 @@ const anthropicAuthPlugin = async ( requestedModelId?: string mainQuotaIdentity?: MainQuotaIdentityBinding }) { - const mainQuotaIdentity = input.mainQuotaIdentity?.accountIdentity - const mainEntry = quotaManager.getMain(mainQuotaIdentity) + const mainQuotaKey = input.mainQuotaIdentity?.quotaKey + const mainEntry = quotaManager.getMain(mainQuotaKey) let mainQuota = mainEntry?.quota if ( !stickyQuotaSnapshotIsFresh( @@ -5887,7 +5914,7 @@ const anthropicAuthPlugin = async ( ) { try { mainQuota = await quotaManager.refreshMain( - mainQuotaIdentity, + mainQuotaKey, input.mainAccessToken, input.mainQuotaIdentity?.generation, ) @@ -6585,7 +6612,9 @@ const anthropicAuthPlugin = async ( trace.done('non_oauth_passthrough', { status: response.status }) return response } - let requestMainQuotaIdentity: MainQuotaIdentityBinding | undefined + let requestMainQuotaIdentity: + | MainQuotaIdentityResolution + | undefined if (auth.access) { const resolution = await resolveMainQuotaAccountIdentity( auth.access, @@ -6970,7 +6999,7 @@ const anthropicAuthPlugin = async ( quota = route.id === STICKY_ROUTING_MAIN_ACCOUNT_ID ? await quotaManager.refreshMain( - requestMainQuotaIdentity?.accountIdentity, + requestMainQuotaIdentity?.quotaKey, route.access, requestMainQuotaIdentity?.generation, ) @@ -7202,7 +7231,7 @@ const anthropicAuthPlugin = async ( function showQuotaToastFromCache() { if (storage?.quota?.showToasts !== true) return const mainEntry = quotaManager.getMain( - requestMainQuotaIdentity?.accountIdentity, + requestMainQuotaIdentity?.quotaKey, ) if (!mainEntry) return // Prefer the shared QuotaManager cache for fallback quota so the @@ -7243,17 +7272,17 @@ const anthropicAuthPlugin = async ( // Identity-aware read prevents routing with a previous main // account's quota after a slot switch. let routingQuotaEntry = quotaManager.getMain( - requestMainQuotaIdentity?.accountIdentity, + requestMainQuotaIdentity?.quotaKey, ) let routingQuota = routingQuotaEntry?.quota if (!routingQuota) { routingQuota = await quotaManager.refreshMain( - requestMainQuotaIdentity?.accountIdentity, + requestMainQuotaIdentity?.quotaKey, auth.access, requestMainQuotaIdentity?.generation, ) routingQuotaEntry = quotaManager.getMain( - requestMainQuotaIdentity?.accountIdentity, + requestMainQuotaIdentity?.quotaKey, ) showQuotaToastFromCache() } else if ( @@ -7276,12 +7305,12 @@ const anthropicAuthPlugin = async ( // below still refuses API-key routes because the entry is not // fresh. routingQuota = await quotaManager.refreshMain( - requestMainQuotaIdentity?.accountIdentity, + requestMainQuotaIdentity?.quotaKey, auth.access, requestMainQuotaIdentity?.generation, ) routingQuotaEntry = quotaManager.getMain( - requestMainQuotaIdentity?.accountIdentity, + requestMainQuotaIdentity?.quotaKey, ) } else { // Stale OR every-N request boundary — background refresh, @@ -7289,7 +7318,7 @@ const anthropicAuthPlugin = async ( // sidebar and show the toast once the new main quota lands. void quotaManager .refreshMain( - requestMainQuotaIdentity?.accountIdentity, + requestMainQuotaIdentity?.quotaKey, auth.access, requestMainQuotaIdentity?.generation, ) @@ -7365,7 +7394,7 @@ const anthropicAuthPlugin = async ( } let mainQuota = quotaManager.getMain( - requestMainQuotaIdentity?.accountIdentity, + requestMainQuotaIdentity?.quotaKey, )?.quota if ( storage?.quota?.failClosedOnUnknownQuota && @@ -7419,7 +7448,7 @@ const anthropicAuthPlugin = async ( ) await Promise.all([ quotaManager.refreshMain( - requestMainQuotaIdentity?.accountIdentity, + requestMainQuotaIdentity?.quotaKey, auth.access, requestMainQuotaIdentity?.generation, ), @@ -7441,7 +7470,7 @@ const anthropicAuthPlugin = async ( // against fresh quota. The initial read above is null on the // first request, before the refresh populates the cache. mainQuota = quotaManager.getMain( - requestMainQuotaIdentity?.accountIdentity, + requestMainQuotaIdentity?.quotaKey, )?.quota } diff --git a/packages/opencode/src/tests/accounts.test.ts b/packages/opencode/src/tests/accounts.test.ts index 6e7581c4..f484f6a1 100644 --- a/packages/opencode/src/tests/accounts.test.ts +++ b/packages/opencode/src/tests/accounts.test.ts @@ -20,6 +20,7 @@ import { buildRefreshOperationError, ClaudeOAuthRefreshError, FallbackAccountManager, + fallbackAccountUuidForLineage, fetchOAuthAccountProfile, fetchOAuthQuotaSnapshot, formatOAuthAccountTier, @@ -50,6 +51,7 @@ import { type OAuthQuotaSnapshot, oauthProfileIsFresh, PROFILE_TTL_MS, + persistFallbackQuotaHeaderPersistent, QuotaManager, quotaFieldSource, quotaSnapshotModelScopeIsExhausted, @@ -6449,6 +6451,109 @@ describe('upsertAccount', () => { expect(merged.lastRefreshError?.message).toBe('new') expect(merged.lastQuotaRefreshError?.message).toBe('new-quota') }) + + test('drops an Anthropic account UUID when an OAuth lineage is replaced', () => { + const storage = baseStorage() + storage.accounts.push({ + id: 'oauth-1', + type: 'oauth', + refresh: 'refresh-a', + authLineageId: 'lineage-a', + anthropicAccountUuid: 'uuid-from-lineage-a', + }) + + upsertAccount(storage, { + id: 'oauth-1', + type: 'oauth', + refresh: 'refresh-b', + authLineageId: 'lineage-b', + }) + + expect( + (storage.accounts[0] as OAuthAccount).anthropicAccountUuid, + ).toBeUndefined() + }) + + test('keeps an Anthropic account UUID across a same-lineage token rotation', () => { + const storage = baseStorage() + storage.accounts.push({ + id: 'oauth-1', + type: 'oauth', + refresh: 'refresh-a', + authLineageId: 'lineage-a', + anthropicAccountUuid: 'uuid-from-lineage-a', + }) + + upsertAccount(storage, { + id: 'oauth-1', + type: 'oauth', + refresh: 'refresh-b', + authLineageId: 'lineage-a', + }) + + expect((storage.accounts[0] as OAuthAccount).anthropicAccountUuid).toBe( + 'uuid-from-lineage-a', + ) + }) +}) + +describe('fallback UUID lineage fences', () => { + test('does not use a persisted UUID from a replaced credential lineage', () => { + const account: OAuthAccount = { + id: 'fallback-1', + type: 'oauth', + refresh: 'refresh-a', + authLineageId: 'lineage-a', + anthropicAccountUuid: 'uuid-from-lineage-a', + } + + expect(fallbackAccountUuidForLineage(account, 'lineage-b')).toBeNull() + expect(fallbackAccountUuidForLineage(account, 'lineage-a')).toBe( + 'uuid-from-lineage-a', + ) + }) + + test('does not persist a harvested UUID after the stored lineage was replaced', async () => { + await saveAccounts( + { + ...baseStorage(), + accounts: [ + { + id: 'fallback-1', + type: 'oauth', + refresh: 'refresh-b', + authLineageId: 'lineage-b', + }, + ], + }, + accountPath, + ) + + expect( + await persistFallbackQuotaHeaderPersistent( + { + accountId: 'fallback-1', + authLineageId: 'lineage-a', + anthropicAccountUuid: 'uuid-from-lineage-a', + quota: { + source: 'headers', + checkedAt: 1_000, + five_hour: { + usedPercent: 25, + remainingPercent: 75, + checkedAt: 1_000, + }, + }, + }, + accountPath, + ), + ).toBe(false) + + expect( + expectOAuthAccount((await loadAccounts(accountPath))?.accounts[0]) + .anthropicAccountUuid, + ).toBeUndefined() + }) }) describe('removeAccount', () => { diff --git a/packages/opencode/src/tests/claude-code.test.ts b/packages/opencode/src/tests/claude-code.test.ts index f01ce4f0..bf247807 100644 --- a/packages/opencode/src/tests/claude-code.test.ts +++ b/packages/opencode/src/tests/claude-code.test.ts @@ -295,12 +295,12 @@ describe('Claude Code fingerprint helpers', () => { const identityA = await pendingA expect(identityA.accountUuid).toBeUndefined() - const compatibility = await resolveClaudeCodeIdentity( - 'compatibility-token-after-late-failure', + const sameCredential = await resolveClaudeCodeIdentity( + tokenB, undefined, 'main-slot', ) - expect(compatibility.accountUuid).toBe('account-b') + expect(sameCredential.accountUuid).toBe('account-b') }) test('orders serialized body fields like captured Claude Code requests', () => { @@ -430,6 +430,31 @@ describe('Claude Code bootstrap identity lookup', () => { expect(fetchMock).toHaveBeenCalledTimes(2) }) + test('drops a bootstrapped UUID when a slot receives a non-oat credential', async () => { + const oatToken = 'sk-ant-oat-main-to-non-oat' + globalThis.fetch = mock(async () => + Response.json({ + oauth_account: { account_uuid: 'main-oat-account-uuid' }, + }), + ) as unknown as typeof fetch + + const oatIdentity = await resolveClaudeCodeIdentity( + oatToken, + undefined, + 'main-slot', + ) + const nonOatIdentity = await resolveClaudeCodeIdentity( + 'main-non-oat-token', + undefined, + 'main-slot', + ) + + expect(oatIdentity.accountUuid).toBe('main-oat-account-uuid') + expect(nonOatIdentity.accountUuid).toBeUndefined() + expect(nonOatIdentity.deviceId).toBe(oatIdentity.deviceId) + expect(nonOatIdentity.sessionId).toBe(oatIdentity.sessionId) + }) + test('keeps device identity stable while refreshing bootstrap UUID per credential', async () => { const accountUuid = 'c7b3bc43-f4d8-48c6-a30f-7fd81a8db03f' const fetchMock = mock( diff --git a/packages/opencode/src/tests/index.test.ts b/packages/opencode/src/tests/index.test.ts index 6a2f0e6e..53e3721e 100644 --- a/packages/opencode/src/tests/index.test.ts +++ b/packages/opencode/src/tests/index.test.ts @@ -7,6 +7,7 @@ import { mock, test, } from 'bun:test' +import { randomUUID } from 'node:crypto' import { readFileSync } from 'node:fs' import { chmod, @@ -3530,7 +3531,96 @@ describe('quota header feed integration', () => { globalThis.fetch = originalFetch }) - test('publishes a direct harvested response with the observation timestamp', async () => { + async function publishFallbackUuid({ + fallbackUuid, + persistedUuid, + bootstrapResponse, + }: { + fallbackUuid: string + persistedUuid?: string + bootstrapResponse: unknown + }) { + const fallbackAccess = `sk-ant-oat-${randomUUID()}` + await useTempAccountFile( + createFallbackStorage({ + quotaHeaderFeed: { enabled: true }, + accounts: [ + { + id: 'fallback-1', + type: 'oauth', + access: fallbackAccess, + refresh: 'fallback-refresh', + expires: Date.now() + 5 * 60 * 60 * 1000, + ...(persistedUuid && { anthropicAccountUuid: persistedUuid }), + quota: { + five_hour: { + usedPercent: 25, + remainingPercent: 75, + checkedAt: Date.now(), + }, + seven_day: { + usedPercent: 30, + remainingPercent: 70, + checkedAt: Date.now(), + }, + }, + }, + ], + }), + ) + globalThis.fetch = mock((input: any, init?: RequestInit) => { + const url = extractUrl(input) + if (url.includes('/claude_cli/bootstrap')) { + return Promise.resolve(Response.json(bootstrapResponse)) + } + if (url.includes('/v1/messages')) { + const authorization = new Headers(init?.headers).get('authorization') + if (authorization === `Bearer ${fallbackAccess}`) { + return Promise.resolve( + new Response('{}', { + status: 200, + headers: { 'anthropic-ratelimit-unified-5h-utilization': '0.25' }, + }), + ) + } + return Promise.resolve(new Response(null, { status: 429 })) + } + return Promise.resolve(Response.json({})) + }) as unknown as typeof fetch + + const plugin = await getPlugin() + const result = await plugin.auth.loader( + () => + Promise.resolve({ + type: 'oauth' as const, + access: 'main-access', + refresh: 'main-refresh', + expires: Date.now() + 100_000, + }), + { models: {} }, + ) + const response = await result.fetch(MESSAGES_URL, EMPTY_POST) + await response.text() + + const [published] = await waitForFeedEntries( + (entries) => + entries.some( + (entry: any) => entry.anthropic_account_uuid === fallbackUuid, + ), + 'a fallback UUID feed entry', + ) + const persisted = await waitForAccountStorage( + (candidate) => + ( + candidate?.accounts.find( + (account) => account.id === 'fallback-1', + ) as any + )?.anthropicAccountUuid === fallbackUuid, + ) + return { published, persisted } + } + + test('non-oat main feed leaves provider UUID null while retaining its quota slot', async () => { const originalNow = Date.now let clock = 1_000_000 Date.now = () => clock @@ -3573,7 +3663,7 @@ describe('quota header feed integration', () => { clock += 1_000 const feedDirectory = process.env.OPENCODE_ANTHROPIC_AUTH_QUOTA_FEED_DIR! - await waitForFeedEntries( + const [feedEntry] = await waitForFeedEntries( (entries) => entries.length === 1, 'one published entry', ) @@ -3588,6 +3678,7 @@ describe('quota header feed integration', () => { expect.objectContaining({ identity_source: 'account_ref', account_ref: 'main-fixed-id', + anthropic_account_uuid: null, configured_account_count: 1, observed_at_ms: 1_000_000, quota: expect.objectContaining({ @@ -3595,6 +3686,7 @@ describe('quota header feed integration', () => { }), }), ) + expect(record.lease_horizon_ms).toBe(180_000) expect(published).not.toHaveProperty('credential_id') for (const secret of [ 'main-access', @@ -3611,11 +3703,251 @@ describe('quota header feed integration', () => { } expect(raw).not.toContain('"credential_id":"main"') expect(raw).not.toContain('"account_ref":"main"') + expect(feedEntry).toEqual( + expect.objectContaining({ + account_ref: 'main-fixed-id', + anthropic_account_uuid: null, + }), + ) + const persisted = await waitForAccountStorage( + (storage) => + storage?.quota?.mainQuota?.accountIdentity === 'main-fixed-id', + ) + expect(persisted?.quota?.mainQuota?.accountIdentity).toBe('main-fixed-id') } finally { Date.now = originalNow } }) + test('oat main feed preserves its bootstrapped Anthropic account UUID', async () => { + const mainAccountId = 'main-oat-slot' + const providerAccountUuid = '33333333-3333-3333-3333-333333333333' + await useTempAccountFile( + createFallbackStorage({ + mainAccountId, + quotaHeaderFeed: { enabled: true }, + accounts: [], + }), + ) + globalThis.fetch = mock((input: any) => { + const url = extractUrl(input) + if (url.includes('/claude_cli/bootstrap')) { + return Promise.resolve( + Response.json({ + oauth_account: { account_uuid: providerAccountUuid }, + }), + ) + } + return Promise.resolve( + new Response('{}', { + status: 200, + headers: { 'anthropic-ratelimit-unified-5h-utilization': '0.25' }, + }), + ) + }) as unknown as typeof fetch + + const plugin = await getPlugin() + const result = await plugin.auth.loader( + () => + Promise.resolve({ + type: 'oauth' as const, + access: 'sk-ant-oat-main-feed', + refresh: 'main-refresh', + expires: Date.now() + 100_000, + }), + { models: {} }, + ) + const response = await result.fetch(MESSAGES_URL, EMPTY_POST) + await response.text() + + const [published] = await waitForFeedEntries( + (entries) => + entries.some( + (entry: any) => entry.anthropic_account_uuid === providerAccountUuid, + ), + 'a bootstrapped main UUID feed entry', + ) + expect(published).toEqual( + expect.objectContaining({ + account_ref: providerAccountUuid, + anthropic_account_uuid: providerAccountUuid, + }), + ) + }) + + test('main feed drops a cached oat UUID after the current credential becomes non-oat', async () => { + const mainAccountId = 'main-oat-to-non-oat-slot' + const providerAccountUuid = '44444444-4444-4444-4444-444444444444' + let liveAccess = 'sk-ant-oat-main-feed-before-rotation' + await useTempAccountFile( + createFallbackStorage({ + mainAccountId, + quotaHeaderFeed: { enabled: true }, + accounts: [], + }), + ) + globalThis.fetch = mock((input: any) => { + const url = extractUrl(input) + if (url.includes('/claude_cli/bootstrap')) { + return Promise.resolve( + Response.json({ + oauth_account: { account_uuid: providerAccountUuid }, + }), + ) + } + return Promise.resolve( + new Response('{}', { + status: 200, + headers: { 'anthropic-ratelimit-unified-5h-utilization': '0.25' }, + }), + ) + }) as unknown as typeof fetch + + const plugin = await getPlugin() + const result = await plugin.auth.loader( + () => + Promise.resolve({ + type: 'oauth' as const, + access: liveAccess, + refresh: `refresh-${liveAccess}`, + expires: Date.now() + 100_000, + }), + { models: {} }, + ) + await (await result.fetch(MESSAGES_URL, EMPTY_POST)).text() + await waitForFeedEntries( + (entries) => + entries.some( + (entry: any) => entry.anthropic_account_uuid === providerAccountUuid, + ), + 'a bootstrapped main UUID feed entry', + ) + + liveAccess = 'main-feed-non-oat-after-rotation' + await (await result.fetch(MESSAGES_URL, EMPTY_POST)).text() + + const feedEntries = await waitForFeedEntries( + (entries) => + entries.some( + (entry: any) => + entry.account_ref === mainAccountId && + entry.anthropic_account_uuid === null, + ), + 'a non-oat main feed entry without a provider UUID', + ) + const published = feedEntries.find( + (entry: any) => + entry.account_ref === mainAccountId && + entry.anthropic_account_uuid === null, + ) + expect(published).toEqual( + expect.objectContaining({ + account_ref: mainAccountId, + anthropic_account_uuid: null, + }), + ) + }) + + test('fresh non-oat main install persists a quota slot and publishes no provider UUID', async () => { + await useTempAccountFile({ + version: 1, + main: { type: 'opencode', provider: 'anthropic' }, + quotaHeaderFeed: { enabled: true }, + accounts: [], + }) + globalThis.fetch = mock(() => + Promise.resolve( + new Response('{}', { + status: 200, + headers: { 'anthropic-ratelimit-unified-5h-utilization': '0.25' }, + }), + ), + ) as unknown as typeof fetch + + const plugin = await getPlugin() + const result = await plugin.auth.loader( + () => + Promise.resolve({ + type: 'oauth' as const, + access: 'fresh-main-non-oat', + refresh: 'main-refresh', + expires: Date.now() + 100_000, + }), + { models: {} }, + ) + const response = await result.fetch(MESSAGES_URL, EMPTY_POST) + await response.text() + + const persisted = await waitForAccountStorage( + (storage) => + storage?.mainAccountId !== undefined && + storage.quota?.mainQuota?.accountIdentity === storage.mainAccountId, + ) + const [published] = await waitForFeedEntries( + (entries) => + entries.some( + (entry: any) => + entry.account_ref === persisted?.mainAccountId && + entry.anthropic_account_uuid === null, + ), + 'a fresh non-oat main feed entry', + ) + expect(persisted?.mainAccountId).toBeDefined() + expect(persisted?.quota?.mainQuota?.accountIdentity).toBe( + persisted?.mainAccountId, + ) + expect(published).toEqual( + expect.objectContaining({ + account_ref: persisted?.mainAccountId, + anthropic_account_uuid: null, + }), + ) + }) + + test('publishes and persists a fallback Anthropic account UUID', async () => { + const fallbackUuid = '11111111-1111-1111-1111-111111111111' + const { published, persisted } = await publishFallbackUuid({ + fallbackUuid, + bootstrapResponse: { oauth_account: { account_uuid: fallbackUuid } }, + }) + expect(published).toEqual( + expect.objectContaining({ + identity_source: 'account_ref', + account_ref: 'fallback-1', + anthropic_account_uuid: fallbackUuid, + }), + ) + expect( + ( + persisted?.accounts.find( + (account) => account.id === 'fallback-1', + ) as any + )?.anthropicAccountUuid, + ).toBe(fallbackUuid) + }) + + test('uses a persisted fallback Anthropic account UUID before bootstrap resolves it', async () => { + const fallbackUuid = '22222222-2222-2222-2222-222222222222' + const { published, persisted } = await publishFallbackUuid({ + fallbackUuid, + persistedUuid: fallbackUuid, + bootstrapResponse: {}, + }) + expect(published).toEqual( + expect.objectContaining({ + account_ref: 'fallback-1', + anthropic_account_uuid: fallbackUuid, + }), + ) + expect( + ( + persisted?.accounts.find( + (account) => account.id === 'fallback-1', + ) as any + )?.anthropicAccountUuid, + ).toBe(fallbackUuid) + }) + test('cache-seeded poll fields survive a later header harvest into the feed', async () => { const originalNow = Date.now let clock = 1_000_000 @@ -4022,12 +4354,11 @@ describe('quota header feed integration', () => { await waitForLogRecord( records, (record) => - record.level === 'trace' && + record.level === 'debug' && record.channel === 'quota' && record.message === 'skipped stale fallback quota persistence after lineage change' && record.payload?.accountId === 'fallback-1' && - record.payload?.storedLineage === 'lineage-b' && record.payload?.servedLineage === 'lineage-a', 'stale fallback quota persistence discard', ) diff --git a/packages/opencode/src/tests/quota-header-feed.test.ts b/packages/opencode/src/tests/quota-header-feed.test.ts index 69ad1a18..ee387cc4 100644 --- a/packages/opencode/src/tests/quota-header-feed.test.ts +++ b/packages/opencode/src/tests/quota-header-feed.test.ts @@ -5,8 +5,10 @@ import { mkdtemp, readdir, readFile, + rename, rm, stat, + utimes, writeFile, } from 'node:fs/promises' import { tmpdir } from 'node:os' @@ -32,6 +34,7 @@ function entry(overrides: Record = {}): QuotaHeaderFeedEntry { provider: 'anthropic', configured_account_count: 1, observed_at_ms: 1_000, + anthropic_account_uuid: null, quota, ...overrides, } as QuotaHeaderFeedEntry @@ -77,6 +80,50 @@ describe('quota header feed', () => { expect(QUOTA_HEADER_FEED_SCHEMA_VERSION).toBe(3) expect(raw.entries.a.schema_version).toBe(3) expect(raw.entries.a.provider).toBe('anthropic') + expect(raw.lease_horizon_ms).toBe(QUOTA_HEADER_FEED_LEASE_MS) + expect(raw.entries.a).toHaveProperty('anthropic_account_uuid', null) + }) + + test('projects only the documented entry keys', async () => { + const registry = new QuotaHeaderFeedRegistry({ + directory, + instanceId: 'entry-allowlist', + }) + await registry.publish({ + ...entry({ + anthropic_account_uuid: 'uuid-1', + unexpected_entry_secret: 'must-not-publish', + }), + accountKey: 'a', + } as unknown as QuotaHeaderFeedPublishEntry) + + const raw = JSON.parse( + await readFile(join(directory, 'entry-allowlist.json'), 'utf8'), + ) + expect(raw.entries.a).toEqual({ + identity_source: 'credential_id', + credential_id: 'cred-1', + schema_version: QUOTA_HEADER_FEED_SCHEMA_VERSION, + provider: 'anthropic', + configured_account_count: 1, + observed_at_ms: 1_000, + anthropic_account_uuid: 'uuid-1', + quota, + }) + expect(JSON.stringify(raw)).not.toContain('must-not-publish') + }) + + test('publishes the configured lease horizon from the registry seam', async () => { + const registry = new QuotaHeaderFeedRegistry({ + directory, + instanceId: 'lease-horizon', + leaseMs: 17, + }) + await registry.publish({ ...entry(), accountKey: 'a' }) + const raw = JSON.parse( + await readFile(join(directory, 'lease-horizon.json'), 'utf8'), + ) + expect(raw.lease_horizon_ms).toBe(17) }) test.each([1, 2, 999])( @@ -510,4 +557,120 @@ describe('quota header feed', () => { entry({ observed_at_ms: 1_002, credential_id: 'new' }), ]) }) + + test('reaps stale sibling process leases without touching fresh or foreign files', async () => { + const now = Date.now() + await mkdir(directory, { recursive: true }) + const staleNames = [ + '101-11111111-1111-1111-1111-111111111111.json', + '102-22222222-2222-2222-2222-222222222222.json', + '103-33333333-3333-3333-3333-333333333333.json', + ] + const freshNames = [ + '104-44444444-4444-4444-4444-444444444444.json', + '105-55555555-5555-5555-5555-555555555555.json', + ] + for (const name of [...staleNames, ...freshNames]) { + const path = join(directory, name) + await writeFile(path, '{}') + const age = staleNames.includes(name) + ? QUOTA_HEADER_FEED_LEASE_MS + 1 + : QUOTA_HEADER_FEED_LEASE_MS - 1 + await utimes(path, (now - age) / 1_000, (now - age) / 1_000) + } + await writeFile(join(directory, 'foreign-named-file.json'), '{}') + await utimes( + join(directory, 'foreign-named-file.json'), + (now - QUOTA_HEADER_FEED_LEASE_MS - 1) / 1_000, + (now - QUOTA_HEADER_FEED_LEASE_MS - 1) / 1_000, + ) + + const registry = new QuotaHeaderFeedRegistry({ + directory, + instanceId: '106-66666666-6666-6666-6666-666666666666', + now: () => now, + }) + await registry.publish({ ...entry(), accountKey: 'a' }) + + const names = await readdir(directory) + for (const name of staleNames) expect(names).not.toContain(name) + for (const name of freshNames) expect(names).toContain(name) + expect(names).toContain('foreign-named-file.json') + expect(names).toContain('106-66666666-6666-6666-6666-666666666666.json') + }) + + test('does not unlink a fresh lease published after the stale lease was inspected', async () => { + const now = Date.now() + const siblingName = '101-11111111-1111-1111-1111-111111111111.json' + const siblingPath = join(directory, siblingName) + const replacementPath = join(directory, 'fresh-publisher.tmp') + await mkdir(directory, { recursive: true }) + await writeFile(siblingPath, '{"stale":true}') + await utimes( + siblingPath, + (now - QUOTA_HEADER_FEED_LEASE_MS - 1) / 1_000, + (now - QUOTA_HEADER_FEED_LEASE_MS - 1) / 1_000, + ) + expect(now - (await stat(siblingPath)).mtimeMs).toBeGreaterThanOrEqual( + QUOTA_HEADER_FEED_LEASE_MS, + ) + + let publisherRan = false + let publisherError: unknown + const registry = new QuotaHeaderFeedRegistry({ + directory, + instanceId: '102-22222222-2222-2222-2222-222222222222', + now: () => now, + beforeRemoveFile: async (path: string) => { + publisherRan = true + try { + await writeFile(replacementPath, '{"fresh":true}') + await utimes(replacementPath, now / 1_000, now / 1_000) + await rename(replacementPath, siblingPath) + expect(path).toBe(siblingPath) + } catch (error) { + publisherError = error + throw error + } + }, + }) + + await registry.publish({ ...entry(), accountKey: 'a' }) + + expect(publisherRan).toBe(true) + expect(publisherError).toBeUndefined() + expect(await readFile(siblingPath, 'utf8')).toBe('{"fresh":true}') + }) + + test('continues publishing when a sibling lease sweep cannot unlink a stale file', async () => { + const now = Date.now() + const stalePath = join( + directory, + '101-11111111-1111-1111-1111-111111111111.json', + ) + await mkdir(directory, { recursive: true }) + await writeFile(stalePath, '{}') + await utimes( + stalePath, + (now - QUOTA_HEADER_FEED_LEASE_MS - 1) / 1_000, + (now - QUOTA_HEADER_FEED_LEASE_MS - 1) / 1_000, + ) + const ownFile = '102-22222222-2222-2222-2222-222222222222.json' + const registry = new QuotaHeaderFeedRegistry({ + directory, + instanceId: ownFile.slice(0, -'.json'.length), + now: () => now, + removeFile: async () => { + throw Object.assign(new Error('permission denied'), { code: 'EACCES' }) + }, + }) + + await expect( + registry.publish({ ...entry(), accountKey: 'a' }), + ).resolves.toBeUndefined() + expect(await readdir(directory)).toContain(ownFile) + expect(await readdir(directory)).toContain( + '101-11111111-1111-1111-1111-111111111111.json', + ) + }) })