fix(rpc): scope loopback servers by project

packages/opencode/src/index.ts

@@ -1004,7 +1004,9 @@ export async function CodexAuthPlugin(
   // command.execute.before reads this; if null (auth not loaded yet),
   // the command is rejected with a message.
   let cmdCtx: CommandContext | null = null
-  let activeRpcServer: RpcServerHandle | null = null
+  const ownedCacheKeepManagers = new Map<string, CacheKeepManager>()
+  const ownedRpcServers = new Map<string, RpcServerHandle>()
+  let activeFallbackManager: FallbackAccountManager | undefined
   let sidebarStateFileForEvents: string | undefined
 
   // Per-loader poller: each plugin invocation owns its timer and callback, so
@@ -1039,18 +1041,33 @@ export async function CodexAuthPlugin(
   return {
     async dispose() {
       backgroundQuotaRefresh.stop()
+      activeFallbackManager?.stopBackgroundRefresh()
+      activeFallbackManager = undefined
       for (const websocketFetch of websocketFetches) websocketFetch.close()
       websocketFetches.length = 0
-      if (activeRpcServer) {
-        await activeRpcServer.stop().catch(() => {})
-        const rpcGlobal = globalThis as {
-          __openaiAuthRpcServer?: RpcServerHandle
+      const cacheKeepGlobal = globalThis as {
+        __openaiAuthCacheKeepManagers?: Map<string, CacheKeepManager>
+      }
+      for (const [key, manager] of ownedCacheKeepManagers) {
+        if (
+          cacheKeepGlobal.__openaiAuthCacheKeepManagers?.get(key) === manager
+        ) {
+          manager.stop()
+          cacheKeepGlobal.__openaiAuthCacheKeepManagers.delete(key)
         }
-        if (rpcGlobal.__openaiAuthRpcServer === activeRpcServer) {
-          rpcGlobal.__openaiAuthRpcServer = undefined
+      }
+      ownedCacheKeepManagers.clear()
+
+      const rpcGlobal = globalThis as {
+        __openaiAuthRpcServers?: Map<string, RpcServerHandle>
+      }
+      for (const [key, rpcServer] of ownedRpcServers) {
+        if (rpcGlobal.__openaiAuthRpcServers?.get(key) === rpcServer) {
+          await rpcServer.stop().catch(() => {})
+          rpcGlobal.__openaiAuthRpcServers.delete(key)
         }
-        activeRpcServer = null
       }
+      ownedRpcServers.clear()
     },
     async event(input) {
       if (input.event.type !== 'session.deleted') return
@@ -1179,6 +1196,11 @@ export async function CodexAuthPlugin(
         const auth = await getAuth()
         if (auth.type !== 'oauth') return {}
 
+        const rpcDir = input.directory
+          ? await resolveRpcDir(input.directory)
+          : undefined
+        const cacheKeepKey = rpcDir?.dir ?? getConfigPath()
+
         // Migration: seed the multi-account store from the existing token (idempotent)
         await migrateIfNeeded(
           {
@@ -1523,9 +1545,12 @@ export async function CodexAuthPlugin(
           return mainRefreshPromise
         }
         const cacheKeepGlobal = globalThis as {
-          __openaiAuthCacheKeepManager?: CacheKeepManager
+          __openaiAuthCacheKeepManagers?: Map<string, CacheKeepManager>
         }
-        cacheKeepGlobal.__openaiAuthCacheKeepManager?.stop()
+        const cacheKeepManagers =
+          cacheKeepGlobal.__openaiAuthCacheKeepManagers ?? new Map()
+        cacheKeepGlobal.__openaiAuthCacheKeepManagers = cacheKeepManagers
+        cacheKeepManagers.get(cacheKeepKey)?.stop()
         const cacheKeepManager = new CacheKeepManager({
           fetchImpl: fetch,
           getMainToken: async () => {
@@ -1563,7 +1588,8 @@ export async function CodexAuthPlugin(
           getWindow: () => cacheKeepWindow,
           getSustain: () => cacheKeepSustain,
         })
-        cacheKeepGlobal.__openaiAuthCacheKeepManager = cacheKeepManager
+        cacheKeepManagers.set(cacheKeepKey, cacheKeepManager)
+        ownedCacheKeepManagers.set(cacheKeepKey, cacheKeepManager)
 
         async function pushQuota(
           snapshot: Record<string, unknown>,
@@ -1817,6 +1843,8 @@ export async function CodexAuthPlugin(
         // Start the loopback RPC server so the TUI can drain notifications and
         // dispatch apply commands.
         // -------------------------------------------------------------------
+        activeFallbackManager?.stopBackgroundRefresh()
+        activeFallbackManager = fallbackManager
         cmdCtx = {
           accountStoragePath: getConfigPath(),
           quotaManager,
@@ -1905,14 +1933,16 @@ export async function CodexAuthPlugin(
         }
 
         let rpcServer: RpcServerHandle | null = null
-        if (input.directory) {
-          const rpcDir = await resolveRpcDir(input.directory)
+        if (rpcDir) {
           const rpcGlobal = globalThis as {
-            __openaiAuthRpcServer?: RpcServerHandle
+            __openaiAuthRpcServers?: Map<string, RpcServerHandle>
           }
-          if (rpcGlobal.__openaiAuthRpcServer) {
-            await rpcGlobal.__openaiAuthRpcServer.stop().catch(() => {})
-            rpcGlobal.__openaiAuthRpcServer = undefined
+          const rpcServers = rpcGlobal.__openaiAuthRpcServers ?? new Map()
+          rpcGlobal.__openaiAuthRpcServers = rpcServers
+          const existingRpcServer = rpcServers.get(rpcDir.dir)
+          if (existingRpcServer) {
+            await existingRpcServer.stop().catch(() => {})
+            rpcServers.delete(rpcDir.dir)
           }
           try {
             rpcServer = await startRpcServer({
@@ -1934,8 +1964,8 @@ export async function CodexAuthPlugin(
                 return { text: payload.text, knobs: payload.knobs }
               },
             })
-            rpcGlobal.__openaiAuthRpcServer = rpcServer
-            activeRpcServer = rpcServer
+            rpcServers.set(rpcDir.dir, rpcServer)
+            ownedRpcServers.set(rpcDir.dir, rpcServer)
           } catch {
             // RPC is best-effort; the plugin must not fail if the port file
             // can't be written (e.g. missing directory in test environments).
@@ -2891,19 +2921,19 @@ export async function CodexAuthPlugin(
         // sidebar shows real numbers shortly after start instead of "checking…".
         // Non-blocking, best-effort — a failure must never crash the loader.
         // -------------------------------------------------------------------
+        // Seed fallback quota from persisted account.quota so the immediate
+        // machine snapshot shows last-known fallback numbers.
+        if (storage) {
+          const oauthAccts: OAuthAccount[] = []
+          for (const a of storage.accounts) {
+            if (isOAuthAccount(a)) oauthAccts.push(a)
+          }
+          quotaManager.seedFallbacksFromAccounts(oauthAccts)
+        }
+
         if (!bootQuotaSeedStarted) {
           bootQuotaSeedStarted = true
 
-          // Seed fallback quota from persisted account.quota so the immediate
-          // The immediate machine snapshot shows last-known fallback numbers.
-          if (storage) {
-            const oauthAccts: OAuthAccount[] = []
-            for (const a of storage.accounts) {
-              if (isOAuthAccount(a)) oauthAccts.push(a)
-            }
-            quotaManager.seedFallbacksFromAccounts(oauthAccts)
-          }
-
           // Immediate: show persisted quota so the sidebar isn't blank
           void writeMachineSidebarState(quotaManager, storage).catch(() => {})
 
@@ -3410,26 +3440,6 @@ export async function CodexAuthPlugin(
             ).catch(() => {})
             return finalResponse
           },
-          async dispose() {
-            backgroundQuotaRefresh.stop()
-            cacheKeepManager.stop()
-            if (
-              cacheKeepGlobal.__openaiAuthCacheKeepManager === cacheKeepManager
-            ) {
-              cacheKeepGlobal.__openaiAuthCacheKeepManager = undefined
-            }
-            fallbackManager.stopBackgroundRefresh()
-            if (activeRpcServer) {
-              await activeRpcServer.stop().catch(() => {})
-              const rpcGlobal = globalThis as {
-                __openaiAuthRpcServer?: RpcServerHandle
-              }
-              if (rpcGlobal.__openaiAuthRpcServer === activeRpcServer) {
-                rpcGlobal.__openaiAuthRpcServer = undefined
-              }
-              activeRpcServer = null
-            }
-          },
         }
       },
       methods: [

packages/opencode/src/rpc/notifications.ts

@@ -5,7 +5,6 @@ const TUI_CONNECTED_WINDOW_MS = 3_000
 
 let queue: RpcNotification[] = []
 let nextId = 1
-let lastDrainAtAny = 0
 const lastDrainAtBySession = new Map<string, number>()
 
 export function pushNotification(
@@ -21,7 +20,6 @@ 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 ||
@@ -30,25 +28,21 @@ export function drainNotifications(
   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()
 }

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,
@@ -78,6 +78,7 @@ export async function startRpcServer(
   // every endpoint holding a dead connection for 90s.
   const handlerTimeoutMs = options.timeoutMs ?? 90_000
   const receiptTimeoutMs = options.receiptTimeoutMs ?? 2_000
+  let warnedMissingNotificationSession = false
   const server = createServer((req, res) => {
     req.setTimeout(handlerTimeoutMs, () => {
       req.socket.destroy()
@@ -107,9 +108,17 @@ export async function startRpcServer(
       const body = await readBody(req)
       const params = JSON.parse(body || '{}') as Record<string, unknown>
       if (method === 'pending-notifications') {
+        const sessionId =
+          typeof params.sessionId === 'string' ? params.sessionId : undefined
+        if (sessionId === undefined && !warnedMissingNotificationSession) {
+          warnedMissingNotificationSession = true
+          log.warn('rpc notification drain missing session id', {
+            pid: process.pid,
+          })
+        }
         const messages = options.drain(
           Number(params.lastReceivedId ?? 0),
-          typeof params.sessionId === 'string' ? params.sessionId : undefined,
+          sessionId,
         )
         return json(200, { messages })
       }
@@ -164,9 +173,12 @@ export async function startRpcServer(
     token,
     async stop() {
       await new Promise<void>((resolve) => server.close(() => resolve()))
-      await unlink(join(options.dir, `port-${process.pid}.json`)).catch(
-        () => {},
-      )
+      const portFile = join(options.dir, `port-${process.pid}.json`)
+      const current = await readFile(portFile, 'utf8')
+        .then((raw) => JSON.parse(raw) as { port?: unknown; token?: unknown })
+        .catch(() => undefined)
+      if (current?.port === port && current.token === token)
+        await unlink(portFile).catch(() => {})
     },
   }
 }

packages/opencode/src/tests/cachekeep.test.ts

@@ -2,6 +2,8 @@ import { afterEach, beforeEach, describe, expect, mock, test } from 'bun:test'
 import { mkdtemp, readdir, rm, writeFile } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
+
+import { getConfigPath } from '../config'
 import type { AccountStorage } from '../core/accounts'
 import {
   buildKeepwarmBody,
@@ -2466,7 +2468,9 @@ describe('CacheKeepManager token resolution', () => {
       if (!loaderResult?.fetch) throw new Error('No fetch override')
 
       const cacheKeepGlobal = globalThis as any
-      const mgr = cacheKeepGlobal.__openaiAuthCacheKeepManager
+      const mgr = cacheKeepGlobal.__openaiAuthCacheKeepManagers?.get(
+        getConfigPath(),
+      )
       expect(mgr).toBeDefined()
 
       const mockFetch = mock(async () => new Response('{}'))
@@ -2530,7 +2534,7 @@ describe('RPC server dispose', () => {
     await rm(tempDir, { recursive: true, force: true })
   })
 
-  test('RPC server stops and unlinks port file on loader dispose', async () => {
+  test('loader options do not expose an RPC lifecycle dispose hook', async () => {
     const originalRpcDir = process.env.OPENCODE_OPENAI_AUTH_RPC_DIR
     process.env.OPENCODE_OPENAI_AUTH_RPC_DIR = tempDir
 
@@ -2564,25 +2568,19 @@ describe('RPC server dispose', () => {
       )
 
       // Verify port file exists in tempDir
-      let files = await readdir(tempDir)
+      const files = await readdir(tempDir)
       expect(
         files.some((f) => f.startsWith('port-') && f.endsWith('.json')),
       ).toBe(true)
 
-      // Dispose the loader
-      await loaderResult?.dispose?.()
-
-      // Verify port file is gone
-      files = await readdir(tempDir)
-      expect(
-        files.some((f) => f.startsWith('port-') && f.endsWith('.json')),
-      ).toBe(false)
+      expect(loaderResult?.dispose).toBeUndefined()
+      await plugin.dispose?.()
     } finally {
       process.env.OPENCODE_OPENAI_AUTH_RPC_DIR = originalRpcDir
     }
   })
 
-  test('RPC server stops and unlinks port file on plugin dispose', async () => {
+  test('plugin dispose clears the RPC registry entry and unlinks the port file', async () => {
     const originalRpcDir = process.env.OPENCODE_OPENAI_AUTH_RPC_DIR
     process.env.OPENCODE_OPENAI_AUTH_RPC_DIR = tempDir
 
@@ -2621,14 +2619,18 @@ describe('RPC server dispose', () => {
         files.some((f) => f.startsWith('port-') && f.endsWith('.json')),
       ).toBe(true)
 
-      // Dispose the plugin
+      const rpcGlobal = globalThis as {
+        __openaiAuthRpcServers?: Map<string, unknown>
+      }
+      expect(rpcGlobal.__openaiAuthRpcServers?.size ?? 0).toBeGreaterThan(0)
+
       await plugin.dispose?.()
 
-      // Verify port file is gone
       files = await readdir(tempDir)
       expect(
         files.some((f) => f.startsWith('port-') && f.endsWith('.json')),
       ).toBe(false)
+      expect(rpcGlobal.__openaiAuthRpcServers?.size ?? 0).toBe(0)
     } finally {
       process.env.OPENCODE_OPENAI_AUTH_RPC_DIR = originalRpcDir
     }

packages/opencode/src/tests/command-session-isolation.test.ts

@@ -127,7 +127,7 @@ describe('command hook session isolation', () => {
       experimentalWebSockets: false,
     })
 
-    const loaderResult = await plugin.auth?.loader?.(
+    await plugin.auth?.loader?.(
       async () => ({
         type: 'oauth',
         provider: 'openai',
@@ -204,6 +204,6 @@ describe('command hook session isolation', () => {
     expect(added).toBeDefined()
     expect(added?.sessionId).toBe('sess-A')
 
-    await loaderResult?.dispose?.()
+    await plugin.dispose?.()
   })
 })