import { join } from 'node:path'; import type { Subprocess } from 'bun'; import type { PiEvent } from '../../api/pi/types'; import type { ClaudeSpawnParams, ClaudeSpawnStreamingParams, ClaudeCodeResult } from '../protocol'; import { buildSandboxPrefix, buildRunuserSuffix, SANDBOX_HOME } from '../sandbox'; import { setClaudeSession, clearClaudeSession, getClaudeSession } from './state'; import { parseStream } from './stream-parser'; const SEND_TIMEOUT_MS = 5 * 60 * 1000; // 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'; })(); const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data'); // Build full bwrap sandbox args for Claude (prefix + Anthropic env + runuser suffix) function buildSandboxArgs(email: string): string[] { const prefix = buildSandboxPrefix(email); // Claude-specific env vars if (process.env.ANTHROPIC_BASE_URL) prefix.push('--setenv', 'ANTHROPIC_BASE_URL', process.env.ANTHROPIC_BASE_URL); if (process.env.ANTHROPIC_API_KEY) prefix.push('--setenv', 'ANTHROPIC_API_KEY', process.env.ANTHROPIC_API_KEY); return [...prefix, ...buildRunuserSuffix()]; } // Active streaming processes const activeProcs = new Map(); // MCP config path, set by user-instance at startup let mcpConfigPath: string | undefined; export function setMcpConfigPath(path: string): void { mcpConfigPath = path; } // ── 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 { prompt, sessionKey, email } = params; const existingSession = getClaudeSession(sessionKey); const claudeArgs = [ CLAUDE_BIN, '-p', prompt, '--dangerously-skip-permissions', '--output-format', 'json', ]; if (mcpConfigPath) claudeArgs.push('--mcp-config', mcpConfigPath); const subModel = params.model?.split('/')[1]; if (subModel) claudeArgs.push('--model', subModel); if (existingSession) { claudeArgs.push('--resume', existingSession); } const isSuperAdmin = params.role === 'Super Admin'; const spawnCmd = isSuperAdmin ? claudeArgs : [...buildSandboxArgs(email), ...claudeArgs]; const spawnCwd = isSuperAdmin ? (process.env.HOME ?? join(DATA_PATH, email, 'home')) : undefined; 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; } } // ── Streaming send ── export async function spawnClaudeStreaming( params: ClaudeSpawnStreamingParams, onEvent: (event: PiEvent) => void, ): Promise { const { prompt, sessionKey, email } = params; const existingSession = getClaudeSession(sessionKey); const claudeArgs = [ CLAUDE_BIN, '-p', prompt, '--dangerously-skip-permissions', '--output-format', 'stream-json', '--verbose', '--include-partial-messages', ]; if (mcpConfigPath) claudeArgs.push('--mcp-config', mcpConfigPath); const subModel = params.model?.split('/')[1]; if (subModel) claudeArgs.push('--model', subModel); if (existingSession) { claudeArgs.push('--resume', existingSession); } const { CLAUDECODE: _, ...cleanEnv } = process.env; const isSuperAdmin = params.role === 'Super Admin'; const spawnCmd = isSuperAdmin ? claudeArgs : [...buildSandboxArgs(email), ...claudeArgs]; const spawnCwd = isSuperAdmin ? (process.env.HOME ?? join(DATA_PATH, email, 'home')) : undefined; const proc = Bun.spawn(spawnCmd, { stdin: 'ignore', stdout: 'pipe', stderr: 'pipe', cwd: spawnCwd, env: cleanEnv as Record, }); 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 callbacks = { onEvent, onSessionId: (sessionId: string) => setClaudeSession(sessionKey, sessionId), }; const state = await parseStream(stdout, callbacks); clearTimeout(timeout); if (!state.gotResult) { const exitCode = await proc.exited; const stderr = await new Response(proc.stderr as ReadableStream).text(); if (state.textBuffer) { onEvent({ type: 'text', text: state.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()); }