Close retired RPC connections after draining replies

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 This repo is a CortexKit-maintained Anthropic auth monorepo for OpenCode and Pi. The OpenCode package is a fork of the original `@ex-machina/opencode-anthropic-auth` plugin, so older entries below the initial CortexKit release are inherited from upstream package history.
 
+## Unreleased
+
+- After a plugin reload, the OpenCode TUI's `/claude` menu no longer briefly gets an "unauthorized" error or a dropped connection from the previous plugin instance.
 ## 2.8.0
 
 - On OpenCode 2, Claude models the plugin supports but OpenCode's own model list doesn't have yet (such as Fable 5.2 on launch day) now appear in the model picker, as on OpenCode 1.

packages/opencode/src/rpc/rpc-server.ts

@@ -5,6 +5,7 @@ import {
   type IncomingMessage,
   type ServerResponse,
 } from 'node:http'
+import type { Socket } from 'node:net'
 import { join } from 'node:path'
 import {
   type CommandApplyRequest,
@@ -59,13 +60,43 @@ export async function startRpcServer(
   options: RpcServerOptions,
 ): Promise<RpcServerHandle> {
   const token = randomBytes(32).toString('hex')
+  const connections = new Set<Socket>()
+  let stopping = false
+  let closing: Promise<void> | undefined
   const server = createServer((req, res) => {
     void dispatch(req, res)
   })
+  server.on('connection', (socket) => {
+    connections.add(socket)
+    socket.once('close', () => connections.delete(socket))
+  })
+
+  function closeServer(): Promise<void> {
+    closing ??= (async () => {
+      stopping = true
+      await new Promise<void>((resolve) => server.close(() => resolve()))
+      // Bun closes connections that were idle when close() ran, but a request
+      // active at that point can leave its keep-alive socket open after replying.
+      // Wait for the replies first, then close every remaining owned socket.
+      await Promise.all(
+        [...connections].map(
+          (socket) =>
+            new Promise<void>((resolve) => {
+              socket.once('close', () => resolve())
+              socket.destroy()
+            }),
+        ),
+      )
+    })()
+    return closing
+  }
 
   async function dispatch(req: IncomingMessage, res: ServerResponse) {
     const json = (status: number, value: unknown) => {
-      res.writeHead(status, { 'content-type': 'application/json' })
+      res.writeHead(status, {
+        'content-type': 'application/json',
+        ...(stopping && { connection: 'close' }),
+      })
       res.end(JSON.stringify(value))
     }
     try {
@@ -129,15 +160,15 @@ export async function startRpcServer(
   try {
     await writePortFile(options.dir, { port, token, pid: process.pid })
   } catch (error) {
-    await new Promise<void>((resolve) => server.close(() => resolve()))
+    await closeServer()
     throw error
   }
 
   return {
     port,
     token,
     async stop() {
-      await new Promise<void>((resolve) => server.close(() => resolve()))
+      await closeServer()
       try {
         const portFile = join(options.dir, `port-${process.pid}.json`)
         const current = JSON.parse(await readFile(portFile, 'utf8')) as {

packages/opencode/src/tests/info-logs.test.ts

@@ -241,6 +241,9 @@ async function migratedClaude() {
         `http://127.0.0.1:${entry.port}/rpc/apply-menu`,
         {
           method: 'POST',
+          // Each test owns a short-lived RPC listener. Do not borrow a pooled
+          // socket from an earlier fixture if the OS reuses its ephemeral port.
+          keepalive: false,
           headers: {
             'content-type': 'application/json',
             authorization: `Bearer ${entry.token}`,

packages/opencode/src/tests/network-guard.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, mock, test } from 'bun:test'
 import { mkdtemp, rm, writeFile } from 'node:fs/promises'
+import { createServer, type Socket } from 'node:net'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { fileURLToPath } from 'node:url'
@@ -60,6 +61,101 @@ describe('test network guard', () => {
     }
   })
 
+  test('preserves the connection-pool opt-out when an ephemeral port is reused', async () => {
+    const sockets = new Set<Socket>()
+    let oldRequests = 0
+    let newRequests = 0
+    function onCompletePost(socket: Socket, handle: () => void) {
+      let request = ''
+      socket.on('data', (chunk) => {
+        request += chunk.toString()
+        if (!request.endsWith('\r\n\r\n{}')) return
+        request = ''
+        handle()
+      })
+    }
+    const previous = createServer((socket) => {
+      sockets.add(socket)
+      onCompletePost(socket, () => {
+        oldRequests++
+        if (oldRequests === 1) {
+          socket.write(
+            'HTTP/1.1 200 OK\r\nContent-Length: 3\r\nConnection: keep-alive\r\n\r\nold',
+          )
+        } else {
+          // Hold the old connection until the next POST to force the race
+          // between server-side closure and the client's pooled socket reuse.
+          socket.resetAndDestroy()
+        }
+      })
+    })
+    const successor = createServer((socket) => {
+      sockets.add(socket)
+      onCompletePost(socket, () => {
+        newRequests++
+        socket.end(
+          'HTTP/1.1 200 OK\r\nContent-Length: 3\r\nConnection: close\r\n\r\nnew',
+        )
+      })
+    })
+    let previousClosed: Promise<void> | undefined
+    try {
+      await new Promise<void>((resolve) =>
+        previous.listen(0, '127.0.0.1', resolve),
+      )
+      const address = previous.address()
+      if (!address || typeof address === 'string') throw new Error('no port')
+      const url = `http://127.0.0.1:${address.port}/rpc/apply-menu`
+      expect(
+        await (await guardedFetch(url, { method: 'POST', body: '{}' })).text(),
+      ).toBe('old')
+      // Stop accepting connections but deliberately delay the old socket's
+      // closure. The successor must only see fresh connections to this port.
+      previousClosed = new Promise<void>((resolve) =>
+        previous.close(() => resolve()),
+      )
+      await new Promise<void>((resolve) =>
+        successor.listen(address.port, '127.0.0.1', resolve),
+      )
+
+      const response = await guardedFetch(url, {
+        method: 'POST',
+        body: '{}',
+        keepalive: false,
+      })
+      expect(await response.text()).toBe('new')
+      expect(oldRequests).toBe(1)
+      expect(newRequests).toBe(1)
+    } finally {
+      for (const socket of sockets) socket.destroy()
+      await (previousClosed ??
+        new Promise<void>((resolve) => previous.close(() => resolve())))
+      await new Promise<void>((resolve) => successor.close(() => resolve()))
+    }
+  })
+
+  test('preserves the connection-pool opt-out across redirects', async () => {
+    const options: unknown[] = []
+    const nativeFetch = Object.assign(
+      async (_request: Request, init?: Parameters<typeof fetch>[1]) => {
+        options.push(init?.keepalive)
+        return options.length === 1
+          ? new Response(null, {
+              status: 307,
+              headers: { location: 'http://127.0.0.1/final' },
+            })
+          : new Response('ok')
+      },
+      { preconnect: () => {} },
+    ) as unknown as typeof fetch
+    const response = await createGuardedFetch(nativeFetch)(
+      'http://127.0.0.1/first',
+      { keepalive: false },
+    )
+    expect(await response.text()).toBe('ok')
+    expect(options).toEqual([false, false])
+  })
+
   test('rejects near-miss IPv4 addresses and loopback-looking hostnames', async () => {
     for (const host of ['128.0.0.1', '27.0.0.1', '127.0.0.1.evil.com']) {
       await expect(

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

@@ -1,5 +1,6 @@
 import { afterEach, describe, expect, test } from 'bun:test'
 import { mkdtemp, rm } from 'node:fs/promises'
+import { createServer } from 'node:http'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import type { NativeMenuDispatchRequest } from '@cortexkit/anthropic-auth-core'
@@ -25,6 +26,71 @@ afterEach(async () => {
 })
 
 describe('rpc-server', () => {
+  test('stop drains an in-flight menu reply and closes its connection before same-port reuse', async () => {
+    dir = await mkdtemp(join(tmpdir(), 'aa-rpc-stop-'))
+    let release!: () => void
+    const gate = new Promise<void>((resolve) => {
+      release = resolve
+    })
+    let entered!: () => void
+    const applying = new Promise<void>((resolve) => {
+      entered = resolve
+    })
+    const server = await startRpcServer({
+      dir,
+      drain: drainNotifications,
+      applyMenu: async () => {
+        entered()
+        await gate
+        return {
+          command: 'claude',
+          ok: true,
+          text: 'applied',
+          menu: { command: 'claude', title: 'Claude', sections: [] },
+        }
+      },
+    })
+    stop = server.stop
+    const successor = createServer((_req, res) => {
+      res.setHeader('content-type', 'application/json')
+      res.end(JSON.stringify({ successor: true }))
+    })
+    try {
+      const url = `http://127.0.0.1:${server.port}/rpc/apply-menu`
+      const pending = fetch(url, {
+        method: 'POST',
+        headers: { authorization: `Bearer ${server.token}` },
+        body: JSON.stringify({
+          command: 'claude',
+          sectionId: 'Diagnostics',
+          actionId: 'logging-level',
+          values: { level: 'warn' },
+          sessionId: 'stopping',
+        }),
+      })
+      await applying
+      const stopping = server.stop()
+      release()
+      const response = await pending
+      expect(response.status).toBe(200)
+      expect(await response.json()).toMatchObject({ ok: true, text: 'applied' })
+      await stopping
+      await new Promise<void>((resolve) =>
+        successor.listen(server.port, '127.0.0.1', resolve),
+      )
+
+      // Without closing the old active connection after its reply, Bun can
+      // send this POST to the disposed RPC server instead of the successor.
+      const next = await fetch(url, { method: 'POST', body: '{}' })
+      expect(await next.json()).toEqual({ successor: true })
+      expect(response.headers.get('connection')).toBe('close')
+    } finally {
+      release()
+      await server.stop()
+      await new Promise<void>((resolve) => successor.close(() => resolve()))
+    }
+  })
+
   test('native menu client/server dispatch preserves structured limits and opaque IDs', async () => {
     dir = await mkdtemp(join(tmpdir(), 'aa-native-rpc-'))
     const calls: NativeMenuDispatchRequest[] = []

packages/opencode/src/tests/setup.ts

@@ -99,6 +99,9 @@ export function createGuardedFetch(
         const replayableRequest = request.clone()
         const response = await Reflect.apply(nativeFetchImpl, globalThis, [
           request,
+          // Bun's connection-pool opt-out is a fetch option, not a Request
+          // property. Preserve it when rebuilding requests, including redirects.
+          { keepalive: init?.keepalive },
         ])
         if (!REDIRECT_STATUSES.has(response.status)) return response
         const location = response.headers.get('location')