phase B: run a turn through the serve, behind a switch

OPENCODE_TURNS=serve picks the new engine; unset keeps the subprocess, which is the default
and stays the default until this has been lived with. A bad evening should cost one restart,
not a revert. Claude is a different sidecar and is untouched.

Verified end to end through the real chat socket:

  session:init -> tool:start(bash) -> tool:result -> assistant:delta x3 -> assistant:text
  -> result, cost in=304 out=73

Those deltas are the first token streaming an opencode turn has ever produced in officer.
Stop is now an INTERRUPT: the turn ends and the session survives — verified by sending a
second prompt to the same session afterwards and getting an answer, which killing a
subprocess could never do.

Reads the LIVE global stream rather than the durable per-session one, because it is a strict
superset — same tool.called, tool.success, step.ended, text.ended, plus the deltas that are
the whole point. Global means one socket carries every session, so everything filters on
sessionID; one subscription is shared for the process rather than one per turn.

A turn ends on step.ended with finish != tool-calls. tool-calls is a step boundary MID-turn,
and treating it as terminal would cut every tool-using conversation in half.

delivery is stated explicitly as queue because it DEFAULTS to steer, which injects into a
running turn — wrong for an ordinary send, where two quick messages would merge into one.
Wiring steer to the button that means it is phase C.

What phase B does not do: read the durable stream. The sidecar still commits every event to
chat_session_events as it arrives, so durability is unchanged, but recovering a turn this
process never saw needs the ?after= cursor and is its own change.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/servers/sidecar/opencode/serve-runner.ts

@@ -0,0 +1,278 @@
+import type { ChatEvent, MessageCost } from '../../api/chat/types';
+import type { OpenCodeRunParams } from '../protocol';
+import type { RunnerMessage } from './runner';
+import { mapServeEvent } from './serve-events';
+
+// Phase B: drive a turn through the serve instead of spawning `opencode run`.
+//
+// OFF BY DEFAULT. `index.ts` picks between this and `runner.ts` on `OPENCODE_TURNS`, and the subprocess
+// stays the default until this has run for a while — a bad day should be one restart from the path that
+// has worked all along, not a rollback.
+//
+// ── Why bother, given the subprocess works ──
+//
+// Everything the subprocess cannot do is a consequence of `stdin: 'ignore'`: no token streaming, no
+// mid-turn injection, no queue, no interrupt that leaves the session alive. The serve offers all four as
+// primitives, verified against 1.18.16 (docs/opencode-fork-decision.md).
+//
+// ── One global stream, demultiplexed ──
+//
+// The serve publishes each turn on two streams (docs/opencode-serve-migration-plan.md). This reads the
+// LIVE one, `GET /api/event`, because it is a strict superset of the durable stream's content — same
+// `tool.called`, `tool.success`, `step.ended`, `text.ended`, PLUS the deltas — and deltas are the point.
+//
+// It is GLOBAL: one socket carries every session on the box, so everything here filters on `sessionID`.
+// Forgetting that would splice one conversation into another. There is exactly one subscription for the
+// process, opened on the first turn and shared, because opening one per turn would multiply the same
+// firehose by the number of turns.
+//
+// What this loses versus the durable stream is `durable.seq`, i.e. replay-after-the-fact. That matters
+// for surviving an officer restart mid-turn and is deliberately NOT in Phase B: the sidecar commits
+// every event to `chat_session_events` as it arrives (unchanged from the subprocess path), which is the
+// same durability guarantee the subprocess had. Reading the durable stream to recover a turn this
+// process never saw is its own change.
+
+type ServeConfig = {
+  /** The serve's base URL, e.g. http://127.0.0.1:53100 */
+  baseUrl: string;
+  /** Used when a turn names no cwd — same fallback the subprocess runner applies. */
+  fallbackCwd: string;
+};
+
+type Emit = (msg: RunnerMessage) => void;
+
+type ServeTurn = {
+  sessionKey: string;
+  openCodeSessionId: string;
+  cost: MessageCost;
+  done: boolean;
+  emit: Emit;
+  finish: (event: ChatEvent) => void;
+};
+
+/** Live turns, by OpenCode session id — the id the event stream speaks. */
+const byOpenCodeId = new Map<string, ServeTurn>();
+/** The same turns by officer's key, which is what `kill` and the Live panel use. */
+const bySessionKey = new Map<string, ServeTurn>();
+
+// ── The one shared subscription ──
+
+let streamStarted = false;
+
+/**
+ * Read `/api/event` forever, routing each event to the turn that owns it.
+ *
+ * Reconnects on drop with a fixed delay. A serve restart, a network blip or the stream simply ending
+ * must not permanently deafen the sidecar — every subsequent turn would hang with no output, which is
+ * the worst failure this path has, because it looks exactly like a slow model.
+ */
+function ensureEventStream(config: ServeConfig): void {
+  if (streamStarted) return;
+  streamStarted = true;
+
+  void (async () => {
+    for (;;) {
+      try {
+        const res = await fetch(`${config.baseUrl}/api/event`, { headers: { accept: 'text/event-stream' } });
+        if (!res.ok || !res.body) throw new Error(`event stream → ${res.status}`);
+
+        const reader = res.body.getReader();
+        const decoder = new TextDecoder();
+        let buffer = '';
+
+        for (;;) {
+          const { done, value } = await reader.read();
+          if (done) break;
+          buffer += decoder.decode(value, { stream: true });
+          const lines = buffer.split('\n');
+          buffer = lines.pop() ?? '';
+          for (const line of lines) {
+            if (!line.startsWith('data:')) continue;
+            try {
+              handleServeEvent(JSON.parse(line.slice(5).trim()));
+            } catch {
+              /* a frame we could not parse is not worth killing the stream over */
+            }
+          }
+        }
+      } catch (err) {
+        console.error('[opencode] event stream dropped, reconnecting:', err instanceof Error ? err.message : err);
+      }
+      await Bun.sleep(1_000);
+    }
+  })();
+}
+
+function handleServeEvent(raw: unknown): void {
+  const mapped = mapServeEvent(raw as never);
+  if (!mapped?.sessionId) return;
+
+  const turn = byOpenCodeId.get(mapped.sessionId);
+  if (!turn || turn.done) return; // another session's, or one we have already finished
+
+  turn.cost = {
+    inputTokens: turn.cost.inputTokens + mapped.costDelta.inputTokens,
+    outputTokens: turn.cost.outputTokens + mapped.costDelta.outputTokens,
+    totalUSD: turn.cost.totalUSD + mapped.costDelta.totalUSD,
+  };
+
+  for (const event of mapped.events) {
+    // An error from the harness ends the turn: nothing follows a failed step, and leaving the turn open
+    // would strand the UI on a spinner.
+    if (event.type === 'error') {
+      turn.finish(event);
+      return;
+    }
+    turn.emit({ type: 'opencode:event', sessionKey: turn.sessionKey, event });
+  }
+
+  // `tool-calls` means the model paused to run a tool and will continue. Anything else is the end.
+  if (mapped.stepFinish && mapped.stepFinish !== 'tool-calls') {
+    turn.finish({ type: 'result', cost: turn.cost });
+  }
+}
+
+// ── HTTP helpers ──
+
+async function serveJson<T>(config: ServeConfig, path: string, init: RequestInit & { cwd: string }): Promise<T | null> {
+  const { cwd, ...rest } = init;
+  const res = await fetch(`${config.baseUrl}${path}`, {
+    ...rest,
+    headers: {
+      'content-type': 'application/json',
+      // The location is per REQUEST on this surface, not a property of the session. A call without it
+      // runs against the serve's own directory, which is not where the user's files are.
+      'x-opencode-directory': cwd,
+      ...(rest.headers ?? {}),
+    },
+    signal: AbortSignal.timeout(30_000),
+  });
+  if (!res.ok) throw new Error(`${path} → ${res.status}`);
+  const text = await res.text();
+  if (!text) return null;
+  // This surface wraps everything in `{data: …}`; the legacy one does not. Reading `.id` off the
+  // envelope silently yields undefined, which is how an entire afternoon disappeared once.
+  const body = JSON.parse(text) as { data?: T } | T;
+  return (body as { data?: T }).data ?? (body as T);
+}
+
+// ── The turn ──
+
+export async function runOpenCodeTurnOnServe(
+  params: OpenCodeRunParams,
+  config: ServeConfig,
+  emit: Emit,
+): Promise<void> {
+  const { sessionKey } = params;
+  const cwd = params.cwd || config.fallbackCwd;
+
+  ensureEventStream(config);
+
+  // Supersede any turn still registered under this key. Unlike the subprocess path there is no process
+  // to kill — the serve owns execution — so this is bookkeeping only, and the old turn is retired
+  // silently rather than reporting an error against a key that now belongs to its replacement.
+  const stale = bySessionKey.get(sessionKey);
+  if (stale) retire(stale, null);
+
+  let openCodeSessionId = params.resumeSessionId ?? '';
+
+  try {
+    if (!openCodeSessionId) {
+      const created = await serveJson<{ id: string }>(config, '/api/session', {
+        method: 'POST',
+        body: JSON.stringify({ location: { directory: cwd } }),
+        cwd,
+      });
+      if (!created?.id) throw new Error('session create returned no id');
+      openCodeSessionId = created.id;
+    }
+
+    if (params.model) {
+      // `providerID/modelID`, the same string the subprocess passes to `--model`.
+      const slash = params.model.indexOf('/');
+      const providerID = slash > 0 ? params.model.slice(0, slash) : 'opencode';
+      const id = slash > 0 ? params.model.slice(slash + 1) : params.model;
+      await serveJson(config, `/api/session/${openCodeSessionId}/model`, {
+        method: 'POST',
+        body: JSON.stringify({ model: { providerID, id } }),
+        cwd,
+      });
+    }
+  } catch (err) {
+    emit({
+      type: 'opencode:event',
+      sessionKey,
+      event: { type: 'error', message: `Could not start an OpenCode session: ${errText(err)}` },
+    });
+    return;
+  }
+
+  const turn: ServeTurn = {
+    sessionKey,
+    openCodeSessionId,
+    cost: { inputTokens: 0, outputTokens: 0, totalUSD: 0 },
+    done: false,
+    emit,
+    finish: (event) => retire(turn, event),
+  };
+  byOpenCodeId.set(openCodeSessionId, turn);
+  bySessionKey.set(sessionKey, turn);
+
+  // Officer learns which `ses_…` to resume next time — a routing fact, not transcript.
+  emit({ type: 'opencode:session', sessionKey, sessionId: openCodeSessionId });
+
+  try {
+    await serveJson(config, `/api/session/${openCodeSessionId}/prompt`, {
+      method: 'POST',
+      // `delivery` is stated explicitly because it DEFAULTS to `"steer"`, which injects into a running
+      // turn. For an ordinary send that is the wrong default — two quick messages would merge into one
+      // turn instead of running in order. `steer` is Phase C's job, wired to the button that means it.
+      body: JSON.stringify({ prompt: { text: params.prompt }, delivery: 'queue' }),
+      cwd,
+    });
+  } catch (err) {
+    retire(turn, { type: 'error', message: `OpenCode refused the prompt: ${errText(err)}` });
+  }
+}
+
+function retire(turn: ServeTurn, event: ChatEvent | null): void {
+  if (turn.done) return;
+  turn.done = true;
+  if (byOpenCodeId.get(turn.openCodeSessionId) === turn) byOpenCodeId.delete(turn.openCodeSessionId);
+  if (bySessionKey.get(turn.sessionKey) === turn) bySessionKey.delete(turn.sessionKey);
+  if (event) turn.emit({ type: 'opencode:event', sessionKey: turn.sessionKey, event });
+}
+
+const errText = (err: unknown): string => (err instanceof Error ? err.message : String(err));
+
+/** The serve analog of `listRunningOpenCodeTurns`. Same shape, so the Live panel needs no changes. */
+export function listRunningServeTurns(): { sessionKey: string }[] {
+  return [...bySessionKey.keys()].map((sessionKey) => ({ sessionKey }));
+}
+
+/**
+ * Stop a turn without destroying its session — the thing the subprocess path cannot do.
+ *
+ * `POST /interrupt` leaves the conversation intact and resumable, where killing a subprocess ended it.
+ */
+export async function killServeTurn(sessionKey: string, config: ServeConfig): Promise<void> {
+  const turn = bySessionKey.get(sessionKey);
+  if (!turn) return;
+  try {
+    await fetch(`${config.baseUrl}/api/session/${turn.openCodeSessionId}/interrupt`, {
+      method: 'POST',
+      headers: { 'x-opencode-directory': config.fallbackCwd },
+      signal: AbortSignal.timeout(10_000),
+    });
+  } catch {
+    /* interrupt is best-effort; the turn is retired either way so the UI is never stuck */
+  }
+  retire(turn, { type: 'stopped' });
+}
+
+/** Retire every live turn, for shutdown. Mirrors `stopAllOpenCodeTurns` on the subprocess path. */
+export function stopAllServeTurns(message: string): number {
+  const turns = [...bySessionKey.values()];
+  for (const turn of turns) retire(turn, { type: 'error', message });
+  return turns.length;
+}