diff --git a/docs/opencode-serve-migration-plan.md b/docs/opencode-serve-migration-plan.md index 8bd94267..594b6ba0 100644 --- a/docs/opencode-serve-migration-plan.md +++ b/docs/opencode-serve-migration-plan.md @@ -15,15 +15,31 @@ works, it is verified end to end, and its limits are all consequences of that on The serve's `/api/session/*` surface offers, and I have run each of these against 1.18.16: -| Capability | How | Verified | -| ------------------------- | -------------------------------------------------------- | -------- | -| Mid-turn injection | `POST /prompt` `{delivery: "steer"}` | yes — steered a running turn | -| Queue behind a turn | `POST /prompt` `{delivery: "queue"}` | yes — "ONE" then "TWO", no errors | -| Token streaming | `GET /event` → `text/event-stream` | yes — `text.started`/`text.ended` observed | -| Reconnect + replay | same, `?after=` | yes — replayed a finished session | -| Interrupt, session lives | `POST /interrupt` → 204 | endpoint only, not exercised | -| Model selection | `POST /model` → 204 | yes — runs on the chosen model | -| Images | `prompt.files` | not exercised (we have images via `--file` already) | +| Capability | How | Verified | +| ------------------------ | --------------------------------------------------- | --------------------------------------------------- | +| Mid-turn injection | `POST /prompt` `{delivery: "steer"}` | yes — steered a running turn | +| Queue behind a turn | `POST /prompt` `{delivery: "queue"}` | yes — "ONE" then "TWO", no errors | +| Token streaming | `GET /api/event` (GLOBAL, live) — `text.delta` | yes — deltas reassemble to the committed text | +| Reconnect + replay | `GET /api/session/{id}/event?after=` (durable) | yes — replayed a finished session | +| Interrupt, session lives | `POST /interrupt` → 204 | endpoint only, not exercised | +| Model selection | `POST /model` → 204 | yes — runs on the chosen model | +| Images | `prompt.files` | not exercised (we have images via `--file` already) | + +## There are TWO streams, and this is the thing to get right + +Corrected after Phase A; the table above originally implied one. The serve publishes each turn twice: + +- **`GET /api/session/{id}/event?after=`** — durable, per session, replayable, every event carrying + `durable.seq`. Whole values only (`text.ended` with the full text). **No deltas.** +- **`GET /api/event`** — live, **global**, ephemeral. Carries `text.delta` and `tool.input.delta`. No cursor. + +Measured on one real turn: 13 events durable, 21 live, the difference being 3 `text.delta` and 5 +`tool.input.delta`. **Reading only the per-session stream — which is what I did first — makes it look +like the serve cannot stream at all**, and would have quietly removed the main reason to migrate. + +The split maps exactly onto what officer already does for Claude: durable → `chat_session_events`, live → +UI deltas. The cost is that the live stream is GLOBAL, so a consumer must filter on `sessionID` and +cannot assume it owns the socket. ## Facts that will bite whoever implements this diff --git a/src/servers/sidecar/opencode/serve-events.test.ts b/src/servers/sidecar/opencode/serve-events.test.ts new file mode 100644 index 00000000..5391ff25 --- /dev/null +++ b/src/servers/sidecar/opencode/serve-events.test.ts @@ -0,0 +1,196 @@ +import { describe, expect, it } from 'bun:test'; +import { isKnownServeEvent, mapServeEvent } from './serve-events'; + +// Fixtures are VERBATIM captures from opencode 1.18.16 — one real turn that ran `echo hello-from-tool` +// through the bash tool and then answered in prose. Ids and timestamps are as they arrived. +// +// They are real for the same reason `runner.test.ts`'s are: this mapping's only job is to match a shape +// nobody documents, so a hand-written fixture would test my imagination rather than the binary. If one +// of these fails after an upgrade, re-capture before editing the expectation. + +const SESSION = 'ses_01346bde5ffeB4hWdcQiq1BQUG'; +const ASSISTANT = 'msg_fecb96b3e0010o8smEyjxT2noD'; +const CALL = 'toolu_01TyLVFSSXus2eoyANdjieWW'; + +describe('mapServeEvent — text', () => { + it('turns a text delta into a delta, because that is what makes output appear as it is typed', () => { + const out = mapServeEvent({ + type: 'session.next.text.delta', + data: { timestamp: 1, sessionID: SESSION, assistantMessageID: ASSISTANT, textID: 'text-0', delta: 'It' }, + } as never); + + expect(out?.events).toEqual([{ type: 'delta', text: 'It' }]); + expect(out?.sessionId).toBe(SESSION); + }); + + it('turns the ended text into the committed block', () => { + const out = mapServeEvent({ + type: 'session.next.text.ended', + data: { sessionID: SESSION, textID: 'text-0', text: 'It printed **hello-from-tool**.' }, + } as never); + + expect(out?.events).toEqual([{ type: 'text', text: 'It printed **hello-from-tool**.' }]); + }); + + it('says nothing for text.started, which carries no text at all', () => { + // Verified against the capture: `text.started` has `{textID}` and nothing else. Emitting an empty + // text here would put a blank assistant bubble on screen before a single token arrived. + const out = mapServeEvent({ + type: 'session.next.text.started', + data: { sessionID: SESSION, textID: 'text-0' }, + } as never); + expect(out?.events).toEqual([]); + }); + + it('drops an empty delta rather than emitting nothing-shaped events', () => { + expect(mapServeEvent({ type: 'session.next.text.delta', data: { delta: '' } } as never)?.events).toEqual([]); + }); +}); + +describe('mapServeEvent — tools', () => { + it('starts the tool row on tool.called, where the input is finally a real object', () => { + const out = mapServeEvent({ + type: 'session.next.tool.called', + data: { + sessionID: SESSION, + assistantMessageID: ASSISTANT, + callID: CALL, + tool: 'bash', + input: { command: 'echo hello-from-tool' }, + }, + } as never); + + expect(out?.events).toEqual([ + { type: 'tool:start', toolCallId: CALL, toolName: 'bash', toolInput: { command: 'echo hello-from-tool' } }, + ]); + }); + + it('stays silent through the input stream, which is half-parsed JSON', () => { + // This is the reason `tool:start` waits for `tool.called`. The real delta below is a fragment of a + // JSON object; rendering a tool row with `{"comman` as its arguments would be worse than waiting. + const started = mapServeEvent({ + type: 'session.next.tool.input.started', + data: { sessionID: SESSION, callID: CALL, name: 'bash' }, + } as never); + const delta = mapServeEvent({ + type: 'session.next.tool.input.delta', + data: { sessionID: SESSION, callID: CALL, delta: '{"comman' }, + } as never); + + expect(started?.events).toEqual([]); + expect(delta?.events).toEqual([]); + }); + + it('flattens the success content blocks into one output string', () => { + const out = mapServeEvent({ + type: 'session.next.tool.success', + data: { + sessionID: SESSION, + callID: CALL, + structured: { exit: 0, truncated: false }, + content: [{ type: 'text', text: 'hello-from-tool' }], + }, + } as never); + + expect(out?.events).toEqual([{ type: 'tool:result', toolCallId: CALL, output: 'hello-from-tool', isError: false }]); + }); + + it('reports a failed tool with its error as the output', () => { + const out = mapServeEvent({ + type: 'session.next.tool.failed', + data: { sessionID: SESSION, callID: CALL, error: { message: 'command not found' } }, + } as never); + + expect(out?.events).toEqual([ + { type: 'tool:result', toolCallId: CALL, output: 'command not found', isError: true }, + ]); + }); + + it('ignores a tool event with no callID, which cannot be correlated', () => { + expect(mapServeEvent({ type: 'session.next.tool.called', data: { tool: 'bash' } } as never)?.events).toEqual([]); + expect(mapServeEvent({ type: 'session.next.tool.success', data: {} } as never)?.events).toEqual([]); + }); +}); + +describe('mapServeEvent — accounting', () => { + it('reports per-step tokens and cost for the caller to accumulate', () => { + const out = mapServeEvent({ + type: 'session.next.step.ended', + data: { + sessionID: SESSION, + finish: 'tool-calls', + cost: 0.0042, + tokens: { input: 3, output: 57, reasoning: 0, cache: { read: 3850, write: 0 } }, + }, + } as never); + + expect(out?.costDelta).toEqual({ inputTokens: 3, outputTokens: 57, totalUSD: 0.0042 }); + }); + + it('emits nothing terminal for a step that merely ended in tool calls', () => { + // `finish: "tool-calls"` is a step boundary MID-turn — the model stopped to run a tool and will + // continue. Treating it as the end of the turn would cut every tool-using conversation in half. + const out = mapServeEvent({ + type: 'session.next.step.ended', + data: { sessionID: SESSION, finish: 'tool-calls', cost: 0, tokens: { input: 3, output: 57 } }, + } as never); + + expect(out?.events).toEqual([]); + }); + + it('treats missing tokens and cost as zero rather than NaN', () => { + const out = mapServeEvent({ type: 'session.next.step.ended', data: { sessionID: SESSION } } as never); + expect(out?.costDelta).toEqual({ inputTokens: 0, outputTokens: 0, totalUSD: 0 }); + }); + + it('surfaces a failed step as an error carrying the provider message', () => { + // The real one, from a probe where the provider was down. + const out = mapServeEvent({ + type: 'session.next.step.failed', + data: { sessionID: SESSION, error: { type: 'unknown', message: 'Provider request failed with HTTP 503' } }, + } as never); + + expect(out?.events).toEqual([{ type: 'error', message: 'Provider request failed with HTTP 503' }]); + }); +}); + +describe('mapServeEvent — envelope', () => { + it('carries the durable cursor when reading the durable stream, and not otherwise', () => { + // The per-session stream stamps `durable.seq`; the global live stream does not. The caller needs + // that difference to know what it may commit and replay. + const durable = mapServeEvent({ + type: 'session.next.text.ended', + durable: { seq: 12 }, + data: { sessionID: SESSION, text: 'hi' }, + } as never); + const live = mapServeEvent({ type: 'session.next.text.ended', data: { sessionID: SESSION, text: 'hi' } } as never); + + expect(durable?.seq).toBe(12); + expect(live?.seq).toBeUndefined(); + }); + + it('always reports the session id, because the live stream is global', () => { + // Not a detail: `/api/event` carries EVERY session's events, so a consumer that forgets to filter + // splices one conversation into another. + for (const type of ['session.next.text.delta', 'session.next.tool.called', 'session.next.step.ended']) { + expect(mapServeEvent({ type, data: { sessionID: SESSION, delta: 'x', callID: CALL } } as never)?.sessionId).toBe( + SESSION, + ); + } + }); + + it('returns null for something that is not an event at all', () => { + expect(mapServeEvent(null)).toBeNull(); + expect(mapServeEvent(undefined)).toBeNull(); + expect(mapServeEvent({} as never)).toBeNull(); + }); + + it('is quiet about an unknown future type, but does not claim to know it', () => { + // Forward compatibility with a silent tell: an unrecognised type produces no events, and + // `isKnownServeEvent` is how a caller notices a release added something worth mapping. + expect(mapServeEvent({ type: 'session.next.something_new_in_1_19', data: {} } as never)?.events).toEqual([]); + expect(isKnownServeEvent('session.next.something_new_in_1_19')).toBe(false); + expect(isKnownServeEvent('session.next.text.delta')).toBe(true); + expect(isKnownServeEvent(undefined)).toBe(false); + }); +}); diff --git a/src/servers/sidecar/opencode/serve-events.ts b/src/servers/sidecar/opencode/serve-events.ts new file mode 100644 index 00000000..93f39fdc --- /dev/null +++ b/src/servers/sidecar/opencode/serve-events.ts @@ -0,0 +1,221 @@ +import type { ChatEvent } from '../../api/chat/types'; + +// `session.next.*` → ChatEvent. Phase A of the serve migration (docs/opencode-serve-migration-plan.md). +// +// NOTHING ROUTES THROUGH THIS YET. Turns still run as `opencode run` subprocesses via runner.ts; this is +// the mapping half of the serve path, written and pinned first so the switch-over is not also the moment +// the parsing is discovered to be wrong. +// +// ── The two streams, which is the part worth understanding ── +// +// The serve publishes the same turn twice, and the split maps exactly onto what officer already does for +// Claude: +// +// • `GET /api/session/{id}/event?after=` — DURABLE, per session, replayable. Every event carries +// `durable.seq`. Carries whole values (`text.ended` with the full text) and NO deltas. This is the +// transcript: what belongs in `chat_session_events`, and what a reconnecting browser replays. +// +// • `GET /api/event` — LIVE, GLOBAL, ephemeral. Carries the deltas +// (`text.delta`, `tool.input.delta`) and no cursor. This is what makes text appear as it is typed. +// +// Measured, not inferred: the same turn produced 13 events on the durable stream and 21 on the live one, +// the difference being 3 `text.delta` and 5 `tool.input.delta`. Reading only the durable stream — which +// is what I did first — makes it look like the serve cannot stream at all, and would have quietly killed +// the main reason for migrating. +// +// The live stream being GLOBAL is the real cost of this design: it carries every session's events, so a +// consumer must filter on `sessionID` and cannot assume it owns the socket. +// +// ── Fixture provenance ── +// +// Every shape here was captured from opencode 1.18.16 running a real turn that used the bash tool and +// then answered in prose. See `serve-events.test.ts`, whose fixtures are verbatim captures. + +/** The envelope every `session.next.*` event arrives in. `durable` is present only on the durable stream. */ +export type ServeEvent = { + type?: string; + durable?: { seq?: number }; + data?: { + sessionID?: string; + messageID?: string; + assistantMessageID?: string; + callID?: string; + textID?: string; + name?: string; + tool?: string; + input?: Record; + delta?: string; + text?: string; + content?: { type?: string; text?: string }[]; + structured?: Record; + error?: { message?: string; type?: string }; + finish?: string; + cost?: number; + tokens?: { input?: number; output?: number }; + model?: { id?: string; providerID?: string }; + }; +}; + +export type MappedServeEvent = { + /** The session this belongs to. The live stream is global, so a consumer MUST filter on it. */ + sessionId?: string; + /** The durable cursor, when this came from the durable stream. Absent on live events. */ + seq?: number; + events: ChatEvent[]; + costDelta: { inputTokens: number; outputTokens: number; totalUSD: number }; +}; + +const NO_COST = { inputTokens: 0, outputTokens: 0, totalUSD: 0 }; + +/** `tool.success` returns content blocks; officer's `tool:result` wants one string. */ +function textOfContent(content: { type?: string; text?: string }[] | undefined): string { + if (!Array.isArray(content)) return ''; + return content + .map((c) => (typeof c?.text === 'string' ? c.text : '')) + .filter(Boolean) + .join('\n'); +} + +/** + * One serve event in, whatever officer should do about it out. + * + * Pure by construction, exactly like `mapRunLine`: no accumulation, no memory, no I/O. The caller owns + * anything that spans events — summing cost across steps, and deciding whether it is reading the live or + * the durable stream. + * + * Returns `null` for an event that is not ours to interpret (another session's, or an envelope with no + * type). Everything recognised but deliberately ignored returns an empty `events` array instead, so + * "we saw it and had nothing to say" stays distinguishable from "we did not understand it". + */ +export function mapServeEvent(evt: ServeEvent | null | undefined): MappedServeEvent | null { + if (!evt || typeof evt.type !== 'string') return null; + + const d = evt.data ?? {}; + const base = { sessionId: d.sessionID, seq: evt.durable?.seq }; + const nothing = (): MappedServeEvent => ({ ...base, events: [], costDelta: NO_COST }); + + switch (evt.type) { + // ── Text ── + case 'session.next.text.delta': { + // Live stream only. `delta` is the increment, never the whole value. + const text = d.delta; + if (typeof text !== 'string' || text.length === 0) return nothing(); + return { ...base, events: [{ type: 'delta', text }], costDelta: NO_COST }; + } + case 'session.next.text.ended': { + // The committed block. Appears on BOTH streams, which is intended: officer's client treats `text` + // as the value that supersedes whatever the deltas built up, the same as the Claude path. + const text = d.text; + if (typeof text !== 'string' || text.length === 0) return nothing(); + return { ...base, events: [{ type: 'text', text }], costDelta: NO_COST }; + } + + // ── Tools ── + // + // `tool:start` is emitted on `tool.called`, NOT on `tool.input.started`, because only `tool.called` + // carries the resolved `input` object. `tool.input.started` knows the name but the arguments are + // still being streamed a few characters at a time (`tool.input.delta`: `{"comman`), and a tool row + // rendered with half-parsed JSON as its arguments is worse than one that appears a moment later. + case 'session.next.tool.called': { + const toolCallId = d.callID; + if (!toolCallId) return nothing(); + return { + ...base, + events: [ + { + type: 'tool:start', + toolCallId, + toolName: d.tool ?? d.name ?? 'tool', + toolInput: (d.input as Record) ?? {}, + }, + ], + costDelta: NO_COST, + }; + } + case 'session.next.tool.success': { + const toolCallId = d.callID; + if (!toolCallId) return nothing(); + return { + ...base, + events: [{ type: 'tool:result', toolCallId, output: textOfContent(d.content), isError: false }], + costDelta: NO_COST, + }; + } + case 'session.next.tool.failed': { + const toolCallId = d.callID; + if (!toolCallId) return nothing(); + // The error replaces the output rather than sitting beside it — same call the run path makes. + const message = d.error?.message ?? textOfContent(d.content) ?? ''; + return { + ...base, + events: [{ type: 'tool:result', toolCallId, output: String(message), isError: true }], + costDelta: NO_COST, + }; + } + + // ── Step accounting ── + case 'session.next.step.ended': { + // Per-step tokens and cost, to be summed by the caller. `finish: "tool-calls"` is a step boundary + // mid-turn, not the end of the turn — the turn has ended when the prompt is fully answered, which + // is a caller-level question, so nothing terminal is emitted here. + const t = d.tokens; + return { + ...base, + events: [], + costDelta: { + inputTokens: t?.input ?? 0, + outputTokens: t?.output ?? 0, + totalUSD: typeof d.cost === 'number' ? d.cost : 0, + }, + }; + } + case 'session.next.step.failed': { + const message = d.error?.message ?? 'OpenCode step failed'; + return { ...base, events: [{ type: 'error', message }], costDelta: NO_COST }; + } + + // ── Seen and deliberately silent ── + // + // Named rather than swept into `default` so that a genuinely NEW event type still lands in the + // unknown bucket, where it can be noticed. + case 'session.next.prompt.admitted': + case 'session.next.prompted': + case 'session.next.step.started': + case 'session.next.text.started': + case 'session.next.tool.input.started': + case 'session.next.tool.input.delta': + case 'session.next.tool.input.ended': + case 'session.next.tool.progress': + case 'session.next.model.switched': + case 'session.next.agent.switched': + case 'server.connected': + return nothing(); + + default: + return nothing(); + } +} + +/** Whether a type is one this mapper knows about — for logging what a new opencode release added. */ +export const isKnownServeEvent = (type: string | undefined): boolean => + typeof type === 'string' && (type === 'server.connected' || KNOWN.has(type)); + +const KNOWN = new Set([ + 'session.next.text.delta', + 'session.next.text.ended', + 'session.next.text.started', + 'session.next.tool.called', + 'session.next.tool.success', + 'session.next.tool.failed', + 'session.next.tool.input.started', + 'session.next.tool.input.delta', + 'session.next.tool.input.ended', + 'session.next.tool.progress', + 'session.next.step.started', + 'session.next.step.ended', + 'session.next.step.failed', + 'session.next.prompt.admitted', + 'session.next.prompted', + 'session.next.model.switched', + 'session.next.agent.switched', +]);