fix(rpc): scope loopback servers by project

packages/opencode/src/index.ts

@@ -1004,7 +1004,8 @@ 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 sidebarStateFileForEvents: string | undefined
 
   // Per-loader poller: each plugin invocation owns its timer and callback, so
@@ -1041,16 +1042,29 @@ export async function CodexAuthPlugin(
       backgroundQuotaRefresh.stop()
       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 +1193,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 +1542,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 +1585,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>,
@@ -1905,14 +1928,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 +1959,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).
@@ -3410,26 +3435,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?.()
   })
 })

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

@@ -10,6 +10,7 @@ import { readFile } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import type { Hooks, PluginInput } from '@opencode-ai/plugin'
+import { getConfigPath } from '../config.ts'
 import type { OAuthAccount } from '../core/accounts.ts'
 import { migrateIfNeeded } from '../core/accounts.ts'
 import { acquireRefreshFileLock } from '../core/refresh-file-lock.ts'
@@ -4952,12 +4953,15 @@ describe('integration: active fallback routing', () => {
       await runCommand(hooks, 'openai-cachekeep', 'sustain on')
       const manager = (
         globalThis as typeof globalThis & {
-          __openaiAuthCacheKeepManager?: {
-            tick(): Promise<void>
-            status(): { tracked: number; sustain: boolean }
-          }
+          __openaiAuthCacheKeepManagers?: Map<
+            string,
+            {
+              tick(): Promise<void>
+              status(): { tracked: number; sustain: boolean }
+            }
+          >
         }
-      ).__openaiAuthCacheKeepManager
+      ).__openaiAuthCacheKeepManagers?.get(getConfigPath())
       if (!manager) throw new Error('missing cachekeep manager')
 
       await manager.tick()
@@ -6900,9 +6904,12 @@ describe('integration: active fallback routing', () => {
           now += 30 * 60_000
           const manager = (
             globalThis as typeof globalThis & {
-              __openaiAuthCacheKeepManager?: { tick(): Promise<void> }
+              __openaiAuthCacheKeepManagers?: Map<
+                string,
+                { tick(): Promise<void> }
+              >
             }
-          ).__openaiAuthCacheKeepManager
+          ).__openaiAuthCacheKeepManagers?.get(getConfigPath())
           if (!manager) throw new Error('missing cachekeep manager')
           await manager.tick()
         },

packages/opencode/src/tests/rpc-notifications.test.ts

@@ -46,6 +46,15 @@ describe('notifications', () => {
     expect(isTuiConnected('s1')).toBe(true)
   })
 
+  test('a drain for one session does not make an unscoped probe connected', () => {
+    drainNotifications(0, 's2')
+    expect(isTuiConnected('s1')).toBe(false)
+    expect(isTuiConnected(undefined as never)).toBe(false)
+  })
+
+  // @ts-expect-error TUI connectivity must always be scoped to a session.
+  isTuiConnected()
+
   test('queue cap evicts oldest beyond 100', () => {
     for (let i = 0; i < 130; i++)
       pushNotification(payload('openai-quota'), 's1')