diff --git a/CHANGELOG.md b/CHANGELOG.md index f9c197b3..c1aa99e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,14 +9,18 @@ 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. -## 2.1.2 +### Accounts -- Requests no longer wait on relay or session-pin bookkeeping locks; relay settings and tokens are checked together before sending. +- Valid tokens no longer wait for proactive refreshes; requests waiting for a rejected-token refresh fail retryably after two seconds while the refresh continues. ### 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 diff --git a/mutations.toml b/mutations.toml index 034fbd2f..e7eb8048 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 (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}"] +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 faaaaa3d..fbb0adbe 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) @@ -594,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), @@ -611,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( @@ -1282,12 +1327,26 @@ 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') { + 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), + ) + } + recordRejection() + if (request.intent === 'serve') { try { + const row = await rowFor(request.binding) + job = jobs.get(key) ?? job + recordRejection() + // 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, - await rowFor(request.binding), + row, 'current', false, ) @@ -1299,8 +1358,14 @@ export function createNativeRefreshCoordinator( 'access-unavailable', 'rejected-access-unchanged', ].includes(current.reason) - ) - return current + ) { + // 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 +1378,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 +1402,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,13 +1422,20 @@ export function createNativeRefreshCoordinator( } : request.binding const row = await rowFor(request.binding) - 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/native-transient-account.ts b/packages/core/src/native-transient-account.ts index e4e91b4c..3c7ea43e 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 9e693504..8277a1ee 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,212 @@ 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') + 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 00000000..5fdffa88 --- /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/index.test.ts b/packages/opencode/src/tests/index.test.ts index cd708bca..6fe23122 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) } } }) diff --git a/packages/opencode/src/tests/native-serving.test.ts b/packages/opencode/src/tests/native-serving.test.ts index 19567ab6..a5d69f5e 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 17e256dd..1bf37d87 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 f483446b..f4fabaeb 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, ) }