import type { ServerWebSocket } from 'bun'; import { randomUUID } from 'crypto'; import type { ClientMessage, ServerMessage, Message, ChatEvent } from './types'; import { sessionManager } from './session-manager'; import { sendClaudeCodeStreaming } from '@@/channels/send-claude-code'; import { sendOpenCodeStreaming } from '@@/channels/send-opencode'; import { ensureGeneralChatSessionsCwd } from './claude-sessions'; import * as sidecar from '@@/sidecar-registry'; import { join } from 'path'; import { getOwnerHomeDir, getEmailAccountsDir } from '../../../servers/data-path'; import { getUserSettings, getEmailAccounts } from 'officerdb'; import { mkdirSync } from 'node:fs'; import { logger } from './logger'; // Default model when no user preference is set const DEFAULT_MODEL = 'claude-code'; // Harness selection: the `claude-code` provider runs through the Claude sidecar; every other provider // (opencode/anthropic/openai/… — all `providerID/modelID` ids) runs through the OpenCode server. const isClaudeModel = (model: string): boolean => model.startsWith('claude-code'); async function getUserDefaultModel(userId: number): Promise { try { const settings = await getUserSettings(userId); const chat = settings?.chat as Record | undefined; return (chat?.defaultModel as string) || null; } catch (err) { logger.error('Failed to read user settings for default model', { userId, error: String(err) }); } return null; } type WSData = { userId: number; email: string; username: string; provider: string; }; const IDLE_TIMEOUT_MS = 60 * 60 * 1000; // 1 hour const resolveCwd = (email: string, cwd?: string) => { const root = getOwnerHomeDir(email); if (!cwd || cwd === '~') return root; if (cwd.startsWith('~/')) return join(root, cwd.slice(2)); // The server owner is the only account — absolute paths are theirs to use. if (cwd.startsWith('/')) return cwd; return join(root, cwd); }; export const resolveBaseCwd = (email: string, cwd?: string) => resolveCwd(email, cwd); // The email chat runs from the selected account's storage dir: // DATA_PATH//email_accounts/ // `accountEmail` will come from the account selector (msg.contextId) later; for now default to the // owner's first enabled account. Falls back to the email_accounts root if there are no accounts. async function resolveEmailCwd(userId: number, ownerEmail: string, accountEmail?: string): Promise { let account = accountEmail?.trim(); if (!account) { try { const accounts = await getEmailAccounts(userId); account = (accounts.find((a) => a.enabled) ?? accounts[0])?.email; } catch (err) { logger.error('Failed to resolve email account for chat cwd', { userId, error: String(err) }); } } const dir = account ? join(getEmailAccountsDir(ownerEmail), account) : getEmailAccountsDir(ownerEmail); mkdirSync(dir, { recursive: true }); return dir; } // The working directory a chat turn runs in, by context: email → the account dir; /chat → a chosen // pwd or the default general_chat_sessions dir; everything else (browser/project/dashboard) → the given cwd. async function resolveChatCwd( msg: { context?: string; contextId?: string; cwd?: string }, email: string, userId: number, ): Promise { if (msg.context === 'email') return resolveEmailCwd(userId, email, msg.contextId); if (msg.context === 'chat') return msg.cwd?.trim() ? resolveCwd(email, msg.cwd) : ensureGeneralChatSessionsCwd(email); return resolveCwd(email, msg.cwd); } const wsToSessionMap = new WeakMap(); // Per-connection heartbeat. Bun closes a WS idle for `idleTimeout` (60s), and its timer only resets // on frames *received* from the client — but during a chat turn the client only receives. So we ping // each connection every 25s; the client auto-pongs at the protocol level, which resets Bun's timer // (and keeps reverse proxies happy). Genuinely dead sockets still time out (no pong). const pingTimers = new WeakMap, ReturnType>(); const PING_INTERVAL_MS = 25_000; function sendToClient(ws: ServerWebSocket | null, msg: ServerMessage): void { if (ws?.readyState === 1) { ws.send(JSON.stringify(msg)); } } export async function open(ws: ServerWebSocket): Promise { const timer = setInterval(() => { try { ws.ping(); // client auto-pongs → resets Bun's idleTimeout } catch { /* socket already gone */ } }, PING_INTERVAL_MS); pingTimers.set(ws, timer); } export function message(ws: ServerWebSocket, raw: string | Buffer): void { const data = typeof raw === 'string' ? raw : raw.toString(); (async () => { try { const clientMsg = JSON.parse(data) as ClientMessage; if (clientMsg.type === 'chat') { await handleChat(ws, clientMsg); } else if (clientMsg.type === 'resume') { await handleResume(ws, clientMsg); } else if (clientMsg.type === 'stop') { await handleStop(ws); } else if (clientMsg.type === 'disconnect') { await handleDisconnect(ws); } } catch (err) { logger.error('Error handling WebSocket message', { email: ws.data.email, error: String(err) }); sendToClient(ws, { type: 'error', message: 'Failed to process message' }); } })(); } export function close(ws: ServerWebSocket): void { const timer = pingTimers.get(ws); if (timer) { clearInterval(timer); pingTimers.delete(ws); } const sessionId = wsToSessionMap.get(ws); if (sessionId) { sessionManager.detachWs(sessionId); sessionManager.setIdleTimeout(sessionId, IDLE_TIMEOUT_MS); } } function createEventHandler(sessionId: string, model: string, cwd: string) { return async (event: ChatEvent): Promise => { const session = sessionManager.getSession(sessionId); if (!session) return; const ws = session.ws as ServerWebSocket | null; switch (event.type) { case 'delta': { sendToClient(ws, { type: 'assistant:delta', text: event.text }); session.streamBuffer += event.text; break; } case 'text': { // Flush streaming buffer as complete text const text = event.text || session.streamBuffer; if (text) { sendToClient(ws, { type: 'assistant:text', text }); const assistantMsg: Message = { id: randomUUID(), timestamp: Date.now(), role: 'assistant', text, model, }; session.messages.push(assistantMsg); session.meta.messageCount += 1; session.streamBuffer = ''; } break; } case 'tool:start': { // Flush any pending streaming text first if (session.streamBuffer) { sendToClient(ws, { type: 'assistant:text', text: session.streamBuffer }); const assistantMsg: Message = { id: randomUUID(), timestamp: Date.now(), role: 'assistant', text: session.streamBuffer, model, }; session.messages.push(assistantMsg); session.meta.messageCount += 1; session.streamBuffer = ''; } sendToClient(ws, { type: 'tool:start', toolCallId: event.toolCallId, toolName: event.toolName, toolInput: event.toolInput, }); const toolMsg: Message = { id: randomUUID(), timestamp: Date.now(), role: 'tool', toolCallId: event.toolCallId, toolName: event.toolName, toolInput: event.toolInput, }; session.messages.push(toolMsg); session.meta.messageCount += 1; break; } case 'tool:result': { sendToClient(ws, { type: 'tool:result', toolCallId: event.toolCallId, output: event.output, isError: event.isError, }); // Update existing tool message with output for (let i = session.messages.length - 1; i >= 0; i--) { const m = session.messages[i]!; if (m.role === 'tool' && m.toolCallId === event.toolCallId) { m.output = event.output; m.isError = event.isError; break; } } break; } case 'result': { // Flush any remaining streaming buffer if (session.streamBuffer) { sendToClient(ws, { type: 'assistant:text', text: session.streamBuffer }); const assistantMsg: Message = { id: randomUUID(), timestamp: Date.now(), role: 'assistant', text: session.streamBuffer, model, cost: event.cost, }; session.messages.push(assistantMsg); session.meta.messageCount += 1; session.streamBuffer = ''; } sendToClient(ws, { type: 'result', sessionId, cost: event.cost }); session.isGenerating = false; session.meta.cost.inputTokens += event.cost.inputTokens; session.meta.cost.outputTokens += event.cost.outputTokens; session.meta.cost.totalUSD += event.cost.totalUSD; session.meta.updatedAt = Date.now(); // No disk persistence — Claude's transcript is the record. break; } case 'error': { sendToClient(ws, { type: 'error', message: event.message }); session.isGenerating = false; break; } case 'stopped': { sendToClient(ws, { type: 'stopped' }); session.isGenerating = false; break; } } }; } async function handleChat( ws: ServerWebSocket, msg: { prompt: string; displayText?: string; sessionId?: string; model?: string; cwd?: string; cwdRoot?: string; groupSlug?: string; attachmentIds?: string[]; thinking?: string; context?: string; contextId?: string; resumeSummary?: string; resumeSessionId?: string; }, ): Promise { const { userId } = ws.data; const sessionId = msg.sessionId || randomUUID(); // Prepend resume summary to the prompt if present const prompt = msg.resumeSummary ? `Here is a summary of a previous conversation to continue from:\n\n${msg.resumeSummary}\n\n---\n\nUser's new message: ${msg.prompt}` : msg.prompt; // Use provided model, or fall back to user default, or the system default. const model = msg.model || (await getUserDefaultModel(userId)) || DEFAULT_MODEL; logger.info('Model selected for chat', { sessionId, model, clientModel: msg.model || null }); // Route by harness: claude-code → Claude sidecar; anything else → OpenCode server. return isClaudeModel(model) ? handleClaudeCodeChat(ws, sessionId, model, msg, prompt) : handleOpenCodeChat(ws, sessionId, model, msg, prompt); } async function handleClaudeCodeChat( ws: ServerWebSocket, sessionId: string, model: string, msg: { prompt: string; displayText?: string; groupSlug?: string; context?: string; contextId?: string; cwd?: string; cwdRoot?: string; resumeSessionId?: string; }, effectivePrompt: string, ): Promise { const { email, username, userId } = ws.data; const cwd = await resolveChatCwd(msg, email, userId); const groupSlug = msg.groupSlug || null; const session = sessionManager.getOrCreate(sessionId, email, cwd, model, groupSlug, msg.context, msg.contextId); session.userId = userId; sessionManager.attachWs(sessionId, ws); wsToSessionMap.set(ws as any, sessionId); sendToClient(ws, { type: 'session:init', sessionId, model, cwd, context: session.meta.context, contextId: session.meta.contextId, }); // Add user message to session const userMsg: Message = { id: randomUUID(), timestamp: Date.now(), role: 'user', text: msg.prompt, }; session.messages.push(userMsg); session.meta.messageCount += 1; session.meta.updatedAt = Date.now(); if (!session.meta.title) { session.meta.title = (msg.displayText ?? msg.prompt).slice(0, 100); } session.isGenerating = true; const onEvent = createEventHandler(sessionId, model, cwd); try { const handle = await sendClaudeCodeStreaming({ userId, email, username, prompt: effectivePrompt, sessionKey: sessionId, cwd, model, resumeSessionId: msg.resumeSessionId, onEvent, }); // 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' }); session.isGenerating = false; } } async function handleOpenCodeChat( ws: ServerWebSocket, sessionId: string, model: string, msg: { prompt: string; displayText?: string; groupSlug?: string; context?: string; contextId?: string; cwd?: string; cwdRoot?: string; resumeSessionId?: string; }, effectivePrompt: string, ): Promise { const { email, username, userId } = ws.data; const cwd = await resolveChatCwd(msg, email, userId); const groupSlug = msg.groupSlug || null; const session = sessionManager.getOrCreate(sessionId, email, cwd, model, groupSlug, msg.context, msg.contextId); session.userId = userId; sessionManager.attachWs(sessionId, ws); wsToSessionMap.set(ws as any, sessionId); sendToClient(ws, { type: 'session:init', sessionId, model, cwd, context: session.meta.context, contextId: session.meta.contextId, }); const userMsg: Message = { id: randomUUID(), timestamp: Date.now(), role: 'user', text: msg.prompt, }; session.messages.push(userMsg); session.meta.messageCount += 1; session.meta.updatedAt = Date.now(); if (!session.meta.title) { session.meta.title = (msg.displayText ?? msg.prompt).slice(0, 100); } session.isGenerating = true; const onEvent = createEventHandler(sessionId, model, cwd); try { const handle = await sendOpenCodeStreaming({ userId, email, username, prompt: effectivePrompt, sessionKey: sessionId, cwd, model, resumeSessionId: msg.resumeSessionId, onEvent, }); // Store the abort handle so handleStop can end the turn (OpenCode is aborted via this handle). session.piProcess = sessionId as any; session._claudeKill = handle.kill; } catch (err) { logger.error('Failed to start OpenCode streaming', { sessionId, error: String(err) }); sendToClient(ws, { type: 'error', message: 'Failed to start OpenCode' }); session.isGenerating = false; } } async function handleResume( ws: ServerWebSocket, msg: { sessionId: string; cwd?: string; cwdRoot?: string }, ): Promise { const { sessionId } = msg; try { const session = sessionManager.getSession(sessionId); if (!session) { // Sessions live in memory for the connection's lifetime; there's no disk store to reload from. sendToClient(ws, { type: 'error', message: 'Session not found', errorCode: 'SESSION_NOT_FOUND' }); return; } sessionManager.attachWs(sessionId, ws); wsToSessionMap.set(ws as any, sessionId); sendToClient(ws, { type: 'session:init', sessionId, model: session.model, cwd: session.cwd, context: session.meta.context, contextId: session.meta.contextId, }); // Claude resumes lazily: the next chat prompt re-attaches via `--resume `, so there's // no long-lived process to spawn here — just replay the stored transcript to the client. sendToClient(ws, { type: 'sync:messages', sessionId, messages: session.messages, isGenerating: session.isGenerating, streamingText: session.streamBuffer, }); logger.info('Session resumed successfully', { sessionId, messageCount: session.messages.length }); } catch (err) { logger.error('Unexpected error in handleResume', { sessionId, error: String(err) }); sendToClient(ws, { type: 'error', message: 'Failed to resume session' }); } } async function handleStop(ws: ServerWebSocket): Promise { const sessionId = wsToSessionMap.get(ws); if (sessionId) { const session = sessionManager.getSession(sessionId); if (session?.piProcess) { try { if (isClaudeModel(session.model)) { sidecar.killClaude(sessionId, session.email); logger.info('Killed Claude Code process via sidecar', { sessionId }); } else { session._claudeKill?.(); // OpenCode: abort the turn via the stored handle logger.info('Aborted OpenCode turn', { sessionId }); } session.isGenerating = false; } catch (err) { logger.error('Failed to stop process', { sessionId, error: String(err) }); } } } sendToClient(ws, { type: 'stopped' }); } // Tear down the whole session (not just the current turn): deleteSession fires _claudeKill (kills any // in-flight Claude/OpenCode turn) + _sidecarUnsub, clears the idle timer, and drops the session from the // manager's maps. The WS stays open so the client can immediately start a fresh session. async function handleDisconnect(ws: ServerWebSocket): Promise { const sessionId = wsToSessionMap.get(ws); if (sessionId) { try { sessionManager.deleteSession(sessionId); wsToSessionMap.delete(ws as any); logger.info('Disconnected chat session', { sessionId }); } catch (err) { logger.error('Failed to disconnect session', { sessionId, error: String(err) }); } } sendToClient(ws, { type: 'disconnected' }); } export const chatWebsocket = { open, message, close, drain() {}, };