phase A: map the serve event stream, routed nowhere

The mapping half of the serve migration, written and pinned before anything depends on it, so the switch-over is not also the moment the parsing turns out to be wrong. Nothing routes through this — turns are still opencode run subprocesses, and the claude path is untouched. The finding that matters: the serve publishes each turn TWICE, and reading the wrong one makes it look like it cannot stream at all. /api/session/{id}/event?after= durable, per session, replayable, durable.seq on every event, whole values only, NO deltas /api/event live, GLOBAL, ephemeral, carries text.delta and tool.input.delta, no cursor Same turn: 13 events durable, 21 live, the difference being 3 text.delta and 5 tool.input.delta. I probed the per-session one first and nearly recorded "no streaming" as a fact — it would have removed the main reason to migrate. The split maps exactly onto what officer already does for claude: durable to chat_session_events, live to UI deltas. The cost is that the live stream is global, so a consumer must filter on sessionID. tool:start is emitted on tool.called, not tool.input.started, because only tool.called has the resolved input object — the input arrives as JSON fragments ({"comman) and a tool row rendered with half-parsed arguments is worse than one that appears a moment later. step.ended with finish tool-calls is a step boundary MID-turn, not the end of the turn, so nothing terminal is emitted for it. Treating it as the end would cut every tool-using conversation in half. Fixtures are verbatim captures from 1.18.16. Replaying both real streams through the mapper reconstructs the turn identically from each, with the reassembled deltas exactly equal to the committed text and identical cost, and zero unrecognised events. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-10 18:33:33 +01:00
parent 38ebe168f0
commit feb9010097
3 changed files with 442 additions and 9 deletions
@@ -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);
+  });
+});
@@ -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=<seq>`  — 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<string, unknown>;
+    delta?: string;
+    text?: string;
+    content?: { type?: string; text?: string }[];
+    structured?: Record<string, unknown>;
+    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<string, unknown>) ?? {},
+          },
+        ],
+        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',
+]);