From 86c2d6333a15e3a79cf7d05f944ae029e0151ebe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Padez?= Date: Thu, 5 Mar 2026 07:36:47 +0000 Subject: [PATCH] process sidecar: independent process manager for long-running work Introduces a separate Bun process (port 5100) that owns all spawned processes and long-running work, so the API server can restart freely without disrupting active sessions. The sidecar owns: - Anthropic proxy (port 5051) with persisted secret across restarts - Claude Code process spawning and session tracking (--resume support) - Pi agent spawning and RPC lifecycle (prompt/abort/thinking) - Job queue engine (lane processing, retries, notifications) The API server becomes a thin client that forwards commands over a single WebSocket connection with auto-reconnect. send-claude-code.ts goes from 550 lines of spawn logic to 73 lines of sidecar delegation. State persisted to data/sidecar/state.json every 30s and on shutdown. Lockfile prevents duplicate instances. See SIDECAR.md for full docs and manual testing procedures. Co-Authored-By: Claude Opus 4.6 --- ecosystem.config.cjs | 6 + src/server.tsx | 9 +- src/servers/api/anthropic-proxy.ts | 4 +- src/servers/api/pi/session-manager.ts | 8 +- src/servers/api/pi/types.ts | 2 + src/servers/api/pi/websocket.ts | 111 +++-- src/servers/api/queue/queue.ts | 10 +- src/servers/bootstrap.ts | 6 +- src/servers/channels/discord/handler.ts | 4 +- src/servers/channels/send-claude-code.ts | 551 +---------------------- src/servers/channels/telegram/handler.ts | 4 +- src/servers/channels/whatsapp/handler.ts | 4 +- src/servers/sidecar-client.ts | 303 +++++++++++++ src/servers/sidecar/SIDECAR.md | 347 ++++++++++++++ src/servers/sidecar/claude-manager.ts | 355 +++++++++++++++ src/servers/sidecar/index.ts | 259 +++++++++++ src/servers/sidecar/pi-manager.ts | 416 +++++++++++++++++ src/servers/sidecar/protocol.ts | 115 +++++ src/servers/sidecar/proxy.ts | 104 +++++ src/servers/sidecar/queue-runner.ts | 266 +++++++++++ src/servers/sidecar/state.ts | 141 ++++++ 21 files changed, 2415 insertions(+), 610 deletions(-) create mode 100644 src/servers/sidecar-client.ts create mode 100644 src/servers/sidecar/SIDECAR.md create mode 100644 src/servers/sidecar/claude-manager.ts create mode 100644 src/servers/sidecar/index.ts create mode 100644 src/servers/sidecar/pi-manager.ts create mode 100644 src/servers/sidecar/protocol.ts create mode 100644 src/servers/sidecar/proxy.ts create mode 100644 src/servers/sidecar/queue-runner.ts create mode 100644 src/servers/sidecar/state.ts diff --git a/ecosystem.config.cjs b/ecosystem.config.cjs index 898746e0..758bb23e 100644 --- a/ecosystem.config.cjs +++ b/ecosystem.config.cjs @@ -1,5 +1,11 @@ module.exports = { apps: [ + { + name: 'officer-sidecar', + script: 'bun', + args: 'run src/servers/sidecar/index.ts', + watch: false, + }, { name: 'officer', script: 'bun', diff --git a/src/server.tsx b/src/server.tsx index 40190e6e..ffde24b5 100644 --- a/src/server.tsx +++ b/src/server.tsx @@ -12,7 +12,7 @@ import { desktopWebsocket } from './servers/api/desktop/websocket'; import { findEntryByProxyId, touchEntry } from './servers/api/dev-server/router'; import officerWeb from './apps/officer-web/index.html'; import { startBrowserRelay } from './servers/api/browser/relay'; -import { startAnthropicProxy } from './servers/api/anthropic-proxy'; +import { initSidecarClient } from './servers/sidecar-client'; import { toShellUsername } from './servers/data-path'; const { PORT = '5000' } = process.env; @@ -229,11 +229,8 @@ try { console.error('[browser-relay] failed to start:', err instanceof Error ? err.message : err); } -try { - startAnthropicProxy(); -} catch (err) { - console.error('[anthropic-proxy] failed to start:', err instanceof Error ? err.message : err); -} +// Connect to process sidecar (owns proxy, Claude Code, Pi, queue) +initSidecarClient(); void initTerminalSidecars(); diff --git a/src/servers/api/anthropic-proxy.ts b/src/servers/api/anthropic-proxy.ts index 0b38f4af..e6e99ea5 100644 --- a/src/servers/api/anthropic-proxy.ts +++ b/src/servers/api/anthropic-proxy.ts @@ -6,8 +6,8 @@ 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 with sk-ant- so Claude Code accepts it as a valid API key format -export const proxySecret = `sk-ant-proxy01-${crypto.randomUUID()}`; +// 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?: { diff --git a/src/servers/api/pi/session-manager.ts b/src/servers/api/pi/session-manager.ts index c46fcae9..511d5f60 100644 --- a/src/servers/api/pi/session-manager.ts +++ b/src/servers/api/pi/session-manager.ts @@ -80,8 +80,12 @@ class SessionManager { clearTimeout(session.idleTimer); } - if (session.piProcess) { - session.piProcess.kill(); + // Clean up sidecar subscriptions + if (session._sidecarUnsub) { + session._sidecarUnsub(); + } + if (session._claudeKill) { + session._claudeKill(); } this.sessions.delete(sessionId); diff --git a/src/servers/api/pi/types.ts b/src/servers/api/pi/types.ts index c4763f94..b1a763f3 100644 --- a/src/servers/api/pi/types.ts +++ b/src/servers/api/pi/types.ts @@ -157,6 +157,8 @@ export type UserSession = { systemContextSent: boolean; messages: Message[]; meta: SessionMeta; + _sidecarUnsub?: () => void; + _claudeKill?: () => void; }; export type ModelInfo = { diff --git a/src/servers/api/pi/websocket.ts b/src/servers/api/pi/websocket.ts index 60e6681a..d2bf9e90 100644 --- a/src/servers/api/pi/websocket.ts +++ b/src/servers/api/pi/websocket.ts @@ -5,6 +5,7 @@ import { sessionManager } from './session-manager'; import * as storage from './storage'; import * as piBridge from './pi-bridge'; import { sendClaudeCodeStreaming } from '@@/channels/send-claude-code'; +import * as sidecar from '@@/sidecar-client'; import { join } from 'path'; import { getHomeDirForRole } from '../../../servers/data-path'; import { getUserSettings } from 'officerdb'; @@ -300,32 +301,38 @@ async function handleChat( const onEvent = createEventHandler(sessionId, model, cwd, homeDir); // If session has history, save to disk and pass --session for context replay - let spawnOptions: { sessionFile?: string; username?: string; role?: string } | undefined; + let sessionFile: string | undefined; if (session.messages.length > 0) { await storage.saveSession(homeDir, sessionId, session.meta, session.messages); const hostPath = await storage.getSessionFilePath(homeDir, sessionId); - if (hostPath) { - spawnOptions = { sessionFile: hostPath, username, role: ws.data.role }; - } + if (hostPath) sessionFile = hostPath; } - if (!spawnOptions) spawnOptions = { username, role: ws.data.role }; - session.piProcess = await piBridge.spawnPi(cwd, model, userId, email, onEvent, spawnOptions); - - // Null out piProcess when the process dies so next message triggers respawn - const proc = session.piProcess; - proc.exited.then(() => { - if (session.piProcess === proc) { - session.piProcess = null; - logger.info('Pi process exited, nulled reference', { sessionId }); - } + // Subscribe to sidecar events for this session + const unsub = sidecar.onPiEvent((evtSessionId, event) => { + if (evtSessionId === sessionId) onEvent(event); }); - logger.info('Spawned Pi process for session', { + await sidecar.spawnPi({ + sessionId, + email, + userId, + username, + role: ws.data.role, + cwd, + model, + sessionFile, + }); + + // Mark session as having a live process (use sessionId as sentinel) + session.piProcess = sessionId as any; + session._sidecarUnsub = unsub; + + logger.info('Spawned Pi process via sidecar', { sessionId, model, cwd, - hasSessionFile: !!spawnOptions?.sessionFile, + hasSessionFile: !!sessionFile, }); } catch (err) { logger.error('Failed to spawn Pi process', { sessionId, model, error: String(err) }); @@ -352,13 +359,13 @@ async function handleChat( // Set thinking level if provided console.log(`[pi] model: ${msg.model ?? 'default'}, thinking: ${msg.thinking ?? 'not set'}`); if (msg.thinking) { - piBridge.setThinkingLevel(session.piProcess, msg.thinking); + sidecar.setPiThinking(sessionId, msg.thinking); } - // Send prompt to Pi + // Send prompt to Pi via sidecar const requestId = randomUUID(); session.isGenerating = true; - piBridge.sendPrompt(session.piProcess, msg.prompt, requestId); + sidecar.sendPiPrompt(sessionId, msg.prompt, requestId); } async function handleClaudeCodeChat( @@ -428,15 +435,9 @@ async function handleClaudeCodeChat( onEvent, }); - // Store proc as piProcess so handleStop can kill it - session.piProcess = handle.proc; - - // Null out when process exits so next message spawns a new one - handle.proc.exited.then(() => { - if (session.piProcess === handle.proc) { - session.piProcess = null; - } - }); + // Store sentinel so handleStop can kill it via sidecar + session.piProcess = sessionId as any; + session._claudeKill = handle.kill; } catch (err) { logger.error('Failed to start Claude Code streaming', { sessionId, error: String(err) }); sendToClient(ws, { type: 'error', message: 'Failed to start Claude Code' }); @@ -490,39 +491,36 @@ async function handleResume( const homeDir = getHomeDirForRole(email, ws.data.role); const onEvent = createEventHandler(sessionId, session.model, session.cwd, homeDir); - // If session has history, save to disk and pass --session for context replay - let spawnOptions: { sessionFile?: string; username?: string; role?: string } | undefined; + let sessionFile: string | undefined; if (session.messages.length > 0) { await storage.saveSession(homeDir, sessionId, session.meta, session.messages); const hostPath = await storage.getSessionFilePath(homeDir, sessionId); - if (hostPath) { - spawnOptions = { sessionFile: hostPath, username: ws.data.username, role: ws.data.role }; - } + if (hostPath) sessionFile = hostPath; } - if (!spawnOptions) spawnOptions = { username: ws.data.username, role: ws.data.role }; - session.piProcess = await piBridge.spawnPi( - session.cwd, - session.model, - session.userId!, - email, - onEvent, - spawnOptions, - ); - - // Null out piProcess when the process dies so next message triggers respawn - const proc = session.piProcess; - proc.exited.then(() => { - if (session.piProcess === proc) { - session.piProcess = null; - logger.info('Pi process exited, nulled reference', { sessionId }); - } + // Subscribe to sidecar events for this session + const unsub = sidecar.onPiEvent((evtSessionId, event) => { + if (evtSessionId === sessionId) onEvent(event); }); - logger.info('Spawned fresh Pi process for resumed session', { + await sidecar.spawnPi({ + sessionId, + email, + userId: session.userId!, + username: ws.data.username, + role: ws.data.role, + cwd: session.cwd, + model: session.model, + sessionFile, + }); + + session.piProcess = sessionId as any; + session._sidecarUnsub = unsub; + + logger.info('Spawned fresh Pi process via sidecar for resumed session', { sessionId, model: session.model, - hasSessionFile: !!spawnOptions?.sessionFile, + hasSessionFile: !!sessionFile, }); } catch (err) { logger.error('Failed to spawn Pi process for resume', { sessionId, error: String(err) }); @@ -555,12 +553,11 @@ async function handleStop(ws: ServerWebSocket): Promise { if (session?.piProcess) { try { if (session.model.startsWith('claude-code')) { - // Claude Code: kill the process directly - session.piProcess.kill(); - logger.info('Killed Claude Code process', { sessionId }); + sidecar.killClaude(sessionId); + logger.info('Killed Claude Code process via sidecar', { sessionId }); } else { - piBridge.abort(session.piProcess, randomUUID()); - logger.info('Sent abort to Pi process', { sessionId }); + sidecar.abortPi(sessionId, randomUUID()); + logger.info('Sent abort to Pi process via sidecar', { sessionId }); } session.isGenerating = false; } catch (err) { diff --git a/src/servers/api/queue/queue.ts b/src/servers/api/queue/queue.ts index 1d33531a..6fc727e3 100644 --- a/src/servers/api/queue/queue.ts +++ b/src/servers/api/queue/queue.ts @@ -1,5 +1,5 @@ import { createRouter } from '../../create-router'; -import { enqueue, cancelJob, readJob, listAllJobs } from '../../queue'; +import * as sidecar from '../../sidecar-client'; import { NOT_FOUND } from '../../custom-errors'; export const queueRouter = createRouter(); @@ -10,7 +10,7 @@ queueRouter.get('/jobs', async (ctx) => { const type = ctx.req.query('type'); const status = ctx.req.query('status'); - let jobs = await listAllJobs(); + let jobs = await sidecar.listJobs(); jobs = jobs.filter((j) => j.userId === user.email); if (lane) jobs = jobs.filter((j) => j.lane === lane); @@ -21,7 +21,7 @@ queueRouter.get('/jobs', async (ctx) => { }); queueRouter.get('/jobs/:id', async (ctx) => { - const job = await readJob(ctx.req.param('id')); + const job = await sidecar.getJob(ctx.req.param('id')); if (!job) throw NOT_FOUND('Job not found'); return ctx.json(job); }); @@ -36,12 +36,12 @@ queueRouter.post('/jobs', async (ctx) => { notify?: boolean; }; - const job = await enqueue({ lane, type, userId: user.email, meta, notify }); + const job = await sidecar.enqueueJob({ lane, type, userId: user.email, meta, notify }); return ctx.json(job, 201); }); queueRouter.delete('/jobs/:id', async (ctx) => { - const job = await cancelJob(ctx.req.param('id')); + const job = await sidecar.cancelJob(ctx.req.param('id')); if (!job) throw NOT_FOUND('Job not found'); return ctx.json(job); }); diff --git a/src/servers/bootstrap.ts b/src/servers/bootstrap.ts index f306a63b..417e150d 100644 --- a/src/servers/bootstrap.ts +++ b/src/servers/bootstrap.ts @@ -8,7 +8,7 @@ import { syncSeedExtensions } from './sync-extensions'; import { syncSeedResources } from './sync-resources'; import { migrateSettingsToResources } from './migrate-resources'; import { generateResourceSkill } from './api/pi/pi-bridge'; -import { initQueue } from './queue'; +// Queue is now owned by the sidecar process import { startDiscordBotIfConfigured } from './channels/discord/bot'; import { startTelegramBotIfConfigured } from './channels/telegram/bot'; import { startWhatsAppBotIfConfigured } from './channels/whatsapp/bot'; @@ -78,9 +78,7 @@ async function installPi(): Promise { await migrateSettingsToResources(); generateResourceSkill(DATA_PATH); - await initQueue().catch((err) => { - console.error('[bootstrap] Failed to initialize queue:', err); - }); + // Queue is initialized by the sidecar process await startDiscordBotIfConfigured().catch((err) => { console.error('[channels] Failed to start Discord bot:', err); diff --git a/src/servers/channels/discord/handler.ts b/src/servers/channels/discord/handler.ts index 0a0b7344..634dae64 100644 --- a/src/servers/channels/discord/handler.ts +++ b/src/servers/channels/discord/handler.ts @@ -4,7 +4,7 @@ import { sendAndAwait, getSessionModel, setSessionModel } from '../send-and-awai import { consumePairingCode } from '../pairing'; import { chunkMessage } from './chunker'; import { listPiModels } from '@@/api/pi/list-models'; -import { enqueue } from '@@/queue/engine'; +import { enqueueJob } from '../../sidecar-client'; import { readJob } from '@@/queue/storage'; import { openEmailDb } from '@@/api/email/email-db'; import type { ModelInfo } from '@@/api/pi/types'; @@ -60,7 +60,7 @@ async function handleEmailSync(ctx: CommandContext): Promise { await channel.send('Syncing emails...'); - const job = await enqueue({ lane: 'google-api', type: 'gmail-sync', userId: email }); + const job = await enqueueJob({ lane: 'google-api', type: 'gmail-sync', userId: email }); // Poll until done const poll = async (): Promise<'completed' | 'failed' | 'cancelled'> => { diff --git a/src/servers/channels/send-claude-code.ts b/src/servers/channels/send-claude-code.ts index 256ce235..64eca243 100644 --- a/src/servers/channels/send-claude-code.ts +++ b/src/servers/channels/send-claude-code.ts @@ -1,37 +1,6 @@ -import { join } from 'node:path'; -import { - getHomeDir, - getNativeToolsDir, - getGlobalToolsDir, - getUserToolsDir, - getNativeSkillsDir, - getGlobalSkillsDir, - getUserSkillsDir, - toShellUsername, -} from '@@/data-path'; -import { readToolDirs, parseFrontmatter as parseToolFrontmatter } from '@@/api/tools/tools'; -import { readSkillDirs, parseFrontmatter as parseSkillFrontmatter } from '@@/api/skills/skills'; -import { buildHostToolEnv } from '@@/api/pi/pi-bridge'; import { logger } from '@@/api/pi/logger'; -import { proxySecret } from '@@/api/anthropic-proxy'; import type { MessageCost, PiEvent } from '@@/api/pi/types'; - -const SEND_TIMEOUT_MS = 5 * 60 * 1000; -const PROXY_PORT = process.env.ANTHROPIC_PROXY_PORT ?? '5051'; - -async function hasOwnCredentials(homeDir: string): Promise { - try { - return await Bun.file(join(homeDir, '.claude', '.credentials.json')).exists(); - } catch { - return false; - } -} - -// Resolve absolute path to claude binary so sudo -u can find it regardless of target user's PATH -const CLAUDE_BIN = (() => { - const result = Bun.spawnSync({ cmd: ['which', 'claude'], stdout: 'pipe', stderr: 'ignore' }); - return result.stdout.toString().trim() || 'claude'; -})(); +import * as sidecar from '@@/sidecar-client'; type ClaudeCodeParams = { userId: number; @@ -49,235 +18,13 @@ type ClaudeCodeResult = { cost: MessageCost; }; -type ClaudeCodeOutput = { - result: string; - session_id: string; - cost_usd: number; - input_tokens: number; - output_tokens: number; - is_error: boolean; -}; - -// Map channel session key → Claude Code session ID for --resume -const claudeCodeSessions = new Map(); - export function clearClaudeCodeSession(sessionKey: string): void { - claudeCodeSessions.delete(sessionKey); -} - -// ── Build dynamic system prompt from available tools & skills ── - -async function buildToolsSystemPrompt(email: string): Promise { - const [nativeTools, globalTools, userTools, nativeSkills, globalSkills, userSkills] = await Promise.all([ - readToolDirs(getNativeToolsDir()), - readToolDirs(getGlobalToolsDir()), - readToolDirs(getUserToolsDir(email)), - readSkillDirs(getNativeSkillsDir()), - readSkillDirs(getGlobalSkillsDir()), - readSkillDirs(getUserSkillsDir(email)), - ]); - - // Merge tools (user overrides global overrides native) - const mergedTools = new Map(nativeTools); - for (const [name, path] of globalTools) mergedTools.set(name, path); - for (const [name, path] of userTools) mergedTools.set(name, path); - - // Merge skills - const mergedSkills = new Map(nativeSkills); - for (const [name, path] of globalSkills) mergedSkills.set(name, path); - for (const [name, path] of userSkills) mergedSkills.set(name, path); - - if (mergedTools.size === 0 && mergedSkills.size === 0) return null; - - const sections: string[] = [ - '# Officer Automation System', - '', - 'You are running inside the Officer platform. Officer has its own automation concepts that are DIFFERENT from your built-in tools. When the user or a task references these, use the definitions below — do NOT map them to your own built-in concepts.', - '', - '- **Task**: A markdown file (TASK.md) with instructions for you to execute. When asked to "run a task", read the TASK.md file and follow its instructions step by step.', - '- **Skill**: A knowledge document (SKILL.md) that describes HOW to do something — APIs, commands, patterns. When a task says "use the X skill", follow the instructions from the matching skill section below.', - '- **Tool**: A capability defined by a TOOL.md and implemented in an index.ts file. Tools are NOT shell commands — do NOT try to call them by name. If the tool has a "Run" line below, execute it using that command, passing inputs as a JSON string argument. Only if there is no Run command should you replicate the behavior manually using the TOOL.md documentation.', - '', - 'Skills and tools listed here are AVAILABLE to you. Follow their documentation directly.', - '', - ]; - - if (mergedTools.size > 0) { - const toolLines: string[] = ['# Tools', '']; - const entries = await Promise.all( - Array.from(mergedTools.entries()).map(async ([dirName, filePath]) => { - const raw = await Bun.file(filePath).text(); - const { frontmatter, body } = parseToolFrontmatter(raw); - const toolDir = join(filePath, '..'); - const hasImpl = await Bun.file(`${toolDir}/index.ts`).exists(); - return { - dirName, - name: frontmatter.name || dirName, - description: frontmatter.description, - body, - toolDir, - hasImpl, - }; - }), - ); - const runnerPath = `${getNativeToolsDir()}/run.ts`; - for (const tool of entries) { - toolLines.push(`## ${tool.name}`); - if (tool.hasImpl) toolLines.push(`Run: \`bun run ${runnerPath} ${tool.toolDir} '{"param":"value"}'\``); - if (tool.description) toolLines.push(tool.description); - if (tool.body.trim()) toolLines.push('', tool.body.trim()); - toolLines.push(''); - } - sections.push(toolLines.join('\n')); - } - - if (mergedSkills.size > 0) { - const skillLines: string[] = ['# Skills', '']; - const entries = await Promise.all( - Array.from(mergedSkills.entries()).map(async ([dirName, filePath]) => { - const raw = await Bun.file(filePath).text(); - const { frontmatter, body } = parseSkillFrontmatter(raw); - return { dirName, name: frontmatter.name || dirName, description: frontmatter.description, body }; - }), - ); - for (const skill of entries) { - skillLines.push(`## ${skill.name}`); - if (skill.description) skillLines.push(skill.description); - if (skill.body.trim()) skillLines.push('', skill.body.trim()); - skillLines.push(''); - } - sections.push(skillLines.join('\n')); - } - - return sections.join('\n\n'); + sidecar.clearClaudeSession(sessionKey); } export async function sendClaudeCode(params: ClaudeCodeParams): Promise { - const { userId, email, username, prompt, sessionKey } = params; - const homeDir = getHomeDir(email); - const shellUsername = toShellUsername(username, email); - const toolEnv = await buildHostToolEnv(userId, email); - - const claudeArgs = [CLAUDE_BIN, '-p', prompt, '--dangerously-skip-permissions', '--output-format', 'json']; - - const subModel = params.model?.split('/')[1]; - if (subModel) claudeArgs.push('--model', subModel); - - const existingSession = claudeCodeSessions.get(sessionKey); - if (existingSession) { - claudeArgs.push('--resume', existingSession); - } - - const isServiceUser = shellUsername === (process.env.USER ?? ''); - const userHasCredentials = !isServiceUser && (await hasOwnCredentials(homeDir)); - - // Service user or users with own credentials use their HOME directly. - // Other users route through the local Anthropic proxy. - const authEnv = isServiceUser - ? { HOME: process.env.HOME ?? '' } - : userHasCredentials - ? { HOME: homeDir } - : { ANTHROPIC_BASE_URL: `http://127.0.0.1:${PROXY_PORT}`, ANTHROPIC_API_KEY: proxySecret }; - - const env: Record = { - ...toolEnv, - ...authEnv, - PATH: process.env.PATH ?? '', - TERM: 'xterm-256color', - }; - - logger.info('Claude Code exec', { - sessionKey, - username: shellUsername, - isServiceUser, - userHasCredentials, - resume: existingSession ?? null, - }); - - const proc = isServiceUser - ? Bun.spawn(claudeArgs, { - stdin: 'pipe', - stdout: 'pipe', - stderr: 'pipe', - env: { ...process.env, ...env }, - }) - : Bun.spawn( - ['sudo', '-u', shellUsername, 'env', ...Object.entries(env).map(([k, v]) => `${k}=${v}`), ...claudeArgs], - { - stdin: 'pipe', - stdout: 'pipe', - stderr: 'pipe', - }, - ); - - const timeout = setTimeout(() => { - try { - proc.kill(); - } catch { - /* already dead */ - } - }, SEND_TIMEOUT_MS); - - try { - const [stdout, stderr] = await Promise.all([new Response(proc.stdout).text(), new Response(proc.stderr).text()]); - - const exitCode = await proc.exited; - clearTimeout(timeout); - - if (stderr.trim()) { - logger.info('Claude Code stderr', { text: stderr.trim().slice(0, 500) }); - } - - if (exitCode !== 0 && !stdout.trim()) { - throw new Error(`claude exited with code ${exitCode}: ${stderr.trim().slice(0, 200)}`); - } - - // Parse JSON output - let output: ClaudeCodeOutput; - try { - output = JSON.parse(stdout) as ClaudeCodeOutput; - } catch { - // Non-JSON output — treat raw stdout as result text - return { - text: stdout.trim() || '(no response)', - sessionId: sessionKey, - model: 'claude-code', - cost: { inputTokens: 0, outputTokens: 0, totalUSD: 0 }, - }; - } - - if (output.is_error) { - throw new Error(output.result || 'Claude Code returned an error'); - } - - // Store session for --resume on next message - if (output.session_id) { - claudeCodeSessions.set(sessionKey, output.session_id); - } - - const cost: MessageCost = { - inputTokens: output.input_tokens ?? 0, - outputTokens: output.output_tokens ?? 0, - totalUSD: output.cost_usd ?? 0, - }; - - logger.info('Claude Code result', { - sessionKey, - sessionId: output.session_id, - cost: cost.totalUSD, - tokens: cost.inputTokens + cost.outputTokens, - }); - - return { - text: output.result || '(no response)', - sessionId: sessionKey, - model: 'claude-code', - cost, - }; - } catch (err) { - clearTimeout(timeout); - throw err; - } + logger.info('Claude Code exec (via sidecar)', { sessionKey: params.sessionKey }); + return sidecar.spawnClaude(params); } // ── Streaming variant for Chat Panel WebSocket ── @@ -294,284 +41,32 @@ type ClaudeCodeStreamingParams = { }; type ClaudeCodeStreamingHandle = { - proc: ReturnType; + kill: () => void; }; export async function sendClaudeCodeStreaming(params: ClaudeCodeStreamingParams): Promise { - const { userId, email, username, prompt, sessionKey, cwd, onEvent } = params; + const { onEvent, ...spawnParams } = params; - const shellUsername = toShellUsername(username, email); - const homeDir = getHomeDir(email); - const workDir = cwd ?? homeDir; + logger.info('Claude Code streaming exec (via sidecar)', { sessionKey: params.sessionKey }); - const claudeArgs = [ - CLAUDE_BIN, - '-p', - prompt, - '--dangerously-skip-permissions', - '--output-format', - 'stream-json', - '--verbose', - '--include-partial-messages', - ]; + // Subscribe to events for this session + const unsub = sidecar.onClaudeEvent((sessionKey, event) => { + if (sessionKey === params.sessionKey) { + onEvent(event); - const subModel = params.model?.split('/')[1]; - if (subModel) claudeArgs.push('--model', subModel); - - const existingSession = claudeCodeSessions.get(sessionKey); - if (existingSession) { - claudeArgs.push('--resume', existingSession); - } - - // Append dynamic system prompt with available tools & skills - const systemPrompt = await buildToolsSystemPrompt(email); - if (systemPrompt) { - claudeArgs.push('--append-system-prompt', systemPrompt); - } - - const toolEnv = await buildHostToolEnv(userId, email); - const { CLAUDECODE: _, ...cleanEnv } = process.env; - - const isServiceUser = shellUsername === (process.env.USER ?? ''); - const userHasCredentials = !isServiceUser && (await hasOwnCredentials(homeDir)); - - // Service user or users with own credentials use their HOME directly. - // Other users route through the local Anthropic proxy. - const authEnv = isServiceUser - ? { HOME: cleanEnv.HOME ?? '' } - : userHasCredentials - ? { HOME: homeDir } - : { ANTHROPIC_BASE_URL: `http://127.0.0.1:${PROXY_PORT}`, ANTHROPIC_API_KEY: proxySecret }; - - const env: Record = { - ...toolEnv, - ...authEnv, - PATH: cleanEnv.PATH ?? '', - TERM: 'xterm-256color', - }; - - logger.info('Claude Code streaming exec', { - sessionKey, - username: shellUsername, - isServiceUser, - userHasCredentials, - cwd: workDir, - resume: existingSession ?? null, + // Unsubscribe when we get a terminal event + if (event.type === 'result' || event.type === 'error') { + unsub(); + } + } }); - const proc = isServiceUser - ? Bun.spawn(claudeArgs, { - cwd: workDir, - stdin: 'ignore', - stdout: 'pipe', - stderr: 'pipe', - env: { ...cleanEnv, ...env }, - }) - : Bun.spawn( - ['sudo', '-u', shellUsername, 'env', ...Object.entries(env).map(([k, v]) => `${k}=${v}`), ...claudeArgs], - { - cwd: workDir, - stdin: 'ignore', - stdout: 'pipe', - stderr: 'pipe', - }, - ); + await sidecar.spawnClaudeStreaming(spawnParams); - const timeout = setTimeout(() => { - try { - proc.kill(); - } catch { - /* already dead */ - } - onEvent({ type: 'error', message: 'Claude Code timed out after 5 minutes' }); - }, SEND_TIMEOUT_MS); - - // Process NDJSON stream in background - (async () => { - try { - const stdout = proc.stdout as ReadableStream; - const reader = stdout.getReader(); - const decoder = new TextDecoder(); - let buffer = ''; - let textBuffer = ''; - let gotResult = false; - - let lineCount = 0; - const processLine = (line: string) => { - if (!line.trim()) return; - lineCount++; - try { - const msg = JSON.parse(line) as Record; - const type = msg.type as string; - if (lineCount <= 5 || type === 'result') { - logger.info('Claude Code NDJSON', { sessionKey, lineCount, type, subtype: msg.subtype ?? null }); - } - - // stream_event — partial streaming (text deltas) - if (type === 'stream_event') { - const event = msg.event as Record | undefined; - if (event?.type === 'content_block_delta') { - const delta = event.delta as Record | undefined; - if (delta?.type === 'text_delta' && typeof delta.text === 'string') { - textBuffer += delta.text; - onEvent({ type: 'delta', text: delta.text }); - } - } - } - - // assistant — complete message with text and tool_use blocks - else if (type === 'assistant') { - const message = msg.message as Record | undefined; - const content = message?.content as Array> | undefined; - if (Array.isArray(content)) { - for (const block of content) { - if (block.type === 'text' && typeof block.text === 'string') { - // Full text block — emit as text event, reset streaming buffer - onEvent({ type: 'text', text: block.text }); - textBuffer = ''; - } else if (block.type === 'tool_use') { - // Flush any pending streamed text before tool - if (textBuffer) { - onEvent({ type: 'text', text: textBuffer }); - textBuffer = ''; - } - onEvent({ - type: 'tool:start', - toolCallId: (block.id as string) ?? '', - toolName: (block.name as string) ?? 'unknown', - toolInput: (block.input as Record) ?? {}, - }); - } - } - } - } - - // user — tool results - else if (type === 'user') { - const message = msg.message as Record | undefined; - const content = message?.content as Array> | undefined; - if (Array.isArray(content)) { - for (const block of content) { - if (block.type === 'tool_result') { - let output = ''; - if (typeof block.content === 'string') { - output = block.content; - } else if (Array.isArray(block.content)) { - output = (block.content as Array>) - .filter((c) => c.type === 'text') - .map((c) => c.text as string) - .join('\n'); - } - onEvent({ - type: 'tool:result', - toolCallId: (block.tool_use_id as string) ?? '', - output, - isError: (block.is_error as boolean) ?? false, - }); - } - } - } - } - - // system init — extract session_id for --resume - else if (type === 'system' && msg.subtype === 'init') { - const sessionId = msg.session_id as string | undefined; - if (sessionId) { - claudeCodeSessions.set(sessionKey, sessionId); - } - } - - // result — final - else if (type === 'result') { - gotResult = true; - clearTimeout(timeout); - - const isError = (msg.is_error as boolean) ?? false; - const resultText = (msg.result as string) ?? ''; - - if (isError) { - logger.info('Claude Code streaming error result', { sessionKey, error: resultText.slice(0, 500) }); - if (textBuffer) { - onEvent({ type: 'text', text: textBuffer }); - textBuffer = ''; - } - onEvent({ type: 'error', message: resultText || 'Claude Code returned an error' }); - return; - } - - if (textBuffer) { - onEvent({ type: 'text', text: textBuffer }); - textBuffer = ''; - } - - const usage = msg.usage as Record | undefined; - const cost: MessageCost = { - inputTokens: usage?.input_tokens ?? 0, - outputTokens: usage?.output_tokens ?? 0, - totalUSD: (msg.total_cost_usd as number) ?? 0, - }; - - const sessionId = msg.session_id as string | undefined; - if (sessionId) { - claudeCodeSessions.set(sessionKey, sessionId); - } - - logger.info('Claude Code streaming result', { sessionKey, cost: cost.totalUSD }); - onEvent({ type: 'result', cost }); - } - } catch { - // Skip malformed JSON lines - } - }; - - while (true) { - const { done, value } = await reader.read(); - if (done) break; - - const chunk = decoder.decode(value, { stream: true }); - buffer += chunk; - const lines = buffer.split('\n'); - buffer = lines.pop()!; - - for (const line of lines) { - processLine(line); - } - } - - // Process any remaining buffer - if (buffer.trim()) { - processLine(buffer); - } - - logger.info('Claude Code stream ended', { sessionKey, totalLines: lineCount, gotResult }); - clearTimeout(timeout); - - // If process exited without a result event, emit error or synthetic result - if (!gotResult) { - const exitCode = await proc.exited; - const stderr = await new Response(proc.stderr as ReadableStream).text(); - logger.info('Claude Code exited without result event', { - sessionKey, - exitCode, - stderr: stderr.trim().slice(0, 500), - }); - if (textBuffer) { - onEvent({ type: 'text', text: textBuffer }); - } - if (exitCode !== 0) { - onEvent({ - type: 'error', - message: `Claude Code exited with code ${exitCode}: ${stderr.trim().slice(0, 200)}`, - }); - } else { - onEvent({ type: 'result', cost: { inputTokens: 0, outputTokens: 0, totalUSD: 0 } }); - } - } - } catch (err) { - clearTimeout(timeout); - onEvent({ type: 'error', message: String(err) }); - } - })(); - - return { proc }; + return { + kill: () => { + sidecar.killClaude(params.sessionKey); + unsub(); + }, + }; } diff --git a/src/servers/channels/telegram/handler.ts b/src/servers/channels/telegram/handler.ts index 5cebf4aa..ae0f33a7 100644 --- a/src/servers/channels/telegram/handler.ts +++ b/src/servers/channels/telegram/handler.ts @@ -5,7 +5,7 @@ import { consumePairingCode } from '../pairing'; import { chunkMessage } from './chunker'; import { getTelegramBot } from './bot'; import { listPiModels } from '@@/api/pi/list-models'; -import { enqueue } from '@@/queue/engine'; +import { enqueueJob } from '../../sidecar-client'; import { readJob } from '@@/queue/storage'; import { openEmailDb } from '@@/api/email/email-db'; import type { ModelInfo } from '@@/api/pi/types'; @@ -60,7 +60,7 @@ async function handleEmailSync(ctx: CommandContext): Promise { await send('Syncing emails...'); - const job = await enqueue({ lane: 'google-api', type: 'gmail-sync', userId: email }); + const job = await enqueueJob({ lane: 'google-api', type: 'gmail-sync', userId: email }); const poll = async (): Promise<'completed' | 'failed' | 'cancelled'> => { for (let i = 0; i < 120; i++) { diff --git a/src/servers/channels/whatsapp/handler.ts b/src/servers/channels/whatsapp/handler.ts index 4ec463cc..5dd309d1 100644 --- a/src/servers/channels/whatsapp/handler.ts +++ b/src/servers/channels/whatsapp/handler.ts @@ -4,7 +4,7 @@ import { sendAndAwait, getSessionModel, setSessionModel } from '../send-and-awai import { consumePairingCode } from '../pairing'; import { getWhatsAppClient } from './bot'; import { listPiModels } from '@@/api/pi/list-models'; -import { enqueue } from '@@/queue/engine'; +import { enqueueJob } from '../../sidecar-client'; import { readJob } from '@@/queue/storage'; import { openEmailDb } from '@@/api/email/email-db'; import type { ModelInfo } from '@@/api/pi/types'; @@ -64,7 +64,7 @@ async function handleEmailSync(ctx: CommandContext): Promise { await send('Syncing emails...'); - const job = await enqueue({ lane: 'google-api', type: 'gmail-sync', userId: email }); + const job = await enqueueJob({ lane: 'google-api', type: 'gmail-sync', userId: email }); const poll = async (): Promise<'completed' | 'failed' | 'cancelled'> => { for (let i = 0; i < 120; i++) { diff --git a/src/servers/sidecar-client.ts b/src/servers/sidecar-client.ts new file mode 100644 index 00000000..8451eeb9 --- /dev/null +++ b/src/servers/sidecar-client.ts @@ -0,0 +1,303 @@ +import type { + SidecarCommand, + SidecarEvent, + SidecarState, + ClaudeSpawnParams, + ClaudeSpawnStreamingParams, + ClaudeCodeResult, + PiSpawnParams, +} from './sidecar/protocol'; +import type { PiEvent } from './api/pi/types'; +import type { Job, EnqueueParams } from './queue/types'; + +const SIDECAR_PORT = Number(process.env.SIDECAR_PORT ?? '5100'); +const SIDECAR_URL = `ws://127.0.0.1:${SIDECAR_PORT}`; + +const RECONNECT_DELAYS = [200, 500, 1000, 2000, 4000, 8000, 15000]; + +type PendingRequest = { + resolve: (value: SidecarEvent) => void; + reject: (error: Error) => void; + timer: Timer; +}; + +type EventHandler = (event: SidecarEvent) => void; + +let ws: WebSocket | null = null; +let connected = false; +let reconnectAttempt = 0; +let reconnectTimer: Timer | null = null; +const pending = new Map(); +const eventHandlers = new Map>(); +let cachedState: SidecarState | null = null; + +// ── Connection management ── + +function connect() { + if (ws && (ws.readyState === WebSocket.CONNECTING || ws.readyState === WebSocket.OPEN)) return; + + try { + ws = new WebSocket(SIDECAR_URL); + } catch { + scheduleReconnect(); + return; + } + + ws.onopen = () => { + connected = true; + reconnectAttempt = 0; + console.log('[sidecar-client] connected'); + + // Sync state on connect + sendCommand({ type: 'state:sync', id: nextId() }).then((res) => { + if (res.type === 'state:sync') { + cachedState = res.state; + console.log('[sidecar-client] state synced'); + } + }).catch(() => { /* best effort */ }); + }; + + ws.onmessage = (event) => { + try { + const msg = JSON.parse(event.data as string) as SidecarEvent; + + // 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); + } catch { + // skip malformed messages + } + }; + + ws.onclose = () => { + connected = false; + ws = null; + rejectAllPending('WebSocket disconnected'); + scheduleReconnect(); + }; + + ws.onerror = () => { + // onclose will fire after this + }; +} + +function scheduleReconnect() { + if (reconnectTimer) return; + const delay = RECONNECT_DELAYS[Math.min(reconnectAttempt, RECONNECT_DELAYS.length - 1)]!; + reconnectAttempt++; + reconnectTimer = setTimeout(() => { + reconnectTimer = null; + connect(); + }, delay); +} + +function rejectAllPending(reason: string) { + for (const [id, req] of pending) { + clearTimeout(req.timer); + req.reject(new Error(reason)); + } + pending.clear(); +} + +// ── Event dispatch ── + +function dispatchEvent(msg: SidecarEvent) { + 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 ── + +let idCounter = 0; +function nextId(): string { + return `sc_${Date.now()}_${++idCounter}`; +} + +const DEFAULT_TIMEOUT_MS = 30_000; +const LONG_TIMEOUT_MS = 6 * 60 * 1000; // 6 min for claude spawn + +function sendCommand(cmd: SidecarCommand, timeoutMs = DEFAULT_TIMEOUT_MS): Promise { + return new Promise((resolve, reject) => { + if (!ws || ws.readyState !== WebSocket.OPEN) { + reject(new Error('Sidecar not connected')); + return; + } + + const timer = setTimeout(() => { + pending.delete(cmd.id); + reject(new Error(`Sidecar command ${cmd.type} timed out`)); + }, timeoutMs); + + pending.set(cmd.id, { resolve, reject, timer }); + ws.send(JSON.stringify(cmd)); + }); +} + +function sendFire(cmd: SidecarCommand): void { + if (ws && ws.readyState === WebSocket.OPEN) { + ws.send(JSON.stringify(cmd)); + } +} + +// ── Public API ── + +export function isConnected(): boolean { + return connected; +} + +export function getCachedState(): SidecarState | null { + return cachedState; +} + +export async function syncState(): Promise { + const res = await sendCommand({ 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({ 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 ── + +export async function spawnClaude(params: ClaudeSpawnParams): Promise { + const res = await sendCommand({ 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 res = await sendCommand({ 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): void { + sendFire({ type: 'claude:kill', id: nextId(), sessionKey }); +} + +export function clearClaudeSession(sessionKey: string): void { + sendFire({ 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.sessionKey, msg.event); + } + }); +} + +// ── Pi ── + +export async function spawnPi(params: PiSpawnParams): Promise { + const res = await sendCommand({ 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({ type: 'pi:prompt', id: nextId(), sessionId, prompt, requestId }); +} + +export function abortPi(sessionId: string, requestId: string): void { + sendFire({ type: 'pi:abort', id: nextId(), sessionId, requestId }); +} + +export function killPi(sessionId: string): void { + sendFire({ type: 'pi:kill', id: nextId(), sessionId }); +} + +export function setPiThinking(sessionId: string, level: string): void { + sendFire({ 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.sessionId, msg.event); + } + }); +} + +// ── Queue ── + +export async function enqueueJob(params: EnqueueParams): Promise { + const res = await sendCommand({ type: 'queue:enqueue', id: nextId(), params }); + if (res.type === 'queue:enqueued') return res.job; + if (res.type === 'queue:error') throw new Error(res.error); + throw new Error('Unexpected response'); +} + +export async function cancelJob(jobId: string): Promise { + const res = await sendCommand({ type: 'queue:cancel', id: nextId(), jobId }); + if (res.type === 'queue:cancelled') return res.job; + if (res.type === 'queue:error') throw new Error(res.error); + throw new Error('Unexpected response'); +} + +export async function listJobs(): Promise { + const res = await sendCommand({ type: 'queue:list', id: nextId() }); + if (res.type === 'queue:list') return res.jobs; + throw new Error('Unexpected response'); +} + +export async function getJob(jobId: string): Promise { + const res = await sendCommand({ type: 'queue:get', id: nextId(), jobId }); + if (res.type === 'queue:get') return res.job; + throw new Error('Unexpected response'); +} + +// ── Health check ── + +export async function isSidecarAlive(): Promise { + try { + const res = await fetch(`http://127.0.0.1:${SIDECAR_PORT}/`, { signal: AbortSignal.timeout(1000) }); + const text = await res.text(); + return text === 'process-sidecar'; + } catch { + return false; + } +} + +// ── Init ── + +export function initSidecarClient(): void { + connect(); +} diff --git a/src/servers/sidecar/SIDECAR.md b/src/servers/sidecar/SIDECAR.md new file mode 100644 index 00000000..9efd73c0 --- /dev/null +++ b/src/servers/sidecar/SIDECAR.md @@ -0,0 +1,347 @@ +# Process Sidecar + +Independent Bun process that owns all long-running work so the API server can restart without disrupting active sessions. + +## Problem + +The API server (port 5000) previously owned all spawned processes: Pi agents, Claude Code sessions, the Anthropic proxy, and the job queue. Restarting the API server would: + +- Kill active Pi and Claude Code conversations mid-response +- Regenerate the Anthropic proxy secret, breaking any Claude Code sessions using it +- Lose in-flight job progress (queue engine ran in-process) + +## Solution + +A separate Bun process ("process sidecar") on port 5100 that owns all spawned processes. The API server communicates with it over a single WebSocket connection. The sidecar is managed by pm2 and starts before the API server. + +## Architecture + +``` +┌─────────────────────────┐ ┌──────────────────────────────┐ +│ API Server (:5000) │ WS │ Process Sidecar (:5100) │ +│ │◄───────►│ │ +│ - Browser WS clients │ │ - Anthropic Proxy (:5051) │ +│ - REST API routes │ │ - Claude Code processes │ +│ - Channel bots │ │ - Pi agent processes │ +│ - sidecar-client.ts │ │ - Job queue engine │ +│ (auto-reconnect) │ │ - State persistence │ +└─────────────────────────┘ └──────────────────────────────┘ +``` + +### What the sidecar owns + +| Concern | Previous location | Sidecar module | +|---------|-------------------|----------------| +| Anthropic proxy (port 5051) | `api/anthropic-proxy.ts` | `sidecar/proxy.ts` | +| Claude Code spawn + session tracking | `channels/send-claude-code.ts` | `sidecar/claude-manager.ts` | +| Pi agent spawn + RPC commands | `api/pi/pi-bridge.ts` | `sidecar/pi-manager.ts` | +| Job queue engine + handlers | `queue/engine.ts` | `sidecar/queue-runner.ts` | + +### What stays in the API server + +- Browser WebSocket connections (ephemeral by nature) +- REST API routes (now thin proxies to sidecar) +- Channel bots (Discord/Telegram/WhatsApp — already reconnect gracefully) +- Browser relay (CDP state is ephemeral) +- PTY sidecar (already its own process, unchanged) + +## Communication Protocol + +Single WebSocket between API server and sidecar. JSON messages with `{ type, id?, ... }` envelopes. + +**Request/response**: Commands include an `id` field. The sidecar responds with a message carrying the same `id`. The client correlates responses via this ID with configurable timeouts. + +**Streaming events**: Pi and Claude Code output events are broadcast to all connected clients without a correlation ID. They carry a `sessionId` or `sessionKey` so the API server can route them to the correct browser WS. + +### Command categories + +``` +ping / pong — health check +state:sync — full state snapshot on connect + +proxy:secret — get persisted proxy secret + +claude:spawn / claude:result — blocking Claude Code exec +claude:spawn-streaming / claude:event — streaming Claude Code exec +claude:kill / claude:clear-session — session management + +pi:spawn / pi:prompt / pi:abort — Pi agent lifecycle +pi:kill / pi:set-thinking — Pi session control + +queue:enqueue / queue:cancel — job management +queue:list / queue:get — job queries +``` + +See `protocol.ts` for the full type definitions. + +## State Persistence + +File: `data/sidecar/state.json` + +Written every 30 seconds (debounced) and on graceful shutdown (SIGTERM/SIGINT). Contains: + +- **proxySecret** — generated once on first boot, reused forever. This is the key fix: the Anthropic proxy secret no longer changes on restart. +- **claudeSessions** — map of `sessionKey → Claude Code session_id` for `--resume` support across restarts. +- **piSessions** — session metadata with PIDs for liveness checking on restart. + +### Lockfile + +`data/sidecar/sidecar.lock` — contains the PID of the running sidecar. On startup, if the lock exists and the PID is alive, the sidecar exits. Stale locks (dead PID) are cleaned up automatically. + +## File Structure + +``` +src/servers/sidecar/ + index.ts — entry point: Bun.serve on :5100, WS dispatch, shutdown + protocol.ts — shared message types (imported by both sides) + proxy.ts — Anthropic proxy server (moved from anthropic-proxy.ts) + claude-manager.ts — Claude Code blocking + streaming spawn, session map + pi-manager.ts — Pi agent spawn, RPC (prompt/abort/thinking), event parsing + queue-runner.ts — Job queue engine (lanes, retries, notifications) + state.ts — File-backed state persistence + lockfile + +src/servers/sidecar-client.ts — API server's WebSocket client (singleton) +``` + +## API Server Integration + +The API server connects to the sidecar on startup via `initSidecarClient()` in `server.tsx`. The client auto-reconnects with exponential backoff (200ms → 15s). + +On connect, it sends `state:sync` to get the current proxy secret and live session info. + +### Modified API server files + +| File | Change | +|------|--------| +| `server.tsx` | `startAnthropicProxy()` → `initSidecarClient()` | +| `bootstrap.ts` | Removed `initQueue()` (sidecar owns it) | +| `channels/send-claude-code.ts` | 550 lines → 73 lines thin client | +| `api/pi/websocket.ts` | Spawn/prompt/abort go through sidecar | +| `api/pi/session-manager.ts` | Cleanup sidecar subscriptions on delete | +| `api/queue/queue.ts` | Routes use sidecar client | +| `channels/discord/handler.ts` | `enqueue` → `enqueueJob` via sidecar | +| `channels/telegram/handler.ts` | Same | +| `channels/whatsapp/handler.ts` | Same | + +## PM2 Configuration + +```js +// ecosystem.config.cjs +module.exports = { + apps: [ + { + name: 'officer-sidecar', // starts first + script: 'bun', + args: 'run src/servers/sidecar/index.ts', + watch: false, + }, + { + name: 'officer', + script: 'bun', + args: 'start', + watch: false, + }, + ], +}; +``` + +The sidecar is listed first so pm2 starts it before the API server. The API server's sidecar client handles the case where the sidecar isn't ready yet (auto-reconnect with backoff). + +## Environment Variables + +| Variable | Default | Description | +|----------|---------|-------------| +| `SIDECAR_PORT` | `5100` | Sidecar HTTP/WS port | +| `ANTHROPIC_PROXY_PORT` | `5051` | Anthropic proxy port (owned by sidecar) | +| `DATA_PATH` | `./data` | Shared data directory | + +## Manual Testing + +### 1. Start the sidecar standalone + +```bash +# From monorepo root +bun run src/servers/sidecar/index.ts +``` + +Expected output: +``` +[sidecar:proxy] listening on 127.0.0.1:5051 +[sidecar:queue] initialized +[sidecar] listening on 127.0.0.1:5100 +``` + +### 2. Health check + +```bash +# Basic liveness +curl http://127.0.0.1:5100/ +# → "process-sidecar" + +# Detailed health +curl http://127.0.0.1:5100/health +# → {"status":"ok","uptime":12345,"piSessions":0,"claudeSessions":0} +``` + +### 3. Verify Anthropic proxy is running + +```bash +# Should reject without valid secret +curl -s http://127.0.0.1:5051/v1/messages +# → {"error":"Unauthorized"} + +# Check state file was created +cat data/sidecar/state.json +# → should show proxySecret, empty claudeSessions, empty piSessions +``` + +### 4. Verify proxy secret persistence + +```bash +# Note the proxySecret from state.json +cat data/sidecar/state.json | jq .proxySecret + +# Stop and restart the sidecar +# Kill the sidecar (Ctrl+C or kill) +bun run src/servers/sidecar/index.ts + +# Check the secret is the same +cat data/sidecar/state.json | jq .proxySecret +# → should be identical to before +``` + +### 5. WebSocket communication test + +```bash +# In one terminal, start the sidecar +bun run src/servers/sidecar/index.ts + +# In another terminal, connect with websocat (or wscat) +# Install: cargo install websocat OR npm install -g wscat +websocat ws://127.0.0.1:5100 + +# Send a ping +{"type":"ping","id":"test1"} +# → should receive: {"type":"pong","id":"test1"} + +# Send state sync +{"type":"state:sync","id":"test2"} +# → should receive: {"type":"state:sync","id":"test2","state":{...}} + +# Get proxy secret +{"type":"proxy:secret","id":"test3"} +# → should receive: {"type":"proxy:secret","id":"test3","secret":"sk-ant-api03-..."} +``` + +### 6. Test lockfile protection + +```bash +# Start sidecar in one terminal +bun run src/servers/sidecar/index.ts + +# Try starting another in a second terminal +bun run src/servers/sidecar/index.ts +# → should exit with: "[sidecar] another instance is already running" + +# Check lockfile +cat data/sidecar/sidecar.lock +# → PID of the running sidecar +``` + +### 7. Full integration test (sidecar + API server) + +```bash +# Start sidecar first +bun run src/servers/sidecar/index.ts & + +# Start API server +bun start +# → should see "[sidecar-client] connected" in logs + +# Test via the dashboard: +# 1. Open a chat panel, send a message with claude-code model +# → should see streaming response (routed through sidecar) +# 2. Open a chat with a Pi model +# → should see streaming response (routed through sidecar) +# 3. Restart the API server (kill + bun start) +# → active Claude Code processes should NOT die +# → sidecar should show "client disconnected" then "client connected" +# → proxy secret should remain the same +``` + +### 8. Test API server restart resilience + +This is the key scenario that motivated the sidecar: + +```bash +# 1. Start sidecar + API server +bun run src/servers/sidecar/index.ts & +bun start & + +# 2. Start a Claude Code streaming session in the dashboard + +# 3. While Claude Code is running, kill the API server +kill $(pgrep -f "bun start") + +# 4. Restart the API server +bun start + +# 5. Check: +# - The Claude Code process should still be running (check with ps) +# - The proxy secret should be the same (check data/sidecar/state.json) +# - The sidecar should show the reconnection in logs +``` + +### 9. Test with pm2 + +```bash +pm2 start ecosystem.config.cjs + +# Check both are running +pm2 status +# → officer-sidecar: online +# → officer: online + +# Restart API server only +pm2 restart officer + +# Check sidecar is still running +pm2 status +curl http://127.0.0.1:5100/health + +# Stop everything +pm2 stop all +``` + +### 10. Queue test + +```bash +# With sidecar running, queue a job via the API +curl -X POST http://localhost:5000/api/queue/jobs \ + -H "Content-Type: application/json" \ + -H "Authorization: Bearer " \ + -d '{"lane":"test","type":"gmail-sync"}' + +# List jobs +curl http://localhost:5000/api/queue/jobs \ + -H "Authorization: Bearer " +``` + +## Graceful Shutdown + +On SIGTERM or SIGINT, the sidecar: +1. Flushes pending state to `data/sidecar/state.json` +2. Releases the lockfile +3. Does **NOT** kill spawned processes — they are independent OS processes + +On restart, the sidecar reads the persisted state and checks which PIDs are still alive. + +## Risks & Mitigations + +| Risk | Mitigation | +|------|------------| +| Sidecar dies with live sessions | pm2 auto-restart + state.json + processes are independent OS processes | +| API↔sidecar WebSocket drops | Auto-reconnect with exponential backoff (200ms → 15s) | +| Two sidecar instances running | Lockfile with PID liveness check on startup | +| Sidecar needs DB access | Imports DB modules directly (same user, same filesystem) | +| Queue handlers need server context | Handlers are self-contained modules imported by the sidecar | diff --git a/src/servers/sidecar/claude-manager.ts b/src/servers/sidecar/claude-manager.ts new file mode 100644 index 00000000..e605b9e4 --- /dev/null +++ b/src/servers/sidecar/claude-manager.ts @@ -0,0 +1,355 @@ +import { join } from 'node:path'; +import type { Subprocess } from 'bun'; +import type { PiEvent, MessageCost } from '../api/pi/types'; +import type { ClaudeSpawnParams, ClaudeSpawnStreamingParams, ClaudeCodeResult } from './protocol'; +import { getState, setClaudeSession, clearClaudeSession, getClaudeSession } from './state'; +import { getProxySecret } from './proxy'; + +const SEND_TIMEOUT_MS = 5 * 60 * 1000; +const PROXY_PORT = process.env.ANTHROPIC_PROXY_PORT ?? '5051'; +const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data'); + +const getHomeDir = (email: string) => join(DATA_PATH, email, 'home'); + +const toShellUsername = (username: string, email: string): string => { + const raw = username || email.split('@')[0]!; + return raw.replace(/@.*$/, '').replace(/[^a-zA-Z0-9._-]/g, '_').toLowerCase().slice(0, 32) || 'officer'; +}; + +async function hasOwnCredentials(homeDir: string): Promise { + try { + return await Bun.file(join(homeDir, '.claude', '.credentials.json')).exists(); + } catch { + return false; + } +} + +// Resolve absolute path to claude binary +const CLAUDE_BIN = (() => { + const result = Bun.spawnSync({ cmd: ['which', 'claude'], stdout: 'pipe', stderr: 'ignore' }); + return result.stdout.toString().trim() || 'claude'; +})(); + +// Active streaming processes +const activeProcs = new Map(); + +function buildAuthEnv(shellUsername: string, homeDir: string, isServiceUser: boolean, userHasCredentials: boolean): Record { + if (isServiceUser) return { HOME: process.env.HOME ?? '' }; + if (userHasCredentials) return { HOME: homeDir }; + return { ANTHROPIC_BASE_URL: `http://127.0.0.1:${PROXY_PORT}`, ANTHROPIC_API_KEY: getProxySecret() }; +} + +// ── Blocking send ── + +type ClaudeCodeOutput = { + result: string; + session_id: string; + cost_usd: number; + input_tokens: number; + output_tokens: number; + is_error: boolean; +}; + +export async function spawnClaude(params: ClaudeSpawnParams): Promise { + const { userId, email, username, prompt, sessionKey } = params; + const homeDir = getHomeDir(email); + const shellUsername = toShellUsername(username, email); + + const claudeArgs = [CLAUDE_BIN, '-p', prompt, '--dangerously-skip-permissions', '--output-format', 'json']; + + const subModel = params.model?.split('/')[1]; + if (subModel) claudeArgs.push('--model', subModel); + + const existingSession = getClaudeSession(sessionKey); + if (existingSession) { + claudeArgs.push('--resume', existingSession); + } + + const isServiceUser = shellUsername === (process.env.USER ?? ''); + const userHasCredentials = !isServiceUser && (await hasOwnCredentials(homeDir)); + const authEnv = buildAuthEnv(shellUsername, homeDir, isServiceUser, userHasCredentials); + + const env: Record = { + ...authEnv, + PATH: process.env.PATH ?? '', + TERM: 'xterm-256color', + }; + + const proc = isServiceUser + ? Bun.spawn(claudeArgs, { stdin: 'pipe', stdout: 'pipe', stderr: 'pipe', env: { ...process.env, ...env } }) + : Bun.spawn( + ['sudo', '-u', shellUsername, 'env', ...Object.entries(env).map(([k, v]) => `${k}=${v}`), ...claudeArgs], + { stdin: 'pipe', stdout: 'pipe', stderr: 'pipe' }, + ); + + const timeout = setTimeout(() => { + try { proc.kill(); } catch { /* already dead */ } + }, SEND_TIMEOUT_MS); + + try { + const [stdout, stderr] = await Promise.all([new Response(proc.stdout).text(), new Response(proc.stderr).text()]); + const exitCode = await proc.exited; + clearTimeout(timeout); + + if (exitCode !== 0 && !stdout.trim()) { + throw new Error(`claude exited with code ${exitCode}: ${stderr.trim().slice(0, 200)}`); + } + + let output: ClaudeCodeOutput; + try { + output = JSON.parse(stdout) as ClaudeCodeOutput; + } catch { + return { + text: stdout.trim() || '(no response)', + sessionId: sessionKey, + model: 'claude-code', + cost: { inputTokens: 0, outputTokens: 0, totalUSD: 0 }, + }; + } + + if (output.is_error) { + throw new Error(output.result || 'Claude Code returned an error'); + } + + if (output.session_id) { + setClaudeSession(sessionKey, output.session_id); + } + + return { + text: output.result || '(no response)', + sessionId: sessionKey, + model: 'claude-code', + cost: { + inputTokens: output.input_tokens ?? 0, + outputTokens: output.output_tokens ?? 0, + totalUSD: output.cost_usd ?? 0, + }, + }; + } catch (err) { + clearTimeout(timeout); + throw err; + } +} + +// ── Streaming send ── + +export async function spawnClaudeStreaming( + params: ClaudeSpawnStreamingParams, + onEvent: (event: PiEvent) => void, +): Promise { + const { userId, email, username, prompt, sessionKey, cwd } = params; + const shellUsername = toShellUsername(username, email); + const homeDir = getHomeDir(email); + const workDir = cwd ?? homeDir; + + const claudeArgs = [ + CLAUDE_BIN, '-p', prompt, + '--dangerously-skip-permissions', + '--output-format', 'stream-json', + '--verbose', '--include-partial-messages', + ]; + + const subModel = params.model?.split('/')[1]; + if (subModel) claudeArgs.push('--model', subModel); + + const existingSession = getClaudeSession(sessionKey); + if (existingSession) { + claudeArgs.push('--resume', existingSession); + } + + const { CLAUDECODE: _, ...cleanEnv } = process.env; + + const isServiceUser = shellUsername === (process.env.USER ?? ''); + const userHasCredentials = !isServiceUser && (await hasOwnCredentials(homeDir)); + const authEnv = buildAuthEnv(shellUsername, homeDir, isServiceUser, userHasCredentials); + + const env: Record = { + ...authEnv, + PATH: cleanEnv.PATH ?? '', + TERM: 'xterm-256color', + }; + + const proc = isServiceUser + ? Bun.spawn(claudeArgs, { cwd: workDir, stdin: 'ignore', stdout: 'pipe', stderr: 'pipe', env: { ...cleanEnv, ...env } }) + : Bun.spawn( + ['sudo', '-u', shellUsername, 'env', ...Object.entries(env).map(([k, v]) => `${k}=${v}`), ...claudeArgs], + { cwd: workDir, stdin: 'ignore', stdout: 'pipe', stderr: 'pipe' }, + ); + + activeProcs.set(sessionKey, proc); + + const timeout = setTimeout(() => { + try { proc.kill(); } catch { /* already dead */ } + onEvent({ type: 'error', message: 'Claude Code timed out after 5 minutes' }); + }, SEND_TIMEOUT_MS); + + // Process NDJSON stream + try { + const stdout = proc.stdout as ReadableStream; + const reader = stdout.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + let textBuffer = ''; + let gotResult = false; + + const processLine = (line: string) => { + if (!line.trim()) return; + try { + const msg = JSON.parse(line) as Record; + const type = msg.type as string; + + if (type === 'stream_event') { + const event = msg.event as Record | undefined; + if (event?.type === 'content_block_delta') { + const delta = event.delta as Record | undefined; + if (delta?.type === 'text_delta' && typeof delta.text === 'string') { + textBuffer += delta.text; + onEvent({ type: 'delta', text: delta.text }); + } + } + } else if (type === 'assistant') { + const message = msg.message as Record | undefined; + const content = message?.content as Array> | undefined; + if (Array.isArray(content)) { + for (const block of content) { + if (block.type === 'text' && typeof block.text === 'string') { + onEvent({ type: 'text', text: block.text }); + textBuffer = ''; + } else if (block.type === 'tool_use') { + if (textBuffer) { + onEvent({ type: 'text', text: textBuffer }); + textBuffer = ''; + } + onEvent({ + type: 'tool:start', + toolCallId: (block.id as string) ?? '', + toolName: (block.name as string) ?? 'unknown', + toolInput: (block.input as Record) ?? {}, + }); + } + } + } + } else if (type === 'user') { + const message = msg.message as Record | undefined; + const content = message?.content as Array> | undefined; + if (Array.isArray(content)) { + for (const block of content) { + if (block.type === 'tool_result') { + let output = ''; + if (typeof block.content === 'string') { + output = block.content; + } else if (Array.isArray(block.content)) { + output = (block.content as Array>) + .filter((c) => c.type === 'text') + .map((c) => c.text as string) + .join('\n'); + } + onEvent({ + type: 'tool:result', + toolCallId: (block.tool_use_id as string) ?? '', + output, + isError: (block.is_error as boolean) ?? false, + }); + } + } + } + } else if (type === 'system' && msg.subtype === 'init') { + const sessionId = msg.session_id as string | undefined; + if (sessionId) { + setClaudeSession(sessionKey, sessionId); + } + } else if (type === 'result') { + gotResult = true; + clearTimeout(timeout); + + const isError = (msg.is_error as boolean) ?? false; + const resultText = (msg.result as string) ?? ''; + + if (isError) { + if (textBuffer) { + onEvent({ type: 'text', text: textBuffer }); + textBuffer = ''; + } + onEvent({ type: 'error', message: resultText || 'Claude Code returned an error' }); + return; + } + + if (textBuffer) { + onEvent({ type: 'text', text: textBuffer }); + textBuffer = ''; + } + + const usage = msg.usage as Record | undefined; + const cost: MessageCost = { + inputTokens: usage?.input_tokens ?? 0, + outputTokens: usage?.output_tokens ?? 0, + totalUSD: (msg.total_cost_usd as number) ?? 0, + }; + + const sessionId = msg.session_id as string | undefined; + if (sessionId) { + setClaudeSession(sessionKey, sessionId); + } + + onEvent({ type: 'result', cost }); + } + } catch { + // Skip malformed JSON lines + } + }; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + const chunk = decoder.decode(value, { stream: true }); + buffer += chunk; + const lines = buffer.split('\n'); + buffer = lines.pop()!; + for (const line of lines) { + processLine(line); + } + } + + if (buffer.trim()) { + processLine(buffer); + } + + clearTimeout(timeout); + + if (!gotResult) { + const exitCode = await proc.exited; + const stderr = await new Response(proc.stderr as ReadableStream).text(); + if (textBuffer) { + onEvent({ type: 'text', text: textBuffer }); + } + if (exitCode !== 0) { + onEvent({ type: 'error', message: `Claude Code exited with code ${exitCode}: ${stderr.trim().slice(0, 200)}` }); + } else { + onEvent({ type: 'result', cost: { inputTokens: 0, outputTokens: 0, totalUSD: 0 } }); + } + } + } catch (err) { + clearTimeout(timeout); + onEvent({ type: 'error', message: String(err) }); + } finally { + activeProcs.delete(sessionKey); + } +} + +export function killClaudeSession(sessionKey: string): boolean { + const proc = activeProcs.get(sessionKey); + if (proc) { + try { proc.kill(); } catch { /* already dead */ } + activeProcs.delete(sessionKey); + return true; + } + return false; +} + +export function clearSession(sessionKey: string): void { + clearClaudeSession(sessionKey); +} + +export function getActiveSessionKeys(): string[] { + return Array.from(activeProcs.keys()); +} diff --git a/src/servers/sidecar/index.ts b/src/servers/sidecar/index.ts new file mode 100644 index 00000000..de0fc274 --- /dev/null +++ b/src/servers/sidecar/index.ts @@ -0,0 +1,259 @@ +import type { ServerWebSocket } from 'bun'; +import type { SidecarCommand, SidecarEvent, SidecarState } from './protocol'; +import { loadState, flushAndSave, acquireLock, releaseLock, getState } from './state'; +import { startAnthropicProxy, getProxySecret, ensureProxySecret } from './proxy'; +import * as claudeManager from './claude-manager'; +import * as piManager from './pi-manager'; +import * as queueRunner from './queue-runner'; + +const PORT = Number(process.env.SIDECAR_PORT ?? '5100'); +const startedAt = Date.now(); + +// ── Startup ── + +if (!acquireLock()) { + console.error('[sidecar] another instance is already running (lock file exists with live PID)'); + process.exit(1); +} + +loadState(); +ensureProxySecret(); + +// Start Anthropic proxy +try { + startAnthropicProxy(); +} catch (err) { + console.error('[sidecar] failed to start Anthropic proxy:', err instanceof Error ? err.message : err); +} + +// Initialize queue +queueRunner.initQueue().catch((err) => { + console.error('[sidecar] failed to initialize queue:', err); +}); + +// ── WebSocket connections ── + +const clients = new Set>(); + +function broadcast(msg: SidecarEvent) { + const data = JSON.stringify(msg); + for (const ws of clients) { + if (ws.readyState === 1) { + ws.send(data); + } + } +} + +function reply(ws: ServerWebSocket, msg: SidecarEvent) { + if (ws.readyState === 1) { + ws.send(JSON.stringify(msg)); + } +} + +function buildState(): SidecarState { + return { + proxySecret: getProxySecret(), + claudeSessions: { ...getState().claudeSessions }, + piSessions: piManager.getAllSessions(), + uptime: Date.now() - startedAt, + }; +} + +// ── Command handlers ── + +async function handleCommand(ws: ServerWebSocket, cmd: SidecarCommand) { + switch (cmd.type) { + case 'ping': + reply(ws, { type: 'pong', id: cmd.id }); + break; + + case 'state:sync': + reply(ws, { type: 'state:sync', id: cmd.id, state: buildState() }); + break; + + case 'proxy:secret': + reply(ws, { type: 'proxy:secret', id: cmd.id, secret: getProxySecret() }); + break; + + // ── Claude Code ── + + case 'claude:spawn': { + try { + const result = await claudeManager.spawnClaude(cmd.params); + reply(ws, { type: 'claude:result', id: cmd.id, result }); + } catch (err) { + reply(ws, { type: 'claude:error', id: cmd.id, error: err instanceof Error ? err.message : String(err) }); + } + break; + } + + case 'claude:spawn-streaming': { + reply(ws, { type: 'claude:spawned', id: cmd.id, sessionKey: cmd.params.sessionKey }); + + const onEvent = (event: import('../api/pi/types').PiEvent) => { + broadcast({ type: 'claude:event', sessionKey: cmd.params.sessionKey, event }); + }; + + claudeManager.spawnClaudeStreaming(cmd.params, onEvent).catch((err) => { + broadcast({ + type: 'claude:event', + sessionKey: cmd.params.sessionKey, + event: { type: 'error', message: err instanceof Error ? err.message : String(err) }, + }); + }); + break; + } + + case 'claude:kill': + claudeManager.killClaudeSession(cmd.sessionKey); + reply(ws, { type: 'claude:killed', id: cmd.id }); + break; + + case 'claude:clear-session': + claudeManager.clearSession(cmd.sessionKey); + reply(ws, { type: 'claude:session-cleared', id: cmd.id }); + break; + + // ── Pi ── + + case 'pi:spawn': { + try { + const onEvent = (event: import('../api/pi/types').PiEvent) => { + broadcast({ type: 'pi:event', sessionId: cmd.params.sessionId, event }); + }; + + await piManager.spawnPi({ + sessionId: cmd.params.sessionId, + email: cmd.params.email, + userId: cmd.params.userId, + username: cmd.params.username, + role: cmd.params.role, + cwd: cmd.params.cwd, + model: cmd.params.model, + sessionFile: cmd.params.sessionFile, + onEvent, + }); + + reply(ws, { type: 'pi:spawned', id: cmd.id, sessionId: cmd.params.sessionId }); + } catch (err) { + reply(ws, { type: 'pi:error', id: cmd.id, error: err instanceof Error ? err.message : String(err) }); + } + break; + } + + case 'pi:prompt': + piManager.sendPrompt(cmd.sessionId, cmd.prompt, cmd.requestId); + break; + + case 'pi:abort': + piManager.abort(cmd.sessionId, cmd.requestId); + break; + + case 'pi:kill': + piManager.killPiSession(cmd.sessionId); + reply(ws, { type: 'pi:killed', id: cmd.id }); + break; + + case 'pi:set-thinking': + piManager.setThinkingLevel(cmd.sessionId, cmd.level); + break; + + // ── Queue ── + + case 'queue:enqueue': { + try { + const job = await queueRunner.enqueue(cmd.params); + reply(ws, { type: 'queue:enqueued', id: cmd.id, job }); + } catch (err) { + reply(ws, { type: 'queue:error', id: cmd.id, error: err instanceof Error ? err.message : String(err) }); + } + break; + } + + case 'queue:cancel': { + try { + const job = await queueRunner.cancelJob(cmd.jobId); + reply(ws, { type: 'queue:cancelled', id: cmd.id, job }); + } catch (err) { + reply(ws, { type: 'queue:error', id: cmd.id, error: err instanceof Error ? err.message : String(err) }); + } + break; + } + + case 'queue:list': { + const jobs = await queueRunner.listAllJobs(); + reply(ws, { type: 'queue:list', id: cmd.id, jobs }); + break; + } + + case 'queue:get': { + const job = await queueRunner.readJob(cmd.jobId); + reply(ws, { type: 'queue:get', id: cmd.id, job }); + break; + } + + default: + reply(ws, { type: 'error', id: (cmd as SidecarCommand).id, error: `Unknown command type: ${(cmd as Record).type}` }); + } +} + +// ── Server ── + +const server = Bun.serve({ + port: PORT, + hostname: '127.0.0.1', + + routes: { + '/': () => new Response('process-sidecar'), + '/health': () => new Response(JSON.stringify({ + status: 'ok', + uptime: Date.now() - startedAt, + piSessions: piManager.getAllSessions().length, + claudeSessions: claudeManager.getActiveSessionKeys().length, + }), { headers: { 'Content-Type': 'application/json' } }), + }, + + fetch(req, server) { + if (req.headers.get('upgrade') === 'websocket') { + const ok = server.upgrade(req); + if (!ok) return new Response('Upgrade failed', { status: 500 }); + return; + } + return new Response('Not found', { status: 404 }); + }, + + websocket: { + open(ws) { + clients.add(ws); + console.log(`[sidecar] client connected (${clients.size} total)`); + }, + message(ws, raw) { + try { + const data = typeof raw === 'string' ? raw : raw.toString(); + const cmd = JSON.parse(data) as SidecarCommand; + handleCommand(ws, cmd); + } catch (err) { + reply(ws, { type: 'error', error: `Invalid message: ${err instanceof Error ? err.message : String(err)}` }); + } + }, + close(ws) { + clients.delete(ws); + console.log(`[sidecar] client disconnected (${clients.size} total)`); + }, + drain() {}, + }, +}); + +console.log(`[sidecar] listening on 127.0.0.1:${PORT}`); + +// ── Graceful shutdown ── + +async function shutdown(signal: string) { + console.log(`[sidecar] ${signal} received, saving state...`); + await flushAndSave(); + releaseLock(); + process.exit(0); +} + +process.on('SIGTERM', () => shutdown('SIGTERM')); +process.on('SIGINT', () => shutdown('SIGINT')); diff --git a/src/servers/sidecar/pi-manager.ts b/src/servers/sidecar/pi-manager.ts new file mode 100644 index 00000000..b01aa01a --- /dev/null +++ b/src/servers/sidecar/pi-manager.ts @@ -0,0 +1,416 @@ +import { join } from 'node:path'; +import { readdirSync, existsSync, mkdirSync, readFileSync } from 'node:fs'; +import type { Subprocess } from 'bun'; +import type { PiEvent, MessageCost } from '../api/pi/types'; +import type { PiSpawnParams, PiSessionInfo } from './protocol'; +import { isPidAlive } from './state'; + +const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data'); +const PI_CONFIG_DIR = join(require('node:os').homedir(), '.pi', 'agent'); +const SEED_PATH = join(import.meta.dir, '../../../seed'); + +const getHomeDir = (email: string) => join(DATA_PATH, email, 'home'); +const getHomeDirForRole = (email: string, role: string | null): string => + role === 'Super Admin' && process.env.HOME_DIR ? process.env.HOME_DIR : getHomeDir(email); +const getGlobalSkillsDir = () => join(DATA_PATH, 'skills'); +const getUserSkillsDir = (email: string) => join(DATA_PATH, email, 'skills'); +const getGlobalExtensionsDir = () => join(DATA_PATH, 'extensions'); +const getUserExtensionsDir = (email: string) => join(DATA_PATH, email, 'extensions'); +const getGlobalToolsDir = () => join(DATA_PATH, 'tools'); +const getUserToolsDir = (email: string) => join(DATA_PATH, email, 'tools'); +const getNativeResourcesDir = () => join(SEED_PATH, 'resources'); +const getGlobalResourcesDir = () => join(DATA_PATH, 'resources'); + +const toShellUsername = (username: string, email: string): string => { + const raw = username || email.split('@')[0]!; + return raw.replace(/@.*$/, '').replace(/[^a-zA-Z0-9._-]/g, '_').toLowerCase().slice(0, 32) || 'officer'; +}; + +// Resolve pi as [node, cli.js] +const PI_CMD = (() => { + const whichResult = Bun.spawnSync({ cmd: ['which', 'pi'], stdout: 'pipe', stderr: 'ignore' }); + const piBin = whichResult.stdout.toString().trim() || 'pi'; + const readlinkResult = Bun.spawnSync({ cmd: ['readlink', '-f', piBin], stdout: 'pipe', stderr: 'ignore' }); + const realPath = readlinkResult.stdout.toString().trim(); + const nodeResult = Bun.spawnSync({ cmd: ['which', 'node'], stdout: 'pipe', stderr: 'ignore' }); + const nodeBin = nodeResult.stdout.toString().trim() || 'node'; + if (realPath && realPath.endsWith('.js')) { + return [nodeBin, realPath]; + } + return [piBin]; +})(); + +// Active Pi processes +type PiSession = { + sessionId: string; + email: string; + userId: number; + model: string; + cwd: string; + proc: Subprocess; + onEvent: (event: PiEvent) => void; +}; + +const sessions = new Map(); + +// ── Helpers ── + +function collectSkillFlags(email: string): string[] { + const flags: string[] = []; + const dirs = [getGlobalSkillsDir(), getUserSkillsDir(email)]; + for (const dir of dirs) { + if (!existsSync(dir)) continue; + for (const entry of readdirSync(dir, { withFileTypes: true })) { + if (!entry.isDirectory()) continue; + if (existsSync(join(dir, entry.name, 'SKILL.md'))) { + flags.push('--skill', `${dir}/${entry.name}`); + } + } + } + return flags; +} + +function collectExtensionFlags(email: string): string[] { + const flags: string[] = []; + const dirs = [getGlobalExtensionsDir(), getUserExtensionsDir(email)]; + for (const dir of dirs) { + if (!existsSync(dir)) continue; + for (const entry of readdirSync(dir, { withFileTypes: true })) { + if (!entry.isDirectory()) continue; + if (existsSync(join(dir, entry.name, 'index.ts'))) { + flags.push('--extension', `${dir}/${entry.name}/index.ts`); + } + } + } + return flags; +} + +function buildResourcesEnv(): string { + const nativeDir = getNativeResourcesDir(); + const globalDir = getGlobalResourcesDir(); + const resourceDirs = new Map(); + for (const dir of [nativeDir, globalDir]) { + if (!existsSync(dir)) continue; + for (const entry of readdirSync(dir, { withFileTypes: true })) { + if (!entry.isDirectory()) continue; + if (existsSync(join(dir, entry.name, 'RESOURCE.md')) || existsSync(join(dir, entry.name, 'config.json'))) { + resourceDirs.set(entry.name, dir); + } + } + } + const result: Record> = {}; + for (const [name] of resourceDirs) { + let nativeConfig: Record = {}; + let globalConfig: Record = {}; + try { nativeConfig = JSON.parse(readFileSync(join(nativeDir, name, 'config.json'), 'utf-8')); } catch {} + try { globalConfig = JSON.parse(readFileSync(join(globalDir, name, 'config.json'), 'utf-8')); } catch {} + const config: Record = {}; + for (const key of Object.keys(nativeConfig)) config[key] = globalConfig[key] ?? nativeConfig[key]!; + for (const key of Object.keys(globalConfig)) if (!(key in config)) config[key] = globalConfig[key]!; + if (Object.values(config).some((v) => v !== '')) { + result[name] = config; + } + } + return JSON.stringify(result); +} + +async function resolveApiKeyForModel(model: string): Promise { + const provider = model.split('/')[0]; + if (!provider) return null; + try { + const authFile = Bun.file(join(PI_CONFIG_DIR, 'auth.json')); + if (!(await authFile.exists())) return null; + const auth = (await authFile.json()) as Record; + return auth[provider]?.key?.trim() || null; + } catch { + return null; + } +} + +// ── Event parsing (mirrors pi-bridge.ts) ── + +function parseErrorMessage(raw: string): string { + try { + const parsed = JSON.parse(raw.replace(/^\d+\s*/, '')); + const inner = parsed?.error; + if (inner?.message) return inner.message; + } catch { /* not JSON */ } + return raw; +} + +function extractMessageError(msg: Record): string | null { + if (msg.stopReason !== 'error') return null; + const raw = msg.errorMessage as string | undefined; + if (!raw) return null; + return parseErrorMessage(raw); +} + +function parsePiEvent(event: Record, currentStreamBuffer: string): PiEvent[] { + const type = event.type as string; + + if (type === 'response') { + if (event.command === 'prompt' && !event.success) { + return [{ type: 'error', message: (event.error as string) ?? 'Prompt failed' }]; + } + return []; + } + + switch (type) { + case 'agent_start': + return []; + + case 'message_update': { + const ame = event.assistantMessageEvent as Record | undefined; + if (ame?.type === 'text_delta') { + return [{ type: 'delta', text: ame.delta as string }]; + } + return []; + } + + case 'message_end': { + const events: PiEvent[] = []; + if (currentStreamBuffer) { + events.push({ type: 'text', text: currentStreamBuffer }); + } + const msg = event.message as Record | undefined; + if (msg) { + const errorText = extractMessageError(msg); + if (errorText) events.push({ type: 'error', message: errorText }); + } + return events; + } + + case 'tool_execution_start': + return [{ + type: 'tool:start', + toolCallId: (event.toolCallId as string) ?? '', + toolName: (event.toolName as string) ?? 'unknown', + toolInput: (event.args as Record) ?? {}, + }]; + + case 'tool_execution_end': { + const toolCallId = (event.toolCallId as string) ?? ''; + const result = event.result; + let resultObj: Record | null = null; + if (typeof result === 'object' && result !== null) { + resultObj = result as Record; + } else if (typeof result === 'string') { + try { resultObj = JSON.parse(result); } catch { /* not JSON */ } + } + const isError = (event.isError as boolean) ?? (resultObj?.isError as boolean) ?? false; + const output = result != null ? (typeof result === 'string' ? result : JSON.stringify(result)) : ''; + return [{ type: 'tool:result', toolCallId, output, isError }]; + } + + case 'agent_end': { + const cost: MessageCost = { inputTokens: 0, outputTokens: 0, totalUSD: 0 }; + const events: PiEvent[] = []; + const messages = event.messages as Array> | undefined; + if (messages) { + for (const msg of messages) { + const usage = msg.usage as Record | undefined; + if (usage) { + cost.inputTokens += (usage.input as number) ?? 0; + cost.outputTokens += (usage.output as number) ?? 0; + const usageCost = usage.cost as Record | undefined; + if (usageCost) cost.totalUSD += (usageCost.total as number) ?? 0; + } + const errorText = extractMessageError(msg); + if (errorText) events.push({ type: 'error', message: errorText }); + } + } + events.push({ type: 'result', cost }); + return events; + } + + default: + return []; + } +} + +function writeRpcCommand(proc: Subprocess, command: Record): void { + const stdin = proc.stdin; + if (!stdin || typeof stdin === 'number') return; + try { + const writer = stdin as { write(data: string): void; flush(): void }; + writer.write(JSON.stringify(command) + '\n'); + writer.flush(); + } catch (err) { + console.error('[sidecar:pi] writeRpcCommand error:', err); + } +} + +// ── Public API ── + +export type PiSpawnOptions = { + sessionId: string; + email: string; + userId: number; + username: string; + role: string; + cwd: string; + model: string; + sessionFile?: string; + onEvent: (event: PiEvent) => void; +}; + +export async function spawnPi(options: PiSpawnOptions): Promise { + const { sessionId, email, userId, username, role, cwd, model, sessionFile, onEvent } = options; + + const skillFlags = collectSkillFlags(email); + const extensionFlags = collectExtensionFlags(email); + + // Generate resource skill + const { generateResourceSkill } = await import('../api/pi/pi-bridge'); + const resourceSkillDir = generateResourceSkill(DATA_PATH); + const resourceSkillFlags = resourceSkillDir ? ['--skill', resourceSkillDir] : []; + + const piArgs = [ + ...PI_CMD, + '--mode', 'rpc', + '--no-skills', '--no-prompt-templates', '--no-themes', + ...skillFlags, + ...extensionFlags, + ...resourceSkillFlags, + ]; + if (model) piArgs.push('--model', model); + if (sessionFile) piArgs.push('--session', sessionFile); + + const apiKey = await resolveApiKeyForModel(model); + if (apiKey) piArgs.push('--api-key', apiKey); + + if (!existsSync(cwd)) { + mkdirSync(cwd, { recursive: true }); + } + + const homeDir = getHomeDirForRole(email, role); + const toolsDirs = [getGlobalToolsDir(), getUserToolsDir(email)].join(':'); + const shellUsername = username ? toShellUsername(username, email) : toShellUsername('', email); + const isServiceUser = shellUsername === (process.env.USER ?? ''); + + const env: Record = { + HOME: isServiceUser ? (process.env.HOME ?? '') : homeDir, + OFFICER_USER_HOME: homeDir, + OFFICER_USER_ROOT: join(DATA_PATH, email), + PI_CODING_AGENT_DIR: isServiceUser ? PI_CONFIG_DIR : join(homeDir, '.pi', 'agent'), + PI_TOOLS_DIRS: toolsDirs, + OFFICER_RESOURCES: buildResourcesEnv(), + OFFICER_EMAIL_DB: join(DATA_PATH, email, 'emails.db'), + TERM: 'xterm-256color', + PATH: process.env.PATH ?? '', + }; + + const proc = isServiceUser + ? Bun.spawn(piArgs, { cwd, stdin: 'pipe', stdout: 'pipe', stderr: 'pipe', env: { ...process.env, ...env } }) + : Bun.spawn( + ['sudo', '-u', shellUsername, 'env', ...Object.entries(env).map(([k, v]) => `${k}=${v}`), ...piArgs], + { cwd, stdin: 'pipe', stdout: 'pipe', stderr: 'pipe' }, + ); + + const session: PiSession = { sessionId, email, userId, model, cwd, proc, onEvent }; + sessions.set(sessionId, session); + + console.log(`[sidecar:pi] spawned Pi for session ${sessionId} (model=${model}, pid=${proc.pid})`); + + // Read stdout JSON event stream + const stdout = proc.stdout as ReadableStream; + const reader = stdout.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + let streamBuffer = ''; + + (async () => { + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split('\n'); + buffer = lines.pop() ?? ''; + for (const line of lines) { + if (!line.trim()) continue; + try { + const event = JSON.parse(line) as Record; + const piEvents = parsePiEvent(event, streamBuffer); + for (const piEvent of piEvents) { + if (piEvent.type === 'delta') { + streamBuffer += piEvent.text; + } else if (piEvent.type === 'text' || piEvent.type === 'tool:start') { + streamBuffer = ''; + } + onEvent(piEvent); + } + } catch { /* skip */ } + } + } + } catch { /* process ended */ } + })(); + + // Stderr → log + const stderr = proc.stderr as ReadableStream; + const stderrReader = stderr.getReader(); + const stderrDecoder = new TextDecoder(); + (async () => { + try { + while (true) { + const { done, value } = await stderrReader.read(); + if (done) break; + const text = stderrDecoder.decode(value, { stream: true }); + if (text.trim()) console.log(`[sidecar:pi:stderr] ${text.trim()}`); + } + } catch { /* process ended */ } + })(); + + // Handle exit + proc.exited.then((code) => { + sessions.delete(sessionId); + if (code !== 0) { + console.error(`[sidecar:pi] Pi process ${sessionId} exited with code ${code}`); + } + }); +} + +export function sendPrompt(sessionId: string, prompt: string, requestId: string): boolean { + const session = sessions.get(sessionId); + if (!session) return false; + writeRpcCommand(session.proc, { type: 'prompt', id: requestId, message: prompt }); + return true; +} + +export function abort(sessionId: string, requestId: string): boolean { + const session = sessions.get(sessionId); + if (!session) return false; + writeRpcCommand(session.proc, { type: 'abort', id: requestId }); + return true; +} + +export function setThinkingLevel(sessionId: string, level: string): boolean { + const session = sessions.get(sessionId); + if (!session) return false; + writeRpcCommand(session.proc, { type: 'set_thinking_level', level }); + return true; +} + +export function killPiSession(sessionId: string): boolean { + const session = sessions.get(sessionId); + if (!session) return false; + try { session.proc.kill(); } catch { /* already dead */ } + sessions.delete(sessionId); + return true; +} + +export function getSession(sessionId: string): PiSession | undefined { + return sessions.get(sessionId); +} + +export function getAllSessions(): PiSessionInfo[] { + return Array.from(sessions.values()).map((s) => ({ + sessionId: s.sessionId, + email: s.email, + userId: s.userId, + model: s.model, + cwd: s.cwd, + pid: s.proc.pid, + alive: isPidAlive(s.proc.pid), + })); +} diff --git a/src/servers/sidecar/protocol.ts b/src/servers/sidecar/protocol.ts new file mode 100644 index 00000000..72756cba --- /dev/null +++ b/src/servers/sidecar/protocol.ts @@ -0,0 +1,115 @@ +import type { MessageCost, PiEvent } from '../api/pi/types'; +import type { Job, EnqueueParams, JobProgress } from '../queue/types'; + +// ── Envelope ── + +export type SidecarMessage = SidecarCommand | SidecarEvent; + +// ── Commands (API server → sidecar) ── + +export type SidecarCommand = + | { type: 'ping'; id: string } + | { type: 'state:sync'; id: string } + // Proxy + | { type: 'proxy:secret'; id: string } + // Claude Code + | { type: 'claude:spawn'; id: string; params: ClaudeSpawnParams } + | { type: 'claude:spawn-streaming'; id: string; params: ClaudeSpawnStreamingParams } + | { type: 'claude:kill'; id: string; sessionKey: string } + | { type: 'claude:clear-session'; id: string; sessionKey: string } + // Pi + | { type: 'pi:spawn'; id: string; params: PiSpawnParams } + | { type: 'pi:prompt'; id: string; sessionId: string; prompt: string; requestId: string } + | { type: 'pi:abort'; id: string; sessionId: string; requestId: string } + | { type: 'pi:kill'; id: string; sessionId: string } + | { type: 'pi:set-thinking'; id: string; sessionId: string; level: string } + // Queue + | { type: 'queue:enqueue'; id: string; params: EnqueueParams } + | { type: 'queue:cancel'; id: string; jobId: string } + | { type: 'queue:list'; id: string } + | { type: 'queue:get'; id: string; jobId: string }; + +// ── Responses/Events (sidecar → API server) ── + +export type SidecarEvent = + | { type: 'pong'; id: string } + | { type: 'state:sync'; id: string; state: SidecarState } + | { type: 'proxy:secret'; id: string; secret: string } + // Claude Code + | { type: 'claude:spawned'; id: string; sessionKey: string } + | { type: 'claude:event'; sessionKey: string; event: PiEvent } + | { type: 'claude:result'; id: string; result: ClaudeCodeResult } + | { type: 'claude:error'; id: string; error: string } + | { type: 'claude:killed'; id: string } + | { type: 'claude:session-cleared'; id: string } + // Pi + | { type: 'pi:spawned'; id: string; sessionId: string } + | { type: 'pi:event'; sessionId: string; event: PiEvent } + | { type: 'pi:error'; id: string; error: string } + | { type: 'pi:killed'; id: string } + // Queue + | { type: 'queue:enqueued'; id: string; job: Job } + | { type: 'queue:cancelled'; id: string; job: Job | null } + | { type: 'queue:list'; id: string; jobs: Job[] } + | { type: 'queue:get'; id: string; job: Job | null } + | { type: 'queue:error'; id: string; error: string } + // Generic + | { type: 'error'; id?: string; error: string }; + +// ── Shared state snapshot ── + +export type SidecarState = { + proxySecret: string; + claudeSessions: Record; // sessionKey → Claude Code session_id + piSessions: PiSessionInfo[]; + uptime: number; +}; + +export type PiSessionInfo = { + sessionId: string; + email: string; + userId: number; + model: string; + cwd: string; + pid: number; + alive: boolean; +}; + +// ── Param types ── + +export type ClaudeSpawnParams = { + userId: number; + email: string; + username: string; + prompt: string; + sessionKey: string; + model?: string; +}; + +export type ClaudeSpawnStreamingParams = { + userId: number; + email: string; + username: string; + prompt: string; + sessionKey: string; + cwd?: string; + model?: string; +}; + +export type ClaudeCodeResult = { + text: string; + sessionId: string; + model: string; + cost: MessageCost; +}; + +export type PiSpawnParams = { + sessionId: string; + email: string; + userId: number; + username: string; + role: string; + cwd: string; + model: string; + sessionFile?: string; +}; diff --git a/src/servers/sidecar/proxy.ts b/src/servers/sidecar/proxy.ts new file mode 100644 index 00000000..8b037d23 --- /dev/null +++ b/src/servers/sidecar/proxy.ts @@ -0,0 +1,104 @@ +import { join } from 'node:path'; +import { homedir } from 'node:os'; +import { getState, updateState } from './state'; + +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'); + +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 getProxySecret(): string { + return getState().proxySecret; +} + +export function ensureProxySecret(): string { + const state = getState(); + if (state.proxySecret) return state.proxySecret; + + // Generate once, persist forever + const secret = `sk-ant-api03-${crypto.randomUUID()}`; + updateState({ proxySecret: secret }); + return secret; +} + +export function startAnthropicProxy() { + const secret = ensureProxySecret(); + + 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 !== secret) { + 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 + 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(); + 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(`[sidecar:proxy] listening on 127.0.0.1:${PROXY_PORT}`); +} diff --git a/src/servers/sidecar/queue-runner.ts b/src/servers/sidecar/queue-runner.ts new file mode 100644 index 00000000..c0b34755 --- /dev/null +++ b/src/servers/sidecar/queue-runner.ts @@ -0,0 +1,266 @@ +import { type Job, type JobProgress, type EnqueueParams, type StepContext, PermanentError } from '../queue/types'; +import { readJob, writeJob, listAllJobs, ensureQueueDir } from '../queue/storage'; +import { getHandler } from '../queue/handler-registry'; + +// Import handlers to register them +import '../queue/handlers'; + +const activeLanes = new Map(); +const PROGRESS_THROTTLE_MS = 1000; + +export async function initQueue() { + await ensureQueueDir(); + await resumeInterruptedJobs(); + console.log('[sidecar:queue] initialized'); +} + +export async function enqueue(params: EnqueueParams): Promise { + const handler = getHandler(params.type); + if (!handler) throw new Error(`No handler registered for job type: ${params.type}`); + + const job: Job = { + id: crypto.randomUUID(), + lane: params.lane, + type: params.type, + userId: params.userId, + status: 'queued', + steps: handler.steps.map((s) => ({ name: s.name, status: 'pending' as const })), + currentStep: 0, + createdAt: Date.now(), + meta: params.meta, + notify: params.notify, + }; + + await writeJob(job); + console.log(`[sidecar:queue] enqueued job ${job.id} (${job.type}) in lane ${job.lane}`); + kickLane(job.lane); + return job; +} + +export async function cancelJob(id: string): Promise { + const job = await readJob(id); + if (!job) return null; + if (job.status === 'completed' || job.status === 'failed' || job.status === 'cancelled') return job; + + job.status = 'cancelled'; + job.completedAt = Date.now(); + for (const step of job.steps) { + if (step.status === 'pending' || step.status === 'running') { + step.status = 'failed'; + step.error = 'Cancelled'; + } + } + await writeJob(job); + console.log(`[sidecar:queue] cancelled job ${job.id}`); + return job; +} + +async function resumeInterruptedJobs() { + const jobs = await listAllJobs(); + const lanesToKick = new Set(); + + for (const job of jobs) { + if (job.status === 'running') { + job.status = 'queued'; + job.startedAt = undefined; + for (const step of job.steps) { + if (step.status === 'running') { + step.status = 'pending'; + step.startedAt = undefined; + } + } + await writeJob(job); + console.log(`[sidecar:queue] reset interrupted job ${job.id} back to queued`); + lanesToKick.add(job.lane); + } else if (job.status === 'queued') { + lanesToKick.add(job.lane); + } + } + + for (const lane of lanesToKick) { + kickLane(lane); + } +} + +function kickLane(lane: string) { + if (activeLanes.get(lane)) return; + activeLanes.set(lane, true); + processNextInLane(lane); +} + +function scheduleRetry(lane: string, delayMs: number) { + setTimeout(() => kickLane(lane), delayMs); +} + +async function processNextInLane(lane: string) { + try { + const jobs = await listAllJobs(); + const now = Date.now(); + const next = jobs + .filter((j) => j.lane === lane && j.status === 'queued' && (!j.retryAt || j.retryAt <= now)) + .sort((a, b) => a.createdAt - b.createdAt)[0]; + + if (!next) { + activeLanes.set(lane, false); + return; + } + + await runJob(next); + } catch (err) { + console.error(`[sidecar:queue] lane ${lane} processing error:`, err); + } finally { + const jobs = await listAllJobs(); + const hasMore = jobs.some( + (j) => j.lane === lane && j.status === 'queued' && (!j.retryAt || j.retryAt <= Date.now()), + ); + if (hasMore) { + processNextInLane(lane); + } else { + activeLanes.set(lane, false); + } + } +} + +async function runJob(job: Job) { + const handler = getHandler(job.type); + if (!handler) { + job.status = 'failed'; + job.error = `No handler for type: ${job.type}`; + job.completedAt = Date.now(); + await writeJob(job); + return; + } + + job.status = 'running'; + job.startedAt = Date.now(); + job.retryAt = undefined; + await writeJob(job); + const isRetry = (job.retries ?? 0) > 0; + console.log( + `[sidecar:queue] ${isRetry ? 'resuming' : 'running'} job ${job.id} (${job.type})${isRetry ? ` retry ${job.retries}` : ''}`, + ); + + const sharedMeta: Record = { ...(job.meta ?? {}) }; + + for (let i = 0; i < handler.steps.length; i++) { + const fresh = await readJob(job.id); + if (!fresh || fresh.status === 'cancelled') { + console.log(`[sidecar:queue] job ${job.id} was cancelled, stopping`); + return; + } + + const handlerStep = handler.steps[i]!; + const step = fresh.steps[i]!; + + if (step.status === 'completed') continue; + + fresh.currentStep = i; + step.status = 'running'; + step.startedAt = Date.now(); + await writeJob(fresh); + + let lastProgressWrite = 0; + let pendingProgress: JobProgress | null = null; + + const updateProgress = async (progress: JobProgress) => { + step.progress = progress; + const now = Date.now(); + if (now - lastProgressWrite >= PROGRESS_THROTTLE_MS) { + lastProgressWrite = now; + pendingProgress = null; + await writeJob(fresh); + } else { + pendingProgress = progress; + } + }; + + const ctx: StepContext = { job: fresh, step, updateProgress, meta: sharedMeta }; + + try { + await handlerStep.run(ctx); + + if (pendingProgress) { + step.progress = pendingProgress; + } + step.status = 'completed'; + step.completedAt = Date.now(); + await writeJob(fresh); + } catch (err) { + const errorMessage = err instanceof Error ? err.message : String(err); + step.status = 'failed'; + step.error = errorMessage; + step.completedAt = Date.now(); + + const isPermanent = err instanceof PermanentError; + const retries = (fresh.retries ?? 0) + 1; + if (!isPermanent && handler.retry && retries <= handler.retry.maxRetries) { + step.status = 'pending'; + step.error = undefined; + step.startedAt = undefined; + step.completedAt = undefined; + step.progress = undefined; + fresh.status = 'queued'; + fresh.error = undefined; + fresh.completedAt = undefined; + fresh.startedAt = undefined; + fresh.retries = retries; + fresh.retryAt = Date.now() + handler.retry.delayMs; + await writeJob(fresh); + console.log( + `[sidecar:queue] job ${fresh.id} will retry (${retries}/${handler.retry.maxRetries}) in ${handler.retry.delayMs / 1000}s`, + ); + scheduleRetry(fresh.lane, handler.retry.delayMs); + return; + } + + fresh.status = 'failed'; + fresh.error = `Step "${step.name}" failed: ${errorMessage}`; + fresh.completedAt = Date.now(); + fresh.meta = { ...fresh.meta, ...sharedMeta }; + await writeJob(fresh); + console.error(`[sidecar:queue] job ${fresh.id} failed at step "${step.name}":`, errorMessage); + await notifyFailure(fresh); + return; + } + } + + const final = await readJob(job.id); + if (final && final.status === 'running') { + final.status = 'completed'; + final.completedAt = Date.now(); + final.meta = { ...final.meta, ...sharedMeta }; + await writeJob(final); + console.log(`[sidecar:queue] job ${final.id} completed`); + await notifyCompletion(final); + } +} + +async function notifyCompletion(job: Job) { + try { + const { sendMail } = await import('emailer'); + await sendMail({ + template: 'JobCompleted', + subject: `Job completed: ${job.type}`, + to: job.userId, + data: { job }, + }); + } catch { + // SMTP might not be configured — non-fatal + } +} + +async function notifyFailure(job: Job) { + try { + const { sendMail } = await import('emailer'); + await sendMail({ + template: 'JobFailed', + subject: `Job failed: ${job.type}`, + to: job.userId, + data: { job }, + }); + } catch { + // SMTP might not be configured — non-fatal + } +} + +export { readJob, listAllJobs }; diff --git a/src/servers/sidecar/state.ts b/src/servers/sidecar/state.ts new file mode 100644 index 00000000..c8e2d712 --- /dev/null +++ b/src/servers/sidecar/state.ts @@ -0,0 +1,141 @@ +import { join } from 'node:path'; +import { mkdirSync, existsSync } from 'node:fs'; + +const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data'); +const STATE_DIR = join(DATA_PATH, 'sidecar'); +const STATE_FILE = join(STATE_DIR, 'state.json'); +const LOCK_FILE = join(STATE_DIR, 'sidecar.lock'); + +export type PersistedState = { + proxySecret: string; + claudeSessions: Record; // sessionKey → Claude Code session_id + piSessions: Array<{ + sessionId: string; + email: string; + userId: number; + model: string; + cwd: string; + pid: number; + }>; +}; + +const DEFAULT_STATE: PersistedState = { + proxySecret: '', + claudeSessions: {}, + piSessions: [], +}; + +let currentState: PersistedState = { ...DEFAULT_STATE }; +let saveTimer: Timer | null = null; + +function ensureDir() { + if (!existsSync(STATE_DIR)) { + mkdirSync(STATE_DIR, { recursive: true }); + } +} + +export function loadState(): PersistedState { + ensureDir(); + try { + const raw = Bun.file(STATE_FILE); + // Synchronous check — Bun.file doesn't have sync exists, use fs + if (!existsSync(STATE_FILE)) { + currentState = { ...DEFAULT_STATE }; + return currentState; + } + // We need to read synchronously at startup + const text = require('node:fs').readFileSync(STATE_FILE, 'utf-8'); + currentState = { ...DEFAULT_STATE, ...JSON.parse(text) }; + return currentState; + } catch { + currentState = { ...DEFAULT_STATE }; + return currentState; + } +} + +export async function saveState(): Promise { + ensureDir(); + await Bun.write(STATE_FILE, JSON.stringify(currentState, null, 2)); +} + +export function getState(): PersistedState { + return currentState; +} + +export function updateState(patch: Partial): void { + Object.assign(currentState, patch); + scheduleSave(); +} + +export function setClaudeSession(sessionKey: string, sessionId: string): void { + currentState.claudeSessions[sessionKey] = sessionId; + scheduleSave(); +} + +export function clearClaudeSession(sessionKey: string): void { + delete currentState.claudeSessions[sessionKey]; + scheduleSave(); +} + +export function getClaudeSession(sessionKey: string): string | undefined { + return currentState.claudeSessions[sessionKey]; +} + +function scheduleSave() { + if (saveTimer) return; + saveTimer = setTimeout(async () => { + saveTimer = null; + await saveState(); + }, 30_000); +} + +export async function flushAndSave(): Promise { + if (saveTimer) { + clearTimeout(saveTimer); + saveTimer = null; + } + await saveState(); +} + +// ── Lockfile ── + +export function acquireLock(): boolean { + ensureDir(); + try { + if (existsSync(LOCK_FILE)) { + const pidStr = require('node:fs').readFileSync(LOCK_FILE, 'utf-8').trim(); + const pid = Number(pidStr); + if (pid && isProcessAlive(pid)) { + return false; // another sidecar is running + } + // Stale lock — remove it + } + require('node:fs').writeFileSync(LOCK_FILE, String(process.pid)); + return true; + } catch { + return false; + } +} + +export function releaseLock(): void { + try { + if (existsSync(LOCK_FILE)) { + require('node:fs').unlinkSync(LOCK_FILE); + } + } catch { + // best effort + } +} + +function isProcessAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + +export function isPidAlive(pid: number): boolean { + return isProcessAlive(pid); +}