officerdev/platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'sidecars-opencode' into sidecars

Browse Source
This commit is contained in:
pastilhas
2026-07-30 05:43:09 +00:00
parent 9318fe8d0e 71e39b7639
commit bd1c8b578e
8 changed files with 93 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/servers/api/chat/opencode/event-mapper.ts
+11 -5
View File
@@ -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;
}
src/servers/api/chat/websocket.ts
+12 -184
View File
@@ -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<WSData> | 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<void> {
const ws = (sessionManager.getSession(sessionId)?.ws ?? null) as ServerWebSocket<WSData> | 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<WSData>): Promise<void> {
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<void> => {
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<WSData>,
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).
src/servers/channels/send-opencode.ts
+15 -10
View File
@@ -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<OpenCodeStreamingHandle> {
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();
src/servers/sidecar-registry.ts
+6 -9
View File
@@ -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);
});
}
src/servers/sidecar/claude/turn-stream.ts
+3 -2
View File
@@ -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.
src/servers/sidecar/opencode/index.ts
+26 -8
View File
@@ -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) {
src/servers/sidecar/opencode/runner.ts
+13 -3
View File
@@ -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 <cwd>` 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();
src/servers/sidecar/protocol.ts
+7 -3
View File
@@ -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 ──