From ad32c7516e8545cfe51ff709ac6c2a42adb34127 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Padez?= Date: Fri, 24 Jul 2026 16:29:55 +0000 Subject: [PATCH] =?UTF-8?q?chat:=20add=20OpenCode=20as=20a=20second=20harn?= =?UTF-8?q?ess=20=E2=80=94=20live=20turn=20(Phase=201)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduces an OpenCode chat harness alongside Claude, driven over HTTP + SSE against a persistent `opencode serve`, emitting the same ChatEvent contract so the entire chat UI and createEventHandler pipeline are unchanged. New servers/api/chat/opencode/: - server-manager.ts — one warm `opencode serve` per cwd (free port, health-gated, respawn on exit; HOME set so it reads the user's ~/.local/share/opencode auth). Binary pinned via OPENCODE_BIN (installed is 1.17.9; the 1.18.4 upgrade never landed). - client.ts — per-server HTTP calls (/session create, /message, /abort) + a single reconnecting `/event` SSE stream demuxed to per-session listeners. - event-mapper.ts — SSE → ChatEvent. Verified live against 1.17.9: message.part.delta → delta, tool parts → tool:start/tool:result, message.updated → cost, session.idle → result. Crucially, deltas are gated on partID being a `text` part (declared before its deltas) so the model's reasoning — which also streams as field:'text' — is dropped, matching the Claude harness hiding thinking. - state.ts — sessionKey ↔ opencode ses_ id map for resume. channels/send-opencode.ts — the OpenCode analog of send-claude-code: ensure serve, create/reuse session, subscribe, post the message, forward mapped events; kill = abort. websocket.ts — replaces the Claude-only coercion with harness routing: provider 'claude-code' → Claude sidecar, everything else → handleOpenCodeChat. handleStop aborts the right harness. Verified end-to-end (streaming text, tool call/result, cost, abort) against a throwaway serve using the free deepseek model — no prod restart involved. UI-level model selection + session history follow in Phases 2–3. Co-Authored-By: Claude Opus 4.8 --- src/servers/api/chat/opencode/client.ts | 137 +++++++++++++++++ src/servers/api/chat/opencode/event-mapper.ts | 140 ++++++++++++++++++ .../api/chat/opencode/server-manager.ts | 118 +++++++++++++++ src/servers/api/chat/opencode/state.ts | 15 ++ src/servers/api/chat/websocket.ts | 113 ++++++++++++-- src/servers/channels/send-opencode.ts | 75 ++++++++++ 6 files changed, 587 insertions(+), 11 deletions(-) create mode 100644 src/servers/api/chat/opencode/client.ts create mode 100644 src/servers/api/chat/opencode/event-mapper.ts create mode 100644 src/servers/api/chat/opencode/server-manager.ts create mode 100644 src/servers/api/chat/opencode/state.ts create mode 100644 src/servers/channels/send-opencode.ts diff --git a/src/servers/api/chat/opencode/client.ts b/src/servers/api/chat/opencode/client.ts new file mode 100644 index 00000000..4100242f --- /dev/null +++ b/src/servers/api/chat/opencode/client.ts @@ -0,0 +1,137 @@ +import type { OpenCodeEvent } from './event-mapper'; +import { logger } from '../logger'; + +// One connection per `opencode serve` base URL: a single shared SSE subscription on `GET /event` +// demultiplexed to per-session listeners, plus the REST calls a turn needs. The SDK-style +// `/session/*` route family is used (feature-complete, incl. DELETE). + +type Listener = (event: OpenCodeEvent) => void; + +class ServerConnection { + private listeners = new Map>(); + private sseRunning = false; + + constructor(private readonly baseUrl: string) {} + + subscribe(sessionId: string, listener: Listener): () => void { + let set = this.listeners.get(sessionId); + if (!set) { + set = new Set(); + this.listeners.set(sessionId, set); + } + set.add(listener); + void this.ensureSse(); + + return () => { + const current = this.listeners.get(sessionId); + if (!current) return; + current.delete(listener); + if (current.size === 0) this.listeners.delete(sessionId); + }; + } + + private dispatch(event: OpenCodeEvent): void { + const sessionId = event.properties?.sessionID; + if (typeof sessionId !== 'string') return; + const set = this.listeners.get(sessionId); + if (!set) return; + for (const listener of set) { + try { + listener(event); + } catch (err) { + logger.error('opencode SSE listener threw', { error: String(err) }); + } + } + } + + /** Keep a single `/event` SSE stream open while any session has listeners; reconnect on drop. */ + private async ensureSse(): Promise { + if (this.sseRunning) return; + this.sseRunning = true; + + void (async () => { + while (this.listeners.size > 0) { + try { + const res = await fetch(`${this.baseUrl}/event`, { headers: { accept: 'text/event-stream' } }); + if (!res.body) throw new Error('no SSE body'); + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + + while (true) { + const { value, done } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + + // SSE frames are separated by a blank line. + let sep = buffer.indexOf('\n\n'); + while (sep !== -1) { + const frame = buffer.slice(0, sep); + buffer = buffer.slice(sep + 2); + this.handleFrame(frame); + sep = buffer.indexOf('\n\n'); + } + } + } catch (err) { + logger.warn('opencode SSE stream error; will reconnect', { baseUrl: this.baseUrl, error: String(err) }); + } + if (this.listeners.size > 0) await new Promise((r) => setTimeout(r, 500)); + } + this.sseRunning = false; + })(); + } + + private handleFrame(frame: string): void { + const data = frame + .split('\n') + .filter((line) => line.startsWith('data:')) + .map((line) => line.slice(5).trim()) + .join(''); + if (!data) return; + try { + this.dispatch(JSON.parse(data) as OpenCodeEvent); + } catch { + /* non-JSON keep-alive or partial frame */ + } + } + + private async postJson(path: string, body: unknown): Promise { + const res = await fetch(`${this.baseUrl}${path}`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(body), + }); + if (!res.ok) throw new Error(`opencode POST ${path} → ${res.status} ${await res.text().catch(() => '')}`); + return (await res.json()) as T; + } + + async createSession(directory: string, title?: string): Promise { + const session = await this.postJson<{ id?: string }>('/session', { directory, title }); + if (!session.id) throw new Error('opencode POST /session returned no id'); + return session.id; + } + + async postMessage(sessionId: string, providerID: string, modelID: string, text: string): Promise { + await this.postJson(`/session/${sessionId}/message`, { + model: { providerID, modelID }, + parts: [{ type: 'text', text }], + }); + } + + async abort(sessionId: string): Promise { + await this.postJson(`/session/${sessionId}/abort`, {}).catch(() => {}); + } +} + +const connections = new Map(); + +export function getConnection(baseUrl: string): ServerConnection { + let conn = connections.get(baseUrl); + if (!conn) { + conn = new ServerConnection(baseUrl); + connections.set(baseUrl, conn); + } + return conn; +} + +export type { ServerConnection }; diff --git a/src/servers/api/chat/opencode/event-mapper.ts b/src/servers/api/chat/opencode/event-mapper.ts new file mode 100644 index 00000000..2c069b22 --- /dev/null +++ b/src/servers/api/chat/opencode/event-mapper.ts @@ -0,0 +1,140 @@ +import type { ChatEvent, MessageCost } from '../types'; + +// Translates OpenCode's SSE events into our internal ChatEvent union — the OpenCode analog of +// sidecar/claude/stream-parser.ts. Shapes verified live against opencode 1.17.9's `GET /event`: +// +// message.part.updated { part:{ id, type } } → learn partID→type +// message.part.delta { partID, delta } → delta (only if partID is a text part) +// message.part.updated { part:{ type:'tool', callID, tool, state } } → tool:start / tool:result +// message.updated { info:{ role:'assistant', cost, tokens } } → stash final cost +// session.idle → result (turn complete) +// session.error → error +// +// IMPORTANT: deltas always carry `field:'text'` even for the model's *reasoning* — reasoning and answer +// are only distinguishable by the delta's part TYPE (a message.part.updated declaring the part as +// `reasoning` vs `text` always precedes that part's deltas). So we gate deltas on partID being a `text` +// part; reasoning-part deltas are dropped (parity with the Claude harness, which hides thinking). +// createEventHandler flushes the assistant text buffer on tool:start and result, so no explicit `text` +// event is needed — the streamed answer deltas are enough. + +export type OpenCodeEvent = { + id?: string; + type: string; + properties?: Record; +}; + +type ToolState = { + status?: string; + input?: Record; + output?: string; + error?: string; +}; + +type OpenCodePart = { + id?: string; + type?: string; + callID?: string; + tool?: string; + field?: string; + state?: ToolState; +}; + +type AssistantInfo = { + role?: string; + cost?: number; + tokens?: { input?: number; output?: number }; +}; + +/** + * Returns a stateful handler that maps raw OpenCode events to ChatEvents, invoking `onEvent` for each. + * State tracks per-callID tool progress (to emit start/result exactly once) and the latest assistant + * cost/tokens (emitted with the terminal `result`). + */ +export function createEventMapper(onEvent: (event: ChatEvent) => void) { + const toolStarted = new Set(); + const toolFinished = new Set(); + const partTypes = new Map(); // partID → type (declared before that part's deltas) + let cost: MessageCost = { inputTokens: 0, outputTokens: 0, totalUSD: 0 }; + let done = false; + + const finish = (event: ChatEvent) => { + if (done) return; + done = true; + onEvent(event); + }; + + return function handle(evt: OpenCodeEvent): void { + const p = evt.properties ?? {}; + + switch (evt.type) { + case 'message.part.delta': { + // Only stream deltas belonging to a `text` part — reasoning parts also emit field:'text' deltas. + const partID = p.partID as string | undefined; + if (partID && partTypes.get(partID) === 'text' && typeof p.delta === 'string' && p.delta.length > 0) { + onEvent({ type: 'delta', text: p.delta }); + } + return; + } + + case 'message.part.updated': { + const part = p.part as OpenCodePart | undefined; + if (!part) return; + if (part.id && part.type) partTypes.set(part.id, part.type); + if (part.type !== 'tool' || !part.callID) return; + const callID = part.callID; + const status = part.state?.status; + + if ((status === 'running' || status === 'completed' || status === 'error') && !toolStarted.has(callID)) { + toolStarted.add(callID); + onEvent({ + type: 'tool:start', + toolCallId: callID, + toolName: part.tool ?? 'tool', + toolInput: part.state?.input ?? {}, + }); + } + + if (status === 'completed' && !toolFinished.has(callID)) { + toolFinished.add(callID); + onEvent({ type: 'tool:result', toolCallId: callID, output: String(part.state?.output ?? ''), isError: false }); + } else if (status === 'error' && !toolFinished.has(callID)) { + toolFinished.add(callID); + onEvent({ + type: 'tool:result', + toolCallId: callID, + output: String(part.state?.error ?? 'Tool failed'), + isError: true, + }); + } + return; + } + + case 'message.updated': { + const info = ((p.info as AssistantInfo | undefined) ?? (p as AssistantInfo)) ?? {}; + if (info.role === 'assistant' && info.tokens) { + cost = { + inputTokens: info.tokens.input ?? 0, + outputTokens: info.tokens.output ?? 0, + totalUSD: info.cost ?? 0, + }; + } + return; + } + + case 'session.idle': { + finish({ type: 'result', cost }); + return; + } + + case 'session.error': { + const error = p.error; + const message = typeof error === 'string' ? error : ((error as { message?: string })?.message ?? 'OpenCode error'); + finish({ type: 'error', message }); + return; + } + + default: + return; + } + }; +} diff --git a/src/servers/api/chat/opencode/server-manager.ts b/src/servers/api/chat/opencode/server-manager.ts new file mode 100644 index 00000000..4706e0c1 --- /dev/null +++ b/src/servers/api/chat/opencode/server-manager.ts @@ -0,0 +1,118 @@ +import type { Subprocess } from 'bun'; +import { homedir } from 'os'; +import { join } from 'path'; +import { mkdirSync } from 'fs'; +import { logger } from '../logger'; + +// The `opencode` binary. Pinned (like CLAUDE_BIN) rather than resolved from PATH; override with +// OPENCODE_BIN. NOTE: the installed binary is 1.17.9 — the 1.18.4 upgrade never landed on disk. +const OPENCODE_BIN = process.env.OPENCODE_BIN || join(homedir(), '.opencode', 'bin', 'opencode'); + +const HEALTH_TIMEOUT_MS = 20_000; +const HEALTH_POLL_MS = 200; +const START_ATTEMPTS = 3; + +type OpenCodeServer = { + baseUrl: string; + proc: Subprocess; + port: number; +}; + +// One warm `opencode serve` per working directory (sessions bind to a directory at creation). +const servers = new Map(); +const starting = new Map>(); + +/** Grab an ephemeral free port by briefly binding one and releasing it. */ +function getFreePort(): number { + const probe = Bun.serve({ port: 0, hostname: '127.0.0.1', fetch: () => new Response('') }); + const port = probe.port; + probe.stop(true); + if (port == null) throw new Error('failed to acquire a free port'); + return port; +} + +async function isHealthy(baseUrl: string): Promise { + try { + const res = await fetch(`${baseUrl}/api/health`, { signal: AbortSignal.timeout(2000) }); + return res.ok; + } catch { + return false; + } +} + +async function waitHealthy(baseUrl: string, timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await isHealthy(baseUrl)) return true; + await new Promise((r) => setTimeout(r, HEALTH_POLL_MS)); + } + return false; +} + +async function startServer(cwd: string, home: string): Promise { + mkdirSync(cwd, { recursive: true }); + + for (let attempt = 1; attempt <= START_ATTEMPTS; attempt += 1) { + const port = getFreePort(); + const baseUrl = `http://127.0.0.1:${port}`; + // Loopback-only + no OPENCODE_SERVER_PASSWORD → the server is open on 127.0.0.1 (single-user box). + // HOME is set to the caller's home so `opencode` reads that user's ~/.local/share/opencode auth. + const proc = Bun.spawn([OPENCODE_BIN, 'serve', '--port', String(port), '--hostname', '127.0.0.1'], { + cwd, + env: { ...process.env, HOME: home }, + stdout: 'ignore', + stderr: 'ignore', + }); + + if (await waitHealthy(baseUrl, HEALTH_TIMEOUT_MS)) { + logger.info('opencode serve started', { cwd, baseUrl }); + proc.exited.then((code) => { + // Drop the cached entry on exit so the next turn respawns. + if (servers.get(cwd)?.proc === proc) servers.delete(cwd); + logger.warn('opencode serve exited', { cwd, code }); + }); + return { baseUrl, proc, port }; + } + + logger.warn('opencode serve failed health check, retrying', { cwd, baseUrl, attempt }); + try { + proc.kill(); + } catch { + /* already gone */ + } + } + + throw new Error(`opencode serve failed to start for cwd ${cwd}`); +} + +/** Ensure a healthy `opencode serve` for `cwd`, returning its base URL. Dedupes concurrent starts. */ +export async function ensureServer(cwd: string, home: string): Promise<{ baseUrl: string }> { + const existing = servers.get(cwd); + if (existing && (await isHealthy(existing.baseUrl))) { + return { baseUrl: existing.baseUrl }; + } + if (existing) { + try { + existing.proc.kill(); + } catch { + /* already gone */ + } + servers.delete(cwd); + } + + const inflight = starting.get(cwd); + if (inflight) { + const s = await inflight; + return { baseUrl: s.baseUrl }; + } + + const promise = startServer(cwd, home); + starting.set(cwd, promise); + try { + const server = await promise; + servers.set(cwd, server); + return { baseUrl: server.baseUrl }; + } finally { + starting.delete(cwd); + } +} diff --git a/src/servers/api/chat/opencode/state.ts b/src/servers/api/chat/opencode/state.ts new file mode 100644 index 00000000..e7edf05c --- /dev/null +++ b/src/servers/api/chat/opencode/state.ts @@ -0,0 +1,15 @@ +// Maps our live-turn sessionKey → OpenCode's own session id (`ses_…`), the analog of +// sidecar/claude/state.ts's sessionKey→claude-uuid map. In-memory only: OpenCode's SQLite store is +// the durable record, and a fresh key just creates a new OpenCode session. + +const sessionKeyToOpenCode = new Map(); + +export const getOpenCodeSession = (sessionKey: string): string | undefined => sessionKeyToOpenCode.get(sessionKey); + +export const setOpenCodeSession = (sessionKey: string, opencodeSessionId: string): void => { + sessionKeyToOpenCode.set(sessionKey, opencodeSessionId); +}; + +export const clearOpenCodeSession = (sessionKey: string): void => { + sessionKeyToOpenCode.delete(sessionKey); +}; diff --git a/src/servers/api/chat/websocket.ts b/src/servers/api/chat/websocket.ts index e20a5737..d9fe9dbc 100644 --- a/src/servers/api/chat/websocket.ts +++ b/src/servers/api/chat/websocket.ts @@ -3,6 +3,7 @@ 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 { ensureClaudeSessionsCwd } from './claude-sessions'; import * as sidecar from '@@/sidecar-registry'; import { join } from 'path'; @@ -13,6 +14,10 @@ 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); @@ -275,18 +280,14 @@ async function handleChat( : msg.prompt; // Use provided model, or fall back to user default, or the system default. - let model = msg.model || (await getUserDefaultModel(userId)) || DEFAULT_MODEL; - - // Claude-only: coerce any legacy/non-Claude model preference to the Claude default so old saved - // settings (Pi/opencode/openrouter model ids) don't break chat. - if (!model.startsWith('claude-code')) { - logger.info('Coercing non-Claude model to Claude default', { sessionId, requested: model }); - model = DEFAULT_MODEL; - } + const model = msg.model || (await getUserDefaultModel(userId)) || DEFAULT_MODEL; logger.info('Model selected for chat', { sessionId, model, clientModel: msg.model || null }); - return handleClaudeCodeChat(ws, sessionId, model, msg, prompt); + // 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( @@ -377,6 +378,91 @@ async function handleClaudeCodeChat( } } +async function handleOpenCodeChat( + ws: ServerWebSocket, + sessionId: string, + model: string, + msg: { + prompt: string; + displayText?: string; + groupSlug?: string; + context?: string; + contextId?: string; + cwd?: string; + cwdRoot?: string; + sandboxed?: boolean; + resumeSessionId?: string; + }, + effectivePrompt: string, +): Promise { + const { email, username, userId } = ws.data; + + // Same cwd resolution as the Claude path: /chat runs from a chosen pwd or the default dir. + const cwd = + msg.context === 'chat' + ? msg.cwd?.trim() + ? resolveCwd(email, ws.data.role, msg.cwd) + : ensureClaudeSessionsCwd(email) + : resolveCwd(email, ws.data.role, msg.cwd); + + 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, + role: ws.data.role, + 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 }, @@ -429,8 +515,13 @@ async function handleStop(ws: ServerWebSocket): Promise { if (session?.piProcess) { try { - sidecar.killClaude(sessionId, session.email); - logger.info('Killed Claude Code process via sidecar', { sessionId }); + 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) }); diff --git a/src/servers/channels/send-opencode.ts b/src/servers/channels/send-opencode.ts new file mode 100644 index 00000000..589af6ca --- /dev/null +++ b/src/servers/channels/send-opencode.ts @@ -0,0 +1,75 @@ +import type { ChatEvent } from '@@/api/chat/types'; +import { logger } from '@@/api/chat/logger'; +import { ensureServer } from '@@/api/chat/opencode/server-manager'; +import { getConnection } from '@@/api/chat/opencode/client'; +import { createEventMapper } from '@@/api/chat/opencode/event-mapper'; +import { getOpenCodeSession, setOpenCodeSession } from '@@/api/chat/opencode/state'; +import { getHomeDirForRole } from '../data-path'; + +// The OpenCode analog of send-claude-code.ts's streaming path. Drives a turn against a warm +// `opencode serve` over HTTP + SSE, mapping events to the shared ChatEvent contract. + +type OpenCodeStreamingParams = { + userId: number; + email: string; + username: string; + prompt: string; + sessionKey: string; + cwd?: string; + model?: string; + role?: string; + resumeSessionId?: string; + onEvent: (event: ChatEvent) => void; +}; + +type OpenCodeStreamingHandle = { + kill: () => void; +}; + +/** Split an OpenCode model id (`providerID/modelID`, e.g. `opencode/claude-opus-4-8`). */ +function splitModel(model: string): { providerID: string; modelID: string } { + const slash = model.indexOf('/'); + if (slash <= 0) return { providerID: 'opencode', modelID: model }; + return { providerID: model.slice(0, slash), modelID: model.slice(slash + 1) }; +} + +export async function sendOpenCodeStreaming(params: OpenCodeStreamingParams): Promise { + const home = getHomeDirForRole(params.email, params.role ?? ''); + const cwd = params.cwd || home; + + logger.info('OpenCode streaming exec', { sessionKey: params.sessionKey, model: params.model }); + + const { baseUrl } = await ensureServer(cwd, home); + const conn = getConnection(baseUrl); + + // Reuse the OpenCode session for this live sessionKey, else create one bound to the cwd. + let opencodeSessionId = getOpenCodeSession(params.sessionKey) ?? params.resumeSessionId; + if (!opencodeSessionId) { + opencodeSessionId = await conn.createSession(cwd); + setOpenCodeSession(params.sessionKey, opencodeSessionId); + } + const sessionId = opencodeSessionId; + + let unsub = () => {}; + const mapper = createEventMapper((event: ChatEvent) => { + params.onEvent(event); + if (event.type === 'result' || event.type === 'error') unsub(); + }); + unsub = conn.subscribe(sessionId, mapper); + + const { providerID, modelID } = splitModel(params.model ?? ''); + + // Fire the turn; assistant tokens + tool calls stream back over the SSE subscription above. + conn.postMessage(sessionId, providerID, modelID, params.prompt).catch((err) => { + logger.error('OpenCode postMessage failed', { sessionKey: params.sessionKey, error: String(err) }); + params.onEvent({ type: 'error', message: 'Failed to send message to OpenCode' }); + unsub(); + }); + + return { + kill: () => { + void conn.abort(sessionId); + unsub(); + }, + }; +}