diff --git a/src/databases/officer_db/src/index.ts b/src/databases/officer_db/src/index.ts index 3ac920c6..c92a3906 100644 --- a/src/databases/officer_db/src/index.ts +++ b/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, diff --git a/src/databases/officer_db/src/queries/chat-events.ts b/src/databases/officer_db/src/queries/chat-events.ts index df454a78..e95cbc83 100644 --- a/src/databases/officer_db/src/queries/chat-events.ts +++ b/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 { - 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 { + 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 { await db.delete(chatSessionEvents).where(lt(chatSessionEvents.createdAt, cutoff)); diff --git a/src/servers/api/chat/types.ts b/src/servers/api/chat/types.ts index a4eca5b8..feb38dd2 100644 --- a/src/servers/api/chat/types.ts +++ b/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 & { prevSeq?: number }; + export type ChatEvent = | { type: 'text'; text: string } | { type: 'delta'; text: string } diff --git a/src/servers/api/chat/websocket.ts b/src/servers/api/chat/websocket.ts index 5291113b..212a9a36 100644 --- a/src/servers/api/chat/websocket.ts +++ b/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): 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 | 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 => { 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; diff --git a/src/servers/api/tasks/pipeline-executor.ts b/src/servers/api/tasks/pipeline-executor.ts index 1d8ba7c9..07df915f 100644 --- a/src/servers/api/tasks/pipeline-executor.ts +++ b/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) { diff --git a/src/servers/channels/send-claude-code.ts b/src/servers/channels/send-claude-code.ts index 8e87ee8e..7f54883c 100644 --- a/src/servers/channels/send-claude-code.ts +++ b/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 { - 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); diff --git a/src/servers/sidecar-registry.ts b/src/servers/sidecar-registry.ts index 3e4fc419..925bb28f 100644 --- a/src/servers/sidecar-registry.ts +++ b/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); }); } diff --git a/src/servers/sidecar/claude/session-log.test.ts b/src/servers/sidecar/claude/session-log.test.ts new file mode 100644 index 00000000..642fdfba --- /dev/null +++ b/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((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']); + }); +}); diff --git a/src/servers/sidecar/claude/session-log.ts b/src/servers/sidecar/claude/session-log.ts new file mode 100644 index 00000000..1d7370ed --- /dev/null +++ b/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; + lastSeq: (sessionId: string) => Promise; +}; + +const postgresWriter: EventWriter = { + append: (sessionId, msg) => appendChatEvent(sessionId, msg), + lastSeq: (sessionId) => getLastChatEventSeq(sessionId), +}; + +type SessionLog = { + stream: ReturnType; + /** Serialises commits so cursor ids are assigned in the order the events actually arrived. */ + tail: Promise; + lastSeq: number | undefined; + resolvedLastSeq: boolean; +}; + +export function createSessionLogStore( + deliver: (d: Delivery) => void, + writer: EventWriter = postgresWriter, +): SessionLogStore { + const logs = new Map(); + + 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 { + // 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); + }, + }; +} diff --git a/src/servers/sidecar/claude/turn-stream.test.ts b/src/servers/sidecar/claude/turn-stream.test.ts new file mode 100644 index 00000000..905d1a6c --- /dev/null +++ b/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' }); + }); +}); diff --git a/src/servers/sidecar/claude/turn-stream.ts b/src/servers/sidecar/claude/turn-stream.ts new file mode 100644 index 00000000..f0461928 --- /dev/null +++ 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 }; +} diff --git a/src/servers/sidecar/claude/user-instance.ts b/src/servers/sidecar/claude/user-instance.ts index aedf51e7..3e7205ed 100644 --- a/src/servers/sidecar/claude/user-instance.ts +++ b/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; diff --git a/src/servers/sidecar/protocol.ts b/src/servers/sidecar/protocol.ts index 2386e42a..5902a974 100644 --- a/src/servers/sidecar/protocol.ts +++ b/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 = { diff --git a/src/workspaces/officerdev/src/hooks/useChat.ts b/src/workspaces/officerdev/src/hooks/useChat.ts index 27031b53..ce68e390 100644 --- a/src/workspaces/officerdev/src/hooks/useChat.ts +++ b/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;