diff --git a/src/servers/api/chat/opencode/event-mapper.ts b/src/servers/api/chat/opencode/event-mapper.ts index 2c069b22..cc3bda08 100644 --- a/src/servers/api/chat/opencode/event-mapper.ts +++ b/src/servers/api/chat/opencode/event-mapper.ts @@ -14,8 +14,8 @@ import type { ChatEvent, MessageCost } from '../types'; // are only distinguishable by the delta's part TYPE (a message.part.updated declaring the part as // `reasoning` vs `text` always precedes that part's deltas). So we gate deltas on partID being a `text` // part; reasoning-part deltas are dropped (parity with the Claude harness, which hides thinking). -// createEventHandler flushes the assistant text buffer on tool:start and result, so no explicit `text` -// event is needed — the streamed answer deltas are enough. +// The turn translator (`sidecar/claude/turn-stream.ts`, shared by both harnesses) flushes the assistant +// text buffer on tool:start and result, so no explicit `text` event is needed — the deltas are enough. export type OpenCodeEvent = { id?: string; @@ -96,7 +96,12 @@ export function createEventMapper(onEvent: (event: ChatEvent) => void) { if (status === 'completed' && !toolFinished.has(callID)) { toolFinished.add(callID); - onEvent({ type: 'tool:result', toolCallId: callID, output: String(part.state?.output ?? ''), isError: false }); + onEvent({ + type: 'tool:result', + toolCallId: callID, + output: String(part.state?.output ?? ''), + isError: false, + }); } else if (status === 'error' && !toolFinished.has(callID)) { toolFinished.add(callID); onEvent({ @@ -110,7 +115,7 @@ export function createEventMapper(onEvent: (event: ChatEvent) => void) { } case 'message.updated': { - const info = ((p.info as AssistantInfo | undefined) ?? (p as AssistantInfo)) ?? {}; + const info = (p.info as AssistantInfo | undefined) ?? (p as AssistantInfo) ?? {}; if (info.role === 'assistant' && info.tokens) { cost = { inputTokens: info.tokens.input ?? 0, @@ -128,7 +133,8 @@ export function createEventMapper(onEvent: (event: ChatEvent) => void) { case 'session.error': { const error = p.error; - const message = typeof error === 'string' ? error : ((error as { message?: string })?.message ?? 'OpenCode error'); + const message = + typeof error === 'string' ? error : ((error as { message?: string })?.message ?? 'OpenCode error'); finish({ type: 'error', message }); return; } diff --git a/src/servers/api/chat/websocket.ts b/src/servers/api/chat/websocket.ts index 212a9a36..7cd4a469 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, TurnMessage, UserSession } from './types'; +import type { ClientMessage, ServerMessage, Message, TurnMessage, UserSession } from './types'; import { sessionManager } from './session-manager'; import { sendClaudeCodeStreaming } from '@@/channels/send-claude-code'; import { sendOpenCodeStreaming } from '@@/channels/send-opencode'; @@ -8,7 +8,7 @@ import { ensureGeneralChatSessionsCwd } from './claude-sessions'; import * as sidecar from '@@/sidecar-registry'; import { join } from 'path'; import { getOwnerHomeDir, getEmailAccountsDir } from '../../../servers/data-path'; -import { getUserSettings, getEmailAccounts, appendChatEvent, getChatEventsSince } from 'officerdb'; +import { getUserSettings, getEmailAccounts, getChatEventsSince } from 'officerdb'; import { mkdirSync } from 'node:fs'; import { logger } from './logger'; @@ -96,24 +96,11 @@ function sendToClient(ws: ServerWebSocket | null, msg: ServerMessage, se } } -// Persist a durable session event to the Postgres queue (for replay across reconnects) and deliver it -// live to the attached socket with its cursor `seq`. Events are queued even while the client is -// DISCONNECTED (ws null) — that's what lets a reconnecting client replay what it missed (e.g. a -// background task:notification). Streaming deltas are ephemeral: delivered live, never persisted. -async function emitToSession(sessionId: string, msg: ServerMessage): Promise { - const ws = (sessionManager.getSession(sessionId)?.ws ?? null) as ServerWebSocket | null; - if (msg.type === 'assistant:delta') { - sendToClient(ws, msg); - return; - } - let seq: number | undefined; - try { - seq = await appendChatEvent(sessionId, msg); - } catch (err) { - logger.error('Failed to persist chat event', { sessionId, error: String(err) }); - } - sendToClient(ws, msg, seq); -} +// Nothing in officer writes to chat_session_events any more. Both harnesses commit their own turn +// output in the sidecar that produced it (`sidecar/claude/session-log.ts`), which is the whole point: +// the durable record does not travel over the socket between the two processes, so officer can restart +// mid-turn without losing it. Officer reads the table on `resume` (getChatEventsSince) and relays what +// the sidecars send. The old `emitToSession` used to live here. export async function open(ws: ServerWebSocket): Promise { const timer = setInterval(() => { @@ -228,7 +215,8 @@ function foldIntoSession(session: UserSession, msg: TurnMessage, model: string): } } -function createClaudeMessageHandler(sessionId: string, model: string) { +// Shared by both harnesses now: whichever sidecar ran the turn has already translated and committed it. +function createMessageHandler(sessionId: string, model: string) { return (msg: TurnMessage, seq?: number): void => { const session = sessionManager.getSession(sessionId); if (!session) return; @@ -237,166 +225,6 @@ function createClaudeMessageHandler(sessionId: string, model: string) { }; } -// ── 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); - if (!session) return; - - // All durable sends go through emitToSession (persist + deliver with cursor seq). Deltas stay live-only. - switch (event.type) { - case 'delta': { - await emitToSession(sessionId, { type: 'assistant:delta', text: event.text }); - session.streamBuffer += event.text; - break; - } - - case 'text': { - // Flush streaming buffer as complete text - const text = event.text || session.streamBuffer; - if (text) { - await emitToSession(sessionId, { type: 'assistant:text', text }); - - const assistantMsg: Message = { - id: randomUUID(), - timestamp: Date.now(), - role: 'assistant', - text, - model, - }; - session.messages.push(assistantMsg); - session.meta.messageCount += 1; - session.streamBuffer = ''; - } - break; - } - - case 'tool:start': { - // Flush any pending streaming text first - if (session.streamBuffer) { - await emitToSession(sessionId, { type: 'assistant:text', text: session.streamBuffer }); - - const assistantMsg: Message = { - id: randomUUID(), - timestamp: Date.now(), - role: 'assistant', - text: session.streamBuffer, - model, - }; - session.messages.push(assistantMsg); - session.meta.messageCount += 1; - session.streamBuffer = ''; - } - - await emitToSession(sessionId, { - type: 'tool:start', - toolCallId: event.toolCallId, - toolName: event.toolName, - toolInput: event.toolInput, - }); - - const toolMsg: Message = { - id: randomUUID(), - timestamp: Date.now(), - role: 'tool', - toolCallId: event.toolCallId, - toolName: event.toolName, - toolInput: event.toolInput, - }; - session.messages.push(toolMsg); - session.meta.messageCount += 1; - break; - } - - case 'tool:result': { - await emitToSession(sessionId, { - type: 'tool:result', - toolCallId: event.toolCallId, - output: event.output, - isError: event.isError, - }); - - // Update existing tool message with output - for (let i = session.messages.length - 1; i >= 0; i--) { - const m = session.messages[i]!; - if (m.role === 'tool' && m.toolCallId === event.toolCallId) { - m.output = event.output; - m.isError = event.isError; - break; - } - } - break; - } - - case 'result': { - // Flush any remaining streaming buffer - if (session.streamBuffer) { - await emitToSession(sessionId, { type: 'assistant:text', text: session.streamBuffer }); - - const assistantMsg: Message = { - id: randomUUID(), - timestamp: Date.now(), - role: 'assistant', - text: session.streamBuffer, - model, - cost: event.cost, - }; - session.messages.push(assistantMsg); - session.meta.messageCount += 1; - session.streamBuffer = ''; - } - - await emitToSession(sessionId, { type: 'result', sessionId, cost: event.cost }); - - session.isGenerating = false; - session.meta.cost.inputTokens += event.cost.inputTokens; - session.meta.cost.outputTokens += event.cost.outputTokens; - session.meta.cost.totalUSD += event.cost.totalUSD; - session.meta.updatedAt = Date.now(); - // No disk persistence — Claude's transcript is the record. - break; - } - - case 'error': { - await emitToSession(sessionId, { type: 'error', message: event.message }); - session.isGenerating = false; - break; - } - - case 'stopped': { - await emitToSession(sessionId, { type: 'stopped' }); - session.isGenerating = false; - break; - } - - 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, - }); - 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, - }); - break; - } - } - }; -} - async function handleChat( ws: ServerWebSocket, msg: { @@ -487,7 +315,7 @@ async function handleClaudeCodeChat( session.isGenerating = true; - const onMessage = createClaudeMessageHandler(sessionId, model); + const onMessage = createMessageHandler(sessionId, model); try { if (!session._claudeKill) { @@ -578,7 +406,7 @@ async function handleOpenCodeChat( session.isGenerating = true; - const onEvent = createEventHandler(sessionId, model, cwd); + const onMessage = createMessageHandler(sessionId, model); try { const handle = await sendOpenCodeStreaming({ @@ -590,7 +418,7 @@ async function handleOpenCodeChat( cwd, model, resumeSessionId: msg.resumeSessionId, - onEvent, + onMessage, }); // Store the abort handle so handleStop can end the turn (OpenCode is aborted via this handle). diff --git a/src/servers/channels/send-opencode.ts b/src/servers/channels/send-opencode.ts index c8059c5a..52f1ece9 100644 --- a/src/servers/channels/send-opencode.ts +++ b/src/servers/channels/send-opencode.ts @@ -1,13 +1,13 @@ -import type { ChatEvent } from '@@/api/chat/types'; +import type { TurnMessage } from '@@/api/chat/types'; import { logger } from '@@/api/chat/logger'; import * as sidecar from '@@/sidecar-registry'; import { getOpenCodeSession } from '@@/api/chat/opencode/state'; // The OpenCode analog of send-claude-code.ts: it drives a turn through the officer-opencode sidecar, -// which spawns `opencode run … --format json` (tools hard-anchored to the chat cwd via --dir) and -// streams mapped ChatEvents back over the sidecar WS. We subscribe to those events (filtered by -// sessionKey) and forward them to the caller's onEvent — the same shared contract the Claude harness -// uses, so createEventHandler and the whole UI pipeline are unchanged. +// which spawns `opencode run … --format json` (tools hard-anchored to the chat cwd via --dir), maps its +// output to TurnMessages, commits each one to chat_session_events and streams the finished message plus +// its cursor id back over the sidecar WS. We subscribe (filtered by sessionKey) and forward to the +// caller's onMessage — the same contract sendClaudeCodeStreaming uses. type OpenCodeStreamingParams = { userId: number; @@ -19,7 +19,9 @@ type OpenCodeStreamingParams = { model?: string; role?: 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 OpenCodeStreamingHandle = { @@ -29,11 +31,13 @@ type OpenCodeStreamingHandle = { export async function sendOpenCodeStreaming(params: OpenCodeStreamingParams): Promise { logger.info('OpenCode streaming exec (via sidecar)', { sessionKey: params.sessionKey, model: params.model }); - // Forward this session's turn events; unsubscribe on the terminal event. - const unsub = sidecar.onOpenCodeEvent((sessionKey, event) => { + // Forward this session's turn messages; unsubscribe on the terminal one. Unlike the Claude harness + // there is nothing after `result` here — `opencode run` exits with the turn, so it has no background + // tasks that could report later. + const unsub = sidecar.onOpenCodeMessage((sessionKey, msg, seq) => { if (sessionKey !== params.sessionKey) return; - params.onEvent(event); - if (event.type === 'result' || event.type === 'error' || event.type === 'stopped') unsub(); + params.onMessage(msg, seq); + if (msg.type === 'result' || msg.type === 'error' || msg.type === 'stopped') unsub(); }); // Resume an existing OpenCode session when we know its id: a stored mapping (set from the sidecar's @@ -51,6 +55,7 @@ export async function sendOpenCodeStreaming(params: OpenCodeStreamingParams): Pr cwd: params.cwd, model: params.model, resumeSessionId, + durable: params.durable, }); } catch (err) { unsub(); diff --git a/src/servers/sidecar-registry.ts b/src/servers/sidecar-registry.ts index 925bb28f..28bc55f1 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, TurnMessage } from './api/chat/types'; +import type { TurnMessage } from './api/chat/types'; // ── Types ── @@ -298,14 +298,11 @@ export function killOpenCode(sessionKey: string): void { sendFire('opencode', { type: 'opencode:kill', id: nextId(), sessionKey }); } -export function onOpenCodeEvent(handler: (sessionKey: string, event: ChatEvent) => void): () => void { - return on('opencode:event', (msg) => { - if (msg.type === 'opencode:event') { - handler( - (msg as SidecarEvent & { type: 'opencode:event' }).sessionKey, - (msg as SidecarEvent & { type: 'opencode:event' }).event, - ); - } +export function onOpenCodeMessage(handler: (sessionKey: string, msg: TurnMessage, seq?: number) => void): () => void { + return on('opencode:message', (ev) => { + if (ev.type !== 'opencode:message') return; + const msg = ev as SidecarEvent & { type: 'opencode:message' }; + handler(msg.sessionKey, msg.msg, msg.seq); }); } diff --git a/src/servers/sidecar/claude/turn-stream.ts b/src/servers/sidecar/claude/turn-stream.ts index f0461928..51bd79b9 100644 --- a/src/servers/sidecar/claude/turn-stream.ts +++ b/src/servers/sidecar/claude/turn-stream.ts @@ -1,7 +1,8 @@ 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 +// Translation from a harness's ChatEvent stream to the browser-facing turn messages, moved here from the +// main server (it was `createEventHandler` in `chat/websocket.ts`, once per harness). Both the agent and +// the opencode sidecar use this one copy. 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. diff --git a/src/servers/sidecar/opencode/index.ts b/src/servers/sidecar/opencode/index.ts index 03e2ccf0..7417eec7 100644 --- a/src/servers/sidecar/opencode/index.ts +++ b/src/servers/sidecar/opencode/index.ts @@ -3,6 +3,7 @@ import { homedir } from 'node:os'; import { join } from 'node:path'; import { DATA_PATH } from '../../data-path'; import { createSidecarConnector } from '../connect'; +import { createSessionLogStore } from '../claude/session-log'; import type { SidecarCommand, SidecarEvent } from '../protocol'; import { runOpenCodeTurn, killOpenCodeTurn } from './runner'; @@ -131,22 +132,33 @@ console.log(`[opencode] serve healthy on port ${port}`); // ── Command handlers ── type ReplyFn = (msg: SidecarEvent) => void; -type SendFn = (msg: SidecarEvent) => void; const RUNNER_CONFIG = { bin: OPENCODE_BIN, fallbackCwd: SERVE_CWD }; -function handleCommand(cmd: SidecarCommand, reply: ReplyFn, send: SendFn) { +function handleCommand(cmd: SidecarCommand, reply: ReplyFn) { switch (cmd.type) { case 'ping': reply({ type: 'pong', id: cmd.id }); break; - case 'opencode:run-streaming': - // Fire the turn; events stream back via `send` (opencode:event / opencode:session / terminal). - runOpenCodeTurn(cmd.params, RUNNER_CONFIG, send); - reply({ type: 'opencode:spawned', id: cmd.id, sessionKey: cmd.params.sessionKey }); + case 'opencode:run-streaming': { + const { sessionKey, durable = true } = cmd.params; + // Turn output goes through the session log: translated to TurnMessages and committed to + // chat_session_events here, in the process that produced it. Officer being down during a turn + // no longer costs the transcript — the browser replays it from its cursor. + runOpenCodeTurn(cmd.params, RUNNER_CONFIG, (msg) => { + if (msg.type === 'opencode:event') { + sessionLog.push(sessionKey, msg.event, durable); + return; + } + // opencode:session is a routing fact (which `ses_…` to resume), not transcript — send it live. + connection.send(msg); + }); + reply({ type: 'opencode:spawned', id: cmd.id, sessionKey }); break; + } case 'opencode:kill': killOpenCodeTurn(cmd.sessionKey); + sessionLog.drop(cmd.sessionKey); break; default: reply({ @@ -164,8 +176,7 @@ const connection = createSidecarConnector({ name: 'opencode', capabilities: ['opencode'], onCommand(cmd, reply) { - // Streaming turn events use a stable send (always the current ws), not the per-command reply. - handleCommand(cmd as SidecarCommand, reply as ReplyFn, (msg) => connection.send(msg)); + handleCommand(cmd as SidecarCommand, reply as ReplyFn); }, onConnected() { // Tell the API where our OpenCode HTTP server is listening, so it can route requests there. @@ -174,6 +185,13 @@ const connection = createSidecarConnector({ }, }); +// Translate → commit → deliver, in that order and one at a time per session. Shared with the agent +// sidecar (`claude/session-log.ts`): both harnesses speak ChatEvents, so the translation and the write +// are the same code, and only the wire event type differs. +const sessionLog = createSessionLogStore((d) => + connection.send({ type: 'opencode:message', sessionKey: d.sessionId, msg: d.msg, seq: d.seq }), +); + // ── Graceful shutdown ── function shutdown(signal: string) { diff --git a/src/servers/sidecar/opencode/runner.ts b/src/servers/sidecar/opencode/runner.ts index a057c7e1..efbbb474 100644 --- a/src/servers/sidecar/opencode/runner.ts +++ b/src/servers/sidecar/opencode/runner.ts @@ -1,7 +1,7 @@ import { existsSync } from 'node:fs'; import type { Subprocess } from 'bun'; import type { ChatEvent, MessageCost } from '../../api/chat/types'; -import type { OpenCodeRunParams, SidecarEvent } from '../protocol'; +import type { OpenCodeRunParams } from '../protocol'; // Drives one chat turn by spawning `opencode run … --format json` and mapping its newline-delimited // JSON events to the shared ChatEvent contract. This is the reliable path: `--dir ` hard-anchors @@ -21,7 +21,14 @@ export type RunnerConfig = { fallbackCwd: string; // used when params.cwd is missing/nonexistent }; -type Emit = (event: SidecarEvent) => void; +// What a turn reports to the sidecar it runs in. `opencode:event` is deliberately not a wire event any +// more: the sidecar translates each one into a TurnMessage and commits it before officer sees anything, +// so the durable record does not depend on officer being up (see index.ts). +export type RunnerMessage = + | { type: 'opencode:event'; sessionKey: string; event: ChatEvent } + | { type: 'opencode:session'; sessionKey: string; sessionId: string }; + +type Emit = (msg: RunnerMessage) => void; type RunHandle = { proc: Subprocess; killedByUser: boolean }; @@ -108,7 +115,10 @@ export function runOpenCodeTurn(params: OpenCodeRunParams, config: RunnerConfig, } catch { /* already gone */ } - finish({ type: 'error', message: `OpenCode turn stalled (no output for ${INACTIVITY_MS / 1000}s) and was stopped` }); + finish({ + type: 'error', + message: `OpenCode turn stalled (no output for ${INACTIVITY_MS / 1000}s) and was stopped`, + }); }, INACTIVITY_MS); }; bumpInactivity(); diff --git a/src/servers/sidecar/protocol.ts b/src/servers/sidecar/protocol.ts index f1eb2af8..21d94757 100644 --- a/src/servers/sidecar/protocol.ts +++ b/src/servers/sidecar/protocol.ts @@ -1,4 +1,4 @@ -import type { MessageCost, ChatEvent, TurnMessage } from '../api/chat/types'; +import type { MessageCost, TurnMessage } from '../api/chat/types'; // ── Envelope ── @@ -54,9 +54,12 @@ export type SidecarEvent = | { type: 'email:new'; userEmail: string } // OpenCode — the sidecar reports where its `opencode serve` is listening (random port) on connect | { type: 'opencode:server'; port: number } - // OpenCode turn streaming (analog of claude:*): spawned ack, per-event stream, session id report + // OpenCode turn streaming (analog of claude:*): spawned ack, per-message stream, session id report | { type: 'opencode:spawned'; id: string; sessionKey: string } - | { type: 'opencode:event'; sessionKey: string; event: ChatEvent } + // Same contract as `claude:message`: a finished turn message the sidecar has already committed to + // chat_session_events, plus the cursor id it landed on. Officer relays it and folds it into its + // in-memory transcript; it does not translate or persist. + | { type: 'opencode:message'; sessionKey: string; msg: TurnMessage; seq?: number } | { type: 'opencode:session'; sessionKey: string; sessionId: string } | { type: 'opencode:error'; id: string; error: string } // Music — the sidecar reports where its audio-streaming HTTP server is listening (random port) on connect @@ -117,6 +120,7 @@ export type OpenCodeRunParams = { cwd?: string; // passed to `opencode run --dir` — hard-re-anchors tools to this directory model?: string; // `providerID/modelID` (e.g. opencode/claude-haiku-4-5); passed to --model verbatim resumeSessionId?: string; // OpenCode `ses_…` id to continue (`--session`) + durable?: boolean; // commit turn output to chat_session_events (default true) — see ClaudeSpawnStreamingParams }; // ── VNC types ──