diff --git a/packages/opencode/src/index.ts b/packages/opencode/src/index.ts index dfc7a6c5..5d0b4d55 100644 --- a/packages/opencode/src/index.ts +++ b/packages/opencode/src/index.ts @@ -625,6 +625,7 @@ async function sendIgnoredMessage( noReply?: boolean beforeActiveAssistant?: boolean canSend?: () => boolean + onMessageId?: (messageId: string) => void } = {}, ): Promise { const session = ctx.client.session as PluginSessionClient | undefined @@ -653,6 +654,7 @@ async function sendIgnoredMessage( // A new user prompt can start while that request is in flight, so re-check the // caller's delivery lease immediately before inserting the ignored message. if (options.canSend && !options.canSend()) return false + if (request.body.messageID) options.onMessageId?.(request.body.messageID) if (typeof session?.promptAsync === 'function') { await session.promptAsync(request) @@ -1087,6 +1089,8 @@ const anthropicAuthPlugin = async ( const desktopNoticeSafeSessions = new Set() const desktopNoticeLatestUserMessages = new Map() const desktopNoticeIdleUserMessages = new Map() + const desktopNoticeMessageIds = new Map>() + const desktopNoticeUserRevisions = new Map() const desktopNoticeProbes = new Map() const stickySessionRouter = new StickySessionRouter({ path: @@ -4044,6 +4048,11 @@ const anthropicAuthPlugin = async ( function queueDesktopNotice(sessionId: string, text: string) { if (isTuiConnected(sessionId)) return + logger.debug('fable-fallback', 'Desktop notification queued', { + session: sessionId, + safe: desktopNoticeSafeSessions.has(sessionId), + text, + }) // OpenCode's prompt endpoints run revert cleanup before honoring noReply. // OpenCode awaits event handlers before it evaluates the loop exit condition. // Escape the post-idle session update, then probe outside that critical section. @@ -4063,6 +4072,36 @@ const anthropicAuthPlugin = async ( } } + function rememberDesktopNoticeMessageId( + sessionId: string, + messageId: string, + ) { + const messageIds = + desktopNoticeMessageIds.get(sessionId) ?? new Set() + messageIds.add(messageId) + while (messageIds.size > 8) { + const oldest = messageIds.values().next().value + if (oldest) messageIds.delete(oldest) + else break + } + desktopNoticeMessageIds.delete(sessionId) + desktopNoticeMessageIds.set(sessionId, messageIds) + while (desktopNoticeMessageIds.size > 128) { + const oldest = desktopNoticeMessageIds.keys().next().value + if (oldest) desktopNoticeMessageIds.delete(oldest) + else break + } + } + + function grantDesktopNoticeLease(sessionId: string) { + desktopNoticeSafeSessions.add(sessionId) + while (desktopNoticeSafeSessions.size > 128) { + const oldest = desktopNoticeSafeSessions.values().next().value + if (oldest) desktopNoticeSafeSessions.delete(oldest) + else break + } + } + function scheduleDesktopNoticeProbe(sessionId: string, attempt = 0) { if ( !pendingDesktopNotices.has(sessionId) || @@ -4090,6 +4129,12 @@ const anthropicAuthPlugin = async ( } async function flushDesktopNoticesIfIdle(sessionId: string, attempt: number) { + logger.debug('fable-fallback', 'Desktop notification flush considered', { + session: sessionId, + attempt, + safe: desktopNoticeSafeSessions.has(sessionId), + pending: pendingDesktopNotices.has(sessionId), + }) if ( !desktopNoticeSafeSessions.has(sessionId) || !pendingDesktopNotices.has(sessionId) @@ -4150,13 +4195,28 @@ const anthropicAuthPlugin = async ( return } try { + const userRevision = desktopNoticeUserRevisions.get(sessionId) ?? 0 const sent = await sendIgnoredMessage(ctx, sessionId, text, { noReply: true, beforeActiveAssistant: true, canSend: () => desktopNoticeSafeSessions.has(sessionId), + onMessageId: (messageId) => + rememberDesktopNoticeMessageId(sessionId, messageId), }) if (!sent) return queue.shift() + if ( + !desktopNoticeSafeSessions.has(sessionId) && + pendingDesktopNotices.get(sessionId)?.[0] && + (desktopNoticeUserRevisions.get(sessionId) ?? 0) === userRevision + ) { + // OpenCode marks its own noReply insertion busy without publishing a + // new idle event. Re-enter only through the same live-status probe; + // a genuine user message changes the revision and blocks this path. + grantDesktopNoticeLease(sessionId) + scheduleDesktopNoticeProbe(sessionId) + return + } } catch (error) { logger.warn('fable-fallback', 'Desktop notification failed', { session: sessionId, @@ -5237,21 +5297,32 @@ const anthropicAuthPlugin = async ( if (!sessionId) return if (value.type === 'message.updated' && info?.role === 'user') { - if (typeof info.id === 'string') { - desktopNoticeLatestUserMessages.set(sessionId, info.id) - if ( - desktopNoticeSafeSessions.has(sessionId) && - desktopNoticeIdleUserMessages.get(sessionId) !== info.id - ) { - // A new user message can precede OpenCode's busy status event. Revoke - // the idle-delivery lease immediately so an ignored notice cannot - // become the active request parent and duplicate a provider turn. - // Repeated updates for the user message that produced the current - // idle event are harmless and must not suppress delivery forever. + // promptAsync publishes ignored notices as user-message updates. They do + // not start a provider turn, so only genuine user messages revoke the lease. + const isDesktopNotice = + typeof info.id === 'string' && + desktopNoticeMessageIds.get(sessionId)?.has(info.id) + if (!isDesktopNotice) { + desktopNoticeUserRevisions.set( + sessionId, + (desktopNoticeUserRevisions.get(sessionId) ?? 0) + 1, + ) + if (typeof info.id === 'string') { + desktopNoticeLatestUserMessages.set(sessionId, info.id) + if ( + desktopNoticeSafeSessions.has(sessionId) && + desktopNoticeIdleUserMessages.get(sessionId) !== info.id + ) { + // A new user message can precede OpenCode's busy status event. Revoke + // the idle-delivery lease immediately so an ignored notice cannot + // become the active request parent and duplicate a provider turn. + // Repeated updates for the user message that produced the current + // idle event are harmless and must not suppress delivery forever. + desktopNoticeSafeSessions.delete(sessionId) + } + } else { desktopNoticeSafeSessions.delete(sessionId) } - } else { - desktopNoticeSafeSessions.delete(sessionId) } } @@ -5274,12 +5345,7 @@ const anthropicAuthPlugin = async ( // live status map is still idle. OpenCode 1.18 no longer guarantees a // session.updated event after session.idle, so that event cannot be used // as the release signal. - desktopNoticeSafeSessions.add(sessionId) - while (desktopNoticeSafeSessions.size > 128) { - const oldest = desktopNoticeSafeSessions.values().next().value - if (oldest) desktopNoticeSafeSessions.delete(oldest) - else break - } + grantDesktopNoticeLease(sessionId) scheduleDesktopNoticeProbe(sessionId) } @@ -5297,6 +5363,8 @@ const anthropicAuthPlugin = async ( desktopNoticeSafeSessions.delete(sessionId) desktopNoticeLatestUserMessages.delete(sessionId) desktopNoticeIdleUserMessages.delete(sessionId) + desktopNoticeMessageIds.delete(sessionId) + desktopNoticeUserRevisions.delete(sessionId) for (const recoveryKey of pendingRecoveryDesktopNotices.keys()) { if (recoveryKey.startsWith(`${sessionId}\0`)) { pendingRecoveryDesktopNotices.delete(recoveryKey) diff --git a/packages/opencode/src/tests/custody-handle-manifest.test.ts b/packages/opencode/src/tests/custody-handle-manifest.test.ts index 38eb88d1..e81dca50 100644 --- a/packages/opencode/src/tests/custody-handle-manifest.test.ts +++ b/packages/opencode/src/tests/custody-handle-manifest.test.ts @@ -1655,27 +1655,30 @@ describe('withCustodyManifestLock', () => { test.serial('reports a held lock as lock_busy', async () => { await withTempDirectory(async (directory) => { const path = join(directory, 'handles.json') - const firstEntered = Promise.withResolvers() - const releaseFirst = Promise.withResolvers() + const lockPath = `${path}.lock` + const times = [0, 29, 30] + let timeIndex = 0 + await fs.mkdir(lockPath, { mode: 0o700 }) + await fs.writeFile( + join(lockPath, 'owner'), + `${JSON.stringify({ + tenant: 'anthropic-auth', + pid: process.pid, + claimed_at_ms: 0, + nonce: 'held-test', + })}\n`, + ) __setCustodyManifestLockTestOptions({ ttlMs: 30, retryMinMs: 1, retryMaxMs: 1, - renewalIntervalMs: 5, - }) - const first = withCustodyManifestLock(path, async () => { - firstEntered.resolve() - await releaseFirst.promise + now: () => times[Math.min(timeIndex++, times.length - 1)]!, }) - try { - await firstEntered.promise - await expect( - withCustodyManifestLock(path, async () => 'acquired'), - ).rejects.toMatchObject({ code: 'lock_busy' }) - } finally { - releaseFirst.resolve() - await first - } + + await expect( + withCustodyManifestLock(path, async () => 'acquired'), + ).rejects.toMatchObject({ code: 'lock_busy' }) + await expect(fs.lstat(lockPath)).resolves.toBeDefined() }) }) diff --git a/packages/opencode/src/tests/index.test.ts b/packages/opencode/src/tests/index.test.ts index 16b6db32..40b58a76 100644 --- a/packages/opencode/src/tests/index.test.ts +++ b/packages/opencode/src/tests/index.test.ts @@ -734,6 +734,26 @@ async function getPlugin( return plugin } +async function withoutClaustrumWarmupDeadline( + fn: () => Promise, +): Promise { + const originalSetTimeout = globalThis.setTimeout + const setTimeoutImpl = (( + ...arguments_: Parameters + ) => + arguments_[1] === 100 + ? ({ unref() {} } as ReturnType) + : originalSetTimeout(...arguments_)) as typeof globalThis.setTimeout + const setTimeoutSpy = spyOn(globalThis, 'setTimeout').mockImplementation( + setTimeoutImpl, + ) + try { + return await fn() + } finally { + setTimeoutSpy.mockRestore() + } +} + function installRelayResponseStart( status: number, errorEvent?: { status?: number; message?: string }, @@ -2442,12 +2462,14 @@ describe('fallback Claustrum credential resolution', () => { }, ) try { - const plugin = await getPlugin(undefined, undefined, { - claustrumConnector: manifestConnector( - [], - new Map([[legacyHandle, 'migration-order-access']]), - ), - }) + const plugin = await withoutClaustrumWarmupDeadline(() => + getPlugin(undefined, undefined, { + claustrumConnector: manifestConnector( + [], + new Map([[legacyHandle, 'migration-order-access']]), + ), + }), + ) expect(stateAtManifestWrite).toContain(legacyHandle) expect(await readFile(accountStatePath, 'utf8')).not.toContain( legacyHandle, @@ -2588,11 +2610,16 @@ describe('fallback Claustrum credential resolution', () => { }, ) - async function withShortManifestLockTiming(fn: () => Promise) { + async function withFixedManifestLockClock( + fn: () => Promise, + now?: () => number, + ) { + const fixedNow = Date.now() __setCustodyManifestLockTestOptions({ ttlMs: 150, retryMinMs: 5, retryMaxMs: 5, + now: now ?? (() => fixedNow), }) try { return await fn() @@ -2621,14 +2648,16 @@ describe('fallback Claustrum credential resolution', () => { const manifestPath = await writeManifest([]) const restore = await configureClaustrumConnection() const calls: CredentialCall[] = [] - const plugin = await getPlugin(undefined, undefined, { - claustrumConnector: - input.connector?.(calls) ?? - manifestConnector( - calls, - new Map([[input.handle, `${input.label}-access`]]), - ), - }) + const plugin = await withoutClaustrumWarmupDeadline(() => + getPlugin(undefined, undefined, { + claustrumConnector: + input.connector?.(calls) ?? + manifestConnector( + calls, + new Map([[input.handle, `${input.label}-access`]]), + ), + }), + ) return { calls, manifestPath, plugin, restore } } @@ -2671,12 +2700,14 @@ describe('fallback Claustrum credential resolution', () => { const manifestPath = await writeManifest([]) const restore = await configureClaustrumConnection() const calls: CredentialCall[] = [] - const plugin = await getPlugin(undefined, undefined, { - claustrumConnector: manifestConnector( - calls, - new Map([[legacyHandle, 'retry-migration-access']]), - ), - }) + const plugin = await withoutClaustrumWarmupDeadline(() => + getPlugin(undefined, undefined, { + claustrumConnector: manifestConnector( + calls, + new Map([[legacyHandle, 'retry-migration-access']]), + ), + }), + ) try { await plugin.__fallbackRefreshReady expect( @@ -2820,72 +2851,59 @@ describe('fallback Claustrum credential resolution', () => { test.serial( 'reports a corrupt manifest lock after its bounded wait and keeps the legacy handle', async () => { - await withShortManifestLockTiming(async () => { - await useTempAccountFile( - manifestStorage({ label: 'fresh-lock', legacy: legacyHandle }), - ) - const manifestPath = await writeManifest([]) - const restore = await configureClaustrumConnection() - const lockPath = `${manifestPath}.lock` - await mkdir(lockPath, { mode: 0o700 }) - await writeFile( - join(lockPath, 'owner'), - `${JSON.stringify({ claimed_at_ms: Date.now(), tenant: 'test' })}\n`, - ) - const logs: LogTestRecord[] = [] - __setLogTestSink((record) => logs.push(record)) - const startedAt = Date.now() - let plugin: Awaited> | undefined - try { - plugin = await Promise.race([ - getPlugin(undefined, undefined, { - claustrumConnector: manifestConnector( - [], - new Map([[legacyHandle, 'fresh-lock-access']]), - ), - }), - Bun.sleep(1_000).then(() => { - throw new Error('manifest lock busy did not respect its deadline') - }), - ]) - for (let attempt = 0; attempt < 100; attempt++) { - if ( + const times = [0, 150] + let timeIndex = 0 + await withFixedManifestLockClock( + async () => { + await useTempAccountFile( + manifestStorage({ label: 'fresh-lock', legacy: legacyHandle }), + ) + const manifestPath = await writeManifest([]) + const restore = await configureClaustrumConnection() + const lockPath = `${manifestPath}.lock` + await mkdir(lockPath, { mode: 0o700 }) + await writeFile( + join(lockPath, 'owner'), + `${JSON.stringify({ claimed_at_ms: 0, tenant: 'test' })}\n`, + ) + const logs: LogTestRecord[] = [] + __setLogTestSink((record) => logs.push(record)) + let plugin: Awaited> | undefined + try { + plugin = await withoutClaustrumWarmupDeadline(() => + getPlugin(undefined, undefined, { + claustrumConnector: manifestConnector( + [], + new Map([[legacyHandle, 'fresh-lock-access']]), + ), + }), + ) + expect( logs.some( (record) => record.message === 'manifest write failed' && record.payload?.reason === 'manifest lock owner invalid', - ) - ) - break - await Bun.sleep(10) + ), + ).toBe(true) + expect( + await readFile( + getAccountStatePath(process.env.OPENCODE_ANTHROPIC_AUTH_FILE!), + 'utf8', + ), + ).toContain(legacyHandle) + } finally { + __setLogTestSink(null) + await plugin?.dispose?.() + restore() } - const elapsedMs = Date.now() - startedAt - expect(elapsedMs).toBeGreaterThanOrEqual(120) - expect(elapsedMs).toBeLessThan(1_000) - expect( - logs.some( - (record) => - record.message === 'manifest write failed' && - record.payload?.reason === 'manifest lock owner invalid', - ), - ).toBe(true) - expect( - await readFile( - getAccountStatePath(process.env.OPENCODE_ANTHROPIC_AUTH_FILE!), - 'utf8', - ), - ).toContain(legacyHandle) - } finally { - __setLogTestSink(null) - await plugin?.dispose?.() - restore() - } - }) + }, + () => times[Math.min(timeIndex++, times.length - 1)]!, + ) }, ) test.serial('renames a stale manifest lock before writing', async () => { - await withShortManifestLockTiming(async () => { + await withFixedManifestLockClock(async () => { await useTempAccountFile( manifestStorage({ label: 'stale-lock', legacy: legacyHandle }), ) @@ -2945,12 +2963,14 @@ describe('fallback Claustrum credential resolution', () => { }, ) try { - const plugin = await getPlugin(undefined, undefined, { - claustrumConnector: manifestConnector( - [], - new Map([[legacyHandle, 'lock-owner-access']]), - ), - }) + const plugin = await withoutClaustrumWarmupDeadline(() => + getPlugin(undefined, undefined, { + claustrumConnector: manifestConnector( + [], + new Map([[legacyHandle, 'lock-owner-access']]), + ), + }), + ) expect(owner).toMatchObject({ tenant: 'anthropic-auth' }) expect(typeof owner?.claimed_at_ms).toBe('number') await expect(fs.stat(lockPath)).rejects.toThrow() @@ -2966,7 +2986,7 @@ describe('fallback Claustrum credential resolution', () => { test.serial( 'preserves two concurrent legacy migrations in one manifest', async () => { - await withShortManifestLockTiming(async () => { + await withFixedManifestLockClock(async () => { const storageA = fallbackWithClaustrum({ id: 'fallback-a', label: 'migration-a', @@ -2976,8 +2996,10 @@ describe('fallback Claustrum credential resolution', () => { }) await useTempAccountFile(storageA) const accountPathA = process.env.OPENCODE_ANTHROPIC_AUTH_FILE! + const accountPathB = join(tempConfigDir!, 'anthropic-auth-b.json') const manifestPath = await writeManifest([]) const restore = await configureClaustrumConnection() + const firstEntered = deferred() const entered = deferred() const release = deferred() let credentialGets = 0 @@ -2988,6 +3010,7 @@ describe('fallback Claustrum credential resolution', () => { if (method !== 'credential.get') throw new Error(`unexpected method: ${method}`) credentialGets += 1 + if (credentialGets === 1) firstEntered.resolve() if (credentialGets === 2) entered.resolve() await release.promise return credentialResponse( @@ -2996,27 +3019,8 @@ describe('fallback Claustrum credential resolution', () => { ) }, ) - const pluginA = await getPlugin(undefined, undefined, { - claustrumConnector: concurrentConnector, - }) - - const accountPathB = join(tempConfigDir!, 'anthropic-auth-b.json') - const storageB = fallbackWithClaustrum({ - id: 'fallback-b', - label: 'migration-b', - enabled: true, - claustrumHandle: `ckh_${'B'.repeat(43)}`, - claustrum: { mode: 'claustrum' }, - }) - await saveAccounts(storageB, accountPathB) - process.env.OPENCODE_ANTHROPIC_AUTH_FILE = accountPathB - process.env.OPENCODE_ANTHROPIC_AUTH_SIDEBAR_STATE_FILE = join( - tempConfigDir!, - 'sidebar-state-b.json', - ) - const pluginB = await getPlugin(undefined, undefined, { - claustrumConnector: concurrentConnector, - }) + let pluginA: Awaited> | undefined + let pluginB: Awaited> | undefined const originalRename = fs.rename const secondManifestRename = deferred() let manifestRenames = 0 @@ -3032,21 +3036,36 @@ describe('fallback Claustrum credential resolution', () => { }, ) try { - await Promise.race([ - entered.promise, - Bun.sleep(1_000).then(() => { - throw new Error( - `concurrent credential calls did not both start (${credentialGets})`, - ) - }), - ]) - release.resolve() - await Promise.race([ - secondManifestRename.promise, - Bun.sleep(1_000).then(() => { - throw new Error('concurrent migrations did not finish') - }), - ]) + await withoutClaustrumWarmupDeadline(async () => { + const pluginAPromise = getPlugin(undefined, undefined, { + claustrumConnector: concurrentConnector, + }) + await firstEntered.promise + + const storageB = fallbackWithClaustrum({ + id: 'fallback-b', + label: 'migration-b', + enabled: true, + claustrumHandle: `ckh_${'B'.repeat(43)}`, + claustrum: { mode: 'claustrum' }, + }) + await saveAccounts(storageB, accountPathB) + process.env.OPENCODE_ANTHROPIC_AUTH_FILE = accountPathB + process.env.OPENCODE_ANTHROPIC_AUTH_SIDEBAR_STATE_FILE = join( + tempConfigDir!, + 'sidebar-state-b.json', + ) + const pluginBPromise = getPlugin(undefined, undefined, { + claustrumConnector: concurrentConnector, + }) + await entered.promise + release.resolve() + ;[pluginA, pluginB] = await Promise.all([ + pluginAPromise, + pluginBPromise, + ]) + }) + await secondManifestRename.promise const manifest = JSON.parse(await readFile(manifestPath, 'utf8')) as { providers: Array<{ provider: string @@ -3065,8 +3084,8 @@ describe('fallback Claustrum credential resolution', () => { ]) } finally { rename.mockRestore() - await pluginA.dispose?.() - await pluginB.dispose?.() + await pluginA?.dispose?.() + await pluginB?.dispose?.() restore() } }) @@ -19021,6 +19040,13 @@ describe('auth.loader', () => { return {} }, ) + const switchNoticeCompletion = deferred() + let holdSwitchNotice = true + mockClient.session.promptAsync = mock(async () => { + if (!holdSwitchNotice) return + holdSwitchNotice = false + await switchNoticeCompletion.promise + }) const plugin = await getPlugin(mockClient) const result = await plugin.auth.loader( () => @@ -19353,6 +19379,38 @@ describe('auth.loader', () => { await restored.text() expect(normalModels.at(-1)).toBe('claude-fable-5') + await waitForSidebarState((state) => + Boolean( + state.fableRecoveries?.some( + (recovery) => + recovery.sessionId === 'ses_fable_filter' && + recovery.mode === 'fable', + ), + ), + ) + await plugin.event?.({ + event: { + type: 'message.updated', + properties: { + info: { + id: switchNotificationMessageId, + sessionID: 'ses_fable_filter', + role: 'user', + }, + }, + }, + }) + await plugin.event?.({ + event: { + type: 'session.status', + properties: { + sessionID: 'ses_fable_filter', + status: { type: 'busy' }, + }, + }, + }) + switchNoticeCompletion.resolve() + await waitForMockCall({ mock: { get calls() {