From a77b6246c873c8eaa0936bb323aa638c8547b51f Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Tue, 20 Oct 2026 22:40:02 +0200 Subject: [PATCH 1/3] mason: join refreshes by rejected credential cause Admit proven unexpired tokens while proactive or different-token recovery is running. Record the initial job cause and additional 401 rejections; bound rejected-token callers to one two-second wait without cancelling shared rotation or altering its leases and retries. Strengthen the prior per-caller adoption test: when another joiner reports a 401 for the would-be adopted token, neither caller may serve it. The old first-caller success expectation intentionally becomes rejected-access-unchanged; the existing rejecting-caller refusal and zero-exchange claims remain. Add red-first OC1/OC2/Pi join coverage and two CAUGHT catalogue controls. Affected selection: 438 pass, one unchanged vault Prime network-error test exceeded its five-second deadline; not retried. Typecheck, builds, lint/format, restored named controls and Pi host smoke pass. --- CHANGELOG.md | 4 + mutations.toml | 34 ++++++ .../core/src/native-refresh-coordinator.ts | 113 ++++++++++++++---- packages/core/src/native-transient-account.ts | 6 +- .../tests/native-refresh-coordinator.test.ts | 103 +++++++++++++++- .../src/tests/native-refresh-join-fixture.ts | 51 ++++++++ .../opencode/src/tests/native-serving.test.ts | 109 +++++++++++++++++ .../src/tests/opencode2-native-source.test.ts | 105 ++++++++++++++++ .../pi/src/tests/native-request-locks.test.ts | 97 ++++++++++++++- 9 files changed, 592 insertions(+), 30 deletions(-) create mode 100644 packages/core/src/tests/native-refresh-join-fixture.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index f9c197b3b..4b2a499ae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ This repo is a CortexKit-maintained Anthropic auth monorepo for OpenCode and Pi. - OpenCode 2 now shows the CLAUDE quota sidebar too, without opening `/claude`. It is the same panel as OpenCode 1 (per-account 5h, 7d and model bars with pacing, collapsed rows, routing, cache, Prime and health), drawn from one shared component in the active OpenCode 2 theme's colors and configured by the same `anthropic-auth` entry in `tui-preferences.jsonc`. The OpenCode 2 server answers it through the native RPC (`sidebar` method) from the account pool, with no file and no credential; the active account is the one that last served the sidebar's session. Turned-off vault accounts stay listed, as on OpenCode 1, and reset countdowns, reading ages and pacing keep advancing between server changes. A TUI talking to an older server shows "Waiting for quota…". - Before an account has served the session, the collapsed sidebar no longer marks `main` as active, matching the expanded view. +### Accounts + +- Valid tokens no longer wait for proactive refreshes; requests waiting for a rejected-token refresh fail retryably after two seconds while the refresh continues. + ## 2.1.2 - Requests no longer wait on relay or session-pin bookkeeping locks; relay settings and tokens are checked together before sending. diff --git a/mutations.toml b/mutations.toml index 034fbd2fe..159445cce 100644 --- a/mutations.toml +++ b/mutations.toml @@ -324,6 +324,40 @@ broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core broad_report = "tmp/mutations/core.xml" broad_id = "{name}" +[[control]] +id = "refresh-join-checks-cause" +guards = "A proactive refresh must not delay a proven unexpired access token." +file = "packages/core/src/native-refresh-coordinator.ts" +old = "if (!jobRejectsAccess(job, row)) {" +new = "if (!job) { // NON-VACUITY BREAK: join every existing refresh" +test_file = "packages/opencode/src/tests/native-serving.test.ts" +runner = "command" +command = ["bun", "scripts/run-mutation-bun.ts", "packages/opencode/src/tests/native-serving.test.ts", "{test}"] +test_count_pattern = "CKDEV executed {count} tests" +expect_red = ["OC1 refresh join proactive preserves valid-token admission"] +expect_message = "error: expect(received).toBe(expected)" +broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/opencode", "tmp/mutations/opencode.xml"] +broad_report = "tmp/mutations/opencode.xml" +broad_id = "{name}" +junit_class = "suites" + +[[control]] +id = "refresh-join-has-overall-deadline" +guards = "A refresh replacing a rejected token has one bounded request wait across all attempts." +file = "packages/core/src/native-refresh-coordinator.ts" +old = "? joinRejectedRefresh(job)" +new = "? job.promise // NON-VACUITY BREAK: wait without an overall deadline" +test_file = "packages/opencode/src/tests/native-serving.test.ts" +runner = "command" +command = ["bun", "scripts/run-mutation-bun.ts", "packages/opencode/src/tests/native-serving.test.ts", "{test}"] +test_count_pattern = "CKDEV executed {count} tests" +expect_red = ["OC1 refresh join deadline preserves valid-token admission"] +expect_message = "error: expect(received).toMatchObject(expected)" +broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/opencode", "tmp/mutations/opencode.xml"] +broad_report = "tmp/mutations/opencode.xml" +broad_id = "{name}" +junit_class = "suites" + [[control]] id = "native-activation-before-401-refresh" guards = "A revoked host activation must not trigger a credential refresh." diff --git a/packages/core/src/native-refresh-coordinator.ts b/packages/core/src/native-refresh-coordinator.ts index faaaaa3da..5fdb2a8fd 100644 --- a/packages/core/src/native-refresh-coordinator.ts +++ b/packages/core/src/native-refresh-coordinator.ts @@ -33,6 +33,7 @@ import { nativeLocalCredentialValidationMatches, } from './native-credential-validation.ts' import { nativeLockObserver } from './native-lock-observer.ts' +import { NativeTransientAccountError } from './native-transient-account.ts' import { isTransientNetworkError } from './network-errors.ts' import { isNativeLocalPoolBinding, @@ -49,6 +50,10 @@ import { tokenFingerprint } from './token-fingerprint.ts' export const NATIVE_REFRESH_MAX_ATTEMPTS = 3 +// Limit a caller's rejected-token refresh wait to two seconds across all attempts. +// The shared refresh may continue, just like an account write after a join expires. +const NATIVE_REFRESH_JOIN_BUDGET_MS = 2_000 + /** * Jobs distinguish credential replacements. This lock must still serialize * all refreshes for the same account. @@ -302,7 +307,7 @@ export interface NativeRefreshRequest { intent: 'serve' | 'refresh' mode: 'local' | 'claustrum' binding: NativeLocalPoolBinding - /** A 401's dispatched access token, used only for replacement adoption. */ + /** The access token dispatched on a 401; it must not be served again. */ rejectedAccessToken?: string signal?: AbortSignal } @@ -348,12 +353,43 @@ type JobResult = interface Job { keys: Set promise: Promise + cause: { kind: 'proactive' } | { kind: '401'; accessFingerprint: string } + additionalRejections: Set } // Shared by every platform/coordinator instance in this process. Deletion is // compare-and-delete: an old job can never dispose a successor registration. const jobs = new Map() +function jobRejectsAccess(job: Job | undefined, row: PoolRow | undefined) { + if (!job || row?.credential?.type !== 'oauth' || !row.credential.access) + return false + const fingerprint = tokenFingerprint(row.credential.access) + return ( + (job.cause.kind === '401' && job.cause.accessFingerprint === fingerprint) || + job.additionalRejections.has(fingerprint) + ) +} + +async function joinRejectedRefresh(job: Job): Promise { + let timer: ReturnType | undefined + try { + // Only the caller's wait ends. The shared exchange retains its leases and + // must still save any consumed single-use refresh token's successor. + return await Promise.race([ + job.promise, + new Promise((_resolve, reject) => { + timer = setTimeout( + () => reject(new NativeTransientAccountError()), + NATIVE_REFRESH_JOIN_BUDGET_MS, + ) + }), + ]) + } finally { + clearTimeout(timer) + } +} + function unregister(job: Job, key: string) { if (jobs.get(key) === job) jobs.delete(key) job.keys.delete(key) @@ -1282,25 +1318,45 @@ export function createNativeRefreshCoordinator( return { status: 'refused', reason: 'cancelled', persisted: false } const key = nativeLocalRefreshJobKey(request.binding) let job = jobs.get(key) - if (!job && request.intent === 'serve') { - try { - const current = await admit( - request, - request.binding, - await rowFor(request.binding), - 'current', - false, + const recordRejection = () => { + // A 401 can arrive after proactive work started. Record its rejected token + // so no caller serves it again, without changing the job's original cause. + if (job && request.rejectedAccessToken) + job.additionalRejections.add( + tokenFingerprint(request.rejectedAccessToken), ) - if ( - current.status !== 'refused' || - ![ - 'proof-missing', - 'identity-unproven', - 'access-unavailable', - 'rejected-access-unchanged', - ].includes(current.reason) - ) - return current + } + recordRejection() + if (request.intent === 'serve') { + try { + const row = await rowFor(request.binding) + job = jobs.get(key) ?? job + recordRejection() + if (!jobRejectsAccess(job, row)) { + const current = await admit( + request, + request.binding, + row, + 'current', + false, + ) + if ( + current.status !== 'refused' || + ![ + 'proof-missing', + 'identity-unproven', + 'access-unavailable', + 'rejected-access-unchanged', + ].includes(current.reason) + ) { + // Admission yields to other callers. A 401 recorded during that + // read must prevent dispatch of the token it rejected. + job = jobs.get(key) ?? job + recordRejection() + if (current.status !== 'usable' || !jobRejectsAccess(job, row)) + return current + } + } } catch (caught) { return { status: 'failed', @@ -1313,10 +1369,18 @@ export function createNativeRefreshCoordinator( // recovery for the same storage, credential epoch and account UUID (or // unknown-identity row). Check for that job before starting duplicate work. job = jobs.get(key) + recordRejection() } if (!job) { job = { keys: new Set([key]), + cause: request.rejectedAccessToken + ? { + kind: '401', + accessFingerprint: tokenFingerprint(request.rejectedAccessToken), + } + : { kind: 'proactive' }, + additionalRejections: new Set(), promise: Promise.resolve({ status: 'refused', reason: 'initializing', @@ -1329,7 +1393,10 @@ export function createNativeRefreshCoordinator( for (const ownedKey of owner.keys) unregister(owner, ownedKey) }) } - const result = await job.promise + const result = await (job.cause.kind === '401' || + job.additionalRejections.size > 0 + ? joinRejectedRefresh(job) + : job.promise) if (request.signal?.aborted) return { status: 'refused', @@ -1346,6 +1413,12 @@ export function createNativeRefreshCoordinator( } : request.binding const row = await rowFor(request.binding) + if (jobRejectsAccess(job, row)) + return { + status: 'refused', + reason: 'rejected-access-unchanged', + persisted: result.persisted, + } return await admit( request, expected, diff --git a/packages/core/src/native-transient-account.ts b/packages/core/src/native-transient-account.ts index e4e91b4ce..3c7ea43e5 100644 --- a/packages/core/src/native-transient-account.ts +++ b/packages/core/src/native-transient-account.ts @@ -23,9 +23,9 @@ const TRANSIENT_ACCOUNT_MESSAGE = const transientAccountErrors = new WeakSet() /** - * A local account could not be authorized because another writer was part - * way through saving it (for example the refresh that records its first - * account UUID) and did not finish within the short wait Core allows. + * A local account could not be authorized because the caller's short wait for + * an account write or a refresh replacing a provider-rejected token expired. + * The shared write or refresh may still finish after the caller stops waiting. * * The request should be retried. `code: 'ECONNRESET'` is how this package * marks an error retryable for its hosts. A host must surface it rather than diff --git a/packages/core/src/tests/native-refresh-coordinator.test.ts b/packages/core/src/tests/native-refresh-coordinator.test.ts index 9e6935047..57c9ab0cc 100644 --- a/packages/core/src/tests/native-refresh-coordinator.test.ts +++ b/packages/core/src/tests/native-refresh-coordinator.test.ts @@ -74,6 +74,7 @@ import { nativePoolStoreLocks, } from '../pool-store.ts' import { tokenFingerprint } from '../token-fingerprint.ts' +import { observeRefreshJoin } from './native-refresh-join-fixture.ts' import { TestLifetime } from './test-lifetime.ts' const roots: string[] = [] @@ -1360,7 +1361,7 @@ test('stale-token 401 adopts pure-store replacement without a provider call and expect(calls).toBe(0) }) -test('adoption is per-caller and never serves the access token that caller rejected', async () => { +test('adoption never serves an access token rejected by any caller in the shared job', async () => { const f = await fixture() await f.add('a', 'A') const binding = await f.binding('a') @@ -1393,9 +1394,8 @@ test('adoption is per-caller and never serves the access token that caller rejec }) finish.release() expect(await first).toMatchObject({ - status: 'usable', - source: 'adopted', - access: 'access-new', + status: 'refused', + reason: 'rejected-access-unchanged', }) expect(await joined).toMatchObject({ status: 'refused', @@ -3180,6 +3180,101 @@ for (const echo of ['captured', 'newer'] as const) { }) } +test('401 recovery admits a proven different access version before the shared job completes', async () => { + const f = await fixture() + await f.add('a', 'A') + const entered = barrier() + const finish = barrier() + const c = f.coordinator({ + reconcile: async (event) => { + if (event.status === 'persisted') { + entered.release() + await finish.promise + } + }, + }) + const binding = await f.binding('a') + const owner = c.refresh({ + mode: 'local', + binding, + rejectedAccessToken: 'access-a', + }) + void owner.catch(() => {}) + try { + await entered.promise + await f.seedProof('a') + expect( + await observeRefreshJoin( + c.authorize({ intent: 'serve', mode: 'local', binding }), + ), + ).toMatchObject({ + status: 'fulfilled', + value: { status: 'usable', access: 'access-new' }, + }) + } finally { + finish.release() + await owner + } +}) + +for (const duringAdmission of [false, true]) { + test(`401 recorded ${duringAdmission ? 'during admission' : 'during proactive refresh'} prevents rejected-token dispatch`, async () => { + const f = await fixture() + await f.add('a', 'A') + await f.seedProof('a') + const binding = await f.binding('a') + const providerEntered = barrier() + const finishProvider = barrier() + const admissionEntered = barrier() + const finishAdmission = barrier() + let firstAdmission = true + const c = f.coordinator({ + refreshToken: async () => { + providerEntered.release() + await finishProvider.promise + return { ...credential('new'), expiresIn: 3600 } + }, + readAdmission: async (subject) => { + if (duringAdmission && firstAdmission) { + firstAdmission = false + admissionEntered.release() + await finishAdmission.promise + } + const proof = f.proofs.get(subject.binding.rowId) + return proof + ? { status: 'proven', validation: proof } + : { status: 'unproven' } + }, + }) + const serve = () => c.authorize({ intent: 'serve', mode: 'local', binding }) + const first = duringAdmission + ? serve() + : c.refresh({ mode: 'local', binding }) + if (duringAdmission) await admissionEntered.promise + else await providerEntered.promise + const rejected = c.refresh({ + mode: 'local', + binding, + rejectedAccessToken: 'access-a', + }) + void rejected.catch(() => {}) + let waiting: Promise | undefined + try { + await providerEntered.promise + finishAdmission.release() + waiting = duringAdmission ? first : serve() + expect((await observeRefreshJoin(waiting, 100)).status).toBe('pending') + finishProvider.release() + for (const result of await Promise.all([first, rejected, waiting])) + expect(result).toMatchObject({ status: 'usable', access: 'access-new' }) + } finally { + finishAdmission.release() + finishProvider.release() + await Promise.allSettled([first, rejected, ...(waiting ? [waiting] : [])]) + } + }) +} + test('each joined caller admits its own independently proven concurrent exact version', async () => { const f = await fixture() await f.add('a', 'A') diff --git a/packages/core/src/tests/native-refresh-join-fixture.ts b/packages/core/src/tests/native-refresh-join-fixture.ts new file mode 100644 index 000000000..5fdffa882 --- /dev/null +++ b/packages/core/src/tests/native-refresh-join-fixture.ts @@ -0,0 +1,51 @@ +import type { createTestLifetimeSuite } from './test-lifetime.ts' + +export const refreshedJoinAccess = 'sk-ant-oat01-refresh-join-successor' +export const refreshJoinMessage = + 'Anthropic account update in progress; retry the request' + +/** Hold the real coordinator inside its provider callback, with its leases intact. */ +export function holdRefreshJoin( + lifetime: ReturnType, +) { + const entered = lifetime.gate() + const release = lifetime.gate() + let completed = false + return { + entered: entered.wait, + release: release.open, + completed: () => completed, + refreshToken: async () => { + entered.open() + await release.wait + completed = true + return { + access: refreshedJoinAccess, + refresh: 'synthetic-refresh-join-successor', + expires: Date.now() + 3600_000, + expiresIn: 3600, + } + }, + } +} + +/** A watchdog only: it never releases the provider or advances a mocked clock. */ +export async function observeRefreshJoin( + work: Promise, + budgetMs = 1_000, +) { + let timer: ReturnType | undefined + try { + return await Promise.race([ + work.then( + (value) => ({ status: 'fulfilled' as const, value }), + (error: unknown) => ({ status: 'rejected' as const, error }), + ), + new Promise<{ status: 'pending' }>((resolve) => { + timer = setTimeout(() => resolve({ status: 'pending' }), budgetMs) + }), + ]) + } finally { + clearTimeout(timer) + } +} diff --git a/packages/opencode/src/tests/native-serving.test.ts b/packages/opencode/src/tests/native-serving.test.ts index 19567ab67..a5d69f5e6 100644 --- a/packages/opencode/src/tests/native-serving.test.ts +++ b/packages/opencode/src/tests/native-serving.test.ts @@ -19,6 +19,7 @@ import { createNativeAccountRuntime, createNativePoolStore, custodyTombstoneOAuth, + getClaudeCodeIdentityForVerifiedAccount, getLogLevel, type LogTestRecord, type NativeCustodyClient, @@ -41,6 +42,12 @@ import { toNativeQuotaMap, } from '../../../core/src/native-quota-codec.ts' import { nativeVaultPaths } from '../../../core/src/native-vault-paths.ts' +import { + holdRefreshJoin, + observeRefreshJoin, + refreshedJoinAccess, + refreshJoinMessage, +} from '../../../core/src/tests/native-refresh-join-fixture.ts' import { holdNativeRelayLock } from '../../../core/src/tests/native-relay-lock-fixture.ts' import { requestLockProbe } from '../../../core/src/tests/native-request-lock-fixture.ts' import { createTestLifetimeSuite } from '../../../core/src/tests/test-lifetime.ts' @@ -1028,6 +1035,108 @@ test('supervised content after real migration refuses foreground and background expect(fixture.records).toHaveLength(0) }) +for (const scenario of ['proactive', '401', 'deadline'] as const) { + test(`OC1 refresh join ${scenario} preserves valid-token admission`, async () => { + const fixture = await migrateServingFixture('local') + const warmup = createNativeAccountRuntime({ + paths: fixture.paths, + host: 'opencode', + }) + let initialAccess: string + try { + const admitted = await warmup.authorizeLocal('main') + if (admitted.status !== 'usable') + throw new Error('Missing initial authorization') + initialAccess = admitted.access + } finally { + await warmup.close() + } + const held = holdRefreshJoin(lifetimes) + const probe = requestLockProbe() + const local = { + onLockEvent: probe.onLockEvent, + refreshToken: held.refreshToken, + resolveIdentity: async () => + getClaudeCodeIdentityForVerifiedAccount( + 'main', + mainIdentity as ProviderAccountUuid, + ), + } + const runtime = createNativeAccountRuntime({ + paths: fixture.paths, + host: 'opencode', + local, + }) + lifetimes.deferCleanup(() => runtime.close()) + const sent: string[] = [] + const completedAtDispatch: boolean[] = [] + network.mockImplementation((input, init) => { + const url = input instanceof Request ? input.url : input.toString() + if (url.includes('/v1/messages')) { + const access = new Headers(init?.headers).get('authorization') ?? '' + sent.push(access) + completedAtDispatch.push(held.completed()) + probe.dispatch() + if (scenario !== 'proactive' && sent.length === 1) + return Promise.resolve(Response.json({}, { status: 401 })) + } + return fixture.fetchProvider(input, init) + }) + const plugin = await createPlugin({ nativeLocal: local }) + const options = await load(plugin, async () => + custodyTombstoneOAuth('anthropic'), + ) + const send = () => + requestFetch(options)('https://api.anthropic.com/v1/messages', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + model: 'claude-sonnet-4-6', + max_tokens: 16, + messages: [{ role: 'user', content: 'Refresh join probe' }], + }), + }) + const owner = + scenario === 'proactive' + ? runtime.authorizeLocal('main', { intent: 'refresh' }) + : send() + void owner.catch(() => {}) + let waiting: Promise | undefined + try { + await held.entered + probe.begin() + waiting = send() + if (scenario === 'proactive') { + expect((await observeRefreshJoin(waiting)).status).toBe('fulfilled') + expect(completedAtDispatch).toEqual([false]) + expect(sent).toEqual([`Bearer ${initialAccess}`]) + expect(probe.foreground).toEqual([]) + } else if (scenario === '401') { + expect((await observeRefreshJoin(waiting, 100)).status).toBe('pending') + expect(sent).toEqual([`Bearer ${initialAccess}`]) + held.release() + expect((await waiting).status).toBe(200) + expect(((await owner) as Response).status).toBe(200) + expect(sent).toEqual([ + `Bearer ${initialAccess}`, + `Bearer ${refreshedJoinAccess}`, + `Bearer ${refreshedJoinAccess}`, + ]) + } else { + expect(await observeRefreshJoin(waiting, 3_500)).toMatchObject({ + status: 'rejected', + error: { message: refreshJoinMessage, code: 'ECONNRESET' }, + }) + expect(held.completed()).toBe(false) + expect(sent).toEqual([`Bearer ${initialAccess}`]) + } + } finally { + held.release() + await Promise.allSettled([owner, ...(waiting ? [waiting] : [])]) + } + }, 8_000) +} + for (const scenario of ['local-healthy', 'vault-healthy', 'refresh'] as const) { test(`OC1 ${scenario} request observes only required foreground locks`, async () => { const fixture = await migrateServingFixture( diff --git a/packages/opencode/src/tests/opencode2-native-source.test.ts b/packages/opencode/src/tests/opencode2-native-source.test.ts index 17e256dd0..1bf37d877 100644 --- a/packages/opencode/src/tests/opencode2-native-source.test.ts +++ b/packages/opencode/src/tests/opencode2-native-source.test.ts @@ -32,6 +32,12 @@ import { fingerprintOf } from '@cortexkit/common-auth/store' import type { IntegrationOAuthMethodRegistration } from '@opencode/plugin/promise/integration' import { nativeQuotaCodec } from '../../../core/src/native-quota-codec' import { nativeVaultPaths } from '../../../core/src/native-vault-paths' +import { + holdRefreshJoin, + observeRefreshJoin, + refreshedJoinAccess, + refreshJoinMessage, +} from '../../../core/src/tests/native-refresh-join-fixture.ts' import { holdNativeRelayLock } from '../../../core/src/tests/native-relay-lock-fixture.ts' import { requestLockProbe } from '../../../core/src/tests/native-request-lock-fixture.ts' import { createTestLifetimeSuite } from '../../../core/src/tests/test-lifetime.ts' @@ -164,6 +170,105 @@ usageTests.test( }, ) +for (const scenario of ['proactive', '401', 'deadline'] as const) { + usageTests.test( + `OC2 refresh join ${scenario} preserves valid-token admission`, + async () => { + const owned: Array<() => Promise | void> = [] + usageTests.deferCleanup(async () => { + for (const cleanup of owned.reverse()) await cleanup() + }) + const held = holdRefreshJoin(usageTests) + const probe = requestLockProbe() + const fixture = await local( + { + onLockEvent: probe.onLockEvent, + refreshToken: held.refreshToken, + resolveIdentity: async () => ({ + accountUuid: identity as ProviderAccountUuid, + deviceId: 'fixture', + sessionId: 'fixture', + }), + }, + owned, + ) + const native = source(fixture.runtime, fixture.paths, owned) + const first = await native.authorizeAccount(scope, 'main') + if (first?.route !== 'oauth') + throw new Error('Missing first authorization') + expect(first.accessToken).toBe('sk-ant-oat01-native-only') + const owner = Promise.resolve( + scenario === 'proactive' + ? fixture.runtime.authorizeLocal('main', { intent: 'refresh' }) + : native.limitFromResponse!({ + ...context(first.provenance), + status: 401, + headers: new Headers(), + body: async () => '', + }), + ) + void owner.catch(() => {}) + let waiting: + | Promise>> + | undefined + try { + await held.entered + probe.begin() + waiting = Promise.resolve( + native.authorizeAccount( + { ...scope, sessionID: 'join-waiter' }, + 'main', + ), + ).then((auth) => { + probe.dispatch() + return auth + }) + if (scenario === 'proactive') { + expect(await observeRefreshJoin(waiting)).toMatchObject({ + status: 'fulfilled', + value: { accessToken: first.accessToken }, + }) + expect(held.completed()).toBe(false) + expect(probe.foreground).toEqual([]) + } else if (scenario === '401') { + expect((await observeRefreshJoin(waiting, 100)).status).toBe( + 'pending', + ) + held.release() + expect(await waiting).toMatchObject({ + accessToken: refreshedJoinAccess, + }) + const limit = await owner + expect(limit).toMatchObject({ status: 401 }) + if ( + !limit || + typeof limit.status !== 'number' || + !('reason' in limit) + ) + throw new Error('Missing 401 retry') + await native.chooseAccount({ + ...scope, + rerouteFrom: { accountId: 'main', limit }, + }) + expect(await native.authorizeAccount(scope, 'main')).toMatchObject({ + accessToken: refreshedJoinAccess, + }) + } else { + expect(await observeRefreshJoin(waiting, 3_500)).toMatchObject({ + status: 'rejected', + error: { message: refreshJoinMessage, code: 'ECONNRESET' }, + }) + expect(held.completed()).toBe(false) + expect(probe.afterDispatch).toEqual([]) + } + } finally { + held.release() + await Promise.allSettled([owner, ...(waiting ? [waiting] : [])]) + } + }, + ) +} + for (const scenario of ['local-healthy', 'vault-healthy', 'refresh'] as const) { usageTests.test( `OC2 ${scenario} request observes only required foreground locks`, diff --git a/packages/pi/src/tests/native-request-locks.test.ts b/packages/pi/src/tests/native-request-locks.test.ts index f483446bb..f4fabaebc 100644 --- a/packages/pi/src/tests/native-request-locks.test.ts +++ b/packages/pi/src/tests/native-request-locks.test.ts @@ -7,6 +7,12 @@ import { type NativeCustodyClient, } from '@cortexkit/anthropic-auth-core' import type { Api, Context, Model } from '@earendil-works/pi-ai' +import { + holdRefreshJoin, + observeRefreshJoin, + refreshedJoinAccess, + refreshJoinMessage, +} from '../../../core/src/tests/native-refresh-join-fixture.ts' import { requestLockProbe } from '../../../core/src/tests/native-request-lock-fixture.ts' import { createTestLifetimeSuite } from '../../../core/src/tests/test-lifetime.ts' import { closePiNativeRuntime, getPiNativeRuntime } from '../native.ts' @@ -49,9 +55,18 @@ const quota = () => ({ resetsAt: new Date(Date.now() + 86_400_000).toISOString(), }, }) -for (const scenario of ['local-healthy', 'vault-healthy', 'refresh'] as const) { +for (const scenario of [ + 'local-healthy', + 'vault-healthy', + 'refresh', + 'join-proactive', + 'join-401', + 'join-deadline', +] as const) { lifetimes.test( - `Pi ${scenario} request observes only required foreground locks`, + scenario.startsWith('join-') + ? `Pi refresh join ${scenario.slice(5)} preserves valid-token admission` + : `Pi ${scenario} request observes only required foreground locks`, () => trackPiTestBody( (async () => { @@ -86,6 +101,7 @@ for (const scenario of ['local-healthy', 'vault-healthy', 'refresh'] as const) { storagePath, ) const probe = requestLockProbe() + const held = holdRefreshJoin(lifetimes) const identity = fixtureAccountIdentity('main') const client: NativeCustodyClient = { async listScoped() { @@ -139,6 +155,13 @@ for (const scenario of ['local-healthy', 'vault-healthy', 'refresh'] as const) { : {}), local: { onLockEvent: probe.onLockEvent, + ...(scenario.startsWith('join-') + ? { + refreshToken: held.refreshToken, + resolveIdentity: async () => + getClaudeCodeIdentityForVerifiedAccount('main', identity), + } + : {}), ...(scenario === 'refresh' ? { now, @@ -168,18 +191,85 @@ for (const scenario of ['local-healthy', 'vault-healthy', 'refresh'] as const) { } await runtime.read() await runtime.drainRoutingPublications() + const sent: string[] = [] + const completedAtDispatch: boolean[] = [] globalThis.fetch = Object.assign( - async (input: Parameters[0]) => { + async (input: Parameters[0], init?: RequestInit) => { const url = input instanceof Request ? input.url : input.toString() if (!url.includes('/v1/messages')) return Response.json({}) + sent.push(new Headers(init?.headers).get('authorization') ?? '') + completedAtDispatch.push(held.completed()) probe.dispatch() + if ( + (scenario === 'join-401' || scenario === 'join-deadline') && + sent.length === 1 + ) + return Response.json( + { + type: 'error', + error: { + type: 'authentication_error', + message: 'Rejected fixture credential', + }, + }, + { status: 401 }, + ) return new Response( 'event: message_start\ndata: {"type":"message_start","message":{"usage":{"input_tokens":1,"output_tokens":0}}}\n\nevent: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":1}}\n\nevent: message_stop\ndata: {"type":"message_stop"}\n\n', ) }, { preconnect: originalFetch.preconnect }, ) + if (scenario.startsWith('join-')) { + const send = (sessionId: string) => + streamCortexKitAnthropic(model, context, { sessionId }).result() + const owner = + scenario === 'join-proactive' + ? runtime.authorizeLocal('main', { intent: 'refresh' }) + : send('join-owner') + void owner.catch(() => {}) + let waiting: ReturnType | undefined + try { + await held.entered + probe.begin() + waiting = send('join-waiter') + if (scenario === 'join-proactive') { + expect(await observeRefreshJoin(waiting)).toMatchObject({ + status: 'fulfilled', + value: { stopReason: 'stop' }, + }) + expect(completedAtDispatch).toEqual([false]) + expect(probe.foreground).toEqual([]) + } else if (scenario === 'join-401') { + expect((await observeRefreshJoin(waiting, 100)).status).toBe( + 'pending', + ) + expect(sent).toHaveLength(1) + held.release() + expect(await waiting).toMatchObject({ stopReason: 'stop' }) + expect(await owner).toMatchObject({ stopReason: 'stop' }) + expect(sent.slice(1)).toEqual([ + `Bearer ${refreshedJoinAccess}`, + `Bearer ${refreshedJoinAccess}`, + ]) + } else { + expect(await observeRefreshJoin(waiting, 3_500)).toMatchObject({ + status: 'fulfilled', + value: { + stopReason: 'error', + errorMessage: refreshJoinMessage, + }, + }) + expect(held.completed()).toBe(false) + expect(sent).toHaveLength(1) + } + } finally { + held.release() + await Promise.allSettled([owner, ...(waiting ? [waiting] : [])]) + } + return + } probe.begin() const result = await streamCortexKitAnthropic(model, context, { sessionId: `lock-proof-${directory}`, @@ -210,5 +300,6 @@ for (const scenario of ['local-healthy', 'vault-healthy', 'refresh'] as const) { ).toBe(true) })(), ), + scenario.startsWith('join-') ? 8_000 : undefined, ) } From 40f8fec148a2f0cc5a272ea5e95de7b2115ae41b Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Tue, 20 Oct 2026 23:31:25 +0200 Subject: [PATCH 2/3] mason: preserve account exclusions before refresh joining Always check serving admission before deciding whether a rejected token joins recovery, including after job completion. Preserve the existing validation-retry exception for explicitly caller-rejected material; refresh backoff, permanent errors and external account policy still win. Add red-first coordinator coverage for held proactive/401 recovery and policy exclusions without altering exchange locks, retries or deadlines. The reported CI assertion failed only in its no-backoff positive control: the original fallback bearer was still valid and could be served while proactive rotation ran. Per the parent's explicit decision, that control now requires dispatch from the selected fallback using either of its two known bearers, rather than requiring rotation to win a race. The excluded-backoff half is unchanged. Ran full Core, OpenCode index and Pi suites once: Core 2033 pass/1 fail (initially exposed rejected-token validation-retry behavior, fixed in production and the 151-test impacted selection passed); OpenCode index 304 pass/1 unrelated profile hook timeout, original CI case passed; Pi 317 pass. No full-suite retries or deadline increases. Final types, 9 host join cases, lint/format, catalogue tests, updated cause mutation and restored named baselines pass. --- mutations.toml | 4 +- .../core/src/native-refresh-coordinator.ts | 78 ++++++------ .../tests/native-refresh-coordinator.test.ts | 111 ++++++++++++++++++ packages/opencode/src/tests/index.test.ts | 11 +- 4 files changed, 165 insertions(+), 39 deletions(-) diff --git a/mutations.toml b/mutations.toml index 159445cce..e7eb8048e 100644 --- a/mutations.toml +++ b/mutations.toml @@ -328,8 +328,8 @@ broad_id = "{name}" id = "refresh-join-checks-cause" guards = "A proactive refresh must not delay a proven unexpired access token." file = "packages/core/src/native-refresh-coordinator.ts" -old = "if (!jobRejectsAccess(job, row)) {" -new = "if (!job) { // NON-VACUITY BREAK: join every existing refresh" +old = "if (current.status !== 'usable' || !jobRejectsAccess(job, row))" +new = "if (current.status !== 'usable' || !job) // NON-VACUITY BREAK: join every existing refresh" test_file = "packages/opencode/src/tests/native-serving.test.ts" runner = "command" command = ["bun", "scripts/run-mutation-bun.ts", "packages/opencode/src/tests/native-serving.test.ts", "{test}"] diff --git a/packages/core/src/native-refresh-coordinator.ts b/packages/core/src/native-refresh-coordinator.ts index 5fdb2a8fd..fbb0adbe4 100644 --- a/packages/core/src/native-refresh-coordinator.ts +++ b/packages/core/src/native-refresh-coordinator.ts @@ -630,8 +630,6 @@ export function createNativeRefreshCoordinator( credential.expires <= now() ) return refuse('access-unavailable') - if (credential.access === request.rejectedAccessToken) - return refuse('rejected-access-unchanged') const subject: NativeKnownCredentialSubject = copySubject({ binding: { ...expected, identity: expected.identity }, credentialFingerprint: fingerprintOf(credential), @@ -647,7 +645,18 @@ export function createNativeRefreshCoordinator( } catch (cause) { throw new Error('Native admission callback failed', { cause }) } - if (admission.status === 'blocked') return refuse(admission.reason) + // A validation retry delays another identity lookup, not replacement of an + // access token the caller already received a 401 for. Other exclusions still win. + if ( + admission.status === 'blocked' && + !( + admission.reason === 'validation-backoff' && + credential.access === request.rejectedAccessToken + ) + ) + return refuse(admission.reason) + if (credential.access === request.rejectedAccessToken) + return refuse('rejected-access-unchanged') if (admission.status !== 'proven') return refuse('proof-missing') if ( !nativeLocalCredentialValidationMatches( @@ -1332,30 +1341,30 @@ export function createNativeRefreshCoordinator( const row = await rowFor(request.binding) job = jobs.get(key) ?? job recordRejection() - if (!jobRejectsAccess(job, row)) { - const current = await admit( - request, - request.binding, - row, - 'current', - false, - ) - if ( - current.status !== 'refused' || - ![ - 'proof-missing', - 'identity-unproven', - 'access-unavailable', - 'rejected-access-unchanged', - ].includes(current.reason) - ) { - // Admission yields to other callers. A 401 recorded during that - // read must prevent dispatch of the token it rejected. - job = jobs.get(key) ?? job - recordRejection() - if (current.status !== 'usable' || !jobRejectsAccess(job, row)) - return current - } + // A pending recovery cannot override custody, quota or refresh-error + // exclusions. Check admission before deciding whether a token must wait. + const current = await admit( + request, + request.binding, + row, + 'current', + false, + ) + if ( + current.status !== 'refused' || + ![ + 'proof-missing', + 'identity-unproven', + 'access-unavailable', + 'rejected-access-unchanged', + ].includes(current.reason) + ) { + // Admission yields to other callers. A 401 recorded during that + // read must prevent dispatch of the token it rejected. + job = jobs.get(key) ?? job + recordRejection() + if (current.status !== 'usable' || !jobRejectsAccess(job, row)) + return current } } catch (caught) { return { @@ -1413,19 +1422,20 @@ export function createNativeRefreshCoordinator( } : request.binding const row = await rowFor(request.binding) - if (jobRejectsAccess(job, row)) - return { - status: 'refused', - reason: 'rejected-access-unchanged', - persisted: result.persisted, - } - return await admit( + const current = await admit( request, expected, row, result.source, result.persisted, ) + if (current.status === 'usable' && jobRejectsAccess(job, row)) + return { + status: 'refused', + reason: 'rejected-access-unchanged', + persisted: result.persisted, + } + return current } catch (error) { return { status: 'failed', diff --git a/packages/core/src/tests/native-refresh-coordinator.test.ts b/packages/core/src/tests/native-refresh-coordinator.test.ts index 57c9ab0cc..8277a1ee5 100644 --- a/packages/core/src/tests/native-refresh-coordinator.test.ts +++ b/packages/core/src/tests/native-refresh-coordinator.test.ts @@ -3180,6 +3180,117 @@ for (const echo of ['captured', 'newer'] as const) { }) } +for (const cause of ['proactive', '401'] as const) { + test(`cause-keyed fast path refuses active refresh backoff during ${cause} refresh`, async () => { + const f = await fixture() + await f.add('a', 'A') + const proof = await f.seedProof('a') + const binding = await f.binding('a') + const entered = barrier() + const finish = barrier() + let backoff = false + let exchanges = 0 + const c = f.coordinator({ + readRestrictions: async (subject) => + backoff + ? { + status: 'blocked', + reason: 'refresh-backoff', + credentialFingerprint: subject.credentialFingerprint, + } + : allowed, + readAdmission: async () => + backoff + ? { status: 'blocked', reason: 'refresh-backoff' } + : { status: 'proven', validation: proof }, + refreshToken: async () => { + exchanges++ + entered.release() + await finish.promise + return { ...credential('new'), expiresIn: 3600 } + }, + }) + const owner = c.refresh({ + mode: 'local', + binding, + ...(cause === '401' ? { rejectedAccessToken: 'access-a' } : {}), + }) + void owner.catch(() => {}) + try { + await entered.promise + backoff = true + expect( + await observeRefreshJoin( + c.authorize({ intent: 'serve', mode: 'local', binding }), + ), + ).toMatchObject({ + status: 'fulfilled', + value: { status: 'refused', reason: 'refresh-backoff' }, + }) + expect(exchanges).toBe(1) + } finally { + finish.release() + await owner + } + }) +} + +for (const reason of [ + 'invalid-grant', + 'quota-ineligible', + 'account-disabled', + 'local-mode-unavailable', +] as const) { + test(`pending 401 refresh never bypasses ${reason} serving exclusion`, async () => { + const f = await fixture() + await f.add('a', 'A') + const proof = await f.seedProof('a') + const binding = await f.binding('a') + const entered = barrier() + const finish = barrier() + let blocked = false + const c = f.coordinator({ + readAdmission: async () => + blocked + ? { status: 'blocked', reason } + : { status: 'proven', validation: proof }, + refreshToken: async () => { + entered.release() + await finish.promise + return { ...credential('new'), expiresIn: 3600 } + }, + }) + const owner = c.refresh({ + mode: 'local', + binding, + rejectedAccessToken: 'access-a', + }) + void owner.catch(() => {}) + try { + await entered.promise + blocked = true + for (const rejectedAccessToken of [undefined, 'access-a']) { + expect( + await observeRefreshJoin( + c.authorize({ + intent: 'serve', + mode: 'local', + binding, + rejectedAccessToken, + }), + ), + ).toMatchObject({ + status: 'fulfilled', + value: { status: 'refused', reason }, + }) + } + } finally { + finish.release() + await owner + } + }) +} + test('401 recovery admits a proven different access version before the shared job completes', async () => { const f = await fixture() await f.add('a', 'A') diff --git a/packages/opencode/src/tests/index.test.ts b/packages/opencode/src/tests/index.test.ts index cd708bcac..6fe231228 100644 --- a/packages/opencode/src/tests/index.test.ts +++ b/packages/opencode/src/tests/index.test.ts @@ -5398,9 +5398,14 @@ describe('quota header feed extended integration', () => { ) } else { expect(response.status).toBe(200) - expect(messageAuthorizations).toContain( - 'Bearer sk-ant-oat01-refreshed-fallback-access', - ) + expect(messageAuthorizations.length).toBeGreaterThan(0) + // The original bearer is still valid, so dispatch need not wait for + // proactive rotation. Both bearers belong to the admitted fallback. + for (const authorization of messageAuthorizations) + expect([ + 'Bearer sk-ant-oat01-unknown-fallback-access', + 'Bearer sk-ant-oat01-refreshed-fallback-access', + ]).toContain(authorization) } } }) From d466263b5173cfeff97284ac9cfcf8fdf7e5ca7f Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Tue, 20 Oct 2026 23:33:47 +0200 Subject: [PATCH 3/3] docs: move the dependency entry back under Unreleased A rebase conflict resolution placed it under the published 2.1.2 heading. --- CHANGELOG.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4b2a499ae..c1aa99e00 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,14 +13,14 @@ This repo is a CortexKit-maintained Anthropic auth monorepo for OpenCode and Pi. - Valid tokens no longer wait for proactive refreshes; requests waiting for a rejected-token refresh fail retryably after two seconds while the refresh continues. -## 2.1.2 - -- Requests no longer wait on relay or session-pin bookkeeping locks; relay settings and tokens are checked together before sending. - ### Dependencies - OpenTUI core and solid move to 0.6.6 together and the OpenCode 2 plugin SDK packages to 2.0.32, so each installs once. solid-js stays at 1.9.12, the exact version OpenTUI Solid 0.6.6 requires. +## 2.1.2 + +- Requests no longer wait on relay or session-pin bookkeeping locks; relay settings and tokens are checked together before sending. + ## 2.1.1 ### Packaging