import { spawn } from 'child_process'; import fsSync from 'node:fs'; import crossSpawn from 'cross-spawn'; import Database from 'better-sqlite3'; import { sessionsService } from './modules/providers/services/sessions.service.js'; import { providerAuthService } from './modules/providers/services/provider-auth.service.js'; import { providerModelsService } from './modules/providers/services/provider-models.service.js'; import { notifyRunFailed, notifyRunStopped } from './services/notification-orchestrator.js'; import { createNormalizedMessage, getOpenCodeDatabasePath } from './shared/utils.js'; const spawnFunction = process.platform === 'win32' ? crossSpawn : spawn; const activeOpenCodeProcesses = new Map(); function readOpenCodeSessionId(event) { if (!event || typeof event !== 'object') { return null; } return event.sessionID || event.sessionId || null; } function readOpenCodeTokenUsage(sessionId) { const dbPath = getOpenCodeDatabasePath(); if (!sessionId || !fsSync.existsSync(dbPath)) { return null; } let db = null; try { db = new Database(dbPath, { readonly: true, fileMustExist: true }); const columns = db.prepare('PRAGMA table_info(session)').all(); const columnNames = new Set(columns.map((column) => column.name)); const requiredColumns = ['tokens_input', 'tokens_output', 'tokens_reasoning', 'tokens_cache_read', 'tokens_cache_write']; if (!requiredColumns.every((column) => columnNames.has(column))) { return null; } const row = db.prepare(` SELECT tokens_input AS inputTokens, tokens_output AS outputTokens, tokens_reasoning AS reasoningTokens, tokens_cache_read AS cacheReadTokens, tokens_cache_write AS cacheWriteTokens FROM session WHERE id = ? `).get(sessionId); if (!row) { return null; } const inputTokens = Number(row.inputTokens || 0) + Number(row.cacheReadTokens || 0); const outputTokens = Number(row.outputTokens || 0); const used = Number(row.inputTokens || 0) + outputTokens + Number(row.reasoningTokens || 0) + Number(row.cacheReadTokens || 0) + Number(row.cacheWriteTokens || 0); if (used <= 0) { return null; } return { used, inputTokens, outputTokens, breakdown: { input: inputTokens, output: outputTokens, }, }; } catch { return null; } finally { if (db) { db.close(); } } } async function spawnOpenCode(command, options = {}, ws) { return new Promise((resolve, reject) => { const { sessionId, projectPath, cwd, model, sessionSummary } = options; const workingDir = cwd || projectPath || process.cwd(); const processKey = sessionId || Date.now().toString(); let capturedSessionId = sessionId || null; let sessionCreatedSent = false; let stdoutLineBuffer = ''; let terminalNotificationSent = false; let opencodeProcess = null; const notifyTerminalState = ({ code = null, error = null } = {}) => { if (terminalNotificationSent) { return; } terminalNotificationSent = true; const finalSessionId = capturedSessionId || sessionId || processKey; if (code === 0 && !error) { notifyRunStopped({ userId: ws?.userId || null, provider: 'opencode', sessionId: finalSessionId, sessionName: sessionSummary, stopReason: 'completed', }); return; } notifyRunFailed({ userId: ws?.userId || null, provider: 'opencode', sessionId: finalSessionId, sessionName: sessionSummary, error: error || `OpenCode CLI exited with code ${code}`, }); }; const registerSession = (nextSessionId) => { if (!nextSessionId || capturedSessionId === nextSessionId) { return; } capturedSessionId = nextSessionId; if (processKey !== capturedSessionId && opencodeProcess) { activeOpenCodeProcesses.delete(processKey); activeOpenCodeProcesses.set(capturedSessionId, opencodeProcess); } if (opencodeProcess) { opencodeProcess.sessionId = capturedSessionId; } if (ws.setSessionId && typeof ws.setSessionId === 'function') { ws.setSessionId(capturedSessionId); } if (!sessionId && !sessionCreatedSent) { sessionCreatedSent = true; ws.send(createNormalizedMessage({ kind: 'session_created', newSessionId: capturedSessionId, sessionId: capturedSessionId, provider: 'opencode', })); } }; const processOpenCodeOutputLine = (line) => { if (!line || !line.trim()) { return; } let response; try { response = JSON.parse(line); } catch { ws.send(createNormalizedMessage({ kind: 'stream_delta', content: line, sessionId: capturedSessionId || sessionId || null, provider: 'opencode', })); return; } try { registerSession(readOpenCodeSessionId(response)); const normalized = sessionsService.normalizeMessage( 'opencode', response, capturedSessionId || sessionId || null, ); for (const msg of normalized) { ws.send(msg); } } catch (error) { const errorContent = error instanceof Error ? error.message : String(error); console.error('[OpenCode] Failed to process JSON output:', errorContent); ws.send(createNormalizedMessage({ kind: 'error', content: errorContent, sessionId: capturedSessionId || sessionId || null, provider: 'opencode', })); } }; void providerModelsService.resolveResumeModel('opencode', sessionId, model).then((resolvedModel) => { const args = ['run', '--format', 'json']; if (sessionId) { args.push('--session', sessionId); } if (resolvedModel) { args.push('--model', resolvedModel); } if (command && command.trim()) { args.push(command.trim()); } opencodeProcess = spawnFunction('opencode', args, { cwd: workingDir, stdio: ['pipe', 'pipe', 'pipe'], env: { ...process.env }, }); activeOpenCodeProcesses.set(processKey, opencodeProcess); opencodeProcess.sessionId = processKey; opencodeProcess.stdin.end(); opencodeProcess.stdout.on('data', (data) => { stdoutLineBuffer += data.toString(); const completeLines = stdoutLineBuffer.split(/\r?\n/); stdoutLineBuffer = completeLines.pop() || ''; completeLines.forEach((line) => { processOpenCodeOutputLine(line.trim()); }); }); opencodeProcess.stderr.on('data', (data) => { const stderrText = data.toString(); if (!stderrText.trim()) { return; } ws.send(createNormalizedMessage({ kind: 'error', content: stderrText, sessionId: capturedSessionId || sessionId || null, provider: 'opencode', })); }); opencodeProcess.on('close', async (code) => { const finalSessionId = capturedSessionId || sessionId || processKey; activeOpenCodeProcesses.delete(finalSessionId); activeOpenCodeProcesses.delete(processKey); if (stdoutLineBuffer.trim()) { processOpenCodeOutputLine(stdoutLineBuffer.trim()); stdoutLineBuffer = ''; } const tokenBudget = readOpenCodeTokenUsage(finalSessionId); if (tokenBudget) { ws.send(createNormalizedMessage({ kind: 'status', text: 'token_budget', tokenBudget, sessionId: finalSessionId, provider: 'opencode', })); } ws.send(createNormalizedMessage({ kind: 'complete', exitCode: code, isNewSession: !sessionId && !!command, sessionId: finalSessionId, provider: 'opencode', })); if (code === 0) { notifyTerminalState({ code }); resolve(); return; } if (code === 127 || code === null) { const installed = await providerAuthService.isProviderInstalled('opencode'); if (!installed) { ws.send(createNormalizedMessage({ kind: 'error', content: 'OpenCode CLI is not installed. Install it from https://opencode.ai/docs/', sessionId: finalSessionId, provider: 'opencode', })); } } notifyTerminalState({ code }); reject(new Error(code === null ? 'OpenCode CLI process was terminated' : `OpenCode CLI exited with code ${code}`)); }); opencodeProcess.on('error', async (error) => { const finalSessionId = capturedSessionId || sessionId || processKey; activeOpenCodeProcesses.delete(finalSessionId); activeOpenCodeProcesses.delete(processKey); const installed = await providerAuthService.isProviderInstalled('opencode'); const errorContent = !installed ? 'OpenCode CLI is not installed. Install it from https://opencode.ai/docs/' : error.message; ws.send(createNormalizedMessage({ kind: 'error', content: errorContent, sessionId: finalSessionId, provider: 'opencode', })); notifyTerminalState({ error }); reject(error); }); }).catch(reject); }); } function abortOpenCodeSession(sessionId) { const process = activeOpenCodeProcesses.get(sessionId); if (!process) { return false; } process.kill('SIGTERM'); activeOpenCodeProcesses.delete(sessionId); return true; } function isOpenCodeSessionActive(sessionId) { return activeOpenCodeProcesses.has(sessionId); } function getActiveOpenCodeSessions() { return Array.from(activeOpenCodeProcesses.keys()); } export { spawnOpenCode, abortOpenCodeSession, isOpenCodeSessionActive, getActiveOpenCodeSessions, };