fail only the disconnected sidecar's in-flight requests
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 <noreply@anthropic.com>
This commit is contained in:
@@ -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<string | null> {
|
|
||||||
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}`);
|
|
||||||
}
|
|
||||||
@@ -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<void> {
|
||||||
|
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<unknown>): 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' });
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -23,6 +23,8 @@ type RegisteredSidecar = {
|
|||||||
};
|
};
|
||||||
|
|
||||||
type PendingRequest = {
|
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;
|
resolve: (value: any) => void;
|
||||||
reject: (error: Error) => void;
|
reject: (error: Error) => void;
|
||||||
timer: Timer;
|
timer: Timer;
|
||||||
@@ -73,8 +75,12 @@ export function unregisterSidecar(id: string): void {
|
|||||||
sidecars.delete(id);
|
sidecars.delete(id);
|
||||||
console.log(`[registry] unregistered sidecar "${sc.name}" (id=${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) {
|
for (const [reqId, req] of pending) {
|
||||||
|
if (req.sidecarId !== id) continue;
|
||||||
clearTimeout(req.timer);
|
clearTimeout(req.timer);
|
||||||
req.reject(new Error(`Sidecar "${sc.name}" disconnected`));
|
req.reject(new Error(`Sidecar "${sc.name}" disconnected`));
|
||||||
pending.delete(reqId);
|
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`));
|
reject(new Error(`Sidecar command ${cmd.type} timed out`));
|
||||||
}, timeoutMs);
|
}, 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));
|
sc.ws.send(JSON.stringify(cmd));
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
@@ -175,7 +181,7 @@ function sendCommandToSidecar(
|
|||||||
reject(new Error(`Sidecar command ${cmd.type} timed out`));
|
reject(new Error(`Sidecar command ${cmd.type} timed out`));
|
||||||
}, timeoutMs);
|
}, 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));
|
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
|
// 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.
|
// `/api/terminal/ws` a byte relay, both keyed off the `pty:server` port like every other HTTP sidecar.
|
||||||
|
|
||||||
|
|
||||||
// ── VNC ──
|
// ── VNC ──
|
||||||
|
|
||||||
export async function startVnc(params: VncStartParams): Promise<{ port: number; display: number }> {
|
export async function startVnc(params: VncStartParams): Promise<{ port: number; display: number }> {
|
||||||
|
|||||||
Reference in New Issue
Block a user