make the agent sidecar the writer of record for chat output

officer's registration socket silently drops sends when it isn't OPEN
(sidecar/connect.ts:send — no queue, no error, no return value). the agent pushed
raw parser events over that socket and officer translated and persisted them, so
everything a turn produced while officer was restarting went nowhere: the turn kept
running, the output was gone, and a reconnecting client replayed a log that simply
had no rows for those seconds. stage 1 kept the agent alive across a restart; this
is what makes its output survive one too.

move the translation and the write into the sidecar:

- turn-stream.ts is the stateful ChatEvent -> browser-message translator lifted out
  of websocket.ts (delta buffering, flush before tool:start and result). pure and
  synchronous, so it is unit tested — 12 tests, 100% lines.
- session-log.ts commits each message to chat_session_events and only then hands it
  to officer, with its cursor id attached. per-session promise chain: translation is
  synchronous and therefore in arrival order, and only the commit is queued, so
  cursor ids are assigned in the order events actually happened. a delta that
  overtook the assistant:text in front of it would make the client commit its stream
  buffer at the wrong point, so deltas go through the same queue even though they are
  never written.
- claude:event on the wire becomes claude:message: a finished browser-facing message
  plus its seq. officer relays it verbatim and folds it into the in-memory session
  for sync:messages. it no longer builds or persists chat messages for this harness.

gap detection, which is what the durable log is for. chat_session_events.id is a
global bigserial, so two consecutive events of one session are not consecutive ids
and a client cannot tell a contiguous replay from one with a hole in it. each durable
message now carries prevSeq — the cursor of the previous message in the same session —
which is inside the persisted payload, so it survives replay. useChat compares it
against the cursor it holds before advancing, and surfaces a visible marker on a
mismatch: a conversation that silently skips a tool call or half an answer reads as
the assistant having done something inexplicable. only checked once a cursor exists,
because opening a session from history legitimately starts mid-chain (events are swept
after 7 days, the transcript is not).

a failed write delivers live with no seq, so the client sees the message but does not
advance past something it cannot replay, and the next successful write chains from the
cursor the client still holds.

pipeline steps pass durable: false. their sessionKey is a throwaway uuid no browser
will ever replay and the job's own event log is its record, so writing those rows only
grows the table.

opencode still goes through officer's createEventHandler, now labelled as such. that
is the sidecars-opencode branch.

this fixes R4 from CLAUDE_SIDECAR_ISOLATION.md. R3 and R5 already worked.

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

src/databases/officer_db/src/index.ts

@@ -78,7 +78,12 @@ export {
   markInterruptedJobs,
 } from './queries/pipeline-jobs';
 
