chat: add OpenCode as a second harness — live turn (Phase 1)

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 <noreply@anthropic.com>

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<string, Set<Listener>>();
+  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<void> {
+    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<T>(path: string, body: unknown): Promise<T> {
+    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<string> {
+    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<void> {
+    await this.postJson(`/session/${sessionId}/message`, {
+      model: { providerID, modelID },
+      parts: [{ type: 'text', text }],
+    });
+  }
+
+  async abort(sessionId: string): Promise<void> {
+    await this.postJson(`/session/${sessionId}/abort`, {}).catch(() => {});
+  }
+}
+
+const connections = new Map<string, ServerConnection>();
+
+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 };

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<string, unknown>;
+};
+
+type ToolState = {
+  status?: string;
+  input?: Record<string, unknown>;
+  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<string>();
+  const toolFinished = new Set<string>();
+  const partTypes = new Map<string, string>(); // 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;
+    }
+  };
+}

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<string, OpenCodeServer>();
+const starting = new Map<string, Promise<OpenCodeServer>>();
+
+/** 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<boolean> {
+  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<boolean> {
+  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<OpenCodeServer> {
+  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);
+  }
+}

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<string, string>();
+
+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);
+};

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<string | null> {
   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<WSData>,
+  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<void> {
+  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<WSData>,
   msg: { sessionId: string; cwd?: string; cwdRoot?: string },
@@ -429,8 +515,13 @@ async function handleStop(ws: ServerWebSocket<WSData>): Promise<void> {
 
     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) });

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<OpenCodeStreamingHandle> {
+  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();
+    },
+  };
+}