From 6ce51045f9dd11188d4d7919a29ed02d3db68349 Mon Sep 17 00:00:00 2001 From: iceteaSA <171169159+iceteaSA@users.noreply.github.com> Date: Fri, 11 Sep 2026 19:18:25 +0200 Subject: [PATCH 1/3] fix(opencode): run one RPC server per project directory in a shared process The plugin kept one RPC server handle per process on `globalThis.__anthropicAuthRpcServer`. Every plugin instantiation stopped the existing handle first, and `stop()` unlinks `port-.json` from the directory that handle was started with. So once a process instantiated the plugin for a second project directory, the first project's port file was deleted and its server stopped: that project's TUI found nothing to poll and `/claude-*` opened no modal for the life of the process, with no error anywhere. One process legitimately serves several directories -- request routing accepts `?directory=`/`x-opencode-directory` and the plugin factory is scoped per directory (opencode `plugin/index.ts:134-179`, `workspace-routing.ts:86-88`). Key the servers by resolved RPC directory instead. Re-instantiating the same directory stops and replaces as before; a different directory starts an additional server and leaves the others alone. Bound the registry with `Hooks.dispose` (declared in `@opencode-ai/plugin` 1.18.21 and invoked by opencode's instance finalizer, `plugin/index.ts:265-278`; disposers also run on per-project reload, `project/instance-store.ts:126-145`). Teardown acts only when the registry entry is still its own handle: the port filename is identical for every server a process starts in one directory, so a dispose arriving after a same-directory replace would otherwise unlink the live successor's port file and put that project back in the dark. Cleanup steps are isolated from the RPC shutdown so a failing one cannot skip it. `stop()` unlinks the port file only when it still names its own port and pid, so a stale handle cannot remove a successor's file even if a future path forgets the identity check. The two defences protect the same observable and would mask each other, so each has its own test and each was mutated alone: removing the identity check reddens only the dispose-behaviour test, removing the port/pid match reddens only the stale-server test. A third test drives real HTTP against both directories' servers and asserts each answers through its own instance's closure -- a regression collapsing them onto one shared closure passes every other test here. --- packages/opencode/src/index.ts | 56 ++- packages/opencode/src/rpc/rpc-server.ts | 15 +- .../src/tests/rpc-multi-project.test.ts | 335 ++++++++++++++++++ .../opencode/src/tests/rpc-server.test.ts | 23 ++ 4 files changed, 416 insertions(+), 13 deletions(-) create mode 100644 packages/opencode/src/tests/rpc-multi-project.test.ts diff --git a/packages/opencode/src/index.ts b/packages/opencode/src/index.ts index 5822c771..e1b75f59 100644 --- a/packages/opencode/src/index.ts +++ b/packages/opencode/src/index.ts @@ -181,7 +181,7 @@ import { stickyRouteFamilyForModel, tokenFingerprint, } from '@cortexkit/anthropic-auth-core' -import type { Plugin } from '@opencode-ai/plugin' +import type { Hooks, Plugin } from '@opencode-ai/plugin' import { applyCacheDiagnosticsOptIn, @@ -2806,20 +2806,31 @@ const anthropicAuthPlugin = async ( } let rpcServer: RpcServerHandle | null = null + let rpcDir: string | null = null if (ctx.directory) { const rpcGlobal = globalThis as { __anthropicAuthRpcServer?: RpcServerHandle + __anthropicAuthRpcServers?: Map } - if (rpcGlobal.__anthropicAuthRpcServer) { - await rpcGlobal.__anthropicAuthRpcServer.stop().catch(() => {}) - rpcGlobal.__anthropicAuthRpcServer = undefined + rpcDir = getRpcDir(ctx.directory) + const rpcServers = + rpcGlobal.__anthropicAuthRpcServers ?? new Map() + rpcGlobal.__anthropicAuthRpcServers = rpcServers + const previousRpcServer = rpcServers.get(rpcDir) + if (previousRpcServer) { + await previousRpcServer.stop().catch(() => {}) + rpcServers.delete(rpcDir) + if (rpcGlobal.__anthropicAuthRpcServer === previousRpcServer) { + rpcGlobal.__anthropicAuthRpcServer = undefined + } } try { rpcServer = await startRpcServer({ - dir: getRpcDir(ctx.directory), + dir: rpcDir, drain: drainNotifications, apply: applyCommand, }) + rpcServers.set(rpcDir, rpcServer) rpcGlobal.__anthropicAuthRpcServer = rpcServer } catch (error) { logger.warn('rpc', 'failed to start', { @@ -2827,6 +2838,36 @@ const anthropicAuthPlugin = async ( }) } } + const dispose: NonNullable = async () => { + try { + await quotaHeaderFeedRegistry?.dispose() + } catch (error) { + logger.warn('quota-header-feed', 'failed to dispose', { + error: error instanceof Error ? error.message : String(error), + }) + } + try { + claustrumCredentialCache?.close() + } catch (error) { + logger.warn('claustrum', 'failed to close credential cache', { + error: error instanceof Error ? error.message : String(error), + }) + } + const rpcServers = ( + globalThis as { + __anthropicAuthRpcServers?: Map + } + ).__anthropicAuthRpcServers + if (!rpcServer || !rpcDir || rpcServers?.get(rpcDir) !== rpcServer) return + try { + await rpcServer.stop() + if (rpcServers.get(rpcDir) === rpcServer) rpcServers.delete(rpcDir) + } catch (error) { + logger.warn('rpc', 'failed to stop', { + error: error instanceof Error ? error.message : String(error), + }) + } + } // Remembers the last explicit routing decision so quota-only sidebar refreshes // (background main/fallback quota landing) do not reset the active account. @@ -7600,10 +7641,6 @@ const anthropicAuthPlugin = async ( return {} }, - dispose: async () => { - await quotaHeaderFeedRegistry?.dispose() - claustrumCredentialCache?.close() - }, methods: [ { label: 'Claude Pro/Max', @@ -7664,6 +7701,7 @@ const anthropicAuthPlugin = async ( }, ], }, + dispose, __primeManager: primeManager, __quotaManager: quotaManager, __persistFallbackQuotaErrorForTest: persistFallbackQuotaError, diff --git a/packages/opencode/src/rpc/rpc-server.ts b/packages/opencode/src/rpc/rpc-server.ts index b886e300..cfdbfcee 100644 --- a/packages/opencode/src/rpc/rpc-server.ts +++ b/packages/opencode/src/rpc/rpc-server.ts @@ -1,5 +1,5 @@ import { randomBytes, timingSafeEqual } from 'node:crypto' -import { unlink } from 'node:fs/promises' +import { readFile, unlink } from 'node:fs/promises' import { createServer, type IncomingMessage, @@ -115,9 +115,16 @@ export async function startRpcServer( token, async stop() { await new Promise((resolve) => server.close(() => resolve())) - await unlink(join(options.dir, `port-${process.pid}.json`)).catch( - () => {}, - ) + try { + const portFile = join(options.dir, `port-${process.pid}.json`) + const current = JSON.parse(await readFile(portFile, 'utf8')) as { + port?: unknown + pid?: unknown + } + if (current.port === port && current.pid === process.pid) { + await unlink(portFile) + } + } catch {} }, } } diff --git a/packages/opencode/src/tests/rpc-multi-project.test.ts b/packages/opencode/src/tests/rpc-multi-project.test.ts new file mode 100644 index 00000000..fa4ec27a --- /dev/null +++ b/packages/opencode/src/tests/rpc-multi-project.test.ts @@ -0,0 +1,335 @@ +import { afterEach, beforeEach, describe, expect, mock, test } from 'bun:test' +import { mkdtemp, readFile, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { + createEmptyStorage, + QuotaHeaderFeedRegistry, + saveAccounts, +} from '@cortexkit/anthropic-auth-core' +import type { Hooks } from '@opencode-ai/plugin' +import { AnthropicAuthPlugin } from '../index' +import { resetNotificationsForTest } from '../rpc/notifications' +import { discoverPortFile } from '../rpc/port-file' +import { getRpcDir } from '../rpc/rpc-dir' +import type { RpcServerHandle } from '../rpc/rpc-server' + +type RpcGlobal = typeof globalThis & { + __anthropicAuthRpcServer?: RpcServerHandle + __anthropicAuthRpcServers?: Map +} + +let testRoot: string +let previousRpcDir: string | undefined +let previousAccountFile: string | undefined +let previousSidebarStateFile: string | undefined +let previousCacheKeepRegistryDir: string | undefined +let previousQuotaFeedDir: string | undefined + +const disabledPluginRuntimeOverrides = { + setInterval: mock( + () => ({ unref() {} }) as unknown as ReturnType, + ) as unknown as typeof setInterval, + clearInterval: mock(() => {}) as unknown as typeof clearInterval, +} + +function createMockClient(applyMarker?: string) { + return { + auth: { set: mock(() => Promise.resolve()) }, + session: { + promptAsync: mock(() => + applyMarker + ? Promise.reject(new Error(applyMarker)) + : Promise.resolve(), + ), + }, + } +} + +async function getPlugin( + directory: string, + applyMarker?: string, +): Promise { + const plugin = AnthropicAuthPlugin as unknown as ( + ctx: Parameters[0], + runtimeOverrides: typeof disabledPluginRuntimeOverrides, + ) => ReturnType + return plugin( + { + // @ts-expect-error: minimal mock for testing + client: createMockClient(applyMarker), + directory, + }, + disabledPluginRuntimeOverrides, + ) +} + +async function applyViaRpc( + entry: { port: number; token: string }, + sessionId: string, +) { + const response = await fetch(`http://127.0.0.1:${entry.port}/rpc/apply`, { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${entry.token}`, + }, + body: JSON.stringify({ + command: 'claude-start', + arguments: '', + sessionId, + }), + }) + expect(response.status).toBe(200) + return (await response.json()) as { text: string } +} + +async function stopRpcServers() { + const rpcGlobal = globalThis as RpcGlobal + const servers = rpcGlobal.__anthropicAuthRpcServers + const handles = new Set([ + ...(servers?.values() ?? []), + ...(rpcGlobal.__anthropicAuthRpcServer + ? [rpcGlobal.__anthropicAuthRpcServer] + : []), + ]) + await Promise.all([...handles].map((server) => server.stop())) + if (servers) { + servers.clear() + rpcGlobal.__anthropicAuthRpcServers = undefined + } + rpcGlobal.__anthropicAuthRpcServer = undefined +} + +beforeEach(async () => { + testRoot = await mkdtemp(join(tmpdir(), 'aa-rpc-multi-project-')) + previousRpcDir = process.env.OPENCODE_ANTHROPIC_AUTH_RPC_DIR + previousAccountFile = process.env.OPENCODE_ANTHROPIC_AUTH_FILE + previousSidebarStateFile = + process.env.OPENCODE_ANTHROPIC_AUTH_SIDEBAR_STATE_FILE + previousCacheKeepRegistryDir = + process.env.OPENCODE_ANTHROPIC_AUTH_CACHEKEEP_REGISTRY_DIR + previousQuotaFeedDir = process.env.OPENCODE_ANTHROPIC_AUTH_QUOTA_FEED_DIR + process.env.OPENCODE_ANTHROPIC_AUTH_RPC_DIR = '.rpc' + process.env.OPENCODE_ANTHROPIC_AUTH_FILE = join( + testRoot, + 'anthropic-auth.json', + ) + process.env.OPENCODE_ANTHROPIC_AUTH_SIDEBAR_STATE_FILE = join( + testRoot, + 'sidebar-state.json', + ) + process.env.OPENCODE_ANTHROPIC_AUTH_CACHEKEEP_REGISTRY_DIR = join( + testRoot, + 'cachekeep-registry', + ) + process.env.OPENCODE_ANTHROPIC_AUTH_QUOTA_FEED_DIR = join( + testRoot, + 'quota-header-feed', + ) + await stopRpcServers() +}) + +afterEach(async () => { + await stopRpcServers() + if (previousRpcDir === undefined) { + delete process.env.OPENCODE_ANTHROPIC_AUTH_RPC_DIR + } else { + process.env.OPENCODE_ANTHROPIC_AUTH_RPC_DIR = previousRpcDir + } + if (previousAccountFile === undefined) { + delete process.env.OPENCODE_ANTHROPIC_AUTH_FILE + } else { + process.env.OPENCODE_ANTHROPIC_AUTH_FILE = previousAccountFile + } + if (previousSidebarStateFile === undefined) { + delete process.env.OPENCODE_ANTHROPIC_AUTH_SIDEBAR_STATE_FILE + } else { + process.env.OPENCODE_ANTHROPIC_AUTH_SIDEBAR_STATE_FILE = + previousSidebarStateFile + } + if (previousCacheKeepRegistryDir === undefined) { + delete process.env.OPENCODE_ANTHROPIC_AUTH_CACHEKEEP_REGISTRY_DIR + } else { + process.env.OPENCODE_ANTHROPIC_AUTH_CACHEKEEP_REGISTRY_DIR = + previousCacheKeepRegistryDir + } + if (previousQuotaFeedDir === undefined) { + delete process.env.OPENCODE_ANTHROPIC_AUTH_QUOTA_FEED_DIR + } else { + process.env.OPENCODE_ANTHROPIC_AUTH_QUOTA_FEED_DIR = previousQuotaFeedDir + } + await rm(testRoot, { recursive: true, force: true }) + resetNotificationsForTest() +}) + +describe('RPC server lifecycle', () => { + test('dispose stops and removes its server when feed cleanup rejects', async () => { + await saveAccounts({ + ...createEmptyStorage(), + quotaHeaderFeed: { enabled: true }, + }) + const originalDispose = QuotaHeaderFeedRegistry.prototype.dispose + QuotaHeaderFeedRegistry.prototype.dispose = async () => { + throw new Error('feed disposal failed') + } + try { + const directory = join(testRoot, 'project') + const plugin = await getPlugin(directory) + const rpcDir = getRpcDir(directory) + const entry = await discoverPortFile(rpcDir) + + expect(entry).not.toBeNull() + await plugin.dispose?.() + + expect(await discoverPortFile(rpcDir)).toBeNull() + expect( + (globalThis as RpcGlobal).__anthropicAuthRpcServers?.get(rpcDir), + ).toBeUndefined() + await expect( + fetch(`http://127.0.0.1:${entry?.port}/health`), + ).rejects.toThrow() + } finally { + QuotaHeaderFeedRegistry.prototype.dispose = originalDispose + } + }) + + test('keeps RPC servers live for distinct project directories', async () => { + const directoryA = join(testRoot, 'project-a') + const directoryB = join(testRoot, 'project-b') + + await getPlugin(directoryA) + await getPlugin(directoryB) + + const entryA = await discoverPortFile(getRpcDir(directoryA)) + const entryB = await discoverPortFile(getRpcDir(directoryB)) + + expect(entryA).not.toBeNull() + expect(entryB).not.toBeNull() + expect(entryA?.port).not.toBe(entryB?.port) + }) + + test('each project RPC server applies through its own plugin instance', async () => { + const directoryA = join(testRoot, 'project-a') + const directoryB = join(testRoot, 'project-b') + await getPlugin(directoryA, 'applied by project-a') + await getPlugin(directoryB, 'applied by project-b') + + const entryA = await discoverPortFile(getRpcDir(directoryA)) + const entryB = await discoverPortFile(getRpcDir(directoryB)) + + expect(entryA).not.toBeNull() + expect(entryB).not.toBeNull() + if (!entryA || !entryB) return + expect( + JSON.parse( + await readFile( + join(getRpcDir(directoryA), `port-${process.pid}.json`), + 'utf8', + ), + ), + ).toMatchObject({ port: entryA.port, token: entryA.token }) + expect( + JSON.parse( + await readFile( + join(getRpcDir(directoryB), `port-${process.pid}.json`), + 'utf8', + ), + ), + ).toMatchObject({ port: entryB.port, token: entryB.token }) + + expect((await applyViaRpc(entryA, 'session-a')).text).toContain( + 'applied by project-a', + ) + expect((await applyViaRpc(entryB, 'session-b')).text).toContain( + 'applied by project-b', + ) + }) + + test('dispose stops its directory while another project remains live', async () => { + const directoryA = join(testRoot, 'project-a') + const directoryB = join(testRoot, 'project-b') + const pluginA = await getPlugin(directoryA) + const pluginB = await getPlugin(directoryB) + const entryB = await discoverPortFile(getRpcDir(directoryB)) + + expect(entryB).not.toBeNull() + expect(pluginA.dispose).toBeFunction() + await pluginA.dispose?.() + + expect(await discoverPortFile(getRpcDir(directoryA))).toBeNull() + expect((await discoverPortFile(getRpcDir(directoryB)))?.port).toBe( + entryB?.port, + ) + expect( + (await fetch(`http://127.0.0.1:${entryB?.port}/health`)).status, + ).toBe(200) + await pluginB.dispose?.() + }) + + test('late disposal cannot remove a same-directory successor port file', async () => { + const directory = join(testRoot, 'project') + const first = await getPlugin(directory) + const second = await getPlugin(directory) + const successor = await discoverPortFile(getRpcDir(directory)) + const successorHandle = ( + globalThis as RpcGlobal + ).__anthropicAuthRpcServers?.get(getRpcDir(directory)) + + expect(successor).not.toBeNull() + expect(successorHandle).toBeDefined() + await first.dispose?.() + + expect( + (globalThis as RpcGlobal).__anthropicAuthRpcServers?.get( + getRpcDir(directory), + ), + ).toBe(successorHandle) + expect((await discoverPortFile(getRpcDir(directory)))?.port).toBe( + successor?.port, + ) + await second.dispose?.() + }) + + test('a dispose whose entry was replaced does not stop the successor server', async () => { + const directory = join(testRoot, 'project') + const first = await getPlugin(directory) + const rpcGlobal = globalThis as RpcGlobal + const rpcDir = getRpcDir(directory) + const firstHandle = rpcGlobal.__anthropicAuthRpcServers?.get(rpcDir) + const successorHandle: RpcServerHandle = { + port: firstHandle?.port ?? 0, + token: firstHandle?.token ?? '', + stop: mock(async () => {}), + } + + expect(firstHandle).toBeDefined() + if (!firstHandle) return + const stopSpy = mock(firstHandle.stop) + firstHandle.stop = stopSpy + rpcGlobal.__anthropicAuthRpcServers?.set(rpcDir, successorHandle) + + await first.dispose?.() + + // D2's port-file check would otherwise mask loss of D1. + expect(stopSpy).not.toHaveBeenCalled() + expect(rpcGlobal.__anthropicAuthRpcServers?.get(rpcDir)).toBe( + successorHandle, + ) + }) + + test('a disposed project can start a discoverable RPC server again', async () => { + const directory = join(testRoot, 'project') + const first = await getPlugin(directory) + + await first.dispose?.() + + const replacement = await getPlugin(directory) + const entry = await discoverPortFile(getRpcDir(directory)) + expect(entry).not.toBeNull() + expect((await fetch(`http://127.0.0.1:${entry?.port}/health`)).status).toBe( + 200, + ) + await replacement.dispose?.() + }) +}) diff --git a/packages/opencode/src/tests/rpc-server.test.ts b/packages/opencode/src/tests/rpc-server.test.ts index d0e2081b..5e256016 100644 --- a/packages/opencode/src/tests/rpc-server.test.ts +++ b/packages/opencode/src/tests/rpc-server.test.ts @@ -7,6 +7,7 @@ import { pushNotification, resetNotificationsForTest, } from '../rpc/notifications' +import { discoverPortFile } from '../rpc/port-file' import { startRpcServer } from '../rpc/rpc-server' let stop: (() => Promise) | null = null @@ -174,4 +175,26 @@ describe('rpc-server', () => { process.removeListener('uncaughtException', onUnhandled) expect(unhandledError).toBeNull() }) + + test('stopping a stale server preserves its successor port file', async () => { + dir = await mkdtemp(join(tmpdir(), 'aa-rpcsrv-')) + const first = await startRpcServer({ + dir, + drain: drainNotifications, + apply: async () => ({ text: 'ok', knobs: {} }), + }) + const second = await startRpcServer({ + dir, + drain: drainNotifications, + apply: async () => ({ text: 'ok', knobs: {} }), + }) + stop = second.stop + + await first.stop() + + expect((await discoverPortFile(dir))?.port).toBe(second.port) + expect((await fetch(`http://127.0.0.1:${second.port}/health`)).status).toBe( + 200, + ) + }) }) From 9d9462396dddc1c65df8c9b5cf2819f00e14ff35 Mon Sep 17 00:00:00 2001 From: iceteaSA <171169159+iceteaSA@users.noreply.github.com> Date: Fri, 11 Sep 2026 19:18:25 +0200 Subject: [PATCH 2/3] fix(rpc): scope the shared notification state to sessions One notification queue serves every RPC server in the process, so anything in it that is not keyed by session is keyed by nothing once a process holds more than one project. Require a session id on queued notices. A notice without one was delivered to every draining TUI and survived one session's ack, which across servers means crossing project boundaries. Require a session id on the connectivity probe and delete the process-wide `lastDrainAtAny` it fell back on. That timestamp was written by every project's drain, so an unscoped call could report a TUI connected for project A because project B's TUI polled -- and the caller skips the desktop fallback when it believes a TUI is present, so the notice would go nowhere. Both call sites already passed an id; this removes the possibility rather than relying on it. Stop an unscoped drain from deleting other sessions' notices. It pruned every acknowledged notice regardless of owner, so one client both swallowed and destroyed other sessions' pending dialogs. This predates the per-directory registry -- with a single server it was cross-session inside one project -- and is fixed as deliver-but-never-prune: the TUI has sent a session id since polling was introduced, so the clients that can omit it are malformed or third-party ones, and rejecting them would fail by silently not delivering, which is the symptom this change exists to fix. With no producer able to queue a session-less notice, the matching branch in the drain filter is unreachable and removed. The wire field stays optional so an older TUI still parses what it is sent. --- packages/opencode/src/rpc/notifications.ts | 38 ++++--- .../src/tests/rpc-notifications.test.ts | 105 +++++++++++++++--- 2 files changed, 116 insertions(+), 27 deletions(-) diff --git a/packages/opencode/src/rpc/notifications.ts b/packages/opencode/src/rpc/notifications.ts index 1464063b..2db233bc 100644 --- a/packages/opencode/src/rpc/notifications.ts +++ b/packages/opencode/src/rpc/notifications.ts @@ -1,16 +1,25 @@ +import { logger } from '@cortexkit/anthropic-auth-core' + import type { OpenDialogPayload, RpcNotification } from './protocol' const QUEUE_CAP = 100 const TUI_CONNECTED_WINDOW_MS = 3_000 +// One queue serves every RPC server in the process, and a process can hold one server per +// project directory. Session ids are globally unique, so a notice that carries one reaches +// only the TUI polling for that session. A notice WITHOUT one broadcasts instead: every +// draining TUI receives it and one session's ack does not prune it for the others — which, +// once a process serves more than one project, would carry it across project boundaries. +// The producer boundary therefore requires a session id; the wire field stays optional so +// an older TUI still parses what it is sent. let queue: RpcNotification[] = [] let nextId = 1 -let lastDrainAtAny = 0 const lastDrainAtBySession = new Map() +let warnedAboutUnscopedDrain = false export function pushNotification( payload: OpenDialogPayload, - sessionId?: string, + sessionId: string, ): void { queue.push({ id: nextId++, type: 'open-dialog', payload, sessionId }) if (queue.length > QUEUE_CAP) queue = queue.slice(queue.length - QUEUE_CAP) @@ -21,34 +30,35 @@ export function drainNotifications( sessionId?: string, ): RpcNotification[] { const now = Date.now() - lastDrainAtAny = now if (sessionId !== undefined) lastDrainAtBySession.set(sessionId, now) const matches = (n: RpcNotification) => - sessionId === undefined || - n.sessionId === undefined || - n.sessionId === sessionId + sessionId === undefined || n.sessionId === sessionId + if (sessionId === undefined && !warnedAboutUnscopedDrain) { + warnedAboutUnscopedDrain = true + logger.warn( + 'rpc.notifications', + 'drain arrived without a session id; delivery is unscoped and the queue is left intact', + ) + } if (lastReceivedId > 0) { queue = queue.filter((n) => { if (n.id > lastReceivedId) return true - if (sessionId === undefined) return false + if (sessionId === undefined) return true return n.sessionId !== sessionId }) } return queue.filter((n) => n.id > lastReceivedId && matches(n)) } -export function isTuiConnected(sessionId?: string): boolean { +export function isTuiConnected(sessionId: string): boolean { const now = Date.now() - if (sessionId !== undefined) { - const at = lastDrainAtBySession.get(sessionId) ?? 0 - return at > 0 && now - at < TUI_CONNECTED_WINDOW_MS - } - return lastDrainAtAny > 0 && now - lastDrainAtAny < TUI_CONNECTED_WINDOW_MS + const at = lastDrainAtBySession.get(sessionId) ?? 0 + return at > 0 && now - at < TUI_CONNECTED_WINDOW_MS } export function resetNotificationsForTest(): void { queue = [] nextId = 1 - lastDrainAtAny = 0 lastDrainAtBySession.clear() + warnedAboutUnscopedDrain = false } diff --git a/packages/opencode/src/tests/rpc-notifications.test.ts b/packages/opencode/src/tests/rpc-notifications.test.ts index e79fd00e..c5df7076 100644 --- a/packages/opencode/src/tests/rpc-notifications.test.ts +++ b/packages/opencode/src/tests/rpc-notifications.test.ts @@ -1,4 +1,8 @@ import { beforeEach, describe, expect, test } from 'bun:test' +import { + __setLogTestSink, + type LogTestRecord, +} from '@cortexkit/anthropic-auth-core' import { drainNotifications, isTuiConnected, @@ -16,6 +20,78 @@ const payload = (command: OpenDialogPayload['command']): OpenDialogPayload => ({ describe('notifications', () => { beforeEach(() => resetNotificationsForTest()) + test('warns once when an unscoped drain leaves the queue intact', () => { + const records: LogTestRecord[] = [] + __setLogTestSink((record) => records.push(record)) + try { + drainNotifications(0) + drainNotifications(0) + expect( + records.filter( + (record) => + record.level === 'warn' && + record.message.includes('drain arrived without a session id'), + ), + ).toHaveLength(1) + } finally { + __setLogTestSink(null) + } + }) + + test('an unscoped drain delivers every pending notice', () => { + pushNotification(payload('claude-quota'), 's1') + pushNotification(payload('claude-dump'), 's2') + + expect( + drainNotifications(0, undefined).map((n) => n.payload.command), + ).toEqual(['claude-quota', 'claude-dump']) + }) + + test('an unscoped drain acknowledges without pruning other sessions', () => { + pushNotification(payload('claude-quota'), 's1') + pushNotification(payload('claude-dump'), 's2') + + // Acknowledged notices are not re-delivered to the client that acked them, + // and an unscoped ack must not speak for the sessions it does not name. + expect(drainNotifications(2, undefined)).toEqual([]) + expect(drainNotifications(0, 's2').map((n) => n.payload.command)).toEqual([ + 'claude-dump', + ]) + expect(drainNotifications(0, 's1').map((n) => n.payload.command)).toEqual([ + 'claude-quota', + ]) + }) + + test('reset re-arms the unscoped-drain warning after an earlier drain', () => { + const records: LogTestRecord[] = [] + __setLogTestSink((record) => records.push(record)) + try { + drainNotifications(0) + resetNotificationsForTest() + drainNotifications(0) + expect( + records.filter( + (record) => + record.level === 'warn' && + record.message.includes('drain arrived without a session id'), + ), + ).toHaveLength(2) + } finally { + __setLogTestSink(null) + } + }) + + test('a session-scoped drain prunes its own acknowledged notices', () => { + pushNotification(payload('claude-quota'), 's1') + pushNotification(payload('claude-dump'), 's2') + + const s1 = drainNotifications(0, 's1') + expect(drainNotifications(s1[0]?.id, 's1')).toEqual([]) + expect(drainNotifications(0, 's2').map((n) => n.payload.command)).toEqual([ + 'claude-dump', + ]) + }) + test('push then drain returns the item once, ordered', () => { pushNotification(payload('claude-quota'), 's1') pushNotification(payload('claude-fast'), 's1') @@ -29,12 +105,14 @@ describe('notifications', () => { expect(second).toEqual([]) }) - test('session scoping: a session only drains its own + global', () => { + test('every queued notice carries its session id and stays scoped to it', () => { pushNotification(payload('claude-quota'), 's1') pushNotification(payload('claude-dump'), 's2') - expect(drainNotifications(0, 's1').map((n) => n.payload.command)).toEqual([ - 'claude-quota', - ]) + const s1 = drainNotifications(0, 's1') + + expect(s1).toHaveLength(1) + expect(s1[0]?.sessionId).toBe('s1') + expect(s1.map((n) => n.payload.command)).toEqual(['claude-quota']) expect(drainNotifications(0, 's2').map((n) => n.payload.command)).toEqual([ 'claude-dump', ]) @@ -46,6 +124,14 @@ describe('notifications', () => { expect(isTuiConnected('s1')).toBe(true) }) + test('a drain only marks its own session as connected', () => { + drainNotifications(0, 's2') + expect(isTuiConnected('s1')).toBe(false) + expect(isTuiConnected('s2')).toBe(true) + // @ts-expect-error isTuiConnected requires a session id + expect(isTuiConnected()).toBe(false) + }) + test('queue cap evicts oldest beyond 100', () => { for (let i = 0; i < 130; i++) pushNotification(payload('claude-quota'), 's1') @@ -53,15 +139,8 @@ describe('notifications', () => { expect(all.length).toBe(100) }) - test('a global notification reaches every session and is not pruned by one ack', () => { - // push a global (no sessionId) notification + test('pushNotification requires a session id at compile time', () => { + // @ts-expect-error pushNotification requires a session id pushNotification(payload('claude-quota')) - const a = drainNotifications(0, 's1') - expect(a.length).toBe(1) - // s1 acks it - drainNotifications(a[0]?.id as number, 's1') - // s2 must STILL receive it - const b = drainNotifications(0, 's2') - expect(b.length).toBe(1) }) }) From 9b63bbea30aaae861905370b77e49fe258cda118 Mon Sep 17 00:00:00 2001 From: iceteaSA <171169159+iceteaSA@users.noreply.github.com> Date: Fri, 11 Sep 2026 23:12:11 +0200 Subject: [PATCH 3/3] refactor(rpc): drop the singular RPC server global The per-directory registry __anthropicAuthRpcServers supersedes the process-wide __anthropicAuthRpcServer handle. Production never read the singular identifier for a decision; the conditional clear was a clear-if-mine that no reader depended on, and the dispose never cleared it on teardown, leaving a dangling reference to a stopped server for the lifetime of the process. The test helper gains a per-directory cleanup proof that, for every rpc dir this file started a server in, the registry holds no entry for it and its port-.json is gone. A post-build gate (packages/opencode/scripts/check-bundle-globals.ts) verifies the bundle still contains the registry identifier and has zero matches for the singular form, wired into the build script and runnable via bun run check:bundle. --- packages/opencode/package.json | 3 +- .../opencode/scripts/check-bundle-globals.ts | 35 +++++++++++++++++++ packages/opencode/src/index.ts | 5 --- .../src/tests/rpc-multi-project.test.ts | 21 ++++++----- 4 files changed, 50 insertions(+), 14 deletions(-) create mode 100644 packages/opencode/scripts/check-bundle-globals.ts diff --git a/packages/opencode/package.json b/packages/opencode/package.json index f349b884..c0bc2ab2 100644 --- a/packages/opencode/package.json +++ b/packages/opencode/package.json @@ -45,9 +45,10 @@ "LICENSE" ], "scripts": { - "build": "rm -rf dist && bun build src/index.ts src/cli.ts src/sidebar-state.ts src/tui-preferences.ts src/rpc/rpc-client.ts src/rpc/port-file.ts src/rpc/protocol.ts src/rpc/rpc-dir.ts --outdir dist --target node --format esm --splitting --external @opencode-ai/plugin --minify && tsc -p tsconfig.build.json --emitDeclarationOnly && bun run build:tui", + "build": "rm -rf dist && bun build src/index.ts src/cli.ts src/sidebar-state.ts src/tui-preferences.ts src/rpc/rpc-client.ts src/rpc/port-file.ts src/rpc/protocol.ts src/rpc/rpc-dir.ts --outdir dist --target node --format esm --splitting --external @opencode-ai/plugin --minify && tsc -p tsconfig.build.json --emitDeclarationOnly && bun run build:tui && bun run check:bundle", "build:tui": "bun scripts/build-tui.ts", "smoke:tui": "bun scripts/smoke-tui-pack-install.ts", + "check:bundle": "bun scripts/check-bundle-globals.ts", "build:dev": "rm -rf dist && tsc -p tsconfig.build.json", "dev": "bun ../../scripts/dev.ts", "dev:clean": "bun ../../scripts/dev-clean.ts", diff --git a/packages/opencode/scripts/check-bundle-globals.ts b/packages/opencode/scripts/check-bundle-globals.ts new file mode 100644 index 00000000..b1c966a6 --- /dev/null +++ b/packages/opencode/scripts/check-bundle-globals.ts @@ -0,0 +1,35 @@ +import { readFile, stat } from 'node:fs/promises' +import { join } from 'node:path' + +const bundlePath = join(import.meta.dir, '..', 'dist', 'index.js') +const minBundleBytes = 1024 + +let size: number +try { + size = (await stat(bundlePath)).size +} catch { + throw new Error(`Bundle artifact check failed: ${bundlePath} is missing`) +} + +if (size <= minBundleBytes) { + throw new Error( + `Bundle artifact check failed: ${bundlePath} is not substantial (${size} bytes)`, + ) +} + +const bundle = await readFile(bundlePath, 'utf8') +const registryMatches = bundle.match(/__anthropicAuthRpcServers/g)?.length ?? 0 +if (registryMatches === 0) { + throw new Error( + 'Bundle positive-control check failed: __anthropicAuthRpcServers is absent', + ) +} + +// This catches one identifier; the positive control makes its zero assertion meaningful, not proof that no other stale global exists. +const singularMatches = + bundle.match(/__anthropicAuthRpcServer(?!s)/g)?.length ?? 0 +if (singularMatches !== 0) { + throw new Error( + `Bundle stale-global check failed: __anthropicAuthRpcServer appears ${singularMatches} time(s)`, + ) +} diff --git a/packages/opencode/src/index.ts b/packages/opencode/src/index.ts index e1b75f59..04f17696 100644 --- a/packages/opencode/src/index.ts +++ b/packages/opencode/src/index.ts @@ -2809,7 +2809,6 @@ const anthropicAuthPlugin = async ( let rpcDir: string | null = null if (ctx.directory) { const rpcGlobal = globalThis as { - __anthropicAuthRpcServer?: RpcServerHandle __anthropicAuthRpcServers?: Map } rpcDir = getRpcDir(ctx.directory) @@ -2820,9 +2819,6 @@ const anthropicAuthPlugin = async ( if (previousRpcServer) { await previousRpcServer.stop().catch(() => {}) rpcServers.delete(rpcDir) - if (rpcGlobal.__anthropicAuthRpcServer === previousRpcServer) { - rpcGlobal.__anthropicAuthRpcServer = undefined - } } try { rpcServer = await startRpcServer({ @@ -2831,7 +2827,6 @@ const anthropicAuthPlugin = async ( apply: applyCommand, }) rpcServers.set(rpcDir, rpcServer) - rpcGlobal.__anthropicAuthRpcServer = rpcServer } catch (error) { logger.warn('rpc', 'failed to start', { error: error instanceof Error ? error.message : String(error), diff --git a/packages/opencode/src/tests/rpc-multi-project.test.ts b/packages/opencode/src/tests/rpc-multi-project.test.ts index fa4ec27a..1ce0a9f4 100644 --- a/packages/opencode/src/tests/rpc-multi-project.test.ts +++ b/packages/opencode/src/tests/rpc-multi-project.test.ts @@ -15,7 +15,6 @@ import { getRpcDir } from '../rpc/rpc-dir' import type { RpcServerHandle } from '../rpc/rpc-server' type RpcGlobal = typeof globalThis & { - __anthropicAuthRpcServer?: RpcServerHandle __anthropicAuthRpcServers?: Map } @@ -25,6 +24,7 @@ let previousAccountFile: string | undefined let previousSidebarStateFile: string | undefined let previousCacheKeepRegistryDir: string | undefined let previousQuotaFeedDir: string | undefined +let startedRpcDirs: Set const disabledPluginRuntimeOverrides = { setInterval: mock( @@ -54,6 +54,7 @@ async function getPlugin( ctx: Parameters[0], runtimeOverrides: typeof disabledPluginRuntimeOverrides, ) => ReturnType + startedRpcDirs.add(getRpcDir(directory)) return plugin( { // @ts-expect-error: minimal mock for testing @@ -87,22 +88,17 @@ async function applyViaRpc( async function stopRpcServers() { const rpcGlobal = globalThis as RpcGlobal const servers = rpcGlobal.__anthropicAuthRpcServers - const handles = new Set([ - ...(servers?.values() ?? []), - ...(rpcGlobal.__anthropicAuthRpcServer - ? [rpcGlobal.__anthropicAuthRpcServer] - : []), - ]) + const handles = new Set(servers?.values() ?? []) await Promise.all([...handles].map((server) => server.stop())) if (servers) { servers.clear() rpcGlobal.__anthropicAuthRpcServers = undefined } - rpcGlobal.__anthropicAuthRpcServer = undefined } beforeEach(async () => { testRoot = await mkdtemp(join(tmpdir(), 'aa-rpc-multi-project-')) + startedRpcDirs = new Set() previousRpcDir = process.env.OPENCODE_ANTHROPIC_AUTH_RPC_DIR previousAccountFile = process.env.OPENCODE_ANTHROPIC_AUTH_FILE previousSidebarStateFile = @@ -132,6 +128,12 @@ beforeEach(async () => { afterEach(async () => { await stopRpcServers() + for (const rpcDir of startedRpcDirs) { + expect( + (globalThis as RpcGlobal).__anthropicAuthRpcServers?.get(rpcDir), + ).toBeUndefined() + expect(await discoverPortFile(rpcDir)).toBeNull() + } if (previousRpcDir === undefined) { delete process.env.OPENCODE_ANTHROPIC_AUTH_RPC_DIR } else { @@ -316,6 +318,9 @@ describe('RPC server lifecycle', () => { expect(rpcGlobal.__anthropicAuthRpcServers?.get(rpcDir)).toBe( successorHandle, ) + // Dispose refused to stop D1 by design; the spy wraps the real stop, so + // invoking it clears the dangling server and its port file before afterEach. + await stopSpy() }) test('a disposed project can start a discoverable RPC server again', async () => {