From 402f95a6403b4117ddee6c5f71cf7992b9443cb2 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Sun, 18 Oct 2026 00:05:25 +0200 Subject: [PATCH 1/9] mason: read vault Claude quota from insula before polling (wip) --- packages/core/src/index.ts | 1 + packages/core/src/insula-quota.ts | 260 +++++++ packages/core/src/native-account-runtime.ts | 11 +- packages/core/src/native-custody.ts | 10 + packages/core/src/native-vault-runtime.ts | 141 +++- packages/core/src/quota-manager.ts | 39 +- .../tests/native-vault-insula-quota.test.ts | 634 ++++++++++++++++++ packages/opencode/src/index.ts | 32 +- packages/opencode/src/v2/lifecycle.ts | 10 +- packages/opencode/src/v2/source.ts | 10 +- packages/pi/src/stream.ts | 18 +- 11 files changed, 1141 insertions(+), 25 deletions(-) create mode 100644 packages/core/src/insula-quota.ts create mode 100644 packages/core/src/tests/native-vault-insula-quota.test.ts diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 8ee59831..b1301ec1 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -234,6 +234,7 @@ export { openNativePiAuthSlot, } from './native-vault-connection.ts' export type { + NativeVaultQuotaPurpose, NativeVaultRuntime, NativeVaultRuntimeOptions, NativeVaultWatch, diff --git a/packages/core/src/insula-quota.ts b/packages/core/src/insula-quota.ts new file mode 100644 index 00000000..2a4e8765 --- /dev/null +++ b/packages/core/src/insula-quota.ts @@ -0,0 +1,260 @@ +import type { + InsulaQuotaContext, + InsulaUsageEntry, + InsulaUsageWindow, +} from '@cortexkit/common-auth/claustrum' +import { + type AccountQuotaWindow, + type AccountScopedQuotaWindow, + type OAuthExtraUsageSnapshot, + type OAuthQuotaSnapshot, + type QuotaFieldSources, + quotaSnapshotCheckedAt, +} from './accounts.ts' + +/** An insula reading in the shape our usage poll produces, before the account identity is attached. */ +export type InsulaClaudeQuota = OAuthQuotaSnapshot & { + five_hour: AccountQuotaWindow + seven_day: AccountQuotaWindow + scoped: AccountScopedQuotaWindow[] + checkedAt: number +} + +function record(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function text(value: unknown): string | undefined { + return typeof value === 'string' && value.trim() ? value.trim() : undefined +} + +// Same bounds as the usage poll's mapping in accounts.ts. +function clampPercent(value: number) { + return Math.min(100, Math.max(0, value)) +} + +/** + * Insula's usedPercent is Anthropic's whole-number utilization, so it is used + * unchanged. When insula names a window's kind or length it must match the + * slot we read it from; anything else is not a window we know how to use. + */ +function windowReading( + window: InsulaUsageWindow | undefined, + checkedAt: number, + expected?: { kind: string; minutes: number }, +): AccountQuotaWindow | undefined { + if (!record(window)) return undefined + const usedPercent = window.usedPercent + if (typeof usedPercent !== 'number' || !Number.isFinite(usedPercent)) + return undefined + if ( + expected && + ((window.windowKind !== undefined && window.windowKind !== expected.kind) || + (window.windowMinutes !== undefined && + window.windowMinutes !== expected.minutes)) + ) + return undefined + const used = clampPercent(usedPercent) + return { + usedPercent: used, + remainingPercent: clampPercent(100 - used), + ...(typeof window.resetsAt === 'string' && { resetsAt: window.resetsAt }), + checkedAt, + } +} + +// Same slug as the usage poll, so an insula reading and a poll reading of +// one model-scoped limit share a window id and replace each other. +function slugForQuotaIdentity(value: string): string { + return value + .toLowerCase() + .replace(/[^a-z0-9]+/g, '-') + .replace(/^-+|-+$/g, '') +} + +/** + * Model-scoped weekly limits. Insula's `id` is Anthropic's model id, or the + * model's display name when Anthropic sends no id (the live value is + * `Fable`). The poll keys the window on the same value, so routing and the + * scoped killswitch floor find it the same way. `title` is display text and + * never identifies a window. + */ +function scopedReadings( + windows: unknown, + checkedAt: number, +): AccountScopedQuotaWindow[] { + if (!Array.isArray(windows)) return [] + const seen = new Set() + const scoped: AccountScopedQuotaWindow[] = [] + for (const candidate of windows) { + if (!record(candidate)) continue + const identity = text(candidate.id) + if (!identity) continue + const slug = slugForQuotaIdentity(identity) + if (!slug) continue + const id = `claude-weekly-scoped-${slug}` + if (seen.has(id)) continue + const reading = windowReading( + candidate.window as InsulaUsageWindow | undefined, + checkedAt, + ) + if (!reading) continue + seen.add(id) + scoped.push({ + ...reading, + id, + title: `${identity} only`, + // A display-name fallback is not a model id; only Anthropic's own ids + // (claude-...) are recorded as one. + ...(identity.startsWith('claude-') && { modelId: identity }), + modelName: identity, + }) + } + return scoped +} + +function money(value: unknown): { minor: number; exponent: number } | null { + if ( + !record(value) || + typeof value.minor !== 'number' || + !Number.isSafeInteger(value.minor) || + value.minor < 0 + ) + return null + const exponent = value.exponent ?? 2 + if ( + typeof exponent !== 'number' || + !Number.isInteger(exponent) || + exponent < 0 || + exponent > 20 + ) + return null + return { minor: value.minor, exponent } +} + +/** + * Insula reports extra usage as a pool titled `Extra usage`, present only + * while overage is enabled, with amounts in cents. A pool whose amounts + * cannot be read is left out, exactly as the poll leaves out an unreadable + * extra_usage block. + */ +function extraUsageReading( + pools: unknown, +): OAuthExtraUsageSnapshot | undefined { + if (!Array.isArray(pools)) return undefined + const pool = pools.find( + (candidate) => record(candidate) && candidate.title === 'Extra usage', + ) + if (!record(pool)) return undefined + const limit = money(pool.total ?? pool.limit) + const remaining = money(pool.remaining) + const usedAmount = money(pool.used) + if (!limit) return undefined + const exponent = limit.exponent + let used: number | undefined + if (usedAmount && usedAmount.exponent === exponent) used = usedAmount.minor + else if (remaining && remaining.exponent === exponent) + used = Math.max(0, limit.minor - remaining.minor) + if (used === undefined) return undefined + const unit = text(pool.unit) ?? text(record(pool.total) && pool.total.unit) + const currency = unit && /^[A-Za-z]{3}$/.test(unit) ? unit : 'USD' + return { + used: { amountMinor: used, currency, exponent }, + limit: { amountMinor: limit.minor, currency, exponent }, + ...(limit.minor > 0 && { + utilizationPercent: (used / limit.minor) * 100, + }), + exhausted: pool.spendable === false || used >= limit.minor, + } +} + +/** + * Convert insula's Claude entry into our usage-poll snapshot. Every window's + * checkedAt is when insula fetched the reading, never when we read its cache, + * so a newer header or poll reading still wins. Both standard windows must be + * present; otherwise this returns undefined and our own poll stays in charge. + * The tertiary slot (legacy seven_day_opus/sonnet) is not a window we model. + */ +export function normalizeInsulaClaudeQuota( + entry: InsulaUsageEntry, + context: InsulaQuotaContext, +): InsulaClaudeQuota | undefined { + const checkedAt = context.fetchedAt + if ( + !Number.isFinite(checkedAt) || + entry.stale !== undefined || + entry.errorClass !== undefined + ) + return undefined + const usage = entry.usage + if (!record(usage)) return undefined + const five_hour = windowReading(usage.primary, checkedAt, { + kind: 'five_hour', + minutes: 300, + }) + const seven_day = windowReading(usage.secondary, checkedAt, { + kind: 'weekly', + minutes: 10_080, + }) + if (!five_hour || !seven_day) return undefined + const scoped = scopedReadings(usage.extraRateWindows, checkedAt) + const extraUsage = extraUsageReading(entry.pools ?? usage.pools) + return { + five_hour, + seven_day, + scoped, + ...(extraUsage && { extraUsage }), + source: 'poll', + checkedAt, + fieldSources: { + five_hour: 'poll', + seven_day: 'poll', + scoped: 'poll', + ...(extraUsage && { extraUsage: 'poll' as const }), + }, + } +} + +function newer( + reading: T | undefined, + stored: T | undefined, +): T | undefined { + if (!stored) return reading + if (!reading) return stored + return stored.checkedAt > reading.checkedAt ? stored : reading +} + +/** + * Insula fetches on its own schedule, so the account's stored quota can hold + * a window read after insula's fetch (a response header, or our own poll). + * Stored quota is replaced as one snapshot, newest wins, so publishing the + * older insula reading as is would be refused and lose its other windows. + * Keep the newer reading of each window instead. + */ +export function mergeInsulaQuota( + reading: InsulaClaudeQuota, + stored: OAuthQuotaSnapshot | undefined, +): OAuthQuotaSnapshot { + if (!stored || quotaSnapshotCheckedAt(stored) <= reading.checkedAt) + return reading + const five_hour = newer(reading.five_hour, stored.five_hour) + const seven_day = newer(reading.seven_day, stored.seven_day) + const scoped = new Map(reading.scoped.map((window) => [window.id, window])) + for (const window of stored.scoped ?? []) { + const merged = newer(scoped.get(window.id), window) + if (merged) scoped.set(window.id, merged) + } + const fieldSources: QuotaFieldSources = { ...reading.fieldSources } + if (five_hour !== reading.five_hour && stored.fieldSources?.five_hour) + fieldSources.five_hour = stored.fieldSources.five_hour + if (seven_day !== reading.seven_day && stored.fieldSources?.seven_day) + fieldSources.seven_day = stored.fieldSources.seven_day + return { + ...reading, + ...(five_hour && { five_hour }), + ...(seven_day && { seven_day }), + scoped: [...scoped.values()], + checkedAt: Math.max(reading.checkedAt, quotaSnapshotCheckedAt(stored)), + fieldSources, + } +} diff --git a/packages/core/src/native-account-runtime.ts b/packages/core/src/native-account-runtime.ts index 9b2fe885..2de93b08 100644 --- a/packages/core/src/native-account-runtime.ts +++ b/packages/core/src/native-account-runtime.ts @@ -88,6 +88,7 @@ import { acquireNativeVaultRuntime, leaseNativeVaultRuntime, type NativeDisplayProfile, + type NativeVaultQuotaPurpose, type NativeVaultRuntime, type NativeVaultRuntimeOptions, nativeDisplayProfile, @@ -247,11 +248,13 @@ export interface NativeAccountRuntime { fence: NativeKnownCredentialSubject | NativeCustodyReceipt, delta?: PrimeUsageDelta, ): Promise + /** `purpose` applies to vault accounts only; local accounts always poll. */ fetchQuota( routeId: string, fetchImpl?: typeof fetch, signal?: AbortSignal, dispatch?: QuotaRequestDispatch, + purpose?: NativeVaultQuotaPurpose, ): Promise fetchProfile( routeId: string, @@ -320,8 +323,8 @@ export function acquireNativeAccountRuntime( use().getOrCreateAuthLineage(id, fence), incrementPrimeUsage: (id, fence, delta) => use().incrementPrimeUsage(id, fence, delta), - fetchQuota: (id, transport, signal, dispatch) => - use().fetchQuota(id, transport, signal, dispatch), + fetchQuota: (id, transport, signal, dispatch, purpose) => + use().fetchQuota(id, transport, signal, dispatch, purpose), fetchProfile: (id, transport, signal) => use().fetchProfile(id, transport, signal), fetchProfileForDisplay: (id, transport, signal) => @@ -1506,10 +1509,10 @@ export function createNativeAccountRuntime( }, })) }, - async fetchQuota(routeId, fetchImpl = fetch, signal, dispatch) { + async fetchQuota(routeId, fetchImpl = fetch, signal, dispatch, purpose) { const snapshot = await runtime.read() if (snapshot.mode === 'claustrum') - return vault.fetchQuota(routeId, fetchImpl, signal, dispatch) + return vault.fetchQuota(routeId, fetchImpl, signal, dispatch, purpose) const before = snapshot.accounts.find((row) => row.id === routeId) const result = await fetchReading( routeId, diff --git a/packages/core/src/native-custody.ts b/packages/core/src/native-custody.ts index 009b56ab..ddb8b02c 100644 --- a/packages/core/src/native-custody.ts +++ b/packages/core/src/native-custody.ts @@ -1,6 +1,7 @@ import { ClaustrumConsumerError, ClaustrumScopedCustody, + type InsulaUsageTransport, } from '@cortexkit/common-auth/claustrum' import { getHostClaustrumEnrollmentPaths } from './claustrum-enrollment.ts' @@ -59,6 +60,12 @@ export interface NativeCustodyClient { reporterSource: NativeCustodyReporterSource }) => Promise close: () => void + /** + * How to read insula's background quota cache on the same daemon. Present + * when the connection was made with connectClaustrumScopedClient; reading + * it opens a separate short-lived socket and never uses this connection. + */ + insula?: InsulaUsageTransport } export type NativeCustodyReporterSource = @@ -219,6 +226,8 @@ export interface NativeCustody { served: NativeCustodyReceipt, signal?: AbortSignal, ) => Promise + /** The shared connection's insula transport, connecting first if needed; undefined when the client has none. */ + insulaTransport: () => Promise close: () => void } @@ -414,6 +423,7 @@ export function createNativeCustody( } return { retry: true, receipt: current } }, + insulaTransport: () => scoped(async (owned) => owned.insula), close: () => { if (closed) return closed = true diff --git a/packages/core/src/native-vault-runtime.ts b/packages/core/src/native-vault-runtime.ts index ade7324f..c1a5d25d 100644 --- a/packages/core/src/native-vault-runtime.ts +++ b/packages/core/src/native-vault-runtime.ts @@ -5,7 +5,10 @@ import { } from '@cortexkit/claustrum-client' import { acceptAccount, + ClaustrumConsumer, ClaustrumConsumerError, + type ClaustrumScopedClient, + connectClaustrumScopedClient, declineAccount, mutateVaultRoster, readVaultRoster, @@ -19,6 +22,12 @@ import { type PrimeUsageDelta, quotaSnapshotCheckedAt, } from './accounts.ts' +import { getHostClaustrumEnrollmentPaths } from './claustrum-enrollment.ts' +import { + type InsulaClaudeQuota, + mergeInsulaQuota, + normalizeInsulaClaudeQuota, +} from './insula-quota.ts' import { logger } from './logger.ts' import { applyNativeMetadataPatch, @@ -76,7 +85,10 @@ import { fetchOAuthAccountProfile } from './oauth-profile.ts' import { NativeAuthorityError } from './pool-authority.ts' import type { NativePoolPaths } from './pool-paths.ts' import { createNativePoolStore } from './pool-store.ts' -import type { QuotaRequestDispatch } from './quota-manager.ts' +import type { + QuotaReadingPurpose, + QuotaRequestDispatch, +} from './quota-manager.ts' export interface NativeVaultRuntimeOptions { paths: NativePoolPaths @@ -90,6 +102,20 @@ export interface NativeVaultRuntimeOptions { now?: () => number } +/** + * Why a vault quota reading is wanted, which decides whether insula's + * background quota cache may answer instead of Anthropic's usage endpoint. + * - 'background': a periodic sweep. Refresh insula's cache (at most once a + * minute per daemon and enrollment), look the route's account up in it, + * and skip the usage GET when it holds a current reading. + * - 'admission': a request needs a reading. Look the served credential's + * account up in insula's cache only, never refreshing it, then poll on a + * miss. + * Omitted: poll Anthropic directly (explicit user refreshes and other + * callers that must see the provider's own reading). + */ +export type NativeVaultQuotaPurpose = QuotaReadingPurpose + export interface NativeVaultWatchOptions { intervalMs: number onRefresh?: () => void | Promise @@ -218,6 +244,7 @@ export interface NativeVaultRuntime { fetchImpl?: typeof fetch, signal?: AbortSignal, dispatch?: QuotaRequestDispatch, + purpose?: NativeVaultQuotaPurpose, ): Promise fetchProfile( routeId: string, @@ -348,13 +375,65 @@ export function createNativeVaultRuntime( now: options.now, connect: options.connect ?? + // The scoped client is a plain ClaustrumClient that also carries the + // insula transport for the same connection file and identity. (() => - ClaustrumClient.connect({ + connectClaustrumScopedClient({ connectionFile, identity, })), env, }) + // Used only for insula's background quota cache: this runtime's own custody + // lists, authorizes and reports. Common-auth shares one insula source per + // resolved connection file and enrollment token path, and the consumer + // never reads it unless the enrollment selects vault custody and its token + // file is readable. + const insula = new ClaustrumConsumer({ + rosterPath: paths.roster, + tokenPath: getHostClaustrumEnrollmentPaths(options.host, env).tokenPath, + family: { + refreshAdapter: 'anthropic', + category: 'anthropic-native', + apiKeys: false, + }, + connect: async (): Promise => { + const transport = await custody.insulaTransport() + const unused = () => Promise.reject(new NativeCustodyError('unavailable')) + return { + listScoped: unused, + getScoped: unused, + reportAuthFailureScoped: unused, + // The connection belongs to this runtime's custody, which closes it. + close: () => {}, + ...(transport && { insula: transport }), + } + }, + isCustodyActive: () => active().catch(() => false), + insula: { provider: 'claude', normalize: normalizeInsulaClaudeQuota }, + ...(options.now && { now: options.now }), + logger: { + warn: () => {}, + debug: () => {}, + }, + }) + // The last source each route's guided quota reading came from, so a change + // is logged once rather than on every sweep. + const quotaSources = new Map() + function noteQuotaSource( + routeId: string, + source: 'insula' | 'fallback', + reason: 'insula-reading' | 'no-insula-reading' | 'account-changed', + ) { + if (quotaSources.get(routeId) === source) return + quotaSources.set(routeId, source) + if (quotaSources.size > 1000) { + const oldest = quotaSources.keys().next().value + if (oldest !== undefined) quotaSources.delete(oldest) + } + // Fixed codes only: no route, credential or account identity. + logger.info('quota', 'vault quota source changed', { source, reason }) + } const issued = new WeakMap< NativeCustodyReceipt, { routeId: string; logicalId: string } @@ -404,6 +483,10 @@ export function createNativeVaultRuntime( if (shutdown.signal.aborted || !allowed.some(Boolean)) return await runtime.refresh(shutdown.signal) if (shutdown.signal.aborted) return + // Keep insula's cache warm for admission lookups, which only read it. + // It refreshes at most once a minute and never throws. + await insula.refreshInsulaQuota() + if (shutdown.signal.aborted) return // Notifications belong to their observers. A slow or disposing host must // not hold discovery for the other leases; hosts drain their own writes. for (const watch of watchers) { @@ -758,8 +841,17 @@ export function createNativeVaultRuntime( }, })) }, - async fetchQuota(routeId, fetchImpl = fetch, signal, dispatch) { + async fetchQuota(routeId, fetchImpl = fetch, signal, dispatch, purpose) { const row = await rowFor(await runtime.read(), routeId) + // A background sweep asks insula about the roster row's account (its + // list_scoped account_id) before any credential is fetched. + let swept: InsulaClaudeQuota | undefined + if (purpose === 'background' && row) { + await insula.refreshInsulaQuota() + check(signal) + swept = await insula.cachedInsulaQuotaForRoute(row.routeId) + check(signal) + } const state = await readNativeRuntime(paths.runtime, paths.storageId) const before = state.status === 'ready' @@ -782,12 +874,44 @@ export function createNativeVaultRuntime( routeId, fetchImpl, signal, - (receipt, transport) => - fetchOAuthQuotaSnapshot({ + async (receipt, transport) => { + // A served receipt is still needed with an insula reading: the + // reading is published bound to this credential version. + let reading: InsulaClaudeQuota | undefined + let miss: 'no-insula-reading' | 'account-changed' = + 'no-insula-reading' + if (purpose === 'background') { + // The sweep joined the roster row's account; publish it only for + // a credential served for that same account. + if ( + swept && + row?.accountIdentity && + receipt.assertedAccountIdentity === row.accountIdentity + ) + reading = swept + else if (swept) miss = 'account-changed' + } else if (purpose === 'admission') { + reading = await insula.cachedInsulaQuota(receipt) + } + if (purpose) + noteQuotaSource( + routeId, + reading ? 'insula' : 'fallback', + reading ? 'insula-reading' : miss, + ) + if (reading) { + const stored = matchingBefore(receipt)?.quota + return mergeInsulaQuota( + reading, + stored ? fromNativeQuotaMap(stored) : undefined, + ) + } + return fetchOAuthQuotaSnapshot({ accessToken: receipt.accessToken, fetchImpl: transport, now: options.now, - }), + }) + }, async (error, receipt) => { const matched = matchingBefore(receipt) const failure = nativeQuotaFailure( @@ -893,6 +1017,7 @@ export function createNativeVaultRuntime( schedulePoll() if (shutdown.signal.aborted) return shutdown.abort(new NativeCustodyError('closed')) + insula.close() custody.close() }, } @@ -1266,8 +1391,8 @@ export function leaseNativeVaultRuntime( prepareRetry: (receipt, signal, site) => use().prepareRetry(receipt, signal, site), publish: (receipt, patch) => use().publish(receipt, patch), - fetchQuota: (id, transport, signal, dispatch) => - use().fetchQuota(id, transport, signal, dispatch), + fetchQuota: (id, transport, signal, dispatch, purpose) => + use().fetchQuota(id, transport, signal, dispatch, purpose), fetchProfile: (id, transport, signal) => use().fetchProfile(id, transport, signal), fetchProfileForDisplay: (id, transport, signal) => diff --git a/packages/core/src/quota-manager.ts b/packages/core/src/quota-manager.ts index 7f3af20a..f1494ae2 100644 --- a/packages/core/src/quota-manager.ts +++ b/packages/core/src/quota-manager.ts @@ -67,6 +67,14 @@ export type QuotaRefreshResult = { fetched: boolean } +/** + * Why a quota reading is wanted, passed to fetchQuotaSnapshot unchanged. + * Vault custody uses it to decide whether insula's background quota cache + * may answer instead of a usage GET (see NativeVaultQuotaPurpose); omitted + * means poll the provider directly. + */ +export type QuotaReadingPurpose = 'background' | 'admission' + /** Pace a physical usage GET and verify the selected credential is still eligible before sending. */ export type QuotaRequestDispatch = ( send: () => Promise, @@ -89,6 +97,7 @@ export type QuotaManagerOptions = { kind: 'main' | 'fallback' accountId: string | undefined accessToken: string + purpose?: QuotaReadingPurpose }, dispatch: QuotaRequestDispatch, ) => Promise @@ -406,6 +415,7 @@ export class QuotaManager { mainAccountIdOrAccessToken: string | undefined, accessToken?: string, expectedIdentityGeneration?: number, + purpose?: QuotaReadingPurpose, ): Promise { const mainAccountId = mainAccountIdOrAccessToken const credential = accessToken ?? mainAccountIdOrAccessToken @@ -416,6 +426,7 @@ export class QuotaManager { mainAccountId, credential, expectedIdentityGeneration, + purpose, ) ).quota } @@ -424,6 +435,7 @@ export class QuotaManager { mainAccountIdOrAccessToken: string | undefined, accessToken?: string, expectedIdentityGeneration?: number, + purpose?: QuotaReadingPurpose, ): Promise { const effectiveAccountId = mainAccountIdOrAccessToken const credential = accessToken ?? mainAccountIdOrAccessToken @@ -466,6 +478,7 @@ export class QuotaManager { credential ?? '', generation, inflightToken, + purpose, ) return this.inflightMain } @@ -474,9 +487,15 @@ export class QuotaManager { accountId: string, accessToken: string, account: Pick | undefined, + purpose?: QuotaReadingPurpose, ): Promise { return ( - await this.refreshFallbackWithMetadata(accountId, accessToken, account) + await this.refreshFallbackWithMetadata( + accountId, + accessToken, + account, + purpose, + ) ).quota } @@ -484,6 +503,7 @@ export class QuotaManager { accountId: string, accessToken: string, account: Pick | undefined, + purpose?: QuotaReadingPurpose, ): Promise { if (!accessToken && !this.canFetchWithoutAccessToken()) { throw new Error('Fallback OAuth access token is unavailable') @@ -506,6 +526,7 @@ export class QuotaManager { accessToken, generation, account, + purpose, ) this.inflightFallbacks.set(inflightKey, promise) void promise.then( @@ -1008,6 +1029,7 @@ export class QuotaManager { accessToken: string, generation: number, inflightToken: number, + purpose?: QuotaReadingPurpose, ): Promise { return this._enqueueApiFetch(async () => { try { @@ -1064,7 +1086,12 @@ export class QuotaManager { ) const quota = this.fetchQuotaSnapshot ? await this.fetchQuotaSnapshot( - { kind: 'main', accountId: mainAccountId, accessToken }, + { + kind: 'main', + accountId: mainAccountId, + accessToken, + ...(purpose && { purpose }), + }, dispatch, ) : await fetchOAuthQuotaSnapshot({ @@ -1163,6 +1190,7 @@ export class QuotaManager { accessToken: string, generation: number, account: Pick | undefined, + purpose?: QuotaReadingPurpose, ): Promise { return this._enqueueApiFetch(async () => { try { @@ -1204,7 +1232,12 @@ export class QuotaManager { ) const quota = this.fetchQuotaSnapshot ? await this.fetchQuotaSnapshot( - { kind: 'fallback', accountId, accessToken }, + { + kind: 'fallback', + accountId, + accessToken, + ...(purpose && { purpose }), + }, dispatch, ) : await fetchOAuthQuotaSnapshot({ diff --git a/packages/core/src/tests/native-vault-insula-quota.test.ts b/packages/core/src/tests/native-vault-insula-quota.test.ts new file mode 100644 index 00000000..95b90c2d --- /dev/null +++ b/packages/core/src/tests/native-vault-insula-quota.test.ts @@ -0,0 +1,634 @@ +import { expect, spyOn } from 'bun:test' +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { + ClaustrumConsumer, + type InsulaUsageEntry, + projectVaultRoster, +} from '@cortexkit/common-auth/claustrum' +import { + type AccountStorage, + getScopedQuotaWindowForModel, + killswitchPassesPolicy, + QUOTA_URL, + quotaSnapshotModelScopeIsExhausted, +} from '../accounts.ts' +import { + mergeInsulaQuota, + normalizeInsulaClaudeQuota, +} from '../insula-quota.ts' +import { + __setLogTestSink, + getLogLevel, + type LogTestRecord, + setLogLevel, +} from '../logger.ts' +import { createNativeAccountRuntime } from '../native-account-runtime.ts' +import type { NativeCustodyClient } from '../native-custody.ts' +import { nativeQuotaCodec } from '../native-quota-codec.ts' +import { readNativeRuntime } from '../native-runtime.ts' +import { nativeVaultPaths } from '../native-vault-paths.ts' +import { createNativeVaultRuntime } from '../native-vault-runtime.ts' +import { resolveNativePoolPaths } from '../pool-paths.ts' +import { createNativePoolStore } from '../pool-store.ts' +import { initializeNativeTestAuthority } from './native-authority-fixture.ts' +import { createTestLifetimeSuite } from './test-lifetime.ts' + +const { test, deferCleanup } = createTestLifetimeSuite() +const uuid = '11111111-2222-4333-8444-555555555555' +const uuid2 = '22222222-2222-4333-8444-555555555555' +const uuid3 = '33333333-2222-4333-8444-555555555555' +const now = 1_000_000 +const fetchedAt = now - 30_000 +type Row = Awaited< + ReturnType +>['rows'][number] +const row = (credentialId: string, accountId: string): Row => ({ + id: credentialId, + accountId, + credentialType: 'oauth', + categories: ['anthropic-native'], + refreshAdapter: 'anthropic', + serves: ['anthropic'], + providerIds: [], + state: 'active', + operations: ['read'], + recordVersion: 7, + createdAtMs: null, +}) +const mainRow = row('oauth:anthropic', uuid) +const workRow = row('oauth:anthropic:work', uuid2) + +/** A claude entry in the shape insula's quota monitor publishes. */ +function entry( + account: string, + overrides: Partial = {}, +): InsulaUsageEntry { + return { + provider: 'claude', + account: ` ${account} `, + fetchedAt: new Date(fetchedAt).toISOString(), + source: 'vault', + usage: { + primary: { + usedPercent: 40, + resetsAt: '1970-01-01T05:00:00.000Z', + windowKind: 'five_hour', + windowMinutes: 300, + }, + secondary: { + usedPercent: 55, + resetsAt: '1970-01-08T00:00:00.000Z', + windowKind: 'weekly', + windowMinutes: 10_080, + breakdown: [{ title: 'ignored' }], + }, + extraRateWindows: [ + { + id: 'Fable', + title: 'Weekly ยท Opus only', + window: { + usedPercent: 97, + resetsAt: '1970-01-08T00:00:00.000Z', + windowKind: 'weekly', + windowMinutes: 10_080, + is_active: false, + }, + }, + ], + }, + ...overrides, + } +} + +const usageBody = { + five_hour: { utilization: 10, resets_at: '1970-01-01T05:00:00.000Z' }, + seven_day: { utilization: 20, resets_at: '1970-01-08T00:00:00.000Z' }, +} + +async function fixture(entries: () => unknown[]) { + const root = await mkdtemp(join(tmpdir(), 'native-vault-insula-')) + deferCleanup(() => rm(root, { recursive: true, force: true })) + const paths = await resolveNativePoolPaths(join(root, 'anthropic-auth.json')) + await initializeNativeTestAuthority(paths) + const vaultPaths = nativeVaultPaths(paths) + const env = { + OPENCODE_ANTHROPIC_AUTH_CLAUSTRUM_ENROLLMENT_FILE: join( + root, + 'opencode-enrollment.json', + ), + } + await writeFile( + env.OPENCODE_ANTHROPIC_AUTH_CLAUSTRUM_ENROLLMENT_FILE, + JSON.stringify({ token: '01'.repeat(32), token_generation: 1 }), + { mode: 0o600 }, + ) + const store = createNativePoolStore({ + paths: vaultPaths, + quota: nativeQuotaCodec, + }) + await store.initialize() + const projected = projectVaultRoster( + undefined, + { + view: 'initial', + credentials: [ + { + credentialId: mainRow.id, + credentialType: 'oauth', + accountIdentity: uuid, + state: 'active', + }, + ], + skipped: [], + }, + { now }, + ) + const primaryRouteId = projected.rows[0]?.routeId + if (!primaryRouteId) throw new Error('Missing fixture primary route') + await store.updateSettings((settings) => { + settings.mainAccountId = primaryRouteId + settings.claustrum = { + primaryAccount: { + credentialId: mainRow.id, + accountId: uuid, + state: 'active', + }, + } + }) + const rows: readonly Row[] = [mainRow, workRow] + const clock = { now } + const counts = { insula: 0, usage: [] as string[] } + const client: NativeCustodyClient = { + listScoped: async () => ({ + view: 'view-2', + rows, + grants: undefined, + grantTuples: undefined, + }), + getScoped: async (input) => ({ + credentialId: input.credentialId, + accountId: rows.find((entry) => entry.id === input.credentialId) + ?.accountId, + material: `synthetic-bearer-${input.credentialId}`, + recordVersion: 7, + expiresAtMs: now + 3_600_000, + }), + reportAuthFailureScoped: async () => {}, + close: () => {}, + insula: { + // Unique per fixture, so common-auth's shared once-a-minute budget + // never carries over between tests. + connectionFile: join(root, 'daemon.json'), + getUsage: async () => { + counts.insula++ + return { result: entries() } + }, + }, + } + const usage: typeof fetch = Object.assign( + async (input: Parameters[0], init?: RequestInit) => { + if (String(input) !== QUOTA_URL) throw new Error('unexpected request') + counts.usage.push(new Headers(init?.headers).get('authorization') ?? '') + return Response.json(usageBody) + }, + { preconnect: fetch.preconnect }, + ) + const runtime = createNativeVaultRuntime({ + paths, + host: 'opencode', + env, + now: () => clock.now, + connect: async () => client, + readToken: async () => ({ token: '01'.repeat(32), token_generation: 1 }), + }) + deferCleanup(async () => { + await runtime.drainQuotaPublications() + runtime.close() + }) + await runtime.refresh() + const roster = await runtime.read() + const routeOf = (account: string) => { + const id = roster?.rows.find((r) => r.accountIdentity === account)?.routeId + if (!id) throw new Error('Missing fixture route') + return id + } + return { runtime, counts, usage, vaultPaths, routeOf, clock } +} + +const bearer = (credentialId: string) => + `Bearer synthetic-bearer-${credentialId}` + +function captureLogs() { + const records: LogTestRecord[] = [] + const previous = getLogLevel() + setLogLevel('debug') + __setLogTestSink((record) => records.push(record)) + deferCleanup(() => { + __setLogTestSink(null) + setLogLevel(previous) + }) + return records +} + +test('background sweep takes a fresh insula reading and sends no usage GET', async () => { + const f = await fixture(() => [entry(uuid), entry(uuid2)]) + const quota = await f.runtime.fetchQuota( + f.routeOf(uuid2), + f.usage, + undefined, + undefined, + 'background', + ) + expect(f.counts.usage).toEqual([]) + expect(f.counts.insula).toBe(1) + expect(quota.accountIdentity).toBe(uuid2) + expect(quota.five_hour).toMatchObject({ + usedPercent: 40, + remainingPercent: 60, + }) + expect(quota.seven_day).toMatchObject({ + usedPercent: 55, + remainingPercent: 45, + }) + // Every window is dated by insula's fetch, never by the sweep's clock. + expect(quota.checkedAt).toBe(fetchedAt) + expect( + [quota.five_hour, quota.seven_day, ...(quota.scoped ?? [])].map( + (window) => window?.checkedAt, + ), + ).toEqual([fetchedAt, fetchedAt, fetchedAt]) + // Published through the native quota path, bound to the served credential. + await f.runtime.drainQuotaPublications() + const state = await readNativeRuntime( + f.vaultPaths.runtime, + f.vaultPaths.storageId, + ) + const stored = + state.status === 'ready' + ? state.state.accounts[f.routeOf(uuid2)] + : undefined + expect(stored?.binding).toMatchObject({ + kind: 'custody', + credentialId: workRow.id, + accountIdentity: uuid2, + recordVersion: 7, + }) + expect(stored?.quotaCheckedAt).toBe(fetchedAt) +}) + +test('a stale, missing or failed insula entry falls back to one usage GET for that account only', async () => { + for (const work of [ + entry(uuid2, { stale: true }), + entry(uuid2, { errorClass: 'rate_limited' }), + entry(uuid2, { fetchedAt: new Date(now - 5 * 60_000).toISOString() }), + undefined, + ]) { + const f = await fixture(() => [entry(uuid), ...(work ? [work] : [])]) + await f.runtime.fetchQuota( + f.routeOf(uuid), + f.usage, + undefined, + undefined, + 'background', + ) + expect(f.counts.usage).toEqual([]) + const quota = await f.runtime.fetchQuota( + f.routeOf(uuid2), + f.usage, + undefined, + undefined, + 'background', + ) + expect(f.counts.usage).toEqual([bearer(workRow.id)]) + expect(quota.five_hour).toMatchObject({ usedPercent: 10, checkedAt: now }) + } +}) + +test('an insula entry for a different account never applies', async () => { + const f = await fixture(() => [ + entry(uuid3), + // Same account, wrong provider. + { ...entry(uuid2), provider: 'codex' }, + ]) + const quota = await f.runtime.fetchQuota( + f.routeOf(uuid2), + f.usage, + undefined, + undefined, + 'background', + ) + expect(f.counts.insula).toBe(1) + expect(f.counts.usage).toEqual([bearer(workRow.id)]) + expect(quota.five_hour?.usedPercent).toBe(10) +}) + +test('admission reads only the insula cache: never refreshes it, polls on a miss', async () => { + const refresh = spyOn(ClaustrumConsumer.prototype, 'refreshInsulaQuota') + const cached = spyOn(ClaustrumConsumer.prototype, 'cachedInsulaQuota') + deferCleanup(() => { + refresh.mockRestore() + cached.mockRestore() + }) + const f = await fixture(() => [entry(uuid2)]) + // Cold cache: no insula fetch on the request path, so the usage poll runs. + await f.runtime.fetchQuota( + f.routeOf(uuid2), + f.usage, + undefined, + undefined, + 'admission', + ) + expect(refresh).toHaveBeenCalledTimes(0) + expect(cached).toHaveBeenCalledTimes(1) + expect(f.counts.insula).toBe(0) + expect(f.counts.usage).toEqual([bearer(workRow.id)]) + // A background tick warms the cache; admission then answers from it. + await f.runtime.fetchQuota( + f.routeOf(uuid), + f.usage, + undefined, + undefined, + 'background', + ) + expect(refresh).toHaveBeenCalledTimes(1) + expect(f.counts.insula).toBe(1) + const callsBefore = refresh.mock.calls.length + const quota = await f.runtime.fetchQuota( + f.routeOf(uuid2), + f.usage, + undefined, + undefined, + 'admission', + ) + expect(refresh.mock.calls.length).toBe(callsBefore) + expect(cached).toHaveBeenCalledTimes(2) + expect(f.counts.insula).toBe(1) + expect(f.counts.usage).toEqual([bearer(workRow.id), bearer(mainRow.id)]) + expect(quota.five_hour?.checkedAt).toBe(fetchedAt) +}) + +test('a quota read without a purpose polls Anthropic and leaves insula alone', async () => { + const f = await fixture(() => [entry(uuid2)]) + await f.runtime.fetchQuota(f.routeOf(uuid2), f.usage) + expect(f.counts.insula).toBe(0) + expect(f.counts.usage).toEqual([bearer(workRow.id)]) +}) + +test('local mode never touches insula', async () => { + const root = await mkdtemp(join(tmpdir(), 'native-local-insula-')) + deferCleanup(() => rm(root, { recursive: true, force: true })) + const paths = await resolveNativePoolPaths(join(root, 'anthropic-auth.json')) + await initializeNativeTestAuthority(paths) + const store = createNativePoolStore({ + paths, + quota: nativeQuotaCodec, + now: () => now, + }) + await store.initialize() + await store.add({ + id: 'imported-main', + identity: uuid, + credential: { + type: 'oauth', + access: 'sk-ant-oat01-synthetic-access', + refresh: 'synthetic-refresh', + expires: now + 3_600_000, + }, + }) + await store.updateSettings((settings) => { + settings.mainAccountId = 'imported-main' + }) + const counts = { insula: 0, connects: 0, usage: 0 } + const runtime = createNativeAccountRuntime({ + paths, + host: 'opencode', + vault: { + env: { + OPENCODE_ANTHROPIC_AUTH_CLAUSTRUM_ENROLLMENT_FILE: join( + root, + 'none.json', + ), + }, + connect: async () => { + counts.connects++ + throw new Error('local mode must not connect') + }, + }, + local: { + now: () => now, + resolveIdentity: async () => ({ + deviceId: 'synthetic-device', + sessionId: 'synthetic-session', + accountUuid: uuid, + }), + refreshToken: async () => { + throw new Error('no refresh expected') + }, + }, + }) + deferCleanup(async () => { + await runtime.drainQuotaPublications() + runtime.close() + }) + const refresh = spyOn(ClaustrumConsumer.prototype, 'refreshInsulaQuota') + const forRoute = spyOn( + ClaustrumConsumer.prototype, + 'cachedInsulaQuotaForRoute', + ) + const cached = spyOn(ClaustrumConsumer.prototype, 'cachedInsulaQuota') + deferCleanup(() => { + refresh.mockRestore() + forRoute.mockRestore() + cached.mockRestore() + }) + const usage: typeof fetch = Object.assign( + async () => { + counts.usage++ + return Response.json(usageBody) + }, + { preconnect: fetch.preconnect }, + ) + for (const purpose of ['background', 'admission'] as const) { + await runtime.fetchQuota('main', usage, undefined, undefined, purpose) + } + expect(counts.usage).toBe(2) + expect(counts.connects).toBe(0) + expect(refresh).toHaveBeenCalledTimes(0) + expect(forRoute).toHaveBeenCalledTimes(0) + expect(cached).toHaveBeenCalledTimes(0) +}) + +test('the Fable window from extraRateWindows drives scoped routing and the scoped killswitch floor', async () => { + const f = await fixture(() => [entry(uuid2)]) + const quota = await f.runtime.fetchQuota( + f.routeOf(uuid2), + f.usage, + undefined, + undefined, + 'background', + ) + expect(f.counts.usage).toEqual([]) + // Keyed exactly as the usage poll keys Anthropic's model-scoped limit. + expect(quota.scoped).toEqual([ + { + id: 'claude-weekly-scoped-fable', + title: 'Fable only', + modelName: 'Fable', + usedPercent: 97, + remainingPercent: 3, + resetsAt: '1970-01-08T00:00:00.000Z', + checkedAt: fetchedAt, + }, + ]) + expect(getScopedQuotaWindowForModel(quota, 'claude-fable-5-1')?.id).toBe( + 'claude-weekly-scoped-fable', + ) + expect( + getScopedQuotaWindowForModel(quota, 'claude-sonnet-4-5'), + ).toBeUndefined() + const storage: AccountStorage = { + version: 1, + main: { type: 'opencode', provider: 'anthropic' }, + accounts: [], + killswitch: { + enabled: true, + main: { five_hour: 5, seven_day: 5, scoped: 10 }, + }, + } + expect( + killswitchPassesPolicy(quota, storage, undefined, 'claude-fable-5-1'), + ).toBe(false) + expect( + killswitchPassesPolicy(quota, storage, undefined, 'claude-sonnet-4-5'), + ).toBe(true) + // An exhausted Fable reading blocks that model's route only. + const exhausted = normalizeInsulaClaudeQuota( + entry(uuid2, { + usage: { + ...entry(uuid2).usage, + extraRateWindows: [{ id: 'Fable', window: { usedPercent: 100 } }], + }, + }), + { fetchedAt }, + ) + expect( + quotaSnapshotModelScopeIsExhausted(exhausted, 'claude-fable-5-1'), + ).toBe(true) + expect(quotaSnapshotModelScopeIsExhausted(exhausted, 'claude-opus-4-8')).toBe( + false, + ) +}) + +test('the source switch is logged once per change with fixed codes and no identity', async () => { + const records = captureLogs() + let entries = [entry(uuid2)] + const f = await fixture(() => entries) + const sweep = () => + f.runtime.fetchQuota( + f.routeOf(uuid2), + f.usage, + undefined, + undefined, + 'background', + ) + await sweep() + await sweep() + const switches = () => + records + .filter((record) => record.message === 'vault quota source changed') + .map((record) => record.payload) + expect(switches()).toEqual([{ source: 'insula', reason: 'insula-reading' }]) + // A minute later insula no longer lists the account: one fallback line. + entries = [] + f.clock.now = now + 61_000 + await sweep() + await sweep() + expect(f.counts.insula).toBe(2) + expect(f.counts.usage).toEqual([bearer(workRow.id), bearer(workRow.id)]) + expect(switches()).toEqual([ + { source: 'insula', reason: 'insula-reading' }, + { source: 'fallback', reason: 'no-insula-reading' }, + ]) + const after = records.filter( + (record) => record.message === 'vault quota source changed', + ) + const text = JSON.stringify(after) + for (const secret of [ + uuid, + uuid2, + mainRow.id, + workRow.id, + f.routeOf(uuid), + f.routeOf(uuid2), + ]) + expect(text).not.toContain(secret) +}) + +test('normalizer: unknown shapes return undefined so the poll stays in charge', () => { + expect( + normalizeInsulaClaudeQuota( + entry(uuid, { usage: { primary: entry(uuid).usage?.primary } }), + { fetchedAt }, + ), + ).toBeUndefined() + expect( + normalizeInsulaClaudeQuota( + entry(uuid, { + usage: { + ...entry(uuid).usage, + primary: { usedPercent: 40, windowKind: 'weekly' }, + }, + }), + { fetchedAt }, + ), + ).toBeUndefined() + expect( + normalizeInsulaClaudeQuota(entry(uuid, { stale: 'old' }), { fetchedAt }), + ).toBeUndefined() + expect( + normalizeInsulaClaudeQuota(entry(uuid), { fetchedAt: Number.NaN }), + ).toBeUndefined() +}) + +test('normalizer maps the Extra usage pool only when present', () => { + const plain = normalizeInsulaClaudeQuota(entry(uuid), { fetchedAt }) + expect(plain?.extraUsage).toBeUndefined() + const pooled = normalizeInsulaClaudeQuota( + entry(uuid, { + pools: [ + { + title: 'Extra usage', + total: { minor: 5_000, exponent: 2, unit: 'USD' }, + remaining: { minor: 1_250, exponent: 2, unit: 'USD' }, + spendable: true, + }, + ], + }), + { fetchedAt }, + ) + expect(pooled?.extraUsage).toEqual({ + used: { amountMinor: 3_750, currency: 'USD', exponent: 2 }, + limit: { amountMinor: 5_000, currency: 'USD', exponent: 2 }, + utilizationPercent: 75, + exhausted: false, + }) +}) + +test('a newer stored window survives an older insula reading', () => { + const reading = normalizeInsulaClaudeQuota(entry(uuid), { fetchedAt }) + if (!reading) throw new Error('expected a reading') + const merged = mergeInsulaQuota(reading, { + five_hour: { usedPercent: 80, remainingPercent: 20, checkedAt: now }, + source: 'headers', + checkedAt: now, + }) + expect(merged.five_hour).toMatchObject({ usedPercent: 80, checkedAt: now }) + expect(merged.seven_day).toMatchObject({ + usedPercent: 55, + checkedAt: fetchedAt, + }) + expect(merged.scoped?.[0]?.checkedAt).toBe(fetchedAt) + expect(merged.checkedAt).toBe(now) +}) diff --git a/packages/opencode/src/index.ts b/packages/opencode/src/index.ts index 95598f93..26ba9bea 100644 --- a/packages/opencode/src/index.ts +++ b/packages/opencode/src/index.ts @@ -172,6 +172,7 @@ import { QuotaHeaderFeedRegistry, QuotaManager, type QuotaManagerOptions, + type QuotaReadingPurpose, type QuotaState, quotaSnapshotCheckedAt, quotaSnapshotHasStandardWindows, @@ -1817,7 +1818,13 @@ const anthropicAuthPlugin = async ( assertNativeEnvironment() const routeId = request.kind === 'main' ? 'main' : request.accountId if (!routeId) throw new Error('Quota request has no account identity') - return nativeAccounts.fetchQuota(routeId, undefined, undefined, dispatch) + return nativeAccounts.fetchQuota( + routeId, + undefined, + undefined, + dispatch, + request.purpose, + ) }, }) function quotaSnapshotIsExhausted( @@ -2834,16 +2841,20 @@ const anthropicAuthPlugin = async ( } let fallbackQuotaRefreshQueue: Promise = Promise.resolve() + // `purpose` tells vault custody whether this sweep is the background tick + // or a request's admission, which decides how insula's quota cache is used. + // Explicit refreshes leave it unset and always poll Anthropic. function refreshFallbackQuotas( force = false, requiredByPolicy = false, modelId?: string, + purpose?: QuotaReadingPurpose, ) { // Startup and request preflight can both observe missing quota before // either poll finishes. Serialize the read/check/poll sequence so the // next sweep sees the preceding result. Explicit refresh still polls. const operation = fallbackQuotaRefreshQueue.then(() => - pollFallbackQuotas(force, requiredByPolicy, modelId), + pollFallbackQuotas(force, requiredByPolicy, modelId, purpose), ) fallbackQuotaRefreshQueue = operation.then( () => {}, @@ -2855,6 +2866,7 @@ const anthropicAuthPlugin = async ( force: boolean, requiredByPolicy: boolean, modelId?: string, + purpose?: QuotaReadingPurpose, ) { const snapshot = await nativeAccounts.read() const unavailable = new Set() @@ -2894,6 +2906,7 @@ const anthropicAuthPlugin = async ( (candidate) => candidate.id === account.id && isOAuthAccount(candidate), ) as OAuthAccount | undefined, + force ? undefined : purpose, ) if (!reading) { unavailable.add(account.id) @@ -3079,7 +3092,7 @@ const anthropicAuthPlugin = async ( } } } - await refreshFallbackQuotas() + await refreshFallbackQuotas(false, false, undefined, 'background') } const canStartNativeServices = initialStorage !== null && process.env.OPENCODE_AUTH_CONTENT === undefined @@ -3101,7 +3114,8 @@ const anthropicAuthPlugin = async ( knownVaultFallbacks = current // Discovery and request preflight share the due sweep, so an arriving // account gets its first reading without duplicating an in-flight poll. - if (discovered) await refreshFallbackQuotas() + if (discovered) + await refreshFallbackQuotas(false, false, undefined, 'background') await refreshSidebarQuota() } const vaultWatch = @@ -7830,6 +7844,7 @@ const anthropicAuthPlugin = async ( false, requiredByPolicy, options.modelId, + 'admission', ) // Polling can also bind a newly discovered account UUID. Filter // the fresh native overlay, not the request's earlier account view. @@ -7932,6 +7947,7 @@ const anthropicAuthPlugin = async ( mainQuotaIdentity, input.mainAccessToken, input.mainQuotaIdentity?.generation, + 'admission', ) } catch {} } @@ -9631,6 +9647,7 @@ const anthropicAuthPlugin = async ( requestMainQuotaIdentity?.quotaKey, auth.access, requestMainQuotaIdentity?.generation, + 'admission', ) routingQuotaEntry = routingMainQuotaEntry( requestMainQuotaIdentity, @@ -9810,7 +9827,12 @@ const anthropicAuthPlugin = async ( ), // Fallback-first may have just polled these accounts. // Keep those fresh readings rather than polling twice. - refreshFallbackQuotas(false, true), + refreshFallbackQuotas( + false, + true, + undefined, + 'admission', + ), ]) } catch (error) { log('[quota] killswitch refresh failed', { diff --git a/packages/opencode/src/v2/lifecycle.ts b/packages/opencode/src/v2/lifecycle.ts index 86b65571..e64fed08 100644 --- a/packages/opencode/src/v2/lifecycle.ts +++ b/packages/opencode/src/v2/lifecycle.ts @@ -41,8 +41,16 @@ export function startNativeAnthropicLifecycle( ) { // Each quota reading or failure is bound to the credential version // used for its request, so a late result cannot affect a newer login. + // The background tick: vault custody takes a current insula + // reading instead of a usage GET when one exists for this account. await runtime - .fetchQuota(account.id, fetchImpl, abort.signal) + .fetchQuota( + account.id, + fetchImpl, + abort.signal, + undefined, + 'background', + ) .catch(() => {}) } if (!oauthProfileIsFresh(account.profile) && !abort.signal.aborted) diff --git a/packages/opencode/src/v2/source.ts b/packages/opencode/src/v2/source.ts index b247c41f..24709613 100644 --- a/packages/opencode/src/v2/source.ts +++ b/packages/opencode/src/v2/source.ts @@ -287,7 +287,15 @@ export function createNativeAnthropicSource( getQuotaCheckIntervalMs(snapshot.policyStorage) if (fresh) continue try { - await runtime.fetchQuota(account.id, options.fetch, active.signal) + // Admission: vault custody may answer from insula's cache, then + // polls on a miss; it never refreshes that cache here. + await runtime.fetchQuota( + account.id, + options.fetch, + active.signal, + undefined, + 'admission', + ) refreshed = true } catch { active.signal.throwIfAborted() diff --git a/packages/pi/src/stream.ts b/packages/pi/src/stream.ts index f52aeb4a..7967c3b6 100644 --- a/packages/pi/src/stream.ts +++ b/packages/pi/src/stream.ts @@ -159,6 +159,7 @@ function getPiRoutingServices( fetchPiOAuth, undefined, dispatch, + request.purpose, ) }, }) @@ -208,7 +209,7 @@ function getPiRoutingServices( !quotaBackoffActive(account.lastQuotaRefreshError, Date.now()) ) { try { - await quotas.refreshFallback(account.id, '', account) + await quotas.refreshFallback(account.id, '', account, 'admission') } catch (error) { // A transient account error means another writer is still saving // this fallback. Rethrow it: swallowing it would leave the fallback @@ -1113,7 +1114,12 @@ async function executeWithFallback(options: { ) ) { try { - mainQuota = await quotaManager.refreshMain(mainAccountId, '') + mainQuota = await quotaManager.refreshMain( + mainAccountId, + '', + undefined, + 'admission', + ) } catch {} } const usableFallbacks = await getUsableFallbackAccounts(storage, { @@ -1145,6 +1151,7 @@ async function executeWithFallback(options: { account.id, '', account, + 'admission', ) } catch {} } @@ -1601,7 +1608,12 @@ async function executeWithFallback(options: { let mainQuota = quotaManager.getMain(mainAccountId)?.quota if (!mainQuota || quotaManager.isMainStale(options.model.id)) { try { - mainQuota = await quotaManager.refreshMain(mainAccountId, '') + mainQuota = await quotaManager.refreshMain( + mainAccountId, + '', + undefined, + 'admission', + ) } catch {} } if ( From 7cab329a9c7c061030a8b25e145bdd1ead2d7190 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Sun, 18 Oct 2026 00:15:34 +0200 Subject: [PATCH 2/9] mason: insula quota tests, mutation rows and comment fixes --- mutations.toml | 26 +++++++++++++++++++ packages/core/src/insula-quota.ts | 2 +- packages/core/src/native-vault-runtime.ts | 3 ++- .../tests/native-vault-insula-quota.test.ts | 19 +++++++++++++- packages/opencode/src/index.ts | 8 +++--- packages/opencode/src/v2/lifecycle.ts | 5 ++-- packages/opencode/src/v2/source.ts | 3 ++- 7 files changed, 57 insertions(+), 9 deletions(-) diff --git a/mutations.toml b/mutations.toml index a4604ce3..578e8691 100644 --- a/mutations.toml +++ b/mutations.toml @@ -1058,6 +1058,18 @@ command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/quota- test_count_pattern = "CKDEV executed {count} tests" expect_red = ["physical quota GET starts stay one second apart when the first GET fails"] expect_message = "error: expect(received).toBeGreaterThanOrEqual(expected)" +[[control]] +id = "vault-insula-joins-roster-route-account" +guards = "A background insula lookup must join the account of the roster row the route resolves to; the primary account's logical id names no roster row and would never join." +file = "packages/core/src/native-vault-runtime.ts" +old = "swept = await insula.cachedInsulaQuotaForRoute(row.routeId)" +new = "swept = await insula.cachedInsulaQuotaForRoute(routeId) // NON-VACUITY BREAK: join on the logical id" +test_file = "packages/core/src/tests/native-vault-insula-quota.test.ts" +runner = "command" +command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/native-vault-insula-quota.test.ts", "{test}"] +test_count_pattern = "CKDEV executed {count} tests" +expect_red = ["the primary account is looked up in insula by its roster route, not its logical id"] +expect_message = "error: expect(received).toEqual(expected)" broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"] broad_report = "tmp/mutations/core.xml" broad_id = "{name}" @@ -1074,6 +1086,20 @@ command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/quota- test_count_pattern = "CKDEV executed {count} tests" expect_red = ["physical quota GET starts stay one second apart when the first GET fails", "the second physical quota GET first requests the rest of the one-second gap"] expect_message = '/error: expect\(received\)\.(toBeGreaterThanOrEqual|toEqual)\(expected\)/' +[[control]] +id = "vault-insula-reading-skips-usage-poll" +guards = "A current insula reading for a vault account must replace that account's usage GET; only a miss polls Anthropic." +file = "packages/core/src/native-vault-runtime.ts" +old = ''' if (reading) { + const stored''' +new = ''' if (reading && false) { // NON-VACUITY BREAK: always poll + const stored''' +test_file = "packages/core/src/tests/native-vault-insula-quota.test.ts" +runner = "command" +command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/native-vault-insula-quota.test.ts", "{test}"] +test_count_pattern = "CKDEV executed {count} tests" +expect_red = ["background sweep takes a fresh insula reading and sends no usage GET", "the primary account is looked up in insula by its roster route, not its logical id", "a stale, missing or failed insula entry falls back to one usage GET for that account only", "admission reads only the insula cache: never refreshes it, polls on a miss", "the Fable window from extraRateWindows drives scoped routing and the scoped killswitch floor", "the source switch is logged once per change with fixed codes and no identity"] +expect_message = "error: expect(received).toEqual(expected)" broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"] broad_report = "tmp/mutations/core.xml" broad_id = "{name}" diff --git a/packages/core/src/insula-quota.ts b/packages/core/src/insula-quota.ts index 2a4e8765..e29dbd2c 100644 --- a/packages/core/src/insula-quota.ts +++ b/packages/core/src/insula-quota.ts @@ -28,7 +28,7 @@ function text(value: unknown): string | undefined { return typeof value === 'string' && value.trim() ? value.trim() : undefined } -// Same bounds as the usage poll's mapping in accounts.ts. +// Keep percentages within 0-100, as the usage poll's mapping in accounts.ts does. function clampPercent(value: number) { return Math.min(100, Math.max(0, value)) } diff --git a/packages/core/src/native-vault-runtime.ts b/packages/core/src/native-vault-runtime.ts index c1a5d25d..68317b84 100644 --- a/packages/core/src/native-vault-runtime.ts +++ b/packages/core/src/native-vault-runtime.ts @@ -404,7 +404,8 @@ export function createNativeVaultRuntime( listScoped: unused, getScoped: unused, reportAuthFailureScoped: unused, - // The connection belongs to this runtime's custody, which closes it. + // Nothing to close here: the connection belongs to this runtime's + // custody, which closes it when the runtime closes. close: () => {}, ...(transport && { insula: transport }), } diff --git a/packages/core/src/tests/native-vault-insula-quota.test.ts b/packages/core/src/tests/native-vault-insula-quota.test.ts index 95b90c2d..e5e303d1 100644 --- a/packages/core/src/tests/native-vault-insula-quota.test.ts +++ b/packages/core/src/tests/native-vault-insula-quota.test.ts @@ -278,6 +278,20 @@ test('background sweep takes a fresh insula reading and sends no usage GET', asy expect(stored?.quotaCheckedAt).toBe(fetchedAt) }) +test('the primary account is looked up in insula by its roster route, not its logical id', async () => { + const f = await fixture(() => [entry(uuid)]) + const quota = await f.runtime.fetchQuota( + 'main', + f.usage, + undefined, + undefined, + 'background', + ) + expect(f.counts.usage).toEqual([]) + expect(quota.accountIdentity).toBe(uuid) + expect(quota.five_hour?.checkedAt).toBe(fetchedAt) +}) + test('a stale, missing or failed insula entry falls back to one usage GET for that account only', async () => { for (const work of [ entry(uuid2, { stale: true }), @@ -366,7 +380,10 @@ test('admission reads only the insula cache: never refreshes it, polls on a miss expect(cached).toHaveBeenCalledTimes(2) expect(f.counts.insula).toBe(1) expect(f.counts.usage).toEqual([bearer(workRow.id), bearer(mainRow.id)]) - expect(quota.five_hour?.checkedAt).toBe(fetchedAt) + // The cold poll above is newer than insula's fetch, so its standard windows + // stay; the Fable window only insula reported is added. + expect(quota.five_hour).toMatchObject({ usedPercent: 10, checkedAt: now }) + expect(quota.scoped?.map((window) => window.checkedAt)).toEqual([fetchedAt]) }) test('a quota read without a purpose polls Anthropic and leaves insula alone', async () => { diff --git a/packages/opencode/src/index.ts b/packages/opencode/src/index.ts index 26ba9bea..acc7d823 100644 --- a/packages/opencode/src/index.ts +++ b/packages/opencode/src/index.ts @@ -2841,9 +2841,11 @@ const anthropicAuthPlugin = async ( } let fallbackQuotaRefreshQueue: Promise = Promise.resolve() - // `purpose` tells vault custody whether this sweep is the background tick - // or a request's admission, which decides how insula's quota cache is used. - // Explicit refreshes leave it unset and always poll Anthropic. + // `purpose` tells the vault runtime whether this sweep is the background + // tick or a request's admission. That decides how it uses insula (the + // vault daemon's quota monitor) instead of a usage GET; see + // NativeVaultQuotaPurpose. Explicit refreshes leave it unset and always + // poll Anthropic. function refreshFallbackQuotas( force = false, requiredByPolicy = false, diff --git a/packages/opencode/src/v2/lifecycle.ts b/packages/opencode/src/v2/lifecycle.ts index e64fed08..944a2ecb 100644 --- a/packages/opencode/src/v2/lifecycle.ts +++ b/packages/opencode/src/v2/lifecycle.ts @@ -41,8 +41,9 @@ export function startNativeAnthropicLifecycle( ) { // Each quota reading or failure is bound to the credential version // used for its request, so a late result cannot affect a newer login. - // The background tick: vault custody takes a current insula - // reading instead of a usage GET when one exists for this account. + // This is the background tick: for a vault account the runtime uses + // a current reading from insula (the vault daemon's quota monitor) + // instead of a usage GET when one exists; see NativeVaultQuotaPurpose. await runtime .fetchQuota( account.id, diff --git a/packages/opencode/src/v2/source.ts b/packages/opencode/src/v2/source.ts index 24709613..75009cc0 100644 --- a/packages/opencode/src/v2/source.ts +++ b/packages/opencode/src/v2/source.ts @@ -287,7 +287,8 @@ export function createNativeAnthropicSource( getQuotaCheckIntervalMs(snapshot.policyStorage) if (fresh) continue try { - // Admission: vault custody may answer from insula's cache, then + // Admission: for a vault account the runtime may answer from its + // cached insula reading (the vault daemon's quota monitor), then // polls on a miss; it never refreshes that cache here. await runtime.fetchQuota( account.id, From 397ef05fab3bbdeade2616f8cefc89d84c8606f1 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Sun, 18 Oct 2026 00:37:08 +0200 Subject: [PATCH 3/9] mason: keep native custody declarations free of common-auth types; settle insula test publication --- packages/core/src/native-custody.ts | 15 ++++++++++++--- .../src/tests/native-vault-insula-quota.test.ts | 5 ++++- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/packages/core/src/native-custody.ts b/packages/core/src/native-custody.ts index ddb8b02c..318f9756 100644 --- a/packages/core/src/native-custody.ts +++ b/packages/core/src/native-custody.ts @@ -1,10 +1,19 @@ import { ClaustrumConsumerError, ClaustrumScopedCustody, - type InsulaUsageTransport, } from '@cortexkit/common-auth/claustrum' import { getHostClaustrumEnrollmentPaths } from './claustrum-enrollment.ts' +/** + * Reads insula's quota cache from the vault daemon. Declared here with the + * same shape as common-auth's InsulaUsageTransport so this module's published + * declarations do not depend on common-auth. + */ +export interface NativeInsulaUsageTransport { + connectionFile: string + getUsage(): Promise +} + /** Claustrum IPC operations for listing accounts, reading tokens, reporting rejection, and closing the connection. */ export interface NativeCustodyClient { listScoped: (enrollmentToken?: string) => Promise<{ @@ -65,7 +74,7 @@ export interface NativeCustodyClient { * when the connection was made with connectClaustrumScopedClient; reading * it opens a separate short-lived socket and never uses this connection. */ - insula?: InsulaUsageTransport + insula?: NativeInsulaUsageTransport } export type NativeCustodyReporterSource = @@ -227,7 +236,7 @@ export interface NativeCustody { signal?: AbortSignal, ) => Promise /** The shared connection's insula transport, connecting first if needed; undefined when the client has none. */ - insulaTransport: () => Promise + insulaTransport: () => Promise close: () => void } diff --git a/packages/core/src/tests/native-vault-insula-quota.test.ts b/packages/core/src/tests/native-vault-insula-quota.test.ts index e5e303d1..674e4db3 100644 --- a/packages/core/src/tests/native-vault-insula-quota.test.ts +++ b/packages/core/src/tests/native-vault-insula-quota.test.ts @@ -11,6 +11,7 @@ import { type AccountStorage, getScopedQuotaWindowForModel, killswitchPassesPolicy, + type ProviderAccountUuid, QUOTA_URL, quotaSnapshotModelScopeIsExhausted, } from '../accounts.ts' @@ -358,6 +359,8 @@ test('admission reads only the insula cache: never refreshes it, polls on a miss expect(cached).toHaveBeenCalledTimes(1) expect(f.counts.insula).toBe(0) expect(f.counts.usage).toEqual([bearer(workRow.id)]) + // Let the poll's reading reach stored quota before insula's is merged in. + await f.runtime.drainQuotaPublications() // A background tick warms the cache; admission then answers from it. await f.runtime.fetchQuota( f.routeOf(uuid), @@ -438,7 +441,7 @@ test('local mode never touches insula', async () => { resolveIdentity: async () => ({ deviceId: 'synthetic-device', sessionId: 'synthetic-session', - accountUuid: uuid, + accountUuid: uuid as ProviderAccountUuid, }), refreshToken: async () => { throw new Error('no refresh expected') From b7fb90f27250300a230ef4dde9fda6d472c8d04e Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Sun, 18 Oct 2026 01:06:24 +0200 Subject: [PATCH 4/9] mason: read insula through acquireInsulaQuotaSource instead of a stub ClaustrumConsumer --- mutations.toml | 18 ++- packages/core/src/insula-quota.ts | 17 +-- packages/core/src/native-vault-runtime.ts | 91 ++++++++------- .../tests/native-vault-insula-quota.test.ts | 110 ++++++++++-------- 4 files changed, 129 insertions(+), 107 deletions(-) diff --git a/mutations.toml b/mutations.toml index 578e8691..af74ac60 100644 --- a/mutations.toml +++ b/mutations.toml @@ -1058,17 +1058,20 @@ command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/quota- test_count_pattern = "CKDEV executed {count} tests" expect_red = ["physical quota GET starts stay one second apart when the first GET fails"] expect_message = "error: expect(received).toBeGreaterThanOrEqual(expected)" +broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"] +broad_report = "tmp/mutations/core.xml" +broad_id = "{name}" [[control]] -id = "vault-insula-joins-roster-route-account" -guards = "A background insula lookup must join the account of the roster row the route resolves to; the primary account's logical id names no roster row and would never join." +id = "vault-insula-joins-vault-account-id" +guards = "A background insula lookup must join on the roster row's vault account id (list_scoped account_id), never another row field such as the credential id." file = "packages/core/src/native-vault-runtime.ts" -old = "swept = await insula.cachedInsulaQuotaForRoute(row.routeId)" -new = "swept = await insula.cachedInsulaQuotaForRoute(routeId) // NON-VACUITY BREAK: join on the logical id" +old = "swept = insulaReading(row.accountIdentity)" +new = "swept = insulaReading(row.credentialId) // NON-VACUITY BREAK: join on the credential id" test_file = "packages/core/src/tests/native-vault-insula-quota.test.ts" runner = "command" command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/native-vault-insula-quota.test.ts", "{test}"] test_count_pattern = "CKDEV executed {count} tests" -expect_red = ["the primary account is looked up in insula by its roster route, not its logical id"] +expect_red = ["EXPECT_RED_PLACEHOLDER"] expect_message = "error: expect(received).toEqual(expected)" broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"] broad_report = "tmp/mutations/core.xml" @@ -1086,6 +1089,9 @@ command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/quota- test_count_pattern = "CKDEV executed {count} tests" expect_red = ["physical quota GET starts stay one second apart when the first GET fails", "the second physical quota GET first requests the rest of the one-second gap"] expect_message = '/error: expect\(received\)\.(toBeGreaterThanOrEqual|toEqual)\(expected\)/' +broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"] +broad_report = "tmp/mutations/core.xml" +broad_id = "{name}" [[control]] id = "vault-insula-reading-skips-usage-poll" guards = "A current insula reading for a vault account must replace that account's usage GET; only a miss polls Anthropic." @@ -1098,7 +1104,7 @@ test_file = "packages/core/src/tests/native-vault-insula-quota.test.ts" runner = "command" command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/native-vault-insula-quota.test.ts", "{test}"] test_count_pattern = "CKDEV executed {count} tests" -expect_red = ["background sweep takes a fresh insula reading and sends no usage GET", "the primary account is looked up in insula by its roster route, not its logical id", "a stale, missing or failed insula entry falls back to one usage GET for that account only", "admission reads only the insula cache: never refreshes it, polls on a miss", "the Fable window from extraRateWindows drives scoped routing and the scoped killswitch floor", "the source switch is logged once per change with fixed codes and no identity"] +expect_red = ["background sweep takes a fresh insula reading and sends no usage GET", "the primary account is looked up in insula by its roster row account id, not its logical id", "a stale, missing or failed insula entry falls back to one usage GET for that account only", "admission reads only the insula cache: never refreshes it, polls on a miss", "the Fable window from extraRateWindows drives scoped routing and the scoped killswitch floor", "the source switch is logged once per change with fixed codes and no identity"] expect_message = "error: expect(received).toEqual(expected)" broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"] broad_report = "tmp/mutations/core.xml" diff --git a/packages/core/src/insula-quota.ts b/packages/core/src/insula-quota.ts index e29dbd2c..2775b408 100644 --- a/packages/core/src/insula-quota.ts +++ b/packages/core/src/insula-quota.ts @@ -179,26 +179,21 @@ export function normalizeInsulaClaudeQuota( entry: InsulaUsageEntry, context: InsulaQuotaContext, ): InsulaClaudeQuota | undefined { + // common-auth's InsulaQuotaSource.current only returns an entry with usage, + // no stale or errorClass field, and a fetchedAt under four minutes old. const checkedAt = context.fetchedAt - if ( - !Number.isFinite(checkedAt) || - entry.stale !== undefined || - entry.errorClass !== undefined - ) - return undefined const usage = entry.usage - if (!record(usage)) return undefined - const five_hour = windowReading(usage.primary, checkedAt, { + const five_hour = windowReading(usage?.primary, checkedAt, { kind: 'five_hour', minutes: 300, }) - const seven_day = windowReading(usage.secondary, checkedAt, { + const seven_day = windowReading(usage?.secondary, checkedAt, { kind: 'weekly', minutes: 10_080, }) if (!five_hour || !seven_day) return undefined - const scoped = scopedReadings(usage.extraRateWindows, checkedAt) - const extraUsage = extraUsageReading(entry.pools ?? usage.pools) + const scoped = scopedReadings(usage?.extraRateWindows, checkedAt) + const extraUsage = extraUsageReading(entry.pools ?? usage?.pools) return { five_hour, seven_day, diff --git a/packages/core/src/native-vault-runtime.ts b/packages/core/src/native-vault-runtime.ts index 68317b84..3d02d8bf 100644 --- a/packages/core/src/native-vault-runtime.ts +++ b/packages/core/src/native-vault-runtime.ts @@ -5,12 +5,13 @@ import { } from '@cortexkit/claustrum-client' import { acceptAccount, - ClaustrumConsumer, + acquireInsulaQuotaSource, ClaustrumConsumerError, - type ClaustrumScopedClient, connectClaustrumScopedClient, declineAccount, + type InsulaQuotaSource, mutateVaultRoster, + readClaustrumEnrollmentToken, readVaultRoster, refreshVaultRoster, } from '@cortexkit/common-auth/claustrum' @@ -384,40 +385,47 @@ export function createNativeVaultRuntime( })), env, }) - // Used only for insula's background quota cache: this runtime's own custody - // lists, authorizes and reports. Common-auth shares one insula source per - // resolved connection file and enrollment token path, and the consumer - // never reads it unless the enrollment selects vault custody and its token - // file is readable. - const insula = new ClaustrumConsumer({ - rosterPath: paths.roster, - tokenPath: getHostClaustrumEnrollmentPaths(options.host, env).tokenPath, - family: { - refreshAdapter: 'anthropic', - category: 'anthropic-native', - apiKeys: false, - }, - connect: async (): Promise => { - const transport = await custody.insulaTransport() - const unused = () => Promise.reject(new NativeCustodyError('unavailable')) - return { - listScoped: unused, - getScoped: unused, - reportAuthFailureScoped: unused, - // Nothing to close here: the connection belongs to this runtime's - // custody, which closes it when the runtime closes. - close: () => {}, - ...(transport && { insula: transport }), + // Insula's background quota cache, shared by common-auth across runtimes + // with the same resolved connection file and enrollment token path (one + // fetch a minute between them). Acquired on the first background tick that + // finds vault custody, released when this runtime closes. + const tokenPath = getHostClaustrumEnrollmentPaths(options.host, env).tokenPath + const clock = options.now ?? Date.now + let insula: { source: InsulaQuotaSource; release(): void } | undefined + /** + * Refresh insula's cache for one background tick. Enrollment is read fresh + * each time: outside vault custody, or with a missing or unreadable token, + * insula is not used at all. An optional source never fails the tick. + */ + async function refreshInsula(): Promise { + try { + if (!(await active())) return + await readClaustrumEnrollmentToken(tokenPath) + if (!insula) { + const transport = await custody.insulaTransport() + check() + if (!transport) return + insula ??= acquireInsulaQuotaSource(transport, tokenPath) } - }, - isCustodyActive: () => active().catch(() => false), - insula: { provider: 'claude', normalize: normalizeInsulaClaudeQuota }, - ...(options.now && { now: options.now }), - logger: { - warn: () => {}, - debug: () => {}, - }, - }) + await insula.source.refresh(clock()) + } catch { + // Insula is optional: without it every account keeps its own poll. + } + } + /** + * Cache-only lookup by the vault's account id for this account (the + * list_scoped or get_scoped account_id), never an email or label. + * common-auth returns only a current entry; anything we cannot map is a miss. + */ + function insulaReading( + accountId: string | undefined, + ): InsulaClaudeQuota | undefined { + if (!insula || !accountId || shutdown.signal.aborted) return undefined + const current = insula.source.current('claude', accountId, clock()) + return current + ? normalizeInsulaClaudeQuota(current.entry, current.context) + : undefined + } // The last source each route's guided quota reading came from, so a change // is logged once rather than on every sweep. const quotaSources = new Map() @@ -486,7 +494,7 @@ export function createNativeVaultRuntime( if (shutdown.signal.aborted) return // Keep insula's cache warm for admission lookups, which only read it. // It refreshes at most once a minute and never throws. - await insula.refreshInsulaQuota() + await refreshInsula() if (shutdown.signal.aborted) return // Notifications belong to their observers. A slow or disposing host must // not hold discovery for the other leases; hosts drain their own writes. @@ -848,10 +856,9 @@ export function createNativeVaultRuntime( // list_scoped account_id) before any credential is fetched. let swept: InsulaClaudeQuota | undefined if (purpose === 'background' && row) { - await insula.refreshInsulaQuota() - check(signal) - swept = await insula.cachedInsulaQuotaForRoute(row.routeId) + await refreshInsula() check(signal) + swept = insulaReading(row.accountIdentity) } const state = await readNativeRuntime(paths.runtime, paths.storageId) const before = @@ -892,7 +899,8 @@ export function createNativeVaultRuntime( reading = swept else if (swept) miss = 'account-changed' } else if (purpose === 'admission') { - reading = await insula.cachedInsulaQuota(receipt) + // Only the account id the vault asserted for this credential. + reading = insulaReading(receipt.assertedAccountIdentity) } if (purpose) noteQuotaSource( @@ -1018,7 +1026,8 @@ export function createNativeVaultRuntime( schedulePoll() if (shutdown.signal.aborted) return shutdown.abort(new NativeCustodyError('closed')) - insula.close() + insula?.release() + insula = undefined custody.close() }, } diff --git a/packages/core/src/tests/native-vault-insula-quota.test.ts b/packages/core/src/tests/native-vault-insula-quota.test.ts index 674e4db3..c46e556d 100644 --- a/packages/core/src/tests/native-vault-insula-quota.test.ts +++ b/packages/core/src/tests/native-vault-insula-quota.test.ts @@ -1,9 +1,8 @@ -import { expect, spyOn } from 'bun:test' +import { expect } from 'bun:test' import { mkdtemp, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { - ClaustrumConsumer, type InsulaUsageEntry, projectVaultRoster, } from '@cortexkit/common-auth/claustrum' @@ -196,18 +195,23 @@ async function fixture(entries: () => unknown[]) { }, { preconnect: fetch.preconnect }, ) - const runtime = createNativeVaultRuntime({ - paths, - host: 'opencode', - env, - now: () => clock.now, - connect: async () => client, - readToken: async () => ({ token: '01'.repeat(32), token_generation: 1 }), - }) - deferCleanup(async () => { - await runtime.drainQuotaPublications() - runtime.close() - }) + // Further runtimes share the paths, enrollment and daemon connection file. + const make = () => { + const runtime = createNativeVaultRuntime({ + paths, + host: 'opencode', + env, + now: () => clock.now, + connect: async () => client, + readToken: async () => ({ token: '01'.repeat(32), token_generation: 1 }), + }) + deferCleanup(async () => { + await runtime.drainQuotaPublications() + runtime.close() + }) + return runtime + } + const runtime = make() await runtime.refresh() const roster = await runtime.read() const routeOf = (account: string) => { @@ -215,7 +219,7 @@ async function fixture(entries: () => unknown[]) { if (!id) throw new Error('Missing fixture route') return id } - return { runtime, counts, usage, vaultPaths, routeOf, clock } + return { runtime, make, counts, usage, vaultPaths, routeOf, clock } } const bearer = (credentialId: string) => @@ -279,7 +283,7 @@ test('background sweep takes a fresh insula reading and sends no usage GET', asy expect(stored?.quotaCheckedAt).toBe(fetchedAt) }) -test('the primary account is looked up in insula by its roster route, not its logical id', async () => { +test('the primary account is looked up in insula by its roster row account id, not its logical id', async () => { const f = await fixture(() => [entry(uuid)]) const quota = await f.runtime.fetchQuota( 'main', @@ -340,12 +344,6 @@ test('an insula entry for a different account never applies', async () => { }) test('admission reads only the insula cache: never refreshes it, polls on a miss', async () => { - const refresh = spyOn(ClaustrumConsumer.prototype, 'refreshInsulaQuota') - const cached = spyOn(ClaustrumConsumer.prototype, 'cachedInsulaQuota') - deferCleanup(() => { - refresh.mockRestore() - cached.mockRestore() - }) const f = await fixture(() => [entry(uuid2)]) // Cold cache: no insula fetch on the request path, so the usage poll runs. await f.runtime.fetchQuota( @@ -355,8 +353,6 @@ test('admission reads only the insula cache: never refreshes it, polls on a miss undefined, 'admission', ) - expect(refresh).toHaveBeenCalledTimes(0) - expect(cached).toHaveBeenCalledTimes(1) expect(f.counts.insula).toBe(0) expect(f.counts.usage).toEqual([bearer(workRow.id)]) // Let the poll's reading reach stored quota before insula's is merged in. @@ -369,9 +365,10 @@ test('admission reads only the insula cache: never refreshes it, polls on a miss undefined, 'background', ) - expect(refresh).toHaveBeenCalledTimes(1) expect(f.counts.insula).toBe(1) - const callsBefore = refresh.mock.calls.length + // Past the one-minute refresh budget, admission still only reads the cache: + // a refresh here would fetch again. + f.clock.now = now + 61_000 const quota = await f.runtime.fetchQuota( f.routeOf(uuid2), f.usage, @@ -379,8 +376,6 @@ test('admission reads only the insula cache: never refreshes it, polls on a miss undefined, 'admission', ) - expect(refresh.mock.calls.length).toBe(callsBefore) - expect(cached).toHaveBeenCalledTimes(2) expect(f.counts.insula).toBe(1) expect(f.counts.usage).toEqual([bearer(workRow.id), bearer(mainRow.id)]) // The cold poll above is newer than insula's fetch, so its standard windows @@ -389,6 +384,42 @@ test('admission reads only the insula cache: never refreshes it, polls on a miss expect(quota.scoped?.map((window) => window.checkedAt)).toEqual([fetchedAt]) }) +test('runtimes on one connection file and token path share one insula source until all release it', async () => { + const f = await fixture(() => [entry(uuid2)]) + const second = f.make() + const sweep = (runtime: typeof f.runtime) => + runtime.fetchQuota( + f.routeOf(uuid2), + f.usage, + undefined, + undefined, + 'background', + ) + await sweep(f.runtime) + // The second runtime's tick falls within the shared minute: no new fetch, + // and it still sees the first runtime's entries. + expect((await sweep(second)).five_hour?.checkedAt).toBe(fetchedAt) + expect(f.counts.insula).toBe(1) + expect(f.counts.usage).toEqual([]) + // Closing one runtime releases only its reference. + await f.runtime.drainQuotaPublications() + f.runtime.close() + const third = f.make() + expect((await sweep(third)).five_hour?.checkedAt).toBe(fetchedAt) + expect(f.counts.usage).toEqual([]) + // Once every holder has closed, the source is dropped. A new runtime gets an + // empty source, and the minute budget, which outlives it, blocks a fetch, + // so its account falls back to the usage poll. + for (const runtime of [second, third]) { + await runtime.drainQuotaPublications() + runtime.close() + } + const fresh = f.make() + expect((await sweep(fresh)).five_hour?.checkedAt).toBe(now) + expect(f.counts.insula).toBe(1) + expect(f.counts.usage).toEqual([bearer(workRow.id)]) +}) + test('a quota read without a purpose polls Anthropic and leaves insula alone', async () => { const f = await fixture(() => [entry(uuid2)]) await f.runtime.fetchQuota(f.routeOf(uuid2), f.usage) @@ -420,7 +451,7 @@ test('local mode never touches insula', async () => { await store.updateSettings((settings) => { settings.mainAccountId = 'imported-main' }) - const counts = { insula: 0, connects: 0, usage: 0 } + const counts = { connects: 0, usage: 0 } const runtime = createNativeAccountRuntime({ paths, host: 'opencode', @@ -452,17 +483,6 @@ test('local mode never touches insula', async () => { await runtime.drainQuotaPublications() runtime.close() }) - const refresh = spyOn(ClaustrumConsumer.prototype, 'refreshInsulaQuota') - const forRoute = spyOn( - ClaustrumConsumer.prototype, - 'cachedInsulaQuotaForRoute', - ) - const cached = spyOn(ClaustrumConsumer.prototype, 'cachedInsulaQuota') - deferCleanup(() => { - refresh.mockRestore() - forRoute.mockRestore() - cached.mockRestore() - }) const usage: typeof fetch = Object.assign( async () => { counts.usage++ @@ -474,10 +494,8 @@ test('local mode never touches insula', async () => { await runtime.fetchQuota('main', usage, undefined, undefined, purpose) } expect(counts.usage).toBe(2) + // No vault connection, so no insula transport was ever obtained. expect(counts.connects).toBe(0) - expect(refresh).toHaveBeenCalledTimes(0) - expect(forRoute).toHaveBeenCalledTimes(0) - expect(cached).toHaveBeenCalledTimes(0) }) test('the Fable window from extraRateWindows drives scoped routing and the scoped killswitch floor', async () => { @@ -604,12 +622,6 @@ test('normalizer: unknown shapes return undefined so the poll stays in charge', { fetchedAt }, ), ).toBeUndefined() - expect( - normalizeInsulaClaudeQuota(entry(uuid, { stale: 'old' }), { fetchedAt }), - ).toBeUndefined() - expect( - normalizeInsulaClaudeQuota(entry(uuid), { fetchedAt: Number.NaN }), - ).toBeUndefined() }) test('normalizer maps the Extra usage pool only when present', () => { From 8e85bb637d8a5bd3949c61ed167898a6b8ad4e03 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Sun, 18 Oct 2026 01:18:56 +0200 Subject: [PATCH 5/9] mason: mutation rows for the insula account join, poll skip and source release --- mutations.toml | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/mutations.toml b/mutations.toml index af74ac60..c58c31ea 100644 --- a/mutations.toml +++ b/mutations.toml @@ -1071,7 +1071,7 @@ test_file = "packages/core/src/tests/native-vault-insula-quota.test.ts" runner = "command" command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/native-vault-insula-quota.test.ts", "{test}"] test_count_pattern = "CKDEV executed {count} tests" -expect_red = ["EXPECT_RED_PLACEHOLDER"] +expect_red = ["background sweep takes a fresh insula reading and sends no usage GET", "the primary account is looked up in insula by its roster row account id, not its logical id", "a stale, missing or failed insula entry falls back to one usage GET for that account only", "runtimes on one connection file and token path share one insula source until all release it", "the Fable window from extraRateWindows drives scoped routing and the scoped killswitch floor", "the source switch is logged once per change with fixed codes and no identity"] expect_message = "error: expect(received).toEqual(expected)" broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"] broad_report = "tmp/mutations/core.xml" @@ -1104,8 +1104,26 @@ test_file = "packages/core/src/tests/native-vault-insula-quota.test.ts" runner = "command" command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/native-vault-insula-quota.test.ts", "{test}"] test_count_pattern = "CKDEV executed {count} tests" -expect_red = ["background sweep takes a fresh insula reading and sends no usage GET", "the primary account is looked up in insula by its roster row account id, not its logical id", "a stale, missing or failed insula entry falls back to one usage GET for that account only", "admission reads only the insula cache: never refreshes it, polls on a miss", "the Fable window from extraRateWindows drives scoped routing and the scoped killswitch floor", "the source switch is logged once per change with fixed codes and no identity"] +expect_red = ["background sweep takes a fresh insula reading and sends no usage GET", "the primary account is looked up in insula by its roster row account id, not its logical id", "a stale, missing or failed insula entry falls back to one usage GET for that account only", "admission reads only the insula cache: never refreshes it, polls on a miss", "runtimes on one connection file and token path share one insula source until all release it", "the Fable window from extraRateWindows drives scoped routing and the scoped killswitch floor", "the source switch is logged once per change with fixed codes and no identity"] expect_message = "error: expect(received).toEqual(expected)" broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"] broad_report = "tmp/mutations/core.xml" broad_id = "{name}" + +[[control]] +id = "vault-insula-source-released-on-close" +guards = "Closing a vault runtime must release its reference to the shared insula source, so the source is dropped once its last holder closes." +file = "packages/core/src/native-vault-runtime.ts" +old = """ insula?.release() +""" +new = """ // NON-VACUITY BREAK: never release +""" +test_file = "packages/core/src/tests/native-vault-insula-quota.test.ts" +runner = "command" +command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/native-vault-insula-quota.test.ts", "{test}"] +test_count_pattern = "CKDEV executed {count} tests" +expect_red = ["runtimes on one connection file and token path share one insula source until all release it"] +expect_message = "error: expect(received).toBe(expected)" +broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"] +broad_report = "tmp/mutations/core.xml" +broad_id = "{name}" From 4189cda014ec41f1fce4eee0245734985b786f4c Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Sun, 18 Oct 2026 02:37:24 +0200 Subject: [PATCH 6/9] mason: adopt common-auth 0.19.1 with insula quota reader --- CHANGELOG.md | 6 +++++- bun.lock | 4 ++-- packages/core/package.json | 2 +- packages/core/scripts/check-native-type-closure.test.ts | 8 ++++---- packages/core/scripts/check-native-type-closure.ts | 4 ++-- 5 files changed, 14 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e20341b2..43671ead 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,9 +13,13 @@ This repo is a CortexKit-maintained Anthropic auth monorepo for OpenCode and Pi. - When vault setup finds no Claude account it can use, it now says why from the vault's own listing (#354). With no read grant for the host, it prints the `ck auth grant` command that adds one. With a grant that reaches no Claude login, it suggests `ck auth reclassify --from-registry` and checking the logins with `ck auth status`. On older vaults that do not report grant reach, the previous message is unchanged. - When setup fails after vault enrollment began, it now prints one line per host: connected to the vault, not connected, or unknown (check with `ck auth status`). A host left connected also gets the Disconnect / run-setup-again remedy. Setup no longer rereads enrollment files for this, so a failure before enrollment began no longer reports a host connected by an earlier run. +### Quota + +- Vault-mode quota now reads Claude's cached quota from insula, with Anthropic usage polls retained as a per-account fallback for stale or missing readings. + ### Compatibility -- Move to `@cortexkit/common-auth` 0.19.0 and `@cortexkit/claustrum-client` 0.10.0. +- Move to `@cortexkit/common-auth` 0.19.1 and `@cortexkit/claustrum-client` 0.10.0. - Update the OpenCode plugin and SDK test dependencies together to 1.19.0. The plugin's peer range and the OpenCode 1.18.18 host used by the process E2E and packed TUI checks are unchanged. - Update the Pi SDK test dependencies (`pi-coding-agent`, `pi-ai`, `pi-agent-core`, `pi-tui`) together to 1.2.0. The Pi peer range is unchanged. diff --git a/bun.lock b/bun.lock index 137b7a05..08c026a2 100644 --- a/bun.lock +++ b/bun.lock @@ -34,7 +34,7 @@ "xxhash-wasm": "^1.1.0", }, "devDependencies": { - "@cortexkit/common-auth": "0.19.0", + "@cortexkit/common-auth": "0.19.1", }, }, "packages/e2e-tests": { @@ -244,7 +244,7 @@ "@cortexkit/claustrum-client": ["@cortexkit/claustrum-client@0.10.0", "", { "dependencies": { "@cortexkit/store": "^0.2.1", "@cortexkit/subc-client": "^0.16.0" } }, "sha512-7mTbq80KPN3x4J6QUR+h8iE12r2/O9vlpcafi4OrUDB14IQglmXXitz+iX0BoqCwlRxYMJehdi31/x55fZhRJg=="], - "@cortexkit/common-auth": ["@cortexkit/common-auth@0.19.0", "", { "dependencies": { "@cortexkit/private-files": "^0.2.1", "jsonc-parser": "^3.3.1" }, "peerDependencies": { "@cortexkit/claustrum-client": ">=0.6.2", "@cortexkit/subc-client": "^0.16.0", "@opencode/plugin": ">=2.0.21", "@opentui/core": ">=0.5.11", "@opentui/solid": ">=0.5.11", "solid-js": "1.9.12" }, "optionalPeers": ["@cortexkit/claustrum-client", "@cortexkit/subc-client", "@opencode/plugin", "@opentui/core", "@opentui/solid", "solid-js"] }, "sha512-p7oy26dBllFF6mkQinsyyqqFP8FKoyVvHyt/aQWEGTljn0AbVRj++Co5Mict4ffQUPrujlQwThilUoPn/AsB8Q=="], + "@cortexkit/common-auth": ["@cortexkit/common-auth@0.19.1", "", { "dependencies": { "@cortexkit/private-files": "^0.2.1", "jsonc-parser": "^3.3.1" }, "peerDependencies": { "@cortexkit/claustrum-client": ">=0.6.2", "@cortexkit/subc-client": "^0.16.0", "@opencode/plugin": ">=2.0.21", "@opentui/core": ">=0.5.11", "@opentui/solid": ">=0.5.11", "solid-js": "1.9.12" }, "optionalPeers": ["@cortexkit/claustrum-client", "@cortexkit/subc-client", "@opencode/plugin", "@opentui/core", "@opentui/solid", "solid-js"] }, "sha512-/tkEwmGliz1V6bUstHYtJqeh8y/hNvHtA4uGO9xMGJRB7LM1+TsLPfAVV1d07gEb8rJkizUyBG9mkTZe2CoA5g=="], "@cortexkit/opencode-anthropic-auth": ["@cortexkit/opencode-anthropic-auth@workspace:packages/opencode"], diff --git a/packages/core/package.json b/packages/core/package.json index b79b4906..858514fb 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -30,7 +30,7 @@ "prepublishOnly": "bun run build" }, "devDependencies": { - "@cortexkit/common-auth": "0.19.0" + "@cortexkit/common-auth": "0.19.1" }, "dependencies": { "@cortexkit/claustrum-client": "0.10.0", diff --git a/packages/core/scripts/check-native-type-closure.test.ts b/packages/core/scripts/check-native-type-closure.test.ts index afc98640..93dba96b 100644 --- a/packages/core/scripts/check-native-type-closure.test.ts +++ b/packages/core/scripts/check-native-type-closure.test.ts @@ -110,7 +110,7 @@ test('rejects unsupported published producer versions before copying declaration producer, 'package.json', JSON.stringify({ - version: '0.19.1', + version: '0.19.0', exports: { './store': { types: './dist/store.d.ts' } }, }), ) @@ -124,7 +124,7 @@ test('rejects unsupported published producer versions before copying declaration ) await expect(closeNativeTypeDeclarations(dist, producer)).rejects.toThrow( - 'Expected published common-auth 0.19.0, got 0.19.1', + 'Expected published common-auth 0.19.1, got 0.19.0', ) expect(await readdir(dist)).not.toContain('internal-types') }) @@ -137,7 +137,7 @@ test('copies the complete published relative declaration graph including cycles producer, 'package.json', JSON.stringify({ - version: '0.19.0', + version: '0.19.1', exports: { './store': { types: './dist/store/index.d.ts' } }, }), ) @@ -273,7 +273,7 @@ test('packed Core types and Node runtime work in a fresh consumer without common const manifest = JSON.parse( await readFile(join(core, 'package.json'), 'utf8'), ) - expect(manifest.devDependencies['@cortexkit/common-auth']).toBe('0.19.0') + expect(manifest.devDependencies['@cortexkit/common-auth']).toBe('0.19.1') expect(manifest.dependencies['@cortexkit/claustrum-client']).toBe('0.10.0') expect(manifest.dependencies['@cortexkit/subc-client']).toBe('^0.16.0') expect(manifest.dependencies['jsonc-parser']).toBe('^3.3.1') diff --git a/packages/core/scripts/check-native-type-closure.ts b/packages/core/scripts/check-native-type-closure.ts index 3449bd4d..80814d91 100644 --- a/packages/core/scripts/check-native-type-closure.ts +++ b/packages/core/scripts/check-native-type-closure.ts @@ -185,9 +185,9 @@ export async function closeNativeTypeDeclarations( version: string exports: Record } - if (manifest.version !== '0.19.0') + if (manifest.version !== '0.19.1') throw new Error( - `Expected published common-auth 0.19.0, got ${manifest.version}`, + `Expected published common-auth 0.19.1, got ${manifest.version}`, ) const producerDist = join(producerRoot, 'dist') const assets = join(distRoot, TYPE_ASSETS) From 33f4bc81f24203edb9d3269a3839416fb7900029 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Sun, 18 Oct 2026 03:10:06 +0200 Subject: [PATCH 7/9] mason: repair quota mutation guard anchors --- mutations.toml | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/mutations.toml b/mutations.toml index c58c31ea..8b073884 100644 --- a/mutations.toml +++ b/mutations.toml @@ -374,6 +374,7 @@ old = ''' const unavailable = await refreshFallbackQuotas( false, requiredByPolicy, options.modelId, + 'admission', )''' new = ''' const unavailable = new Set() // NON-VACUITY BREAK: choose a route without polling due fallbacks''' test_file = "packages/opencode/src/tests/quota-sweep.test.ts" @@ -392,9 +393,14 @@ id = "fallback-quota-sweeps-do-not-overlap" guards = "A fallback quota sweep started while another sweep's poll is outstanding must read the pool after that reading is stored, not poll the same account again from an older snapshot." file = "packages/opencode/src/index.ts" old = ''' const operation = fallbackQuotaRefreshQueue.then(() => - pollFallbackQuotas(force, requiredByPolicy, modelId), + pollFallbackQuotas(force, requiredByPolicy, modelId, purpose), )''' -new = ''' const operation = pollFallbackQuotas(force, requiredByPolicy, modelId) // NON-VACUITY BREAK: start sweeps without waiting for the previous one''' +new = ''' const operation = pollFallbackQuotas( + force, + requiredByPolicy, + modelId, + purpose, + ) // NON-VACUITY BREAK: start sweeps without waiting for the previous one''' test_file = "packages/opencode/src/tests/quota-sweep.test.ts" runner = "command" command = ["bun", "scripts/run-mutation-bun.ts", "packages/opencode/src/tests/quota-sweep.test.ts", "{test}"] @@ -555,7 +561,13 @@ broad_id = "{name}" id = "pi-forwards-physical-quota-dispatch" guards = "Pi forwards each native usage GET through the shared quota dispatcher rather than bypassing physical-request spacing." file = "packages/pi/src/stream.ts" -old = "fetchQuota(\n id,\n fetchPiOAuth,\n undefined,\n dispatch,\n )" +old = '''fetchQuota( + id, + fetchPiOAuth, + undefined, + dispatch, + request.purpose, + )''' new = "fetchQuota(id, fetchPiOAuth)" test_file = "packages/pi/src/tests/quota-dispatch.test.ts" runner = "command" @@ -1098,7 +1110,7 @@ guards = "A current insula reading for a vault account must replace that account file = "packages/core/src/native-vault-runtime.ts" old = ''' if (reading) { const stored''' -new = ''' if (reading && false) { // NON-VACUITY BREAK: always poll +new = ''' if (reading && !reading) { // NON-VACUITY BREAK: always poll instead of using an insula reading const stored''' test_file = "packages/core/src/tests/native-vault-insula-quota.test.ts" runner = "command" From 097966cb033c579c4c3b21760a2195c62e23223b Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Sun, 18 Oct 2026 03:18:44 +0200 Subject: [PATCH 8/9] test: a sidebar write notifies listeners that started during the write writeSidebarStateAtomic read its test hooks once, when the write began, and reused that snapshot after the rename. A test that installed its listener mid-write, after reading the old file, never heard the rename and waited until its 5 s timeout. That flaked "feed publication failure does not affect response, quota persistence, or sidebar update" about 3 in 40 runs on main and this branch alike. The after-rename hook is now looked up at rename time; production runs without hooks. --- packages/opencode/src/sidebar-state.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/opencode/src/sidebar-state.ts b/packages/opencode/src/sidebar-state.ts index 09074503..bfe56e25 100644 --- a/packages/opencode/src/sidebar-state.ts +++ b/packages/opencode/src/sidebar-state.ts @@ -656,7 +656,9 @@ async function writeSidebarStateAtomic( return 'lock-lost-before-rename' } await rename(tempFile, stateFile) - await testHooks?.afterRename?.(stateFile) + // Look the hooks up again: a test may start listening while this write is + // in flight, after reading the old file, and must still hear this rename. + await sidebarStateWriteHooksFor(stateFile)?.afterRename?.(stateFile) if (!(await ownsLock())) return 'lock-lost-after-rename' return 'written' } catch (error) { From 5f1724dc95f7c9a2064b511ed769bd8b8b3781f9 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Sun, 18 Oct 2026 03:31:46 +0200 Subject: [PATCH 9/9] mutations: accept toBe failures for the two insula rows The shared-source test catches both mutants at its first assertion, a toBe on the reading time, so its failure message is toBe rather than toEqual. --- mutations.toml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/mutations.toml b/mutations.toml index 8b073884..81f65d32 100644 --- a/mutations.toml +++ b/mutations.toml @@ -1084,7 +1084,7 @@ runner = "command" command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/native-vault-insula-quota.test.ts", "{test}"] test_count_pattern = "CKDEV executed {count} tests" expect_red = ["background sweep takes a fresh insula reading and sends no usage GET", "the primary account is looked up in insula by its roster row account id, not its logical id", "a stale, missing or failed insula entry falls back to one usage GET for that account only", "runtimes on one connection file and token path share one insula source until all release it", "the Fable window from extraRateWindows drives scoped routing and the scoped killswitch floor", "the source switch is logged once per change with fixed codes and no identity"] -expect_message = "error: expect(received).toEqual(expected)" +expect_message = '/error: expect\(received\)\.(toBe|toEqual)\(expected\)/' broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"] broad_report = "tmp/mutations/core.xml" broad_id = "{name}" @@ -1117,7 +1117,7 @@ runner = "command" command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/native-vault-insula-quota.test.ts", "{test}"] test_count_pattern = "CKDEV executed {count} tests" expect_red = ["background sweep takes a fresh insula reading and sends no usage GET", "the primary account is looked up in insula by its roster row account id, not its logical id", "a stale, missing or failed insula entry falls back to one usage GET for that account only", "admission reads only the insula cache: never refreshes it, polls on a miss", "runtimes on one connection file and token path share one insula source until all release it", "the Fable window from extraRateWindows drives scoped routing and the scoped killswitch floor", "the source switch is logged once per change with fixed codes and no identity"] -expect_message = "error: expect(received).toEqual(expected)" +expect_message = '/error: expect\(received\)\.(toBe|toEqual)\(expected\)/' broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"] broad_report = "tmp/mutations/core.xml" broad_id = "{name}"