import { resolve } from 'node:path'; import type { ServerWebSocket } from 'bun'; import type { Subprocess } from 'bun'; import type { SidecarCommand, SidecarEvent, ClaudeState, ClaudeSpawnParams, ClaudeSpawnStreamingParams, ClaudeCodeResult, PiSpawnParams, PtyCommand, PtyEvent, VncStartParams, VncSessionInfo, } from './sidecar/protocol'; import type { SidecarRegistration } from './sidecar/registration-protocol'; import type { PiEvent } from './api/pi/types'; // ── Types ── type RegisteredSidecar = { id: string; name: string; capabilities: string[]; ws: ServerWebSocket; }; type PendingRequest = { resolve: (value: any) => void; reject: (error: Error) => void; timer: Timer; }; type EventHandler = (event: SidecarEvent | PtyEvent) => void; // ── State ── const sidecars = new Map(); const pending = new Map(); const eventHandlers = new Map>(); let cachedState: ClaudeState | null = null; let idCounter = 0; function nextId(): string { return `sr_${Date.now()}_${++idCounter}`; } // ── Registration ── export function registerSidecar(ws: ServerWebSocket, registration: SidecarRegistration): string { // If a sidecar with the same name is already registered, unregister it first for (const [id, sc] of sidecars) { if (sc.name === registration.name) { console.log(`[registry] replacing existing sidecar "${registration.name}" (id=${id})`); unregisterSidecar(id); break; } } const id = `sc_${registration.name}_${Date.now()}`; sidecars.set(id, { id, name: registration.name, capabilities: registration.capabilities, ws, }); console.log( `[registry] registered sidecar "${registration.name}" (id=${id}, capabilities=[${registration.capabilities.join(', ')}])`, ); return id; } export function unregisterSidecar(id: string): void { const sc = sidecars.get(id); if (!sc) return; sidecars.delete(id); console.log(`[registry] unregistered sidecar "${sc.name}" (id=${id})`); // Reject all pending requests for this sidecar for (const [reqId, req] of pending) { clearTimeout(req.timer); req.reject(new Error(`Sidecar "${sc.name}" disconnected`)); pending.delete(reqId); } // Clear cached state if the claude sidecar disconnects if (sc.capabilities.includes('proxy')) { cachedState = null; } } export function handleSidecarMessage(id: string, msg: SidecarEvent | PtyEvent): void { // Check if this is a response to a pending request if ('id' in msg && msg.id && pending.has(msg.id)) { const req = pending.get(msg.id)!; pending.delete(msg.id); clearTimeout(req.timer); req.resolve(msg); return; } // Otherwise dispatch as event dispatchEvent(msg); } // ── Lookup ── function findSidecarByCapability(cap: string): RegisteredSidecar | undefined { for (const sc of sidecars.values()) { if (sc.capabilities.includes(cap)) return sc; } return undefined; } function findSidecarByName(name: string): RegisteredSidecar | undefined { for (const sc of sidecars.values()) { if (sc.name === name) return sc; } return undefined; } // ── Event dispatch ── function dispatchEvent(msg: SidecarEvent | PtyEvent) { const handlers = eventHandlers.get(msg.type); if (handlers) { for (const handler of handlers) { try { handler(msg); } catch { /* ignore */ } } } } export function on(eventType: string, handler: EventHandler): () => void { if (!eventHandlers.has(eventType)) { eventHandlers.set(eventType, new Set()); } eventHandlers.get(eventType)!.add(handler); return () => { eventHandlers.get(eventType)?.delete(handler); }; } // ── Command sending ── const DEFAULT_TIMEOUT_MS = 30_000; const LONG_TIMEOUT_MS = 6 * 60 * 1000; function sendCommand(cap: string, cmd: SidecarCommand | PtyCommand, timeoutMs = DEFAULT_TIMEOUT_MS): Promise { return new Promise((resolve, reject) => { const sc = findSidecarByCapability(cap); if (!sc) { reject(new Error(`No sidecar with capability "${cap}" is connected`)); return; } const timer = setTimeout(() => { pending.delete((cmd as any).id); reject(new Error(`Sidecar command ${cmd.type} timed out`)); }, timeoutMs); pending.set((cmd as any).id, { resolve, reject, timer }); sc.ws.send(JSON.stringify(cmd)); }); } function sendFire(cap: string, cmd: SidecarCommand | PtyCommand): void { const sc = findSidecarByCapability(cap); if (sc) { sc.ws.send(JSON.stringify(cmd)); } } function sendCommandToSidecar( sc: RegisteredSidecar, cmd: SidecarCommand | PtyCommand, timeoutMs = DEFAULT_TIMEOUT_MS, ): Promise { return new Promise((resolve, reject) => { const timer = setTimeout(() => { pending.delete((cmd as any).id); reject(new Error(`Sidecar command ${cmd.type} timed out`)); }, timeoutMs); pending.set((cmd as any).id, { resolve, reject, timer }); sc.ws.send(JSON.stringify(cmd)); }); } function sendFireToSidecar(sc: RegisteredSidecar, cmd: SidecarCommand | PtyCommand): void { sc.ws.send(JSON.stringify(cmd)); } // ── On-demand Claude sidecar spawning ── const USER_INSTANCE_SCRIPT = resolve(import.meta.dir, 'sidecar/claude/user-instance.ts'); const SIDECAR_SPAWN_TIMEOUT_MS = 15_000; const claudeProcs = new Map(); const claudeSpawnWaiters = new Map>(); async function ensureClaudeSidecar(email: string): Promise { const name = `claude:${email}`; // Already registered? const existing = findSidecarByName(name); if (existing) return existing; // Already spawning? const waiter = claudeSpawnWaiters.get(email); if (waiter) return waiter; // Spawn and wait for registration const promise = spawnAndWaitForRegistration(email, name); claudeSpawnWaiters.set(email, promise); try { return await promise; } finally { claudeSpawnWaiters.delete(email); } } async function spawnAndWaitForRegistration(email: string, name: string): Promise { const proxySecret = await getProxySecret(); const proxyPort = process.env.ANTHROPIC_PROXY_PORT ?? '5051'; const env: Record = { ...(process.env as Record), CLAUDE_USER_EMAIL: email, ANTHROPIC_BASE_URL: `http://127.0.0.1:${proxyPort}`, ANTHROPIC_API_KEY: proxySecret, }; const proc = Bun.spawn(['bun', 'run', USER_INSTANCE_SCRIPT], { env, stdout: 'inherit', stderr: 'inherit', }); claudeProcs.set(email, proc); // Clean up on exit proc.exited.then(() => { claudeProcs.delete(email); }); // Wait for the sidecar to register return new Promise((resolve, reject) => { const timeout = setTimeout(() => { unsub(); reject(new Error(`Claude sidecar for ${email} failed to register within ${SIDECAR_SPAWN_TIMEOUT_MS}ms`)); }, SIDECAR_SPAWN_TIMEOUT_MS); // Poll for registration (the sidecar connects via WebSocket and registerSidecar is called) const check = () => { const sc = findSidecarByName(name); if (sc) { clearTimeout(timeout); clearInterval(interval); resolve(sc); } }; const interval = setInterval(check, 50); const unsub = () => { clearTimeout(timeout); clearInterval(interval); }; }); } // ── Public API ── export function isConnected(): boolean { return findSidecarByCapability('proxy') !== undefined; } export function getCachedState(): ClaudeState | null { return cachedState; } export async function syncState(): Promise { const res = await sendCommand('proxy', { type: 'state:sync', id: nextId() }); if (res.type === 'state:sync') { cachedState = res.state; return res.state; } throw new Error('Unexpected response'); } export async function getProxySecret(): Promise { if (cachedState?.proxySecret) return cachedState.proxySecret; const res = await sendCommand('proxy', { type: 'proxy:secret', id: nextId() }); if (res.type === 'proxy:secret') return res.secret; throw new Error('Failed to get proxy secret'); } export function getProxySecretSync(): string { return cachedState?.proxySecret ?? ''; } // ── Claude Code (per-user routing) ── export async function spawnClaude(params: ClaudeSpawnParams): Promise { const sc = await ensureClaudeSidecar(params.email); const res = await sendCommandToSidecar(sc, { type: 'claude:spawn', id: nextId(), params }, LONG_TIMEOUT_MS); if (res.type === 'claude:result') return res.result; if (res.type === 'claude:error') throw new Error(res.error); throw new Error('Unexpected response'); } export async function spawnClaudeStreaming(params: ClaudeSpawnStreamingParams): Promise { const sc = await ensureClaudeSidecar(params.email); const res = await sendCommandToSidecar(sc, { type: 'claude:spawn-streaming', id: nextId(), params }); if (res.type === 'claude:spawned') return; if (res.type === 'claude:error') throw new Error(res.error); throw new Error('Unexpected response'); } export function killClaude(sessionKey: string, email: string): void { const sc = findSidecarByName(`claude:${email}`); if (sc) sendFireToSidecar(sc, { type: 'claude:kill', id: nextId(), sessionKey }); } export function clearClaudeSession(sessionKey: string, email?: string): void { if (email) { const sc = findSidecarByName(`claude:${email}`); if (sc) sendFireToSidecar(sc, { type: 'claude:clear-session', id: nextId(), sessionKey }); } else { // Broadcast to all claude sidecars (used when email is not available) for (const sc of sidecars.values()) { if (sc.capabilities.includes('claude')) { sendFireToSidecar(sc, { type: 'claude:clear-session', id: nextId(), sessionKey }); } } } } export function onClaudeEvent(handler: (sessionKey: string, event: PiEvent) => void): () => void { return on('claude:event', (msg) => { if (msg.type === 'claude:event') { handler( (msg as SidecarEvent & { type: 'claude:event' }).sessionKey, (msg as SidecarEvent & { type: 'claude:event' }).event, ); } }); } // ── Pi ── export async function spawnPi(params: PiSpawnParams): Promise { const res = await sendCommand('pi', { type: 'pi:spawn', id: nextId(), params }); if (res.type === 'pi:spawned') return; if (res.type === 'pi:error') throw new Error(res.error); throw new Error('Unexpected response'); } export function sendPiPrompt(sessionId: string, prompt: string, requestId: string): void { sendFire('pi', { type: 'pi:prompt', id: nextId(), sessionId, prompt, requestId }); } export function abortPi(sessionId: string, requestId: string): void { sendFire('pi', { type: 'pi:abort', id: nextId(), sessionId, requestId }); } export function killPi(sessionId: string): void { sendFire('pi', { type: 'pi:kill', id: nextId(), sessionId }); } export function setPiThinking(sessionId: string, level: string): void { sendFire('pi', { type: 'pi:set-thinking', id: nextId(), sessionId, level }); } export function onPiEvent(handler: (sessionId: string, event: PiEvent) => void): () => void { return on('pi:event', (msg) => { if (msg.type === 'pi:event') { handler( (msg as SidecarEvent & { type: 'pi:event' }).sessionId, (msg as SidecarEvent & { type: 'pi:event' }).event, ); } }); } // ── Terminal (PTY sidecar) ── export function sendPtyCommand(cmd: PtyCommand): void { sendFire('terminal', cmd); } export async function sendPtyCommandAsync(cmd: PtyCommand, timeoutMs = DEFAULT_TIMEOUT_MS): Promise { return sendCommand('terminal', cmd, timeoutMs); } export function isTerminalConnected(): boolean { return findSidecarByCapability('terminal') !== undefined; } // ── VNC ── export async function startVnc(params: VncStartParams): Promise<{ port: number; display: number }> { const res = await sendCommand('vnc', { type: 'vnc:start', id: nextId(), params }); if (res.type === 'vnc:started') return { port: res.port, display: res.display }; if (res.type === 'vnc:error') throw new Error(res.error); throw new Error('Unexpected response'); } export function stopVnc(email: string): void { sendFire('vnc', { type: 'vnc:stop', id: nextId(), email }); } export async function getVncStatus(email: string): Promise { const res = await sendCommand('vnc', { type: 'vnc:status', id: nextId(), email }); if (res.type === 'vnc:status') return res.session; throw new Error('Unexpected response'); } export function isVncConnected(): boolean { return findSidecarByCapability('vnc') !== undefined; }