From 9c2c746967f9f480382b3151a153570c634ea0d7 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Mon, 9 Nov 2026 22:08:09 +0100 Subject: [PATCH 1/2] mason: close retired RPC connections after draining replies Bun closes already-idle HTTP sockets on server.close, but sockets active at stop can remain reusable after replying. A TUI request following a reload can then reach the retired RPC dispatcher and receive unauthorized instead of reaching the successor. Mark draining replies Connection: close and destroy/await all remaining owned sockets after the native close callback drains replies. Keep short-lived info-log RPC fixtures out of Bun's pool and preserve that fetch option through the loopback guard, including redirects. Add deterministic regressions: old RPC source reaches the retired dispatcher after stop; old guard loses the pool opt-out and produces ECONNRESET. The original release log does not record port/socket reuse, so its exact ordering is not claimed as observed. --- CHANGELOG.md | 1 + packages/opencode/src/rpc/rpc-server.ts | 37 ++++++- packages/opencode/src/tests/info-logs.test.ts | 3 + .../opencode/src/tests/network-guard.test.ts | 96 +++++++++++++++++++ .../opencode/src/tests/rpc-server.test.ts | 66 +++++++++++++ packages/opencode/src/tests/setup.ts | 3 + 6 files changed, 203 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 043c8dd53..16aec07f7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ This repo is a CortexKit-maintained Anthropic auth monorepo for OpenCode and Pi. ## 2.8.0 +- 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. - 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. - Adds Claude Fable 5.2 (`claude-fable-5-2`) on OpenCode 1, OpenCode 2 and Pi, sent the way Claude Code sends it: adaptive thinking, medium default effort, per-message effort and Anthropic's safety fallback. Fast mode stays off for Fable 5.2 until Anthropic enables it for subscription accounts. - Claude subscription requests no longer carry a `service_tier` field a host adds, keeping them in the shape Claude Code sends; API-key requests keep it. diff --git a/packages/opencode/src/rpc/rpc-server.ts b/packages/opencode/src/rpc/rpc-server.ts index fc40d2edc..f2ec3f776 100644 --- a/packages/opencode/src/rpc/rpc-server.ts +++ b/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 { const token = randomBytes(32).toString('hex') + const connections = new Set() + let stopping = false + let closing: Promise | 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 { + closing ??= (async () => { + stopping = true + await new Promise((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((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,7 +160,7 @@ export async function startRpcServer( try { await writePortFile(options.dir, { port, token, pid: process.pid }) } catch (error) { - await new Promise((resolve) => server.close(() => resolve())) + await closeServer() throw error } @@ -137,7 +168,7 @@ export async function startRpcServer( port, token, async stop() { - await new Promise((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 { diff --git a/packages/opencode/src/tests/info-logs.test.ts b/packages/opencode/src/tests/info-logs.test.ts index 41038586b..5f12361ff 100644 --- a/packages/opencode/src/tests/info-logs.test.ts +++ b/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}`, diff --git a/packages/opencode/src/tests/network-guard.test.ts b/packages/opencode/src/tests/network-guard.test.ts index 64573f6ea..aca8798e6 100644 --- a/packages/opencode/src/tests/network-guard.test.ts +++ b/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() + 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 | undefined + try { + await new Promise((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((resolve) => + previous.close(() => resolve()), + ) + await new Promise((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((resolve) => previous.close(() => resolve()))) + await new Promise((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[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( diff --git a/packages/opencode/src/tests/rpc-server.test.ts b/packages/opencode/src/tests/rpc-server.test.ts index 52be5c07b..ed87b732a 100644 --- a/packages/opencode/src/tests/rpc-server.test.ts +++ b/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((resolve) => { + release = resolve + }) + let entered!: () => void + const applying = new Promise((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((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((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[] = [] diff --git a/packages/opencode/src/tests/setup.ts b/packages/opencode/src/tests/setup.ts index 0232875f1..314410e76 100644 --- a/packages/opencode/src/tests/setup.ts +++ b/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') From 5e4ba507eb7d8715f0ebc8fd2d497a9ef703172e Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Mon, 9 Nov 2026 22:10:17 +0100 Subject: [PATCH 2/2] docs: changelog lines under Unreleased --- CHANGELOG.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 16aec07f7..7edfdc2ae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,9 +2,11 @@ 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. -## 2.8.0 +## 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. - Adds Claude Fable 5.2 (`claude-fable-5-2`) on OpenCode 1, OpenCode 2 and Pi, sent the way Claude Code sends it: adaptive thinking, medium default effort, per-message effort and Anthropic's safety fallback. Fast mode stays off for Fable 5.2 until Anthropic enables it for subscription accounts. - Claude subscription requests no longer carry a `service_tier` field a host adds, keeping them in the shape Claude Code sends; API-key requests keep it.