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

move opencode's turn output into its own sidecar

Browse Source 
The second copy of the same problem. The opencode sidecar reported raw
ChatEvents and officer translated them, buffered the assistant text and wrote
every durable message to chat_session_events — so an officer restart mid-turn
lost whatever the model had produced since the last write, and `connect.ts`
dropped the events that arrived while it was down without a word.

Both harnesses speak ChatEvents, so the sidecar reuses the agent's session log
verbatim: translate, commit, then deliver the finished message with its cursor
id as `opencode:message`. Officer folds it into the in-memory transcript and
relays it, exactly as it now does for claude — `createEventHandler` (166 lines,
a duplicate of turn-stream.ts) and `emitToSession` are gone, and nothing in
officer writes to chat_session_events any more.

`opencode:event` stops being a wire event; it is the runner's internal report to
the sidecar it runs in, typed as such so it cannot leak back onto the socket.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
pastilhas
2026-07-30 05:18:09 +00:00
co-authored by Claude Opus 4.8
parent 238c3b8097
commit 71e39b7639
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/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).