From 4eeaa3c93db1694b6d9f89ceb0ffc4149843a865 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Padez?= Date: Tue, 4 Aug 2026 01:33:29 +0000 Subject: [PATCH] fail only the disconnected sidecar's in-flight requests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit unregisterSidecar rejected every entry in the pending map, not just the ones belonging to the sidecar that went away. Restarting any single sidecar failed in-flight work on every other: `pm2 restart officer-music` could kill a running agent turn with `Sidecar "music" disconnected` — a message pointing at a process that had nothing to do with it. Pending entries now carry their owning sidecar id and the rejection loop skips the rest. Also deletes src/servers/api/anthropic-proxy.ts. It had no importers; PM2's officer-anthropic-proxy runs src/servers/sidecar/claude/index.ts. Co-Authored-By: Claude Opus 5 --- src/servers/api/anthropic-proxy.ts | 92 ---------------------- src/servers/sidecar-registry.test.ts | 109 +++++++++++++++++++++++++++ src/servers/sidecar-registry.ts | 13 +++- 3 files changed, 118 insertions(+), 96 deletions(-) delete mode 100644 src/servers/api/anthropic-proxy.ts create mode 100644 src/servers/sidecar-registry.test.ts diff --git a/src/servers/api/anthropic-proxy.ts b/src/servers/api/anthropic-proxy.ts deleted file mode 100644 index e6e99ea5..00000000 --- a/src/servers/api/anthropic-proxy.ts +++ /dev/null @@ -1,92 +0,0 @@ -import { join } from 'node:path'; -import { homedir } from 'node:os'; - -const PROXY_PORT = Number(process.env.ANTHROPIC_PROXY_PORT ?? '5051'); -const ANTHROPIC_API_BASE = 'https://api.anthropic.com'; -const CREDENTIALS_PATH = join(homedir(), '.claude', '.credentials.json'); - -// Generate a random proxy secret at startup — shared with Claude Code spawns -// Prefix must match a real Anthropic API key format (sk-ant-api03-*) so Claude Code accepts it -export const proxySecret = `sk-ant-api03-${crypto.randomUUID()}`; - -type CredentialsFile = { - claudeAiOauth?: { - accessToken?: string; - }; -}; - -async function readOAuthToken(): Promise { - try { - const file = Bun.file(CREDENTIALS_PATH); - if (!(await file.exists())) return null; - const data = (await file.json()) as CredentialsFile; - return data.claudeAiOauth?.accessToken?.trim() || null; - } catch { - return null; - } -} - -export function startAnthropicProxy() { - Bun.serve({ - port: PROXY_PORT, - hostname: '127.0.0.1', - idleTimeout: 0, - - async fetch(req) { - const url = new URL(req.url); - - // Validate proxy secret - const incomingKey = req.headers.get('x-api-key'); - if (incomingKey !== proxySecret) { - return new Response(JSON.stringify({ error: 'Unauthorized' }), { - status: 401, - headers: { 'Content-Type': 'application/json' }, - }); - } - - // Read fresh OAuth token - const token = await readOAuthToken(); - if (!token) { - return new Response(JSON.stringify({ error: 'No OAuth token available' }), { - status: 502, - headers: { 'Content-Type': 'application/json' }, - }); - } - - // Build upstream URL - const upstream = `${ANTHROPIC_API_BASE}${url.pathname}${url.search}`; - - // Clone headers, replace proxy secret with real OAuth token - const headers = new Headers(req.headers); - headers.set('x-api-key', token); - headers.delete('host'); - - // Read request body fully before forwarding — avoids stream-in-stream issues - const body = req.body ? await req.arrayBuffer() : null; - - // Forward request - const upstreamRes = await fetch(upstream, { - method: req.method, - headers, - body, - }); - - // Build clean response headers - const resHeaders = new Headers(); - for (const [key, value] of upstreamRes.headers) { - const lower = key.toLowerCase(); - // Skip hop-by-hop and encoding headers that may have been consumed by fetch - if (lower === 'content-encoding' || lower === 'content-length' || lower === 'transfer-encoding') continue; - resHeaders.set(key, value); - } - - return new Response(upstreamRes.body, { - status: upstreamRes.status, - statusText: upstreamRes.statusText, - headers: resHeaders, - }); - }, - }); - - console.log(`[anthropic-proxy] listening on 127.0.0.1:${PROXY_PORT}`); -} diff --git a/src/servers/sidecar-registry.test.ts b/src/servers/sidecar-registry.test.ts new file mode 100644 index 00000000..c429dee8 --- /dev/null +++ b/src/servers/sidecar-registry.test.ts @@ -0,0 +1,109 @@ +// A disconnect must fail only the work that was in flight to the sidecar that actually went away. +// +// This exists because it did not. `unregisterSidecar` walked the whole `pending` map and rejected every +// entry, so `pm2 restart officer-music` would fail a running agent turn with `Sidecar "music" disconnected` +// — a message that points at the wrong process entirely and sends anyone debugging it in the wrong +// direction. Sidecars restart independently by design; that is the whole point of them being PM2 peers. + +import { afterEach, describe, expect, test } from 'bun:test'; +import { registerSidecar, unregisterSidecar, spawnClaude, startVnc } from './sidecar-registry'; + +/** A socket that records what was written and never answers, so requests stay pending. */ +function silentSocket() { + const sent: string[] = []; + return { sent, ws: { send: (data: string) => void sent.push(data) } as never }; +} + +const registered: string[] = []; + +/** Register a fake sidecar and remember it, so a failed assertion can't leak it into the next test. */ +function register(name: string, capabilities: string[]) { + const socket = silentSocket(); + const id = registerSidecar(socket.ws, { type: 'register', name, capabilities }); + registered.push(id); + return { ...socket, id }; +} + +afterEach(() => { + while (registered.length) unregisterSidecar(registered.pop()!); +}); + +/** Wait until a request has actually reached the socket — spawnClaude awaits waitForCapability first. */ +async function awaitSend(socket: { sent: string[] }): Promise { + for (let i = 0; i < 50 && socket.sent.length === 0; i++) await Bun.sleep(1); + expect(socket.sent).not.toHaveLength(0); +} + +/** Resolves to 'pending' when the promise has not settled shortly after. */ +async function settleState(promise: Promise): Promise<'pending' | 'resolved' | { rejected: string }> { + const marker = Symbol('pending'); + const outcome = await Promise.race([ + promise.then( + () => 'resolved' as const, + (err: Error) => ({ rejected: err.message }), + ), + Bun.sleep(25).then(() => marker), + ]); + return outcome === marker ? 'pending' : (outcome as 'resolved' | { rejected: string }); +} + +const CLAUDE_PARAMS = { + userId: 1, + email: 'owner@example.com', + username: 'owner', + prompt: 'hello', + sessionKey: 'test-session', +}; + +const VNC_PARAMS = { email: 'owner@example.com', username: 'owner' }; + +describe('unregisterSidecar', () => { + test('an unrelated sidecar disconnecting leaves other sidecars’ requests in flight', async () => { + const agent = register('agent', ['claude']); + const vnc = register('vnc', ['vnc']); + + const turn = spawnClaude(CLAUDE_PARAMS); + // Take ownership of the eventual rejection immediately, so it is never an unhandled one. + const turnState = settleState(turn); + await awaitSend(agent); + + unregisterSidecar(vnc.id); + + // The agent never disconnected, so its turn is still running. + expect(await turnState).toBe('pending'); + + // And when the agent itself goes, its own request fails — naming the right process. + unregisterSidecar(agent.id); + expect(await settleState(turn)).toEqual({ rejected: 'Sidecar "agent" disconnected' }); + }); + + test('a sidecar’s own disconnect still rejects its own request', async () => { + const vnc = register('vnc', ['vnc']); + + const start = startVnc(VNC_PARAMS); + const state = settleState(start); + await awaitSend(vnc); + + unregisterSidecar(vnc.id); + expect(await state).toEqual({ rejected: 'Sidecar "vnc" disconnected' }); + }); + + test('re-registering under the same name only clears that name’s work', async () => { + // registerSidecar unregisters an existing same-named sidecar first — a reconnect after a restart. + // That path runs the same rejection loop, so it has the same blast radius if it is not scoped. + register('agent', ['claude']); + const vnc = register('vnc', ['vnc']); + + const desktop = startVnc(VNC_PARAMS); + const desktopState = settleState(desktop); + await awaitSend(vnc); + + // The agent reconnects — a different process entirely. + register('agent', ['claude']); + + expect(await desktopState).toBe('pending'); + + unregisterSidecar(vnc.id); + expect(await settleState(desktop)).toEqual({ rejected: 'Sidecar "vnc" disconnected' }); + }); +}); diff --git a/src/servers/sidecar-registry.ts b/src/servers/sidecar-registry.ts index 62b078ef..27286141 100644 --- a/src/servers/sidecar-registry.ts +++ b/src/servers/sidecar-registry.ts @@ -23,6 +23,8 @@ type RegisteredSidecar = { }; type PendingRequest = { + /** Which sidecar this request was sent to, so a disconnect only fails that sidecar's own work. */ + sidecarId: string; resolve: (value: any) => void; reject: (error: Error) => void; timer: Timer; @@ -73,8 +75,12 @@ export function unregisterSidecar(id: string): void { sidecars.delete(id); console.log(`[registry] unregistered sidecar "${sc.name}" (id=${id})`); - // Reject all pending requests for this sidecar + // Reject the pending requests belonging to THIS sidecar, and only those. Until the pending entries + // carried a sidecarId this loop rejected the whole map, so restarting any one sidecar failed in-flight + // work on every other — `pm2 restart officer-music` could kill a running agent turn with the message + // `Sidecar "music" disconnected`, which is the sort of failure nobody traces back to its cause. for (const [reqId, req] of pending) { + if (req.sidecarId !== id) continue; clearTimeout(req.timer); req.reject(new Error(`Sidecar "${sc.name}" disconnected`)); pending.delete(reqId); @@ -152,7 +158,7 @@ function sendCommand(cap: string, cmd: SidecarCommand, timeoutMs = DEFAULT_TIMEO reject(new Error(`Sidecar command ${cmd.type} timed out`)); }, timeoutMs); - pending.set((cmd as any).id, { resolve, reject, timer }); + pending.set((cmd as any).id, { sidecarId: sc.id, resolve, reject, timer }); sc.ws.send(JSON.stringify(cmd)); }); } @@ -175,7 +181,7 @@ function sendCommandToSidecar( reject(new Error(`Sidecar command ${cmd.type} timed out`)); }, timeoutMs); - pending.set((cmd as any).id, { resolve, reject, timer }); + pending.set((cmd as any).id, { sidecarId: sc.id, resolve, reject, timer }); sc.ws.send(JSON.stringify(cmd)); }); } @@ -320,7 +326,6 @@ export function onOpenCodeSession(handler: (sessionKey: string, sessionId: strin // Nothing here any more. The pty sidecar serves its own listener; `/api/terminal/*` is a proxy and // `/api/terminal/ws` a byte relay, both keyed off the `pty:server` port like every other HTTP sidecar. - // ── VNC ── export async function startVnc(params: VncStartParams): Promise<{ port: number; display: number }> {