-export { appendChatEvent, getChatEventsSince, pruneChatEventsOlderThan } from './queries/chat-events';
+export {
+  appendChatEvent,
+  getChatEventsSince,
+  getLastChatEventSeq,
+  pruneChatEventsOlderThan,
+} from './queries/chat-events';
 
 export {
   getMusicFavorites,

src/databases/officer_db/src/queries/chat-events.ts

@@ -1,13 +1,10 @@
-import { eq, and, gt, asc, lt } from 'drizzle-orm';
+import { eq, and, gt, asc, desc, lt } from 'drizzle-orm';
 import { db } from '../db';
 import { chatSessionEvents } from '../schema';
 
 /** Append one outbound event to a session's durable log; returns its global cursor id. */
 export async function appendChatEvent(sessionId: string, event: unknown): Promise<number> {
-  const [row] = await db
-    .insert(chatSessionEvents)
-    .values({ sessionId, event })
-    .returning({ id: chatSessionEvents.id });
+  const [row] = await db.insert(chatSessionEvents).values({ sessionId, event }).returning({ id: chatSessionEvents.id });
   return row!.id;
 }
 
@@ -23,6 +20,21 @@ export async function getChatEventsSince(
     .orderBy(asc(chatSessionEvents.id));
 }
 
+/**
+ * The newest cursor id for one session, or undefined if it has no events yet. The writer uses this to
+ * pick its `prevSeq` chain back up after its own restart, so a client can still tell a contiguous
+ * replay from one with a hole in it.
+ */
+export async function getLastChatEventSeq(sessionId: string): Promise<number | undefined> {
+  const [row] = await db
+    .select({ id: chatSessionEvents.id })
+    .from(chatSessionEvents)
+    .where(eq(chatSessionEvents.sessionId, sessionId))
+    .orderBy(desc(chatSessionEvents.id))
+    .limit(1);
+  return row?.id;
+}
+
 /** Retention: drop events older than the cutoff (called periodically). */
 export async function pruneChatEventsOlderThan(cutoff: Date): Promise<void> {
   await db.delete(chatSessionEvents).where(lt(chatSessionEvents.createdAt, cutoff));

src/servers/api/chat/types.ts

@@ -138,6 +138,27 @@ export type ServerMessage =
   | { type: 'task:started'; taskId: string; description: string; taskType?: string }
   | { type: 'task:notification'; taskId: string; status: 'completed' | 'failed' | 'stopped'; summary: string };
 
+// The turn-output subset of ServerMessage — everything the agent sidecar produces on its own. The
+// remaining members (session:init, sync:messages, disconnected, connection-level errors) are officer's:
+// they describe the browser's connection, not the turn.
+//
+// The sidecar builds these, commits them to chat_session_events, and hands officer a finished message
+// plus its cursor id; officer relays it verbatim. `prevSeq` is the writer's continuity claim — the
+// cursor of the previous durable message in the same session — which lets a reconnecting client tell a
+// contiguous replay from one with a hole in it. Absent when the writer cannot vouch for it.
+export type TurnMessageType =
+  | 'assistant:delta'
+  | 'assistant:text'
+  | 'tool:start'
+  | 'tool:result'
+  | 'result'
+  | 'error'
+  | 'stopped'
+  | 'task:started'
+  | 'task:notification';
+
+export type TurnMessage = Extract<ServerMessage, { type: TurnMessageType }> & { prevSeq?: number };
+
 export type ChatEvent =
   | { type: 'text'; text: string }
   | { type: 'delta'; text: string }

src/servers/api/chat/websocket.ts

@@ -1,6 +1,6 @@
 import type { ServerWebSocket } from 'bun';
 import { randomUUID } from 'crypto';
-import type { ClientMessage, ServerMessage, Message, ChatEvent } from './types';
+import type { ClientMessage, ServerMessage, Message, ChatEvent, TurnMessage, UserSession } from './types';
 import { sessionManager } from './session-manager';
 import { sendClaudeCodeStreaming } from '@@/channels/send-claude-code';
 import { sendOpenCodeStreaming } from '@@/channels/send-opencode';
@@ -165,6 +165,82 @@ export function close(ws: ServerWebSocket<WSData>): void {
   }
 }
 
+// ── Claude Code: relay, don't rebuild ──
+
+// The agent sidecar owns Claude's turn output end to end — it translates the parser stream, commits each
+// message to chat_session_events and hands us a finished message plus its cursor id. Officer relays it.
+// That is what makes a restart survivable: the durable record no longer travels over the socket between
+// the two processes, so if this one is down the output is already written and the client replays it.
+//
+// Officer keeps only the in-memory transcript, which exists to answer a `resume` with sync:messages —
+// Claude's own transcript is the real record — so it is folded from the same messages, not rebuilt.
+function foldIntoSession(session: UserSession, msg: TurnMessage, model: string): void {
+  switch (msg.type) {
+    case 'assistant:delta':
+      session.streamBuffer += msg.text;
+      break;
+
+    case 'assistant:text':
+      session.messages.push({ id: randomUUID(), timestamp: Date.now(), role: 'assistant', text: msg.text, model });
+      session.meta.messageCount += 1;
+      session.streamBuffer = '';
+      break;
+
+    case 'tool:start':
+      session.messages.push({
+        id: randomUUID(),
+        timestamp: Date.now(),
+        role: 'tool',
+        toolCallId: msg.toolCallId,
+        toolName: msg.toolName,
+        toolInput: msg.toolInput,
+      });
+      session.meta.messageCount += 1;
+      break;
+
+    case 'tool:result':
+      for (let i = session.messages.length - 1; i >= 0; i--) {
+        const m = session.messages[i]!;
+        if (m.role === 'tool' && m.toolCallId === msg.toolCallId) {
+          m.output = msg.output;
+          m.isError = msg.isError;
+          break;
+        }
+      }
+      break;
+
+    case 'result': {
+      session.isGenerating = false;
+      session.meta.cost.inputTokens += msg.cost.inputTokens;
+      session.meta.cost.outputTokens += msg.cost.outputTokens;
+      session.meta.cost.totalUSD += msg.cost.totalUSD;
+      session.meta.updatedAt = Date.now();
+      // The turn's cost belongs to the assistant message it paid for (as it did when officer built these).
+      const last = session.messages[session.messages.length - 1];
+      if (last?.role === 'assistant' && !last.cost) last.cost = msg.cost;
+      break;
+    }
+
+    case 'error':
+    case 'stopped':
+      session.isGenerating = false;
+      break;
+  }
+}
+
+function createClaudeMessageHandler(sessionId: string, model: string) {
+  return (msg: TurnMessage, seq?: number): void => {
+    const session = sessionManager.getSession(sessionId);
+    if (!session) return;
+    foldIntoSession(session, msg, model);
+    sendToClient(session.ws as ServerWebSocket<WSData> | null, msg, seq);
+  };
+}
+
+// ── OpenCode: officer still translates and persists ──
+
+// Unchanged from before the split, and still correct for OpenCode: that sidecar reports raw ChatEvents,
+// so officer does the translation and owns the durable write. Moving it is the `sidecars-opencode` branch.
 function createEventHandler(sessionId: string, model: string, cwd: string) {
   return async (event: ChatEvent): Promise<void> => {
     const session = sessionManager.getSession(sessionId);
@@ -298,13 +374,23 @@ function createEventHandler(sessionId: string, model: string, cwd: string) {
 
       case 'task:started': {
         // Background task launched (run_in_background / Monitor). Independent of turn state.
-        await emitToSession(sessionId, { type: 'task:started', taskId: event.taskId, description: event.description, taskType: event.taskType });
+        await emitToSession(sessionId, {
+          type: 'task:started',
+          taskId: event.taskId,
+          description: event.description,
+          taskType: event.taskType,
+        });
         break;
       }
 
       case 'task:notification': {
         // Background task finished — arrives AFTER the turn's 'result'; the whole point of the fix.
-        await emitToSession(sessionId, { type: 'task:notification', taskId: event.taskId, status: event.status, summary: event.summary });
+        await emitToSession(sessionId, {
+          type: 'task:notification',
+          taskId: event.taskId,
+          status: event.status,
+          summary: event.summary,
+        });
         break;
       }
     }
@@ -401,7 +487,7 @@ async function handleClaudeCodeChat(
 
   session.isGenerating = true;
 
-  const onEvent = createEventHandler(sessionId, model, cwd);
+  const onMessage = createClaudeMessageHandler(sessionId, model);
 
   try {
     if (!session._claudeKill) {
@@ -416,7 +502,7 @@ async function handleClaudeCodeChat(
         cwd,
         model,
         resumeSessionId: msg.resumeSessionId,
-        onEvent,
+        onMessage,
       });
       session.piProcess = sessionId as any;
       session._claudeKill = handle.kill;

src/servers/api/tasks/pipeline-executor.ts

@@ -8,7 +8,7 @@ import { getTaskByDirName } from './task-files';
 import { getHomeDir } from '../../data-path';
 import { resolveBaseCwd } from '../chat/websocket';
 import { sendClaudeCodeStreaming } from '../../channels/send-claude-code';
-import type { ChatEvent, MessageCost } from '../chat/types';
+import type { TurnMessage, MessageCost } from '../chat/types';
 import * as jobManager from './pipeline-job-manager';
 
 const DEFAULT_MODEL = 'claude-code';
@@ -148,23 +148,26 @@ async function runAgenticStep({
       fn();
     };
 
-    const onEvent = (event: ChatEvent) => {
+    // The agent sidecar now hands over finished turn messages rather than raw parser events, so this is a
+    // re-label onto the pipeline's own event stream. The `seq` is ignored: pipeline steps have their own
+    // durable record (the job's events), not the chat cursor.
+    const onMessage = (msg: TurnMessage) => {
       if (abortSignal.aborted) return;
       lastActivity = Date.now();
 
-      switch (event.type) {
-        case 'delta':
-          emit({ type: 'assistant:delta', text: event.text, stepIndex, iterationLabel });
+      switch (msg.type) {
+        case 'assistant:delta':
+          emit({ type: 'assistant:delta', text: msg.text, stepIndex, iterationLabel });
           break;
-        case 'text':
-          emit({ type: 'assistant:text', text: event.text, stepIndex, iterationLabel });
+        case 'assistant:text':
+          emit({ type: 'assistant:text', text: msg.text, stepIndex, iterationLabel });
           break;
         case 'tool:start':
           emit({
             type: 'tool:start',
-            toolCallId: event.toolCallId,
-            toolName: event.toolName,
-            toolInput: event.toolInput,
+            toolCallId: msg.toolCallId,
+            toolName: msg.toolName,
+            toolInput: msg.toolInput,
             stepIndex,
             iterationLabel,
           });
@@ -172,25 +175,29 @@ async function runAgenticStep({
         case 'tool:result':
           emit({
             type: 'tool:result',
-            toolCallId: event.toolCallId,
-            output: event.output,
-            isError: event.isError,
+            toolCallId: msg.toolCallId,
+            output: msg.output,
+            isError: msg.isError,
             stepIndex,
             iterationLabel,
           });
           break;
-        case 'result':
+        case 'result': {
+          const cost = msg.cost;
           settle(() => {
             cleanup?.();
-            resolve(event.cost);
+            resolve(cost);
           });
           break;
-        case 'error':
+        }
+        case 'error': {
+          const message = msg.message;
           settle(() => {
             cleanup?.();
-            reject(new Error(event.message));
+            reject(new Error(message));
           });
           break;
+        }
         case 'stopped':
           settle(() => {
             cleanup?.();
@@ -236,7 +243,8 @@ async function runAgenticStep({
         sessionKey: sessionId,
         cwd,
         model,
-        onEvent,
+        durable: false,
+        onMessage,
       });
       cleanup = handle.kill;
     } catch (err) {

src/servers/channels/send-claude-code.ts

@@ -1,5 +1,5 @@
 import { logger } from '@@/api/chat/logger';
-import type { MessageCost, ChatEvent } from '@@/api/chat/types';
+import type { MessageCost, TurnMessage } from '@@/api/chat/types';
 import * as sidecar from '@@/sidecar-registry';
 
 type ClaudeCodeParams = {
@@ -38,7 +38,9 @@ type ClaudeCodeStreamingParams = {
   cwd?: string;
   model?: string;
   resumeSessionId?: string;
-  onEvent: (event: ChatEvent) => void;
+  durable?: boolean;
+  // Finished turn messages, already committed by the sidecar; `seq` is the cursor id to deliver them under.
+  onMessage: (msg: TurnMessage, seq?: number) => void;
 };
 
 type ClaudeCodeStreamingHandle = {
@@ -46,15 +48,15 @@ type ClaudeCodeStreamingHandle = {
 };
 
 export async function sendClaudeCodeStreaming(params: ClaudeCodeStreamingParams): Promise<ClaudeCodeStreamingHandle> {
-  const { onEvent, ...spawnParams } = params;
+  const { onMessage, ...spawnParams } = params;
 
   logger.info('Claude Code streaming exec (via sidecar)', { sessionKey: params.sessionKey });
 
   // Session-scoped subscription. The persistent session outlives the turn, so background task events
   // (task:notification) arrive AFTER 'result' — do NOT unsubscribe on a terminal turn event; only on
   // an explicit kill/teardown (the returned handle, called from deleteSession/disconnect).
-  const unsub = sidecar.onClaudeEvent((sessionKey, event) => {
-    if (sessionKey === params.sessionKey) onEvent(event);
+  const unsub = sidecar.onClaudeMessage((sessionKey, msg, seq) => {
+    if (sessionKey === params.sessionKey) onMessage(msg, seq);
   });
 
   await sidecar.spawnClaudeStreaming(spawnParams);

src/servers/sidecar-registry.ts

@@ -13,7 +13,7 @@ import type {
   VncSessionInfo,
 } from './sidecar/protocol';
 import type { SidecarRegistration } from './sidecar/registration-protocol';
-import type { ChatEvent } from './api/chat/types';
+import type { ChatEvent, TurnMessage } from './api/chat/types';
 
 // ── Types ──
 
@@ -274,14 +274,14 @@ export function clearClaudeSession(sessionKey: string): void {
   sendFire('claude', { type: 'claude:clear-session', id: nextId(), sessionKey });
 }
 
-export function onClaudeEvent(handler: (sessionKey: string, event: ChatEvent) => void): () => void {
-  return on('claude:event', (msg) => {
-    if (msg.type === 'claude:event') {
-      handler(
-        (msg as SidecarEvent & { type: 'claude:event' }).sessionKey,
-        (msg as SidecarEvent & { type: 'claude:event' }).event,
-      );
-    }
+// Turn output arrives finished and already durable: the agent translated it and committed it to
+// chat_session_events, and `seq` is its cursor id there. Officer relays it — it no longer builds or
+// persists chat messages for this harness.
+export function onClaudeMessage(handler: (sessionKey: string, msg: TurnMessage, seq?: number) => void): () => void {
+  return on('claude:message', (ev) => {
+    if (ev.type !== 'claude:message') return;
+    const msg = ev as SidecarEvent & { type: 'claude:message' };
+    handler(msg.sessionKey, msg.msg, msg.seq);
   });
 }
 

src/servers/sidecar/claude/session-log.test.ts

@@ -0,0 +1,182 @@
+import { describe, test, expect } from 'bun:test';
+import { createSessionLogStore, type Delivery, type EventWriter } from './session-log';
+import type { ChatEvent, MessageCost } from '../../api/chat/types';
+
+const COST: MessageCost = { inputTokens: 1, outputTokens: 2, totalUSD: 0.0001 };
+
+/**
+ * A writer that finishes its appends in the WORST possible order — the last one first — so the test
+ * fails unless the store serialises commits itself. `lastSeq` starts empty (a fresh session) unless a
+ * seed is given.
+ */
+function reverseOrderWriter(opts: { seed?: number; failOn?: (n: number) => boolean } = {}) {
+  const pending: Array<{ resolve: (seq: number) => void; reject: (err: Error) => void }> = [];
+  let next = 100;
+  const writes: Array<{ sessionId: string; type: string; prevSeq?: number }> = [];
+
+  const writer: EventWriter = {
+    lastSeq: async () => opts.seed,
+    append: (sessionId, msg) => {
+      writes.push({ sessionId, type: msg.type, prevSeq: msg.prevSeq });
+      return new Promise<number>((resolve, reject) => pending.push({ resolve, reject }));
+    },
+  };
+
+  // Settle everything queued so far, newest first.
+  function settleAll() {
+    const batch = pending.splice(0).reverse();
+    for (const p of batch) {
+      const n = next++;
+      if (opts.failOn?.(n)) p.reject(new Error(`write ${n} failed`));
+      else p.resolve(n);
+    }
+  }
+
+  return { writer, settleAll, writes, pendingCount: () => pending.length };
+}
+
+function collector() {
+  const got: Delivery[] = [];
+  return { got, deliver: (d: Delivery) => got.push(d) };
+}
+
+/** Let the store's promise chain drain, settling writes as they queue up. */
+async function drain(settleAll: () => void, pendingCount: () => number) {
+  for (let i = 0; i < 50; i++) {
+    await Promise.resolve();
+    if (pendingCount() > 0) settleAll();
+  }
+}
+
+const TURN: ChatEvent[] = [
+  { type: 'delta', text: 'Let me look.' },
+  { type: 'tool:start', toolCallId: 't1', toolName: 'Read', toolInput: { file: 'a.ts' } },
+  { type: 'tool:result', toolCallId: 't1', output: 'ok', isError: false },
+  { type: 'delta', text: 'Found it.' },
+  { type: 'result', cost: COST },
+];
+
+describe('createSessionLogStore', () => {
+  test('deliveries keep push order even when the writes finish backwards', async () => {
+    const { got, deliver } = collector();
+    const { writer, settleAll, pendingCount } = reverseOrderWriter();
+    const store = createSessionLogStore(deliver, writer);
+
+    for (const e of TURN) store.push('s1', e);
+    await drain(settleAll, pendingCount);
+
+    expect(got.map((d) => d.msg.type)).toEqual([
+      'assistant:delta',
+      'assistant:text',
+      'tool:start',
+      'tool:result',
+      'assistant:delta',
+      'assistant:text',
+      'result',
+    ]);
+  });
+
+  test('cursors ascend and each durable message chains to the one before it', async () => {
+    const { got, deliver } = collector();
+    const { writer, settleAll, pendingCount } = reverseOrderWriter();
+    const store = createSessionLogStore(deliver, writer);
+
+    for (const e of TURN) store.push('s1', e);
+    await drain(settleAll, pendingCount);
+
+    const durable = got.filter((d) => d.seq !== undefined);
+    expect(durable).toHaveLength(5);
+
+    const seqs = durable.map((d) => d.seq!);
+    expect([...seqs].sort((a, b) => a - b)).toEqual(seqs);
+
+    // First durable message of a fresh session makes no continuity claim; the rest point at their predecessor.
+    expect(durable[0]!.msg.prevSeq).toBeUndefined();
+    for (let i = 1; i < durable.length; i++) {
+      expect(durable[i]!.msg.prevSeq).toBe(durable[i - 1]!.seq);
+    }
+  });
+
+  test('a session that predates this process picks the chain up from the stored cursor', async () => {
+    const { got, deliver } = collector();
+    const { writer, settleAll, pendingCount } = reverseOrderWriter({ seed: 42 });
+    const store = createSessionLogStore(deliver, writer);
+
+    store.push('s1', { type: 'text', text: 'after a restart' });
+    await drain(settleAll, pendingCount);
+
+    expect(got).toHaveLength(1);
+    expect(got[0]!.msg.prevSeq).toBe(42);
+  });
+
+  test('deltas are delivered without a cursor', async () => {
+    const { got, deliver } = collector();
+    const { writer, settleAll, pendingCount } = reverseOrderWriter();
+    const store = createSessionLogStore(deliver, writer);
+
+    store.push('s1', { type: 'delta', text: 'hi' });
+    await drain(settleAll, pendingCount);
+
+    expect(got).toEqual([{ sessionId: 's1', msg: { type: 'assistant:delta', text: 'hi' } }]);
+  });
+
+  test('durable:false skips the writer entirely but still delivers everything', async () => {
+    const { got, deliver } = collector();
+    const { writer, settleAll, writes, pendingCount } = reverseOrderWriter();
+    const store = createSessionLogStore(deliver, writer);
+
+    for (const e of TURN) store.push('job-uuid', e, false);
+    await drain(settleAll, pendingCount);
+
+    expect(writes).toHaveLength(0);
+    expect(got).toHaveLength(7);
+    expect(got.every((d) => d.seq === undefined)).toBe(true);
+  });
+
+  test('a failed write delivers live without a cursor and leaves the chain intact', async () => {
+    const { got, deliver } = collector();
+    // Fail the second successful-cursor slot, i.e. one write in the middle of the turn.
+    const { writer, settleAll, pendingCount } = reverseOrderWriter({ failOn: (n) => n === 101 });
+    const store = createSessionLogStore(deliver, writer);
+
+    store.push('s1', { type: 'text', text: 'one' });
+    await drain(settleAll, pendingCount);
+    store.push('s1', { type: 'text', text: 'two' });
+    await drain(settleAll, pendingCount);
+    store.push('s1', { type: 'text', text: 'three' });
+    await drain(settleAll, pendingCount);
+
+    expect(got.map((d) => d.msg.type)).toEqual(['assistant:text', 'assistant:text', 'assistant:text']);
+
+    const [first, failed, third] = got as [Delivery, Delivery, Delivery];
+    expect(first.seq).toBe(100);
+    // The message the client cannot replay carries neither a cursor nor a continuity claim.
+    expect(failed.seq).toBeUndefined();
+    // ...and the next write chains from the cursor the client actually still holds, not from the hole.
+    expect(third.msg.prevSeq).toBe(first.seq);
+    expect(third.seq).toBe(102);
+  });
+
+  test('sessions are independent, and drop forgets a session', async () => {
+    const { got, deliver } = collector();
+    const { writer, settleAll, pendingCount } = reverseOrderWriter();
+    const store = createSessionLogStore(deliver, writer);
+
+    store.push('a', { type: 'delta', text: 'from-a' });
+    store.push('b', { type: 'delta', text: 'from-b' });
+    store.push('a', { type: 'result', cost: COST });
+    await drain(settleAll, pendingCount);
+
+    // 'a' flushed only its own buffer.
+    const texts = got.filter((d) => d.msg.type === 'assistant:text');
+    expect(texts).toHaveLength(1);
+    expect(texts[0]).toMatchObject({ sessionId: 'a', msg: { text: 'from-a' } });
+
+    // After a drop, a new turn on the same key starts from a clean buffer and re-reads the stored cursor.
+    store.drop('a');
+    got.length = 0;
+    store.push('a', { type: 'result', cost: COST });
+    await drain(settleAll, pendingCount);
+    expect(got.map((d) => d.msg.type)).toEqual(['result']);
+  });
+});

src/servers/sidecar/claude/session-log.ts

@@ -0,0 +1,121 @@
+import type { ChatEvent, TurnMessage } from '../../api/chat/types';
+import { appendChatEvent, getLastChatEventSeq } from 'officerdb';
+import { createTurnStream, type TurnOutput } from './turn-stream';
+
+// The agent sidecar is the writer of record for chat output.
+//
+// It used to push raw ChatEvents at officer over the registration socket and let officer translate and
+// persist them. That socket silently drops when officer is down (`connect.ts:send` — no queue, no
+// error), so everything the agent produced during a restart was lost: the turn kept running here and
+// its output went nowhere. Writing to Postgres here instead means an officer restart costs a replay
+// rather than the output, because the durable record no longer travels over the socket that died.
+//
+// Officer still gets every message live — it just gets it already written, with its cursor id attached,
+// and relays it verbatim.
+
+export type Delivery = {
+  sessionId: string;
+  msg: TurnMessage;
+  /** The cursor id under which this message is durable. Absent = ephemeral delta, or the write failed. */
+  seq?: number;
+};
+
+export type SessionLogStore = {
+  /**
+   * Translate one parser event and commit + deliver whatever it produces, strictly in order.
+   * `durable: false` skips the write (see ClaudeSpawnStreamingParams.durable) — the messages are still
+   * delivered, just without a cursor, because nothing will ever replay them.
+   */
+  push: (sessionId: string, event: ChatEvent, durable?: boolean) => void;
+  /** Forget a session's buffer and cursor chain (on kill / clear-session). */
+  drop: (sessionId: string) => void;
+};
+
+// The durable store, behind an interface so the ordering guarantee below can be tested against a writer
+// whose writes finish out of order. Defaults to Postgres.
+export type EventWriter = {
+  append: (sessionId: string, msg: TurnMessage) => Promise<number>;
+  lastSeq: (sessionId: string) => Promise<number | undefined>;
+};
+
+const postgresWriter: EventWriter = {
+  append: (sessionId, msg) => appendChatEvent(sessionId, msg),
+  lastSeq: (sessionId) => getLastChatEventSeq(sessionId),
+};
+
+type SessionLog = {
+  stream: ReturnType<typeof createTurnStream>;
+  /** Serialises commits so cursor ids are assigned in the order the events actually arrived. */
+  tail: Promise<void>;
+  lastSeq: number | undefined;
+  resolvedLastSeq: boolean;
+};
+
+export function createSessionLogStore(
+  deliver: (d: Delivery) => void,
+  writer: EventWriter = postgresWriter,
+): SessionLogStore {
+  const logs = new Map<string, SessionLog>();
+
+  function logFor(sessionId: string): SessionLog {
+    let log = logs.get(sessionId);
+    if (!log) {
+      log = {
+        stream: createTurnStream(sessionId),
+        tail: Promise.resolve(),
+        lastSeq: undefined,
+        resolvedLastSeq: false,
+      };
+      logs.set(sessionId, log);
+    }
+    return log;
+  }
+
+  async function commit(sessionId: string, log: SessionLog, out: TurnOutput, durable: boolean): Promise<void> {
+    // Deltas are live-only, but still go through the queue: a delta that overtook the `assistant:text`
+    // or `tool:start` in front of it would make the client commit its stream buffer at the wrong point.
+    if (!out.durable || !durable) {
+      deliver({ sessionId, msg: out.msg });
+      return;
+    }
+
+    // Pick the chain back up after a restart of this process, so `prevSeq` stays meaningful for a
+    // session that started before it. Once per session; failure just means no continuity claim.
+    if (!log.resolvedLastSeq) {
+      log.resolvedLastSeq = true;
+      try {
+        log.lastSeq = await writer.lastSeq(sessionId);
+      } catch (err) {
+        console.error(`[agent] could not read last event cursor for ${sessionId}:`, err);
+      }
+    }
+
+    const msg: TurnMessage = log.lastSeq === undefined ? out.msg : { ...out.msg, prevSeq: log.lastSeq };
+
+    try {
+      const seq = await writer.append(sessionId, msg);
+      log.lastSeq = seq;
+      deliver({ sessionId, msg, seq });
+    } catch (err) {
+      // The write IS the durability guarantee, so don't pretend. Deliver live without a cursor: the
+      // client sees the message but won't advance its cursor past something it cannot replay, and the
+      // next successful write chains from the last cursor the client actually holds.
+      console.error(`[agent] failed to persist chat event for ${sessionId}:`, err);
+      deliver({ sessionId, msg: out.msg });
+    }
+  }
+
+  return {
+    push(sessionId, event, durable = true) {
+      const log = logFor(sessionId);
+      // Translation is synchronous and therefore in arrival order; only the commit is queued.
+      for (const out of log.stream.push(event)) {
+        log.tail = log.tail.then(() => commit(sessionId, log, out, durable));
+      }
+    },
+
+    drop(sessionId) {
+      logs.delete(sessionId);
+    },
+  };
+}

src/servers/sidecar/claude/turn-stream.test.ts

@@ -0,0 +1,147 @@
+import { describe, test, expect } from 'bun:test';
+import { createTurnStream } from './turn-stream';
+import type { ChatEvent, MessageCost } from '../../api/chat/types';
+
+const SESSION = 'sess-1';
+const COST: MessageCost = { inputTokens: 10, outputTokens: 20, totalUSD: 0.001 };
+
+// Feed a whole event sequence through one stream and collect what came out, so the assertions read as
+// "this turn produced this transcript" rather than per-call plumbing.
+function run(events: ChatEvent[]) {
+  const stream = createTurnStream(SESSION);
+  const out = events.flatMap((e) => stream.push(e));
+  return {
+    all: out,
+    durable: out.filter((o) => o.durable).map((o) => o.msg),
+    types: out.map((o) => o.msg.type),
+  };
+}
+
+describe('createTurnStream', () => {
+  test('deltas are live-only and never durable', () => {
+    const { all, durable } = run([
+      { type: 'delta', text: 'he' },
+      { type: 'delta', text: 'llo' },
+    ]);
+    expect(all).toHaveLength(2);
+    expect(all.every((o) => o.durable)).toBe(false);
+    expect(durable).toHaveLength(0);
+  });
+
+  test('an explicit text event wins over the deltas that produced it', () => {
+    const { durable } = run([
+      { type: 'delta', text: 'par' },
+      { type: 'delta', text: 'tial' },
+      { type: 'text', text: 'partial and complete' },
+    ]);
+    expect(durable).toEqual([{ type: 'assistant:text', text: 'partial and complete' }]);
+  });
+
+  test('a text event with no text falls back to the accumulated buffer', () => {
+    const { durable } = run([
+      { type: 'delta', text: 'buffered' },
+      { type: 'text', text: '' },
+    ]);
+    expect(durable).toEqual([{ type: 'assistant:text', text: 'buffered' }]);
+  });
+
+  test('an empty text event with an empty buffer produces nothing', () => {
+    expect(run([{ type: 'text', text: '' }]).all).toHaveLength(0);
+  });
+
+  test('the buffer is flushed as one message before a tool call', () => {
+    const { durable } = run([
+      { type: 'delta', text: 'Let me ' },
+      { type: 'delta', text: 'check.' },
+      { type: 'tool:start', toolCallId: 't1', toolName: 'Read', toolInput: { file: 'a.ts' } },
+    ]);
+    expect(durable).toEqual([
+      { type: 'assistant:text', text: 'Let me check.' },
+      { type: 'tool:start', toolCallId: 't1', toolName: 'Read', toolInput: { file: 'a.ts' } },
+    ]);
+  });
+
+  test('the buffer is flushed before the turn result, and result carries the session id', () => {
+    const { durable } = run([
+      { type: 'delta', text: 'Done.' },
+      { type: 'result', cost: COST },
+    ]);
+    expect(durable).toEqual([
+      { type: 'assistant:text', text: 'Done.' },
+      { type: 'result', sessionId: SESSION, cost: COST },
+    ]);
+  });
+
+  test('a flushed buffer is not emitted twice', () => {
+    const { durable } = run([
+      { type: 'delta', text: 'once' },
+      { type: 'tool:start', toolCallId: 't1', toolName: 'Read', toolInput: {} },
+      { type: 'tool:result', toolCallId: 't1', output: 'ok', isError: false },
+      { type: 'result', cost: COST },
+    ]);
+    expect(durable.filter((m) => m.type === 'assistant:text')).toEqual([{ type: 'assistant:text', text: 'once' }]);
+  });
+
+  test('deltas after a flush start a fresh block', () => {
+    const { durable } = run([
+      { type: 'delta', text: 'first' },
+      { type: 'tool:start', toolCallId: 't1', toolName: 'Read', toolInput: {} },
+      { type: 'delta', text: 'second' },
+      { type: 'result', cost: COST },
+    ]);
+    expect(durable.filter((m) => m.type === 'assistant:text')).toEqual([
+      { type: 'assistant:text', text: 'first' },
+      { type: 'assistant:text', text: 'second' },
+    ]);
+  });
+
+  test('a full turn keeps the client-visible order', () => {
+    const { types } = run([
+      { type: 'delta', text: 'a' },
+      { type: 'tool:start', toolCallId: 't1', toolName: 'Bash', toolInput: { cmd: 'ls' } },
+      { type: 'tool:result', toolCallId: 't1', output: 'a.ts', isError: false },
+      { type: 'delta', text: 'b' },
+      { type: 'result', cost: COST },
+    ]);
+    expect(types).toEqual([
+      'assistant:delta',
+      'assistant:text',
+      'tool:start',
+      'tool:result',
+      'assistant:delta',
+      'assistant:text',
+      'result',
+    ]);
+  });
+
+  test('errors and stops are durable, and do not flush a partial answer away', () => {
+    const { durable } = run([
+      { type: 'delta', text: 'half' },
+      { type: 'error', message: 'boom' },
+    ]);
+    expect(durable).toEqual([{ type: 'error', message: 'boom' }]);
+    expect(run([{ type: 'stopped' }]).durable).toEqual([{ type: 'stopped' }]);
+  });
+
+  test('background task events pass through and are durable — the reason the queue exists', () => {
+    const { durable } = run([
+      { type: 'result', cost: COST },
+      { type: 'task:started', taskId: 'bg1', description: 'long job', taskType: 'local_agent' },
+      { type: 'task:notification', taskId: 'bg1', status: 'completed', summary: 'all good' },
+    ]);
+    expect(durable).toEqual([
+      { type: 'result', sessionId: SESSION, cost: COST },
+      { type: 'task:started', taskId: 'bg1', description: 'long job', taskType: 'local_agent' },
+      { type: 'task:notification', taskId: 'bg1', status: 'completed', summary: 'all good' },
+    ]);
+  });
+
+  test('streams are independent', () => {
+    const a = createTurnStream('a');
+    const b = createTurnStream('b');
+    a.push({ type: 'delta', text: 'from-a' });
+    b.push({ type: 'delta', text: 'from-b' });
+    expect(a.push({ type: 'result', cost: COST })[0]!.msg).toEqual({ type: 'assistant:text', text: 'from-a' });
+    expect(b.push({ type: 'result', cost: COST })[0]!.msg).toEqual({ type: 'assistant:text', text: 'from-b' });
+  });
+});

src/servers/sidecar/claude/turn-stream.ts

@@ -0,0 +1,101 @@
+import type { ChatEvent, TurnMessage } from '../../api/chat/types';
+
+// Translation from the parser's ChatEvent stream to the browser-facing turn messages, moved here from
+// the main server (`chat/websocket.ts:createEventHandler`). It lives with the process that produces the
+// stream because it is stateful: `delta` events accumulate into a buffer that has to be flushed as one
+// `assistant:text` at the next boundary (a tool call, or the end of the turn). A consumer downstream of
+// a socket that can drop cannot hold that state correctly.
+//
+// Pure apart from the buffer, and the buffer is the whole of it — which is what makes it testable.
+
+export type TurnOutput = {
+  msg: TurnMessage;
+  /** false only for `assistant:delta`: superseded by the `assistant:text` that follows, never persisted. */
+  durable: boolean;
+};
+
+export type TurnStream = {
+  push: (event: ChatEvent) => TurnOutput[];
+};
+
+export function createTurnStream(sessionId: string): TurnStream {
+  let buffer = '';
+
+  // Emit whatever deltas have accumulated as one complete message. Called at every boundary where the
+  // assistant stops talking, so the transcript holds text blocks rather than a thousand fragments.
+  function flush(): TurnOutput[] {
+    if (!buffer) return [];
+    const text = buffer;
+    buffer = '';
+    return [{ msg: { type: 'assistant:text', text }, durable: true }];
+  }
+
+  function push(event: ChatEvent): TurnOutput[] {
+    switch (event.type) {
+      case 'delta':
+        buffer += event.text;
+        return [{ msg: { type: 'assistant:delta', text: event.text }, durable: false }];
+
+      case 'text': {
+        // An explicit full text block wins over the accumulated deltas that produced it.
+        const text = event.text || buffer;
+        buffer = '';
+        return text ? [{ msg: { type: 'assistant:text', text }, durable: true }] : [];
+      }
+
+      case 'tool:start':
+        return [
+          ...flush(),
+          {
+            msg: {
+              type: 'tool:start',
+              toolCallId: event.toolCallId,
+              toolName: event.toolName,
+              toolInput: event.toolInput,
+            },
+            durable: true,
+          },
+        ];
+
+      case 'tool:result':
+        return [
+          {
+            msg: { type: 'tool:result', toolCallId: event.toolCallId, output: event.output, isError: event.isError },
+            durable: true,
+          },
+        ];
+
+      case 'result':
+        return [...flush(), { msg: { type: 'result', sessionId, cost: event.cost }, durable: true }];
+
+      case 'error':
+        return [{ msg: { type: 'error', message: event.message }, durable: true }];
+
+      case 'stopped':
+        return [{ msg: { type: 'stopped' }, durable: true }];
+
+      case 'task:started':
+        return [
+          {
+            msg: {
+              type: 'task:started',
+              taskId: event.taskId,
+              description: event.description,
+              taskType: event.taskType,
+            },
+            durable: true,
+          },
+        ];
+
+      case 'task:notification':
+        return [
+          {
+            msg: { type: 'task:notification', taskId: event.taskId, status: event.status, summary: event.summary },
+            durable: true,
+          },
+        ];
+    }
+  }
+
+  return { push };
+}

src/servers/sidecar/claude/user-instance.ts

@@ -3,6 +3,7 @@ import { join, resolve } from 'node:path';
 import { homedir } from 'node:os';
 import type { SidecarCommand, SidecarEvent } from '../protocol';
 import { initPaths, loadState, flushAndSave, acquireLock, releaseLock, readProxySecretFromDisk } from './state';
+import { createSessionLogStore } from './session-log';
 import { setMcpConfigPath } from './claude-manager';
 import * as claudeManager from './claude-manager';
 import { createSidecarConnector } from '../connect';
@@ -129,6 +130,14 @@ ensureAnthropicEnv();
 
 console.log(`[agent] started for ${email} (HOME=${homeDir})`);
 
+// ── Turn output ──
+
+// Every message a turn produces is translated, committed to chat_session_events and only then pushed to
+// officer. `connection` is initialised below, before any command can arrive to invoke this.
+const sessionLog = createSessionLogStore((d) =>
+  connection.send({ type: 'claude:message', sessionKey: d.sessionId, msg: d.msg, seq: d.seq }),
+);
+
 // ── Command handlers ──
 
 type ReplyFn = (msg: SidecarEvent) => void;
@@ -154,22 +163,20 @@ async function handleCommand(cmd: SidecarCommand, reply: ReplyFn) {
       ensureAnthropicEnv();
       reply({ type: 'claude:spawned', id: cmd.id, sessionKey: cmd.params.sessionKey });
 
-      const onEvent = (event: import('../../api/chat/types').ChatEvent) => {
-        connection.send({ type: 'claude:event', sessionKey: cmd.params.sessionKey, event });
-      };
-
-      claudeManager.spawnClaudeStreaming(cmd.params, onEvent).catch((err) => {
-        connection.send({
-          type: 'claude:event',
-          sessionKey: cmd.params.sessionKey,
-          event: { type: 'error', message: err instanceof Error ? err.message : String(err) },
-        });
+      const { sessionKey, durable = true } = cmd.params;
+      claudeManager
+        .spawnClaudeStreaming(cmd.params, (event) => sessionLog.push(sessionKey, event, durable))
+        .catch((err) => {
+          // Through the log like any other output, so a failure to start is durable and replayable too.
+          const message = err instanceof Error ? err.message : String(err);
+          sessionLog.push(sessionKey, { type: 'error', message }, durable);
         });
       break;
     }
 
     case 'claude:kill':
       claudeManager.killClaudeSession(cmd.sessionKey);
+      sessionLog.drop(cmd.sessionKey);
       reply({ type: 'claude:killed', id: cmd.id });
       break;
 
@@ -180,6 +187,7 @@ async function handleCommand(cmd: SidecarCommand, reply: ReplyFn) {
 
     case 'claude:clear-session':
       claudeManager.clearSession(cmd.sessionKey);
+      sessionLog.drop(cmd.sessionKey);
       reply({ type: 'claude:session-cleared', id: cmd.id });
       break;
 

src/servers/sidecar/protocol.ts

@@ -1,4 +1,4 @@
-import type { MessageCost, ChatEvent } from '../api/chat/types';
+import type { MessageCost, ChatEvent, TurnMessage } from '../api/chat/types';
 
 // ── Envelope ──
 
@@ -35,7 +35,10 @@ export type SidecarEvent =
   | { type: 'proxy:secret'; id: string; secret: string }
   // Claude Code
   | { type: 'claude:spawned'; id: string; sessionKey: string }
-  | { type: 'claude:event'; sessionKey: string; event: ChatEvent }
+  // A finished, browser-facing turn message. The agent has already committed it to chat_session_events
+  // and `seq` is its cursor id there; officer relays it verbatim. No `seq` means it is not durable —
+  // an `assistant:delta` (superseded by the text that follows) or a message whose write failed.
+  | { type: 'claude:message'; sessionKey: string; msg: TurnMessage; seq?: number }
   | { type: 'claude:result'; id: string; result: ClaudeCodeResult }
   | { type: 'claude:error'; id: string; error: string }
   | { type: 'claude:killed'; id: string }
@@ -93,6 +96,10 @@ export type ClaudeSpawnStreamingParams = {
   cwd?: string;
   model?: string;
   resumeSessionId?: string; // resume this Claude session uuid (from the /chat session list)
+  // Whether turn output should be committed to chat_session_events (default true). A chat session wants
+  // it — that is what survives an officer restart. A pipeline step does not: its sessionKey is a throwaway
+  // uuid no browser will ever replay, and the job's own event log is its record.
+  durable?: boolean;
 };
 
 export type ClaudeCodeResult = {

src/workspaces/officerdev/src/hooks/useChat.ts

@@ -141,6 +141,20 @@ export function useChat(initialSessionId?: string, initialModel?: string | null,
   function handleMessage(data: unknown) {
     const msg = data as ServerMessage;
 
+    // Continuity check, before the cursor moves. The writer stamps every durable message with the cursor
+    // of the one before it in the same session, so a mismatch against what we last saw means something is
+    // missing — pruned by retention, or a write that failed. Surface it: a conversation that silently
+    // skips a tool call or half an answer reads as the assistant having done something inexplicable.
+    // Only checked once we actually hold a cursor; opening a session from history starts mid-chain by
+    // design (events are swept after 7 days, the transcript itself is not).
+    const prevSeq = (data as { prevSeq?: number }).prevSeq;
+    if (typeof prevSeq === 'number' && cursorRef.current > 0 && prevSeq !== cursorRef.current) {
+      setMessages((prev) => [
+        ...prev,
+        { role: 'error', text: '⚠️ Some output could not be recovered — part of this conversation is missing above.' },
+      ]);
+    }
+
     // Advance the resume cursor for any durable (seq-carrying) event.
     const seq = (data as { seq?: number }).seq;
     if (typeof seq === 'number' && seq > cursorRef.current) cursorRef.current = seq;