import { existsSync } from 'node:fs'; import { homedir } from 'node:os'; import { join } from 'node:path'; import { query, type Query } from '@anthropic-ai/claude-agent-sdk'; import type { ChatEvent, PromptImage } from '../../api/chat/types'; import type { ClaudeSpawnParams, ClaudeSpawnStreamingParams, ClaudeCodeResult } from '../protocol'; import { setClaudeSession, clearClaudeSession, getClaudeSession } from './state'; import { createParseState, processMessage } from './stream-parser'; const SEND_TIMEOUT_MS = 30 * 60 * 1000; // Where the Claude Code CLI lives. This was hardcoded to /usr/local/bin/claude, which dated from the // bwrap-sandboxed architecture: the jail ro-bound /usr and saw nothing else, so the installer's real // target (~/.local/bin/claude) had to be symlinked into a path the sandbox could reach. That sandbox // is gone, and the hardcoded path made the sidecar unrunnable anywhere it does not exist — a stock // macOS host has no /usr/local/bin at all. // // Resolution order mirrors OPENCODE_BIN in the opencode sidecar: an explicit pin, then PATH, then the // locations Anthropic's installer actually writes to. function resolveClaudeBin(): string { const pinned = process.env.CLAUDE_BIN; if (pinned) return pinned; const onPath = Bun.which('claude'); if (onPath) return onPath; const candidates = [join(homedir(), '.local', 'bin', 'claude'), '/usr/local/bin/claude', '/opt/homebrew/bin/claude']; return candidates.find((candidate) => existsSync(candidate)) ?? 'claude'; } const CLAUDE_BIN = resolveClaudeBin(); console.log(`[claude] CLI resolved to ${CLAUDE_BIN}`); // Capture original HOME before user-instance overrides it const HOST_HOME = process.env.HOME!; const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data'); // Tear a persistent session down after this long with no new turn (see PersistentSession below). const IDLE_TIMEOUT_MS = 30 * 60 * 1000; // MCP config paths, set by user-instance at startup let mcpHostPath: string | undefined; // path on the host filesystem export function setMcpConfigPath(hostPath: string): void { mcpHostPath = hostPath; } // ── Blocking send (one-shot; used by the non-streaming 'claude:spawn' command) ── 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 { prompt, sessionKey } = params; const existingSession = getClaudeSession(sessionKey); const claudeArgs = [CLAUDE_BIN, '-p', prompt, '--dangerously-skip-permissions', '--output-format', 'json']; const mcpConfig = mcpHostPath; if (mcpConfig) claudeArgs.push('--mcp-config', mcpConfig); const subModel = params.model?.split('/')[1]; if (subModel) claudeArgs.push('--model', subModel); if (existingSession) { claudeArgs.push('--resume', existingSession); } const spawnCmd = claudeArgs; const spawnCwd = params.cwd ?? HOST_HOME; const proc = Bun.spawn(spawnCmd, { stdin: 'pipe', stdout: 'pipe', stderr: 'pipe', cwd: spawnCwd, env: process.env as Record, }); 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; } } // ── Persistent streaming sessions (Agent SDK) ── // // Each sessionKey gets ONE long-lived `query()` driven by a streaming-input queue. The process stays // alive BETWEEN turns, so background work (Bash run_in_background, Monitor) and its `task_notification` // return to a live harness instead of being orphaned when the turn ends. A turn = one user message // pushed onto the input queue; the single consumer loop maps every SDK message (assistant text/tools, // tool results, turn `result`, and — crucially — post-turn `task_started`/`task_notification`) to a // ChatEvent and forwards it. Idle policy: torn down after IDLE_TIMEOUT_MS with no new turn, or on an // explicit kill (user "disconnect"). "Stop" is interrupt() — it ends the turn but keeps the session. /** * `content` is an Anthropic `MessageParam` content: a bare string, or the block array we build when the * turn carries images. Blocks are what make an attached screenshot actually reach the model — the string * form can only carry the client's `[Attached image: …]` placeholder, which is a label, not the picture. */ type ContentBlock = | { type: 'text'; text: string } | { type: 'image'; source: { type: 'base64'; media_type: string; data: string } }; type SdkUserMessage = { type: 'user'; message: { role: 'user'; content: string | ContentBlock[] }; parent_tool_use_id: null; session_id: string; }; type PersistentSession = { sessionKey: string; query: Query; pushTurn: (prompt: string, images?: PromptImage[]) => void; closeInput: () => void; abort: AbortController; emit: (event: ChatEvent) => void; isGenerating: boolean; pendingTasks: Set; // background tasks started but not yet notified; suppress idle-GC while non-empty /** * The user pressed stop and we are waiting for the turn to fall over. The SDK reports an interrupt as * an ordinary failed `result` — `is_error` with no text — which is indistinguishable downstream from * the harness actually breaking, and reached the user as "Claude Code returned an error". Only the * side that called `interrupt()` knows better, so it says so here. */ interrupted: boolean; idleTimer?: ReturnType; }; const sessions = new Map(); /** A hand-rolled async iterable we can push turns onto and close on teardown. */ function makeInputQueue() { const buf: SdkUserMessage[] = []; let wake: (() => void) | null = null; let closed = false; async function* gen(): AsyncGenerator { while (true) { if (buf.length) { yield buf.shift()!; continue; } if (closed) return; await new Promise((r) => { wake = r; }); } } return { gen: gen(), push(m: SdkUserMessage) { buf.push(m); wake?.(); wake = null; }, close() { closed = true; wake?.(); wake = null; }, }; } function armIdle(session: PersistentSession): void { if (session.idleTimer) clearTimeout(session.idleTimer); session.idleTimer = setTimeout(() => { // Heartbeat: never GC a session that's mid-turn or still has background tasks running — a long // silent run_in_background job would otherwise be killed along with its pending task_notification. if (session.isGenerating || session.pendingTasks.size > 0) { armIdle(session); return; } killClaudeSession(session.sessionKey); }, IDLE_TIMEOUT_MS); } function createSession(params: ClaudeSpawnStreamingParams, onEvent: (event: ChatEvent) => void): PersistentSession { const { sessionKey } = params; const input = makeInputQueue(); const abort = new AbortController(); const session: PersistentSession = { sessionKey, query: undefined as unknown as Query, pushTurn: () => {}, closeInput: () => input.close(), abort, emit: onEvent, isGenerating: false, pendingTasks: new Set(), interrupted: false, }; // Strip the nested-session guard vars so the SDK can spawn `claude` (mirrors the old spawn env clean). const { CLAUDECODE: _c, CLAUDE_CODE_ENTRYPOINT: _e, CLAUDE_CODE_SSE_PORT: _s, ...cleanEnv } = process.env; const resumeId = getClaudeSession(sessionKey) ?? params.resumeSessionId; const subModel = params.model?.split('/')[1]; const q = query({ prompt: input.gen as AsyncIterable, options: { cwd: params.cwd ?? HOST_HOME, permissionMode: 'bypassPermissions', allowDangerouslySkipPermissions: true, includePartialMessages: true, abortController: abort, pathToClaudeCodeExecutable: CLAUDE_BIN, settingSources: ['user', 'project', 'local'], env: cleanEnv as Record, stderr: (d: string) => { if (d.trim()) console.error(`[claude:stream:${sessionKey}] ${d.slice(0, 300)}`); }, ...(subModel ? { model: subModel } : {}), ...(resumeId ? { resume: resumeId } : {}), ...(mcpHostPath ? { extraArgs: { 'mcp-config': mcpHostPath } } : {}), }, }); session.query = q; session.pushTurn = (prompt: string, images?: PromptImage[]) => { if (session.idleTimer) clearTimeout(session.idleTimer); session.isGenerating = true; // Images first, then the text: the model reads what it is looking at before what to do about it. const content: string | ContentBlock[] = images?.length ? [ ...images.map( (img): ContentBlock => ({ type: 'image', source: { type: 'base64', media_type: img.mediaType, data: img.data }, }), ), { type: 'text', text: prompt }, ] : prompt; input.push({ type: 'user', message: { role: 'user', content }, parent_tool_use_id: null, session_id: sessionKey, }); }; sessions.set(sessionKey, session); // Single consumer loop for the session's whole life. Turn-end (`result`) and errors flip isGenerating // and (re)arm the idle timer; the session process stays alive so later task_notifications still flow. void (async () => { const state = createParseState(); const emit = (raw: ChatEvent) => { // A turn we interrupted ends in a failed `result`. That is the stop landing, not a fault. const event: ChatEvent = raw.type === 'error' && session.interrupted ? { type: 'stopped' } : raw; if (event.type === 'task:started') { // Work is running — hold off idle-GC until it finishes. session.pendingTasks.add(event.taskId); if (session.idleTimer) { clearTimeout(session.idleTimer); session.idleTimer = undefined; } } else if (event.type === 'task:notification') { session.pendingTasks.delete(event.taskId); if (!session.isGenerating && session.pendingTasks.size === 0) armIdle(session); } else if (event.type === 'result' || event.type === 'error' || event.type === 'stopped') { session.isGenerating = false; // Whatever ended the turn, the interrupt is spent — a later genuine error must not wear it. session.interrupted = false; if (session.pendingTasks.size === 0) armIdle(session); } session.emit(event); }; try { for await (const msg of q as AsyncGenerator>) { processMessage(msg, state, { onEvent: emit, onSessionId: (id: string) => setClaudeSession(sessionKey, id), }); } } catch (err) { if (!abort.signal.aborted) { session.emit({ type: 'error', message: err instanceof Error ? err.message : String(err) }); } } finally { if (session.idleTimer) clearTimeout(session.idleTimer); sessions.delete(sessionKey); } })(); return session; } /** Start a turn: create the persistent session if needed, then push the prompt as a user message. */ export async function spawnClaudeStreaming( params: ClaudeSpawnStreamingParams, onEvent: (event: ChatEvent) => void, ): Promise { let session = sessions.get(params.sessionKey); if (session) { session.emit = onEvent; // adopt the latest emitter (equivalent across turns; keeps events flowing) } else { session = createSession(params, onEvent); } session.pushTurn(params.prompt, params.images); } /** Interrupt the current turn but KEEP the session alive (the "stop" button). */ export async function interruptClaudeSession(sessionKey: string): Promise { const session = sessions.get(sessionKey); if (!session) return false; // Set before the await: the failed `result` can arrive while interrupt() is still resolving, and the // consumer loop reads this flag to tell a stop from a fault. session.interrupted = true; try { await session.query.interrupt(); session.isGenerating = false; return true; } catch { session.interrupted = false; return false; } } /** Fully tear the session down (the "disconnect" action / idle GC): abort the query + close input. */ export function killClaudeSession(sessionKey: string): boolean { const session = sessions.get(sessionKey); if (!session) return false; if (session.idleTimer) clearTimeout(session.idleTimer); try { session.abort.abort(); } catch { /* already aborted */ } try { session.closeInput(); } catch { /* noop */ } sessions.delete(sessionKey); return true; } export function clearSession(sessionKey: string): void { clearClaudeSession(sessionKey); } export function getActiveSessionKeys(): string[] { return Array.from(sessions.keys()); } /** * Is a turn actually in flight for this session, right now, in this process? * * This is the ground truth a reconnecting browser has no way to work out for itself. Officer's own view * dies with `pm2 restart officer` while the turn keeps running here, so "officer doesn't remember" means * nothing — and if *this* process was the one that restarted, the session is simply absent and the turn * it was running is gone, however alive the client still believes it to be. */ export function isSessionGenerating(sessionKey: string): boolean { return sessions.get(sessionKey)?.isGenerating ?? false; }