diff --git a/src/servers/sidecar/opencode/index.ts b/src/servers/sidecar/opencode/index.ts index 93c12314..b5e6c283 100644 --- a/src/servers/sidecar/opencode/index.ts +++ b/src/servers/sidecar/opencode/index.ts @@ -7,7 +7,9 @@ import { sweepRecordedServe } from './serve-sweep'; import { connectProviderCredential } from './connect-credential'; import { createSessionLogStore } from '../claude/session-log'; import type { SidecarCommand, SidecarEvent } from '../protocol'; +import type { RunnerMessage } from './runner'; import { runOpenCodeTurn, killOpenCodeTurn, listRunningOpenCodeTurns, stopAllOpenCodeTurns } from './runner'; +import { runOpenCodeTurnOnServe, killServeTurn, listRunningServeTurns, stopAllServeTurns } from './serve-runner'; // The OpenCode sidecar (officer-opencode). Same philosophy as officer-claude: a singleton process that // OWNS its runtime — here, an `opencode serve` — registers with the API server, and answers commands. It @@ -143,6 +145,17 @@ void connectProviderCredential(baseUrl); type ReplyFn = (msg: SidecarEvent) => void; const RUNNER_CONFIG = { bin: OPENCODE_BIN, fallbackCwd: SERVE_CWD }; +const SERVE_RUNNER_CONFIG = { baseUrl, fallbackCwd: SERVE_CWD }; + +// Which engine runs a turn. `subprocess` (the default) spawns `opencode run`; `serve` drives the +// serve's /api/session surface, which is the only way to get streaming, steer, queue and a stop that +// leaves the session alive. +// +// A switch rather than a replacement, and defaulted to the old path on purpose: the subprocess has +// worked all day and the serve path has not been lived with yet. A bad evening should cost one restart +// with OPENCODE_TURNS unset, not a revert. +const USE_SERVE_TURNS = (process.env.OPENCODE_TURNS ?? 'subprocess').toLowerCase() === 'serve'; +console.log(`[opencode] turn engine: ${USE_SERVE_TURNS ? 'serve (/api/session)' : 'subprocess (opencode run)'}`); function handleCommand(cmd: SidecarCommand, reply: ReplyFn) { switch (cmd.type) { @@ -154,23 +167,35 @@ function handleCommand(cmd: SidecarCommand, reply: ReplyFn) { // Turn output goes through the session log: translated to TurnMessages and committed to // chat_session_events here, in the process that produced it. Officer being down during a turn // no longer costs the transcript — the browser replays it from its cursor. - runOpenCodeTurn(cmd.params, RUNNER_CONFIG, (msg) => { + // Same emit contract either way, which is what makes the switch a switch: the durable commit and + // the routing fact behave identically whether a subprocess or the serve produced the event. + const onMessage = (msg: RunnerMessage) => { if (msg.type === 'opencode:event') { sessionLog.push(sessionKey, msg.event, durable); return; } // opencode:session is a routing fact (which `ses_…` to resume), not transcript — send it live. connection.send(msg); - }); + }; + + if (USE_SERVE_TURNS) void runOpenCodeTurnOnServe(cmd.params, SERVE_RUNNER_CONFIG, onMessage); + else runOpenCodeTurn(cmd.params, RUNNER_CONFIG, onMessage); reply({ type: 'opencode:spawned', id: cmd.id, sessionKey }); break; } case 'opencode:list': - reply({ type: 'opencode:sessions', id: cmd.id, sessions: listRunningOpenCodeTurns() }); + reply({ + type: 'opencode:sessions', + id: cmd.id, + sessions: USE_SERVE_TURNS ? listRunningServeTurns() : listRunningOpenCodeTurns(), + }); break; case 'opencode:kill': - killOpenCodeTurn(cmd.sessionKey); + // On the serve this is an INTERRUPT: the turn stops and the session survives, so the conversation + // can be continued rather than only re-opened. + if (USE_SERVE_TURNS) void killServeTurn(cmd.sessionKey, SERVE_RUNNER_CONFIG); + else killOpenCodeTurn(cmd.sessionKey); sessionLog.drop(cmd.sessionKey); break; default: @@ -221,7 +246,8 @@ async function shutdown(signal: string) { // Before the connection goes: killing a turn produces an event, and that event's durable write travels // over this socket. Tearing it down first would stop every turn silently — the exact outcome this is // here to prevent. - const stopped = stopAllOpenCodeTurns(`The OpenCode sidecar restarted (${signal}), so this turn stopped.`); + const message = `The OpenCode sidecar restarted (${signal}), so this turn stopped.`; + const stopped = USE_SERVE_TURNS ? stopAllServeTurns(message) : stopAllOpenCodeTurns(message); if (stopped > 0) { console.log(`[opencode] ${signal} received, stopping ${stopped} in-flight turn(s)...`); await Promise.race([sessionLog.flush(), Bun.sleep(SHUTDOWN_FLUSH_MS)]); diff --git a/src/servers/sidecar/opencode/serve-events.ts b/src/servers/sidecar/opencode/serve-events.ts index 93f39fdc..7b1162dc 100644 --- a/src/servers/sidecar/opencode/serve-events.ts +++ b/src/servers/sidecar/opencode/serve-events.ts @@ -63,6 +63,15 @@ export type MappedServeEvent = { seq?: number; events: ChatEvent[]; costDelta: { inputTokens: number; outputTokens: number; totalUSD: number }; + /** + * Present only on `step.ended`, and the answer to "is the turn over". + * + * A subprocess turn ends by exiting; a serve turn has no such moment, so this is the only signal. + * `"tool-calls"` means the model paused to run a tool and WILL continue — treating it as terminal cuts + * every tool-using conversation in half. `"stop"` is the real end. Anything else (e.g. `"length"`) is + * also an ending, so the caller should test for "not tool-calls" rather than for "stop". + */ + stepFinish?: string; }; const NO_COST = { inputTokens: 0, outputTokens: 0, totalUSD: 0 }; @@ -162,6 +171,7 @@ export function mapServeEvent(evt: ServeEvent | null | undefined): MappedServeEv return { ...base, events: [], + stepFinish: typeof d.finish === 'string' ? d.finish : undefined, costDelta: { inputTokens: t?.input ?? 0, outputTokens: t?.output ?? 0, diff --git a/src/servers/sidecar/opencode/serve-runner.ts b/src/servers/sidecar/opencode/serve-runner.ts new file mode 100644 index 00000000..7b7834a8 --- /dev/null +++ b/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(); +/** The same turns by officer's key, which is what `kill` and the Live panel use. */ +const bySessionKey = new Map(); + +// ── 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(config: ServeConfig, path: string, init: RequestInit & { cwd: string }): Promise { + 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 { + 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 { + 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; +}