diff --git a/CLAUDE.md b/CLAUDE.md index 73323ea1..6d4de067 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -23,8 +23,15 @@ One Bun process (`src/server.tsx`) serves everything: Long-running and privileged work lives in **sidecars**: separate processes that dial back in over `/api/sidecar/register` and are tracked in `src/servers/sidecar-registry.ts`. PM2 runs them -(`ecosystem.config.cjs`): `officer` (the server), `officer-claude`, `officer-opencode`, -`officer-email`, `officer-pty`, `officer-vnc`. +(`ecosystem.config.cjs`): `officer` (the server), `officer-anthropic-proxy`, `officer-agent`, +`officer-opencode`, `officer-email`, `officer-pty`, `officer-vnc`, `officer-music`, `officer-vault`, +`officer-slskd`. + +**`officer-anthropic-proxy` and `officer-agent` are not the same thing.** The proxy holds the Anthropic +credential and forwards API traffic; the agent is the process that spawns `claude`. They were one entry +named `officer-claude` until the sidecar-isolation work — which is exactly how the false claim that +"restarting officer doesn't disturb the agent" survived so long. Every sidecar is a PM2 peer of +`officer`, so **no sidecar is a child of the server and restarting the server does not kill one.** Agents run **unsandboxed as the server owner**, with `--dangerously-skip-permissions`. This is deliberate — it is the owner's own machine. Do not add a jail without being asked. diff --git a/docs/working-on-officer.md b/docs/working-on-officer.md index c510355a..82b45452 100644 --- a/docs/working-on-officer.md +++ b/docs/working-on-officer.md @@ -57,8 +57,13 @@ Commit messages: simple lowercase, no prefixes, explaining *why*. ## Running and checking your work -The server runs under pm2 as `officer`, plus sidecars (`officer-claude`, `officer-opencode`, -`officer-email`, `officer-pty`, `officer-vnc`). `pm2 list` shows them; `pm2 logs officer` follows. +The server runs under pm2 as `officer`, plus sidecars (`officer-anthropic-proxy`, `officer-agent`, +`officer-opencode`, `officer-email`, `officer-pty`, `officer-vnc`, `officer-music`, `officer-vault`, +`officer-slskd`). `pm2 list` shows them; `pm2 logs officer` follows. + +Two of those names are worth knowing apart: **`officer-anthropic-proxy` holds the Anthropic credential +and proxies API traffic; `officer-agent` is the process that actually runs `claude`.** They used to be +one confusingly-named entry (`officer-claude`) that was only the proxy. **Don't restart the owner's server to test.** Boot your own on a spare port instead — the running instance holds `PORT` from `.env` (9010): diff --git a/ecosystem.config.cjs b/ecosystem.config.cjs index 15c25a28..c32430d6 100644 --- a/ecosystem.config.cjs +++ b/ecosystem.config.cjs @@ -6,12 +6,25 @@ module.exports = { args: 'start', watch: false, }, + // The Anthropic credential proxy. Despite the old name (`officer-claude`) this process does NOT + // run agents — it holds the proxy secret and forwards to api.anthropic.com. The process that runs + // agents is `officer-agent` below. { - name: 'officer-claude', + name: 'officer-anthropic-proxy', script: 'bun', args: 'run src/servers/sidecar/claude/index.ts', watch: false, }, + // The process that actually runs `claude`. It used to be spawned on demand by the main server, + // which made every agent session a grandchild of `officer` and killed it on every restart. As a PM2 + // peer it survives them. It resolves the owner from the database and the proxy secret from the + // proxy's state file, so it needs nothing from `officer` in order to start. + { + name: 'officer-agent', + script: 'bun', + args: 'run src/servers/sidecar/claude/user-instance.ts', + watch: false, + }, { name: 'officer-opencode', script: 'bun', @@ -24,10 +37,13 @@ module.exports = { args: 'run src/servers/sidecar/email/index.ts', watch: false, }, + // The only sidecar run by `node` rather than `bun`, and the only one that is not TypeScript: node-pty + // is a native addon. It also does not use sidecar/connect.ts, and carries its own copy of the + // reconnect loop. { name: 'officer-pty', script: 'node', - args: 'src/servers/api/terminal/pty-sidecar.mjs', + args: 'src/servers/sidecar/pty/index.mjs', watch: false, }, { diff --git a/src/databases/officer_db/src/index.ts b/src/databases/officer_db/src/index.ts index 90072fc0..c92a3906 100644 --- a/src/databases/officer_db/src/index.ts +++ b/src/databases/officer_db/src/index.ts @@ -2,6 +2,7 @@ export { getUsers, getUserById, getUserByEmail, + getOwnerUser, getUserCount, createUser, updateUser, @@ -77,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/auth.ts b/src/databases/officer_db/src/queries/auth.ts index febf9dec..3422bf07 100644 --- a/src/databases/officer_db/src/queries/auth.ts +++ b/src/databases/officer_db/src/queries/auth.ts @@ -19,6 +19,14 @@ export async function getUserByEmail(email: string): Promise { + const [user] = await db.select().from(users).orderBy(users.id).limit(1); + return user; +} + export async function getUserCount(): Promise { const [result] = await db.select({ count: sql`count(*)::int` }).from(users); return result?.count ?? 0; 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/server.tsx b/src/server.tsx index 2fca4e2b..a6dacbe9 100644 --- a/src/server.tsx +++ b/src/server.tsx @@ -8,8 +8,7 @@ import { terminalWebsocket } from './servers/api/terminal/websocket'; import { chatWebsocket } from './servers/api/chat/websocket'; import { taskRunnerWebsocket } from './servers/api/tasks/task-executor'; import { pipelineWebsocket } from './servers/api/tasks/pipeline-executor'; -import { cliampWebsocket } from './servers/api/cliamp/websocket'; -import { cliampAudioWebsocket } from './servers/api/cliamp/audio-ws'; +import { cliampWebsocket, cliampAudioWebsocket } from './servers/api/cliamp/relay'; import { desktopWebsocket } from './servers/api/desktop/websocket'; import { vaultWebsocket, upgradeVaultWs } from './servers/api/vault/websocket'; import { findEntryByProxyId, touchEntry } from './servers/api/dev-server/router'; @@ -50,7 +49,7 @@ type WSData = { command?: string; cols?: number; rows?: number; - files?: string; + search?: string; // raw query string, for providers that relay it to a sidecar devServerPort?: number; devServerSlug?: string; wsProxyPath?: string; @@ -250,7 +249,6 @@ async function upgradeWs( const command = url.searchParams.get('command') ?? undefined; const cols = url.searchParams.get('cols') ? Number(url.searchParams.get('cols')) : undefined; const rows = url.searchParams.get('rows') ? Number(url.searchParams.get('rows')) : undefined; - const files = url.searchParams.get('files') ?? undefined; const ok = server.upgrade(req, { data: { userId: user.id, @@ -262,7 +260,7 @@ async function upgradeWs( command, cols, rows, - files, + search: url.search, }, }); if (!ok) return new Response('Upgrade failed', { status: 500 }); @@ -388,51 +386,8 @@ initQueue().catch((err) => console.error('[queue] failed to initialize:', err)); import { cleanupOnStartup } from './servers/api/tasks/pipeline-job-manager'; cleanupOnStartup().catch((err) => console.error('[pipeline-jobs] startup cleanup failed:', err)); -// Ensure PulseAudio is running with virtual sink for cliamp audio streaming -(async () => { - const pulseaudio = Bun.which('pulseaudio'); - const pactl = Bun.which('pactl'); - if (!pulseaudio || !pactl) { - console.log('[cliamp] pulseaudio not installed, skipping audio setup'); - return; - } - - // Start PulseAudio daemon if not running - const check = Bun.spawnSync({ cmd: [pulseaudio, '--check'], stdout: 'ignore', stderr: 'ignore' }); - if (check.exitCode !== 0) { - const start = Bun.spawnSync({ cmd: [pulseaudio, '--start', '-D'], stdout: 'ignore', stderr: 'ignore' }); - if (start.exitCode !== 0) { - console.error('[cliamp] failed to start pulseaudio'); - return; - } - console.log('[cliamp] pulseaudio started'); - } else { - console.log('[cliamp] pulseaudio already running'); - } - - // Load null sink if not already loaded - const sinks = Bun.spawnSync({ cmd: [pactl, 'list', 'short', 'sinks'], stdout: 'pipe', stderr: 'ignore' }); - const sinkList = sinks.stdout.toString(); - if (!sinkList.includes('virtual_out')) { - const load = Bun.spawnSync({ - cmd: [ - pactl, - 'load-module', - 'module-null-sink', - 'sink_name=virtual_out', - 'sink_properties=device.description=Virtual_Output', - ], - stdout: 'pipe', - stderr: 'pipe', - }); - if (load.exitCode !== 0) { - console.error('[cliamp] failed to load null sink:', load.stderr.toString().trim()); - } else { - console.log('[cliamp] virtual_out null sink loaded'); - } - } else { - console.log('[cliamp] virtual_out sink already exists'); - } -})(); +// PulseAudio and the `virtual_out` sink used to be set up here, at every boot of a process that has no +// audio responsibilities. They belong to the music sidecar, which owns both cliamp halves now +// (sidecar/music/pulse-audio.ts). // Pi check/install is handled by bootstrap.ts (imported above) 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/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 5b439508..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 } 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(() => { @@ -165,149 +152,76 @@ export function close(ws: ServerWebSocket): void { } } -function createEventHandler(sessionId: string, model: string, cwd: string) { - return async (event: ChatEvent): Promise => { +// ── 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; + } +} + +// 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; - - // 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; - } - } + foldIntoSession(session, msg, model); + sendToClient(session.ws as ServerWebSocket | null, msg, seq); }; } @@ -401,7 +315,7 @@ async function handleClaudeCodeChat( session.isGenerating = true; - const onEvent = createEventHandler(sessionId, model, cwd); + const onMessage = createMessageHandler(sessionId, model); try { if (!session._claudeKill) { @@ -416,7 +330,7 @@ async function handleClaudeCodeChat( cwd, model, resumeSessionId: msg.resumeSessionId, - onEvent, + onMessage, }); session.piProcess = sessionId as any; session._claudeKill = handle.kill; @@ -492,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({ @@ -504,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). @@ -572,7 +486,7 @@ async function handleStop(ws: ServerWebSocket): Promise { if (isClaudeModel(session.model)) { // Interrupt the current turn but KEEP the persistent session alive (background tasks + the // warm worker survive). Full teardown is 'disconnect' → deleteSession → _claudeKill. - void sidecar.interruptClaude(sessionId, session.email); + void sidecar.interruptClaude(sessionId); logger.info('Interrupted Claude Code turn via sidecar (session stays warm)', { sessionId }); } else { session._claudeKill?.(); // OpenCode: abort the turn via the stored handle diff --git a/src/servers/api/cliamp/audio-ws.ts b/src/servers/api/cliamp/audio-ws.ts deleted file mode 100644 index 5fc6bd40..00000000 --- a/src/servers/api/cliamp/audio-ws.ts +++ /dev/null @@ -1,91 +0,0 @@ -import type { ServerWebSocket } from 'bun'; -import { spawn, type Subprocess } from 'bun'; - -type WSData = { - userId: number; - email: string; - username: string; - role: string; -}; - -type AudioSession = { - proc: Subprocess; - closed: boolean; -}; - -const sessions = new Map, AudioSession>(); - -export const cliampAudioWebsocket = { - async open(ws: ServerWebSocket) { - const parecPath = Bun.which('parec'); - if (!parecPath) { - ws.close(4000, 'parec not found on host'); - return; - } - - let proc: Subprocess<'ignore', 'pipe', 'ignore'>; - try { - proc = spawn({ - cmd: [parecPath, '--format=s16le', '--rate=44100', '--channels=2', '-d', 'virtual_out.monitor'], - stdout: 'pipe', - stderr: 'ignore', - stdin: 'ignore', - }); - } catch { - ws.close(4000, 'Failed to start audio capture'); - return; - } - - const session: AudioSession = { proc, closed: false }; - sessions.set(ws, session); - console.log('[cliamp-audio] parec started, streaming to WS'); - - // Stream stdout chunks as binary WS frames - const reader = proc.stdout.getReader(); - let totalBytes = 0; - const pump = async () => { - try { - while (!session.closed) { - const { done, value } = await reader.read(); - if (done) break; - if (value && !session.closed) { - totalBytes += value.byteLength; - if (totalBytes <= value.byteLength) { - console.log(`[cliamp-audio] first chunk: ${value.byteLength} bytes`); - } - try { - ws.sendBinary(value); - } catch { - break; - } - } - } - } catch { - // stream ended or error - } finally { - if (!session.closed) { - session.closed = true; - try { ws.close(); } catch { /* ignore */ } - } - } - }; - - pump(); - }, - - message() { - // No client-to-server messages expected - }, - - close(ws: ServerWebSocket) { - console.log('[cliamp-audio] WS closed'); - const session = sessions.get(ws); - if (session) { - session.closed = true; - try { session.proc.kill(); } catch { /* ignore */ } - sessions.delete(ws); - } - }, - - drain() {}, -}; diff --git a/src/servers/api/cliamp/relay.ts b/src/servers/api/cliamp/relay.ts new file mode 100644 index 00000000..7a4fb1fd --- /dev/null +++ b/src/servers/api/cliamp/relay.ts @@ -0,0 +1,110 @@ +import type { ServerWebSocket } from 'bun'; +import { getMusicServerWsUrl } from '../music/sidecar-server'; + +// Platform side of the two cliamp sockets. Both used to spawn processes here — the `cliamp` player and a +// `parec` capture — which put the whole local-audio pipeline inside the thin proxy. They now live in the +// music sidecar (`sidecar/music/cliamp-ws.ts`), and this is what is left of them: authenticate the browser +// (done before the upgrade, in server.tsx), then pass frames through in both directions without reading +// them. Text or binary, no inspection — same dumb-pipe shape as the vault notifications relay. + +export type CliampWSData = { + provider: 'cliamp' | 'cliamp-audio'; + search?: string; // the browser's query string, forwarded minus the platform token +}; + +type UpstreamState = { + ws: WebSocket | null; + queue: (string | Uint8Array)[]; + ready: boolean; +}; + +// Bun hands frames over as `string | Buffer`; a Buffer is a Uint8Array at runtime, so forward as-is +// rather than copying every PCM chunk. +const asPayload = (raw: string | Buffer): string | Uint8Array => + typeof raw === 'string' ? raw : (raw as Uint8Array); + +// The sidecar has no use for the platform JWT and should not see it. +const forwardedQuery = (search: string | undefined): string => { + const params = new URLSearchParams(search ?? ''); + params.delete('token'); + const qs = params.toString(); + return qs ? `?${qs}` : ''; +}; + +function createCliampRelay(path: string) { + const upstreams = new Map, UpstreamState>(); + + return { + open(ws: ServerWebSocket) { + const base = getMusicServerWsUrl(); + if (!base) { + try { + ws.close(1011, 'Music sidecar not available'); + } catch { + /* already closed */ + } + return; + } + + const state: UpstreamState = { ws: null, queue: [], ready: false }; + upstreams.set(ws, state); + + const upstream = new WebSocket(`${base}${path}${forwardedQuery(ws.data.search)}`); + upstream.binaryType = 'arraybuffer'; + state.ws = upstream; + + upstream.addEventListener('open', () => { + state.ready = true; + for (const m of state.queue) upstream.send(m); + state.queue.length = 0; + }); + upstream.addEventListener('message', (ev) => { + try { + ws.send(ev.data as string | ArrayBuffer); + } catch { + /* client gone */ + } + }); + upstream.addEventListener('close', (ev) => { + upstreams.delete(ws); + try { + ws.close(ev.code || 1000, ev.reason || ''); + } catch { + /* already closed */ + } + }); + upstream.addEventListener('error', () => { + upstreams.delete(ws); + try { + ws.close(1011, 'upstream error'); + } catch { + /* already closed */ + } + }); + }, + + message(ws: ServerWebSocket, raw: string | Buffer) { + const state = upstreams.get(ws); + if (!state) return; + const payload = asPayload(raw); + if (state.ready && state.ws) state.ws.send(payload); + else state.queue.push(payload); // buffer until the upstream socket opens + }, + + close(ws: ServerWebSocket) { + const state = upstreams.get(ws); + if (!state) return; + try { + state.ws?.close(); + } catch { + /* already closed */ + } + upstreams.delete(ws); + }, + + drain() {}, + }; +} + +export const cliampWebsocket = createCliampRelay('/cliamp/ws'); +export const cliampAudioWebsocket = createCliampRelay('/cliamp/audio/ws'); diff --git a/src/servers/api/cliamp/websocket.ts b/src/servers/api/cliamp/websocket.ts deleted file mode 100644 index aabdc7f0..00000000 --- a/src/servers/api/cliamp/websocket.ts +++ /dev/null @@ -1,201 +0,0 @@ -import type { ServerWebSocket } from 'bun'; -import { spawn, type Subprocess } from 'bun'; -import { resolve, normalize, dirname, join } from 'node:path'; -import { fileURLToPath } from 'node:url'; -import { getOwnerHomeDir } from '@@/data-path'; - -const __dirname = dirname(fileURLToPath(import.meta.url)); -const ASOUNDRC_PATH = join(__dirname, 'asoundrc'); - -type WSData = { - userId: number; - email: string; - username: string; - files: string; -}; - -type CliampSession = { - proc: Subprocess<'pipe', 'pipe', 'pipe'>; - closed: boolean; -}; - -const sessions = new Map, CliampSession>(); - -const sendOutput = (ws: ServerWebSocket, data: string) => { - try { - ws.send(JSON.stringify({ type: 'output', data })); - } catch { - // ws already closed - } -}; - -const sendExit = (ws: ServerWebSocket) => { - try { - ws.send(JSON.stringify({ type: 'exit' })); - } catch { - // ws already closed - } -}; - -const validateFilePaths = (files: string[], homeDir: string): string[] | null => { - const resolved = files.map((f) => normalize(resolve(homeDir, f))); - for (const p of resolved) { - if (!p.startsWith(homeDir)) return null; - } - return resolved; -}; - -const findCliamp = (): string | null => { - const which = Bun.which('cliamp'); - if (which) return which; - const candidates = [ - process.env.GOPATH ? `${process.env.GOPATH}/bin/cliamp` : null, - `${process.env.HOME}/.local/go-path/bin/cliamp`, - `${process.env.HOME}/go/bin/cliamp`, - ]; - for (const bin of candidates) { - if (!bin) continue; - try { - const stat = Bun.spawnSync({ cmd: ['test', '-x', bin], stdout: 'ignore', stderr: 'ignore' }); - if (stat.exitCode === 0) return bin; - } catch { - /* ignore */ - } - } - return null; -}; - -const shellEscape = (s: string) => `'${s.replace(/'/g, "'\\''")}'`; - -export const cliampWebsocket = { - async open(ws: ServerWebSocket) { - const { email, files: filesParam } = ws.data; - - if (!filesParam) { - sendOutput(ws, '\r\n[Error] No files specified.\r\n'); - return; - } - - const cliampPath = findCliamp(); - if (!cliampPath) { - sendOutput(ws, '\r\n[Error] cliamp not found on host.\r\n'); - return; - } - - const homeDir = getOwnerHomeDir(email); - const rawFiles = [filesParam]; - - // Resolve paths relative to user home dir - const absoluteFiles = rawFiles.map((f) => { - if (f.startsWith('/')) return `${homeDir}${f}`; - return `${homeDir}/${f}`; - }); - - const validated = validateFilePaths(absoluteFiles, homeDir); - if (!validated) { - sendOutput(ws, '\r\n[Error] Invalid file path.\r\n'); - return; - } - - // Use `script` to allocate a PTY for cliamp (avoids node-pty dependency) - // script -qfc ' ' /dev/null - const cliampCmd = [shellEscape(cliampPath), ...validated.map(shellEscape)].join(' '); - console.log(`[cliamp] spawning: ${cliampCmd}`); - let proc: Subprocess<'pipe', 'pipe', 'pipe'>; - try { - proc = spawn({ - cmd: ['script', '-qfc', cliampCmd, '/dev/null'], - stdin: 'pipe', - stdout: 'pipe', - stderr: 'pipe', - cwd: homeDir, - env: { ...process.env, TERM: 'xterm-256color', PULSE_SINK: 'virtual_out', ALSA_CONFIG_PATH: ASOUNDRC_PATH }, - }); - } catch (err) { - const message = err instanceof Error ? err.message : 'Failed to start cliamp'; - sendOutput(ws, `\r\n[Error] ${message}\r\n`); - return; - } - - const session: CliampSession = { proc, closed: false }; - sessions.set(ws, session); - - // Pump stdout → WS - const reader = proc.stdout.getReader(); - const pump = async () => { - try { - while (!session.closed) { - const { done, value } = await reader.read(); - if (done) break; - if (value && !session.closed) { - sendOutput(ws, new TextDecoder().decode(value)); - } - } - } catch { - // stream ended - } finally { - if (!session.closed) { - session.closed = true; - sendExit(ws); - } - } - }; - pump(); - - // Also read stderr (cliamp may write there) - const stderrReader = proc.stderr.getReader(); - const pumpStderr = async () => { - try { - while (!session.closed) { - const { done, value } = await stderrReader.read(); - if (done) break; - if (value && !session.closed) { - sendOutput(ws, new TextDecoder().decode(value)); - } - } - } catch { - // stream ended - } - }; - pumpStderr(); - - // On process exit → notify client - proc.exited.then((code) => { - console.log(`[cliamp] process exited code=${code}`); - if (!session.closed) { - session.closed = true; - sendExit(ws); - } - sessions.delete(ws); - }); - }, - - message(ws: ServerWebSocket, raw: string | Buffer) { - const session = sessions.get(ws); - if (!session || session.closed) return; - - try { - const msg = JSON.parse(typeof raw === 'string' ? raw : raw.toString()); - if (msg.type === 'input' && msg.data) { - session.proc.stdin.write(msg.data); - } - } catch { - // ignore - } - }, - - close(ws: ServerWebSocket) { - const session = sessions.get(ws); - if (session) { - session.closed = true; - try { - session.proc.kill(); - } catch { - /* ignore */ - } - sessions.delete(ws); - } - }, - - drain() {}, -}; diff --git a/src/servers/api/desktop/rest.ts b/src/servers/api/desktop/rest.ts index 3cac087a..671b06a3 100644 --- a/src/servers/api/desktop/rest.ts +++ b/src/servers/api/desktop/rest.ts @@ -1,19 +1,19 @@ import { createRouter } from '../../create-router'; -import { getVncPassword } from './vnc-config'; import * as sidecar from '@@/sidecar-registry'; export const desktopRouter = createRouter(); // The desktop UI asks for the password before it can open the WebSocket — and that WebSocket is what // starts the VNC session. So this cannot wait for a session to exist: on a fresh install nothing has -// ever written the password, and answering "not configured" deadlocked the page permanently. Ask the -// sidecar to provision it instead; it owns the .vnc directory and the call is idempotent. +// ever written the password, and answering "not configured" deadlocked the page permanently. +// +// Officer does not read the password file itself. It lives in the owner's ~/.vnc, next to the rfbauth +// file x11vnc authenticates against, and the sidecar is the process that writes both — so it is the +// process that answers for them too. `vnc:ensure-password` is idempotent: it returns the existing +// pair when both files are already there, and provisions them when they are not. desktopRouter.get('/vnc-password', async (ctx) => { const user = ctx.get('user'); - const existing = await getVncPassword(user.email); - if (existing) return ctx.json({ password: existing }); - if (!sidecar.isVncConnected()) { return ctx.json({ error: 'VNC sidecar is not connected' }, 503); } diff --git a/src/servers/api/desktop/vnc-config.ts b/src/servers/api/desktop/vnc-config.ts deleted file mode 100644 index f808255f..00000000 --- a/src/servers/api/desktop/vnc-config.ts +++ /dev/null @@ -1,10 +0,0 @@ -import { join } from 'node:path'; -import { getOwnerHomeDir } from '@@/data-path'; - -const getVncDir = (email: string): string => join(getOwnerHomeDir(email), '.vnc'); - -export async function getVncPassword(email: string): Promise { - const file = Bun.file(join(getVncDir(email), 'password')); - if (!(await file.exists())) return null; - return (await file.text()).trim(); -} diff --git a/src/servers/api/email/email.ts b/src/servers/api/email/email.ts index dddd0ebf..3066fa4e 100644 --- a/src/servers/api/email/email.ts +++ b/src/servers/api/email/email.ts @@ -14,7 +14,9 @@ export const emailRouter = createRouter(); emailRouter.route('/accounts', accountsRouter); // ── Sending (SMTP) — sends as the connected account using its app password ── -async function getSmtpTransport(userId: number): Promise<{ transport: ReturnType; from: string }> { +async function getSmtpTransport( + userId: number, +): Promise<{ transport: ReturnType; from: string }> { const accounts = await getEmailAccounts(userId); const acct = accounts.find((a) => a.enabled) ?? accounts[0]; if (!acct) throw errors.BAD_REQUEST('No email account configured'); @@ -35,9 +37,14 @@ emailRouter.post('/send', async (ctx) => { const to = str(form.to).trim(); if (!to) throw errors.BAD_REQUEST('At least one recipient is required'); - const toFiles = (raw: unknown) => (Array.isArray(raw) ? raw : raw ? [raw] : []).filter((f): f is File => f instanceof File); + const toFiles = (raw: unknown) => + (Array.isArray(raw) ? raw : raw ? [raw] : []).filter((f): f is File => f instanceof File); const attachments = await Promise.all( - toFiles(form.files).map(async (f) => ({ filename: f.name, content: Buffer.from(await f.arrayBuffer()), contentType: f.type || undefined })), + toFiles(form.files).map(async (f) => ({ + filename: f.name, + content: Buffer.from(await f.arrayBuffer()), + contentType: f.type || undefined, + })), ); // Inline images: cid `inline-` matches the `` the composer put in the html. const inline = await Promise.all( @@ -164,14 +171,24 @@ emailRouter.get('/search', async (ctx) => { } }); +// `folder` is request input and was being interpolated straight into the SQL. It is bound now — the +// fragment and its parameters travel together because each call site builds several statements from the +// same fragment and has to spread the params in the right order. +type FolderFilter = { where: string; params: string[] }; +const folderFilter = (folder: string): FolderFilter => + folder === 'all' + ? { where: 'deleted = 0', params: [] } + : { where: 'deleted = 0 AND labels LIKE ?', params: [`%${folder}%`] }; + emailRouter.get('/messages', async (ctx) => { const user = ctx.get('user'); - const page = Number(ctx.req.query('page') ?? '1'); - const limit = Number(ctx.req.query('limit') ?? '50'); + // `|| n` also catches NaN from a non-numeric query param, which used to reach the bindings as NaN. + const page = Math.max(Number(ctx.req.query('page') ?? '1') || 1, 1); + const limit = Math.max(Number(ctx.req.query('limit') ?? '50') || 50, 1); const folder = ctx.req.query('folder') ?? 'inbox'; const offset = (page - 1) * limit; - const folderWhere = folder === 'all' ? 'deleted = 0' : `deleted = 0 AND labels LIKE '%${folder}%'`; + const { where: folderWhere, params: folderParams } = folderFilter(folder); const db = await openUserEmailDb(user.email, user.id); if (!db) return ctx.json({ messages: [], total: 0 }); @@ -190,10 +207,10 @@ emailRouter.get('/messages', async (ctx) => { ) WHERE rn = 1 ORDER BY date DESC LIMIT ? OFFSET ?`, ) - .all(limit, offset) as Record[]; + .all(...folderParams, limit, offset) as Record[]; const countRow = db .query(`SELECT COUNT(DISTINCT COALESCE(thread_id, id)) as total FROM emails WHERE ${folderWhere}`) - .get() as { total: number }; + .get(...folderParams) as { total: number }; const messages = rows.map(rowToSummary); return ctx.json({ messages, total: countRow.total }); } finally { @@ -202,9 +219,9 @@ emailRouter.get('/messages', async (ctx) => { }); function buildMessage(db: ReturnType, row: Record): EmailMessage { - const attachmentRows = db.query('SELECT * FROM attachments WHERE email_id = ? ORDER BY idx').all(row.id as string) as Array< - Record - >; + const attachmentRows = db + .query('SELECT * FROM attachments WHERE email_id = ? ORDER BY idx') + .all(row.id as string) as Array>; const from = row.from_name ? `${row.from_name} <${row.from_address}>` : (row.from_address as string); return { id: row.id as string, @@ -250,9 +267,10 @@ emailRouter.get('/thread/:id', async (ctx) => { const db = await openUserEmailDb(user.email, user.id); if (!db) return ctx.text('Not found', 404); try { - const head = db.query('SELECT thread_id, subject FROM emails WHERE id = ? AND deleted = 0').get(id) as - | { thread_id: string | null; subject: string } - | null; + const head = db.query('SELECT thread_id, subject FROM emails WHERE id = ? AND deleted = 0').get(id) as { + thread_id: string | null; + subject: string; + } | null; if (!head) return ctx.text('Not found', 404); const threadKey = head.thread_id ?? id; @@ -363,23 +381,24 @@ emailRouter.get('/stats', async (ctx) => { const user = ctx.get('user'); const folder = ctx.req.query('folder') ?? 'inbox'; - const folderWhere = folder === 'all' ? 'deleted = 0' : `deleted = 0 AND labels LIKE '%${folder}%'`; + const { where: folderWhere, params: folderParams } = folderFilter(folder); const db = await openUserEmailDb(user.email, user.id); if (!db) return ctx.json({ total: 0, byDomain: [], bySender: [] }); try { - const total = (db.query(`SELECT COUNT(*) as count FROM emails WHERE ${folderWhere}`).get() as { count: number }) - .count; + const total = ( + db.query(`SELECT COUNT(*) as count FROM emails WHERE ${folderWhere}`).get(...folderParams) as { count: number } + ).count; const byDomain = db .query( `SELECT from_domain, COUNT(*) as count FROM emails WHERE ${folderWhere} GROUP BY from_domain ORDER BY count DESC LIMIT 20`, ) - .all() as Array<{ from_domain: string; count: number }>; + .all(...folderParams) as Array<{ from_domain: string; count: number }>; const bySender = db .query( `SELECT from_address, from_name, COUNT(*) as count FROM emails WHERE ${folderWhere} GROUP BY from_address ORDER BY count DESC LIMIT 20`, ) - .all() as Array<{ from_address: string; from_name: string; count: number }>; + .all(...folderParams) as Array<{ from_address: string; from_name: string; count: number }>; return ctx.json({ total, byDomain, bySender }); } finally { diff --git a/src/servers/api/music/router.ts b/src/servers/api/music/router.ts index c3cd6420..fac0dba8 100644 --- a/src/servers/api/music/router.ts +++ b/src/servers/api/music/router.ts @@ -24,15 +24,14 @@ musicRouter.all('/*', async (ctx) => { const target = `${baseUrl}${subpath}${url.search}`; // A from-scratch reindex holds this proxied connection open for minutes with no bytes flowing, which - // the main server's 60s idle timeout would drop. Extend it to 30 min for the build/progress endpoints - // (Bun passes the server as Hono's env). Matches the sidecar's own per-request extension. - if (subpath === '/reindex' || subpath === '/reindex/stream') { - const server = ctx.env as { timeout?: (req: Request, seconds: number) => void } | undefined; - try { - server?.timeout?.(ctx.req.raw, 1800); - } catch { - /* older Bun / no per-request timeout — the build still completes in the background */ - } + // the main server's 60s idle timeout would drop. Extend every request under this prefix to 30 min (Bun + // passes the server as Hono's env) — the proxy must not know which of the sidecar's routes are slow, and + // the sidecar applies its own per-request extension anyway. + const server = ctx.env as { timeout?: (req: Request, seconds: number) => void } | undefined; + try { + server?.timeout?.(ctx.req.raw, 1800); + } catch { + /* older Bun / no per-request timeout — the build still completes in the background */ } const method = ctx.req.method; diff --git a/src/servers/api/music/sidecar-server.ts b/src/servers/api/music/sidecar-server.ts index 2b92611a..22cd23f6 100644 --- a/src/servers/api/music/sidecar-server.ts +++ b/src/servers/api/music/sidecar-server.ts @@ -16,3 +16,8 @@ sidecar.on('music:server', (msg) => { export function getMusicServerUrl(): string | null { return serverPort ? `http://127.0.0.1:${serverPort}` : null; } + +/** Same server, ws:// scheme — for the cliamp sockets the platform relays (see api/cliamp/relay.ts). */ +export function getMusicServerWsUrl(): string | null { + return serverPort ? `ws://127.0.0.1:${serverPort}` : null; +} 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/api/terminal/websocket.ts b/src/servers/api/terminal/websocket.ts index 682059b3..97dcb95e 100644 --- a/src/servers/api/terminal/websocket.ts +++ b/src/servers/api/terminal/websocket.ts @@ -1,5 +1,4 @@ import type { ServerWebSocket } from 'bun'; -import { join } from 'node:path'; import { sendPtyCommand, sendPtyCommandAsync, on, isTerminalConnected } from '@@/sidecar-registry'; import type { PtyInitConfig } from '../../sidecar/protocol'; @@ -16,8 +15,7 @@ type WSData = { type BridgeSession = { client: ServerWebSocket; sessionId: string; - unsubOutput: (() => void) | null; - unsubExit: (() => void) | null; + unsubs: Array<() => void>; }; const sessions = new Map, BridgeSession>(); @@ -35,13 +33,6 @@ const sendOutput = (ws: ServerWebSocket, data: string) => { } }; -const resolveCwd = (home: string, cwd?: string) => { - if (!cwd || cwd === '~') return home; - if (cwd.startsWith('~/')) return join(home, cwd.slice(2)); - if (cwd.startsWith('/')) return cwd; - return home; -}; - export const terminalWebsocket = { async open(ws: ServerWebSocket) { const { email, username } = ws.data; @@ -55,40 +46,27 @@ export const terminalWebsocket = { const sessionId = ws.data.sessionId ?? `host-${ws.data.userId}`; - // The server owner is the only account, so the terminal is always a plain host shell. - const config: PtyInitConfig = { + // Everything this bridge knows: which session, which folder the panel was opened on, and how big the + // client's viewport is. The shell, its arguments and the home directory are the sidecar's — it is the + // process that spawns them, and officer has no business reading the owner's SHELL and HOME to guess. + const config: PtyInitConfig = { sessionId, cwd: ws.data.cwd, cols: ws.data.cols, rows: ws.data.rows }; + + // The sidecar emits one global stream, so each frame is filtered down to this session and relabelled. + const relay = (event: 'pty:output' | 'pty:replay' | 'pty:exit', clientType: string) => + on(event, (msg) => { + if (msg.type !== event || msg.sessionId !== sessionId) return; + try { + ws.send(JSON.stringify({ type: clientType, data: 'data' in msg ? msg.data : undefined })); + } catch { + // ws already closed + } + }); + + const session: BridgeSession = { + client: ws, sessionId, - host: true, - shell: { command: process.env.SHELL ?? '/bin/zsh', args: ['-i'] }, - cwd: resolveCwd(process.env.HOME!, ws.data.cwd), - homeDir: process.env.HOME!, - userLabel: email, - cols: ws.data.cols, - rows: ws.data.rows, + unsubs: [relay('pty:output', 'output'), relay('pty:replay', 'replay'), relay('pty:exit', 'exit')], }; - - // Subscribe to events for this session - const unsubOutput = on('pty:output', (msg) => { - if (msg.type === 'pty:output' && msg.sessionId === sessionId) { - try { - ws.send(JSON.stringify({ type: 'output', data: msg.data })); - } catch { - // ws already closed - } - } - }); - - const unsubExit = on('pty:exit', (msg) => { - if (msg.type === 'pty:exit' && msg.sessionId === sessionId) { - try { - ws.send(JSON.stringify({ type: 'exit' })); - } catch { - // ws already closed - } - } - }); - - const session: BridgeSession = { client: ws, sessionId, unsubOutput, unsubExit }; sessions.set(ws, session); // Send init command to PTY sidecar @@ -97,8 +75,7 @@ export const terminalWebsocket = { } catch (err) { const message = err instanceof Error ? err.message : 'Failed to initialize terminal'; sendOutput(ws, `\r\n[Terminal error] ${message}\r\n`); - unsubOutput(); - unsubExit(); + for (const unsub of session.unsubs) unsub(); sessions.delete(ws); } }, @@ -126,16 +103,9 @@ export const terminalWebsocket = { }); } break; - case 'cwd': - if (msg.path) { - sendPtyCommand({ - type: 'pty:input', - id: nextId(), - sessionId: session.sessionId, - data: `cd ${JSON.stringify(msg.path)}\r`, - }); - } - break; + // There was a 'cwd' case here that typed `cd \r` into the user's shell. No frontend sends + // that message — the browser composes its own `cd` (Terminal.tsx / CommandTerminalWrapper.tsx) — + // so it was unreachable, and synthesizing keystrokes is not a thing a proxy should do. } } catch { // ignore malformed messages @@ -145,8 +115,7 @@ export const terminalWebsocket = { close(ws: ServerWebSocket) { const session = sessions.get(ws); if (session) { - session.unsubOutput?.(); - session.unsubExit?.(); + for (const unsub of session.unsubs) unsub(); // Don't kill PTY — it can be reattached sessions.delete(ws); } diff --git a/src/servers/api/users/provision.ts b/src/servers/api/users/provision.ts index 69b70b4e..9cdcebcf 100644 --- a/src/servers/api/users/provision.ts +++ b/src/servers/api/users/provision.ts @@ -3,7 +3,7 @@ import { join } from 'node:path'; import { DATA_PATH, getHomeDir, toShellUsername } from '@@/data-path'; import { generateContainerContext, generateClaudeSettings } from '@@/generate-container-context'; -const TEMPLATE_DIR = join(import.meta.dir, '../terminal/templates'); +const TEMPLATE_DIR = join(import.meta.dir, 'templates'); const copyTemplate = async (src: string, dest: string) => { if (existsSync(dest)) return; diff --git a/src/servers/api/terminal/templates/.tmux.conf b/src/servers/api/users/templates/.tmux.conf similarity index 100% rename from src/servers/api/terminal/templates/.tmux.conf rename to src/servers/api/users/templates/.tmux.conf diff --git a/src/servers/api/terminal/templates/.zshenv b/src/servers/api/users/templates/.zshenv similarity index 100% rename from src/servers/api/terminal/templates/.zshenv rename to src/servers/api/users/templates/.zshenv diff --git a/src/servers/api/terminal/templates/.zshrc b/src/servers/api/users/templates/.zshrc similarity index 100% rename from src/servers/api/terminal/templates/.zshrc rename to src/servers/api/users/templates/.zshrc diff --git a/src/servers/api/terminal/templates/starship-officer.toml b/src/servers/api/users/templates/starship-officer.toml similarity index 100% rename from src/servers/api/terminal/templates/starship-officer.toml rename to src/servers/api/users/templates/starship-officer.toml diff --git a/src/servers/channels/send-claude-code.ts b/src/servers/channels/send-claude-code.ts index 639ca7c8..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 = { @@ -18,8 +18,8 @@ type ClaudeCodeResult = { cost: MessageCost; }; -export function clearClaudeCodeSession(sessionKey: string, email?: string): void { - sidecar.clearClaudeSession(sessionKey, email); +export function clearClaudeCodeSession(sessionKey: string): void { + sidecar.clearClaudeSession(sessionKey); } export async function sendClaudeCode(params: ClaudeCodeParams): Promise { @@ -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,22 +48,22 @@ 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); return { kill: () => { - sidecar.killClaude(params.sessionKey, params.email); + sidecar.killClaude(params.sessionKey); unsub(); }, }; 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 75b41a2d..28bc55f1 100644 --- a/src/servers/sidecar-registry.ts +++ b/src/servers/sidecar-registry.ts @@ -1,6 +1,4 @@ -import { resolve } from 'node:path'; import type { ServerWebSocket } from 'bun'; -import type { Subprocess } from 'bun'; import type { SidecarCommand, SidecarEvent, @@ -15,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 { TurnMessage } from './api/chat/types'; // ── Types ── @@ -113,13 +111,6 @@ function findSidecarByCapability(cap: string): RegisteredSidecar | undefined { return undefined; } -function findSidecarByName(name: string): RegisteredSidecar | undefined { - for (const sc of sidecars.values()) { - if (sc.name === name) return sc; - } - return undefined; -} - // ── Event dispatch ── function dispatchEvent(msg: SidecarEvent | PtyEvent) { @@ -195,82 +186,27 @@ function sendFireToSidecar(sc: RegisteredSidecar, cmd: SidecarCommand | PtyComma sc.ws.send(JSON.stringify(cmd)); } -// ── On-demand Claude sidecar spawning ── +// ── Waiting for a sidecar to appear ── -const USER_INSTANCE_SCRIPT = resolve(import.meta.dir, 'sidecar/claude/user-instance.ts'); -const SIDECAR_SPAWN_TIMEOUT_MS = 15_000; +// Officer no longer spawns any sidecar; PM2 owns every one of them. The only thing left to handle is +// startup order — PM2 brings `officer` and its peers up together, so the first request after a boot can +// arrive a beat before the sidecar has finished dialling in. Wait briefly rather than failing the +// request. (This replaces ~77 lines of spawn-and-poll: `ensureClaudeSidecar`, +// `spawnAndWaitForRegistration`, and the per-email `claudeProcs`/`claudeSpawnWaiters` maps.) +const CAPABILITY_WAIT_MS = 15_000; +const CAPABILITY_POLL_MS = 100; -const claudeProcs = new Map(); -const claudeSpawnWaiters = new Map>(); - -async function ensureClaudeSidecar(email: string): Promise { - const name = `claude:${email}`; - - // Already registered? - const existing = findSidecarByName(name); +async function waitForCapability(cap: string, timeoutMs = CAPABILITY_WAIT_MS): Promise { + const existing = findSidecarByCapability(cap); if (existing) return existing; - // Already spawning? - const waiter = claudeSpawnWaiters.get(email); - if (waiter) return waiter; - - // Spawn and wait for registration - const promise = spawnAndWaitForRegistration(email, name); - claudeSpawnWaiters.set(email, promise); - try { - return await promise; - } finally { - claudeSpawnWaiters.delete(email); + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + await Bun.sleep(CAPABILITY_POLL_MS); + const sc = findSidecarByCapability(cap); + if (sc) return sc; } -} - -async function spawnAndWaitForRegistration(email: string, name: string): Promise { - const proxySecret = await getProxySecret(); - const proxyPort = process.env.ANTHROPIC_PROXY_PORT ?? '5051'; - - const env: Record = { - ...(process.env as Record), - CLAUDE_USER_EMAIL: email, - ANTHROPIC_BASE_URL: `http://127.0.0.1:${proxyPort}`, - ANTHROPIC_API_KEY: proxySecret, - }; - - const proc = Bun.spawn(['bun', 'run', USER_INSTANCE_SCRIPT], { - env, - stdout: 'inherit', - stderr: 'inherit', - }); - - claudeProcs.set(email, proc); - - // Clean up on exit - proc.exited.then(() => { - claudeProcs.delete(email); - }); - - // Wait for the sidecar to register - return new Promise((resolve, reject) => { - const timeout = setTimeout(() => { - unsub(); - reject(new Error(`Claude sidecar for ${email} failed to register within ${SIDECAR_SPAWN_TIMEOUT_MS}ms`)); - }, SIDECAR_SPAWN_TIMEOUT_MS); - - // Poll for registration (the sidecar connects via WebSocket and registerSidecar is called) - const check = () => { - const sc = findSidecarByName(name); - if (sc) { - clearTimeout(timeout); - clearInterval(interval); - resolve(sc); - } - }; - const interval = setInterval(check, 50); - - const unsub = () => { - clearTimeout(timeout); - clearInterval(interval); - }; - }); + throw new Error(`No sidecar with capability "${cap}" registered within ${timeoutMs}ms`); } // ── Public API ── @@ -303,10 +239,14 @@ export function getProxySecretSync(): string { return cachedState?.proxySecret ?? ''; } -// ── Claude Code (per-user routing) ── +// ── Claude Code (the `officer-agent` sidecar, capability 'claude') ── + +// Single-user platform, so there is exactly one agent sidecar and it is found by capability like every +// other one. The `email` on the params is still passed through to the sidecar — it needs it to resolve +// paths — but officer no longer uses it to *locate* anything. export async function spawnClaude(params: ClaudeSpawnParams): Promise { - const sc = await ensureClaudeSidecar(params.email); + const sc = await waitForCapability('claude'); const res = await sendCommandToSidecar(sc, { type: 'claude:spawn', id: nextId(), params }, LONG_TIMEOUT_MS); if (res.type === 'claude:result') return res.result; if (res.type === 'claude:error') throw new Error(res.error); @@ -314,46 +254,34 @@ export async function spawnClaude(params: ClaudeSpawnParams): Promise { - const sc = await ensureClaudeSidecar(params.email); + const sc = await waitForCapability('claude'); const res = await sendCommandToSidecar(sc, { type: 'claude:spawn-streaming', id: nextId(), params }); if (res.type === 'claude:spawned') return; if (res.type === 'claude:error') throw new Error(res.error); throw new Error('Unexpected response'); } -export function killClaude(sessionKey: string, email: string): void { - const sc = findSidecarByName(`claude:${email}`); - if (sc) sendFireToSidecar(sc, { type: 'claude:kill', id: nextId(), sessionKey }); +export function killClaude(sessionKey: string): void { + sendFire('claude', { type: 'claude:kill', id: nextId(), sessionKey }); } // Interrupt the current turn but keep the persistent session warm (the "stop" button). -export function interruptClaude(sessionKey: string, email: string): void { - const sc = findSidecarByName(`claude:${email}`); - if (sc) sendFireToSidecar(sc, { type: 'claude:interrupt', id: nextId(), sessionKey }); +export function interruptClaude(sessionKey: string): void { + sendFire('claude', { type: 'claude:interrupt', id: nextId(), sessionKey }); } -export function clearClaudeSession(sessionKey: string, email?: string): void { - if (email) { - const sc = findSidecarByName(`claude:${email}`); - if (sc) sendFireToSidecar(sc, { type: 'claude:clear-session', id: nextId(), sessionKey }); - } else { - // Broadcast to all claude sidecars (used when email is not available) - for (const sc of sidecars.values()) { - if (sc.capabilities.includes('claude')) { - sendFireToSidecar(sc, { type: 'claude:clear-session', id: nextId(), sessionKey }); - } - } - } +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); }); } @@ -370,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/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/state.ts b/src/servers/sidecar/claude/state.ts index 60151b10..b55e63ec 100644 --- a/src/servers/sidecar/claude/state.ts +++ b/src/servers/sidecar/claude/state.ts @@ -101,6 +101,31 @@ export async function flushAndSave(): Promise { await saveState(); } +/** + * Read the Anthropic proxy secret out of the *proxy* sidecar's state file. + * + * The proxy (`officer-anthropic-proxy`) and the agent (`officer-agent`) keep separate state — see + * `initPaths`: `DATA_PATH/sidecar/` versus `DATA_PATH//sidecar/` — so the agent cannot reach + * the secret through `getState()`. It used to be handed the secret in env by the main server, and + * needing that handoff is precisely why the agent had to be spawned by `officer` (and therefore died + * with it). Reading it off disk keeps the two processes independent, with the proxy still the only + * writer. + * + * Returns '' when the secret is not on disk yet: `ensureProxySecret` persists through a 30s debounce, + * so a brand-new install has a window where the file exists without it. Callers should treat '' as + * "retry later" rather than fatal. + */ +export function readProxySecretFromDisk(): string { + try { + const proxyStateFile = join(DATA_PATH, 'sidecar', 'claude-state.json'); + if (!existsSync(proxyStateFile)) return ''; + const parsed = JSON.parse(readFileSync(proxyStateFile, 'utf-8')) as Partial; + return parsed.proxySecret ?? ''; + } catch { + return ''; + } +} + // ── Lockfile ── export function acquireLock(): boolean { 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..51bd79b9 --- /dev/null +++ b/src/servers/sidecar/claude/turn-stream.ts @@ -0,0 +1,102 @@ +import type { ChatEvent, TurnMessage } from '../../api/chat/types'; + +// 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. +// +// 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 59e9c5b3..3e7205ed 100644 --- a/src/servers/sidecar/claude/user-instance.ts +++ b/src/servers/sidecar/claude/user-instance.ts @@ -2,30 +2,40 @@ import { existsSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs'; import { join, resolve } from 'node:path'; import { homedir } from 'node:os'; import type { SidecarCommand, SidecarEvent } from '../protocol'; -import { initPaths, loadState, flushAndSave, acquireLock, releaseLock } from './state'; +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'; import { sign } from '../../jwt'; -import { getUserByEmail, getEmailAccounts } from 'officerdb'; +import { getUserByEmail, getOwnerUser, getEmailAccounts } from 'officerdb'; -const email = process.env.CLAUDE_USER_EMAIL; -if (!email) { - console.error('[user-instance] CLAUDE_USER_EMAIL is required'); - process.exit(1); +// PM2 starts this sidecar with no user in its env. Single-user platform, so resolve the owner from the +// database rather than being told who to run as by the main server — one less thing that has to come +// from `officer` before this process can work. CLAUDE_USER_EMAIL still wins when set, for manual runs. +async function resolveOwner() { + const explicit = process.env.CLAUDE_USER_EMAIL?.trim(); + for (;;) { + const user = explicit ? await getUserByEmail(explicit) : await getOwnerUser(); + if (user) return user; + // Fresh install: wait for POST /auth/bootstrap instead of exiting into a PM2 restart loop. + console.log(`[agent] no ${explicit ? `user "${explicit}"` : 'owner account'} yet — retrying in 5s`); + await Bun.sleep(5_000); + } } +const dbUser = await resolveOwner(); +const email = dbUser.email; + const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data'); -const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${process.env.PORT ?? '5000'}`; -const OFFICER_API_URL = process.env.OFFICER_API_URL ?? `http://127.0.0.1:${process.env.PORT ?? '9010'}`; +// Same officer instance for both, so the fallback port has to agree. It used to default to 5000 for the +// WebSocket and 9010 for the REST base, which would have split them apart if PORT were ever unset. +const OFFICER_PORT = process.env.PORT ?? '9010'; +const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${OFFICER_PORT}`; +const OFFICER_API_URL = process.env.OFFICER_API_URL ?? `http://127.0.0.1:${OFFICER_PORT}`; const MCP_SERVER_SCRIPT = resolve(import.meta.dir, '../../mcp-tool-server.ts'); // Mint a long-lived JWT for this user so tools (e.g. gmail) can call back to dev-platform as them -const dbUser = await getUserByEmail(email); -if (!dbUser) { - console.error(`[user-instance] no user found for ${email}`); - process.exit(1); -} const OFFICER_AUTH_TOKEN = await sign({ id: dbUser.id, email, username: dbUser.username }, '30d'); // Single-user platform: the owner runs Claude with no isolation — real HOME, real ~/.claude — so @@ -49,7 +59,7 @@ process.env.HOME = homeDir; initPaths(email); if (!acquireLock()) { - console.error(`[claude:${email}] another instance is already running (lock file exists with live PID)`); + console.error(`[agent] another instance is already running for ${email} (lock file exists with live PID)`); process.exit(1); } @@ -85,6 +95,30 @@ function generateMcpConfig(): string { return join(contextDir, 'mcp-host.json'); } +// ── Anthropic credentials ── + +// The `claude` CLI inherits this process's env (claude-manager spawns with `process.env`), so the proxy +// endpoint and secret have to be set here. Officer used to inject both when it spawned this process; +// reading them ourselves is what lets this sidecar be a PM2 peer instead of a child of the server. +// +// Resolved lazily rather than once at boot: PM2 starts the proxy and the agent together, and +// `ensureProxySecret` persists on a 30s debounce, so on a first-ever boot the secret can be briefly +// absent. Re-checked before every spawn until it lands. +const ANTHROPIC_PROXY_PORT = process.env.ANTHROPIC_PROXY_PORT ?? '5051'; + +function ensureAnthropicEnv(): void { + process.env.ANTHROPIC_BASE_URL ??= `http://127.0.0.1:${ANTHROPIC_PROXY_PORT}`; + if (process.env.ANTHROPIC_API_KEY) return; + + const secret = readProxySecretFromDisk(); + if (secret) { + process.env.ANTHROPIC_API_KEY = secret; + console.log('[agent] anthropic proxy secret loaded from disk'); + } else { + console.warn('[agent] anthropic proxy secret not on disk yet — retrying before next spawn'); + } +} + // ── Startup ── // The owner runs un-isolated with HOME as their real home, so the generated container CLAUDE.md is @@ -92,8 +126,17 @@ function generateMcpConfig(): string { // terminal `claude` loads too. setMcpConfigPath(generateMcpConfig()); +ensureAnthropicEnv(); -console.log(`[claude:${email}] started (HOME=${homeDir})`); +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 ── @@ -106,6 +149,7 @@ async function handleCommand(cmd: SidecarCommand, reply: ReplyFn) { break; case 'claude:spawn': { + ensureAnthropicEnv(); try { const result = await claudeManager.spawnClaude(cmd.params); reply({ type: 'claude:result', id: cmd.id, result }); @@ -116,24 +160,23 @@ async function handleCommand(cmd: SidecarCommand, reply: ReplyFn) { } case 'claude:spawn-streaming': { + 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; @@ -144,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; @@ -158,9 +202,12 @@ async function handleCommand(cmd: SidecarCommand, reply: ReplyFn) { // ── Connect to API server ── +// A stable identity, not a per-email one. Officer looks this sidecar up by the 'claude' capability, so +// it no longer needs to know which user is running to find it — that was the last thing tying the +// registry's claude verbs to an email argument. const connection = createSidecarConnector({ apiUrl: `${API_URL}/api/sidecar/register`, - name: `claude:${email}`, + name: 'agent', capabilities: ['claude'], onCommand(cmd, reply) { handleCommand(cmd as SidecarCommand, reply as ReplyFn); @@ -170,7 +217,7 @@ const connection = createSidecarConnector({ // ── Graceful shutdown ── async function shutdown(signal: string) { - console.log(`[claude:${email}] ${signal} received, saving state...`); + console.log(`[agent] ${signal} received, saving state...`); connection.destroy(); await flushAndSave(); releaseLock(); diff --git a/src/servers/sidecar/email-cron.ts b/src/servers/sidecar/email-cron.ts deleted file mode 100644 index 63eac8ba..00000000 --- a/src/servers/sidecar/email-cron.ts +++ /dev/null @@ -1,92 +0,0 @@ -import { getAllSyncedAccounts, getUserById } from 'officerdb'; -import { getValidGoogleAccessToken } from '../api/integrations/google-auth'; -import * as queueRunner from './queue-runner'; - -const INTERVAL_MS = 10 * 60 * 1000; // 10 minutes - -let timer: ReturnType | null = null; - -async function tick() { - try { - const accounts = await getAllSyncedAccounts(); - if (accounts.length === 0) return; - - const allJobs = await queueRunner.listAllJobs(); - const activeEmailSyncIds = new Set( - allJobs - .filter((j) => j.type === 'email-sync' && (j.status === 'queued' || j.status === 'running')) - .map((j) => (j.meta as Record | undefined)?.emailAccountId), - ); - - for (const account of accounts) { - if (activeEmailSyncIds.has(account.id)) continue; - - const user = await getUserById(account.userId); - if (!user) continue; - - // Resolve IMAP auth - const imapAuth: Record = { user: account.email }; - if (account.authType === 'oauth') { - let accessToken: string | null = null; - try { - accessToken = await getValidGoogleAccessToken(account.userId); - } catch (err) { - console.log(`[email-cron] Skipping ${account.email}: token refresh failed —`, err instanceof Error ? err.message : err); - continue; - } - if (!accessToken) { - console.log(`[email-cron] Skipping ${account.email}: no OAuth access token`); - continue; - } - imapAuth.accessToken = accessToken; - } else { - const creds = account.credentials as Record; - imapAuth.pass = creds.password; - } - - try { - await queueRunner.enqueue({ - lane: 'email', - type: 'email-sync', - userId: user.email, - meta: { - emailAccountId: account.id, - userEmail: user.email, - account: { - id: account.id, - userId: account.userId, - email: account.email, - imapHost: account.imapHost, - imapPort: account.imapPort, - imapSecure: account.imapSecure, - provider: account.provider, - authType: account.authType, - credentials: account.credentials, - }, - imapAuth, - }, - }); - console.log(`[email-cron] Enqueued incremental sync for ${account.email}`); - } catch (err) { - console.error(`[email-cron] Failed to enqueue sync for ${account.email}:`, err instanceof Error ? err.message : err); - } - } - } catch (err) { - console.error('[email-cron] Error:', err instanceof Error ? err.message : err); - } -} - -export function initEmailCron() { - if (timer) return; - console.log(`[email-cron] Starting email sync cron (every ${INTERVAL_MS / 60_000} min)`); - timer = setInterval(tick, INTERVAL_MS); - // Run first tick after a short delay to let the queue initialize - setTimeout(tick, 30_000); -} - -export function stopEmailCron() { - if (timer) { - clearInterval(timer); - timer = null; - } -} diff --git a/src/servers/api/cliamp/asoundrc b/src/servers/sidecar/music/asoundrc similarity index 100% rename from src/servers/api/cliamp/asoundrc rename to src/servers/sidecar/music/asoundrc diff --git a/src/servers/sidecar/music/cliamp-ws.test.ts b/src/servers/sidecar/music/cliamp-ws.test.ts new file mode 100644 index 00000000..2521c2f8 --- /dev/null +++ b/src/servers/sidecar/music/cliamp-ws.test.ts @@ -0,0 +1,66 @@ +import { describe, expect, it } from 'bun:test'; +import { cliampUpgradeData, musicWebsocket } from './cliamp-ws'; + +// The player socket refuses a path before it spawns anything, so these two cases exercise the whole +// server → handler → frame path without starting cliamp. Anything that would actually play needs a real +// file and a real audio sink, so it is not tested here. + +function serveOnce() { + const server = Bun.serve({ + port: 0, + hostname: '127.0.0.1', + fetch(req, srv) { + const url = new URL(req.url); + const data = cliampUpgradeData(url.pathname, url.searchParams); + if (data && srv.upgrade(req, { data })) return undefined as unknown as Response; + return new Response('nope', { status: 400 }); + }, + websocket: musicWebsocket, + }); + return server; +} + +function firstFrame(url: string): Promise { + return new Promise((resolve, reject) => { + const ws = new WebSocket(url); + const timer = setTimeout(() => reject(new Error('no frame')), 3000); + ws.addEventListener('message', (ev) => { + clearTimeout(timer); + ws.close(); + resolve(String(ev.data)); + }); + ws.addEventListener('error', () => { + clearTimeout(timer); + reject(new Error('socket error')); + }); + }); +} + +describe('cliamp player socket', () => { + it('rejects a path that escapes the owner home', async () => { + const server = serveOnce(); + try { + const frame = await firstFrame(`ws://127.0.0.1:${server.port}/cliamp/ws?files=../../etc/passwd`); + expect(JSON.parse(frame)).toEqual({ type: 'output', data: '\r\n[Error] Invalid file path.\r\n' }); + } finally { + server.stop(true); + } + }); + + it('reports a missing files param instead of spawning', async () => { + const server = serveOnce(); + try { + const frame = await firstFrame(`ws://127.0.0.1:${server.port}/cliamp/ws`); + expect(JSON.parse(frame)).toEqual({ type: 'output', data: '\r\n[Error] No files specified.\r\n' }); + } finally { + server.stop(true); + } + }); + + it('routes only the two cliamp paths', () => { + const q = new URLSearchParams(); + expect(cliampUpgradeData('/cliamp/ws', q)).toEqual({ kind: 'player', files: '' }); + expect(cliampUpgradeData('/cliamp/audio/ws', q)).toEqual({ kind: 'capture' }); + expect(cliampUpgradeData('/stream', q)).toBeNull(); + }); +}); diff --git a/src/servers/sidecar/music/cliamp-ws.ts b/src/servers/sidecar/music/cliamp-ws.ts new file mode 100644 index 00000000..7249d166 --- /dev/null +++ b/src/servers/sidecar/music/cliamp-ws.ts @@ -0,0 +1,232 @@ +import type { ServerWebSocket } from 'bun'; +import { spawn, type Subprocess } from 'bun'; +import { homedir } from 'node:os'; +import { join, normalize, resolve, sep } from 'node:path'; +import { VIRTUAL_SINK } from './pulse-audio'; + +// Local playback, both halves of it, owned by the process that owns the audio pipeline: +// +// /cliamp/ws — runs the `cliamp` TUI player against a file and pipes its terminal both ways +// /cliamp/audio/ws — captures what the sink hears and streams it to the browser as raw PCM +// +// Officer relays these two sockets and nothing else: it authenticates the browser and forwards frames. +// Every fact below — where the binary is, what a legal path is, which sink to play into, the ALSA config, +// the capture format — is pipeline knowledge and stays here. The frame shapes are the browser's contract +// ({type:'output'|'exit'} / {type:'input'} as JSON text, PCM as binary), so they are unchanged by the move. +// +// Both sockets are loopback-only, like the rest of this server: officer is the only client. + +const ASOUNDRC_PATH = join(import.meta.dir, 'asoundrc'); + +// Single super user, so the owner's home is the root every path is resolved against — same convention as +// stream-audio.ts and the indexer. +const ROOT_DIR = process.env.HOME_DIR ?? homedir(); + +// parec's output format IS the contract with the browser's AudioWorklet: signed 16-bit LE, 44.1kHz, stereo. +const CAPTURE_ARGS = ['--format=s16le', '--rate=44100', '--channels=2', '-d', `${VIRTUAL_SINK}.monitor`]; + +export type MusicWSData = { kind: 'player'; files: string } | { kind: 'capture' }; + +type Session = { + proc: Subprocess; + closed: boolean; +}; + +const sessions = new Map, Session>(); + +const sendOutput = (ws: ServerWebSocket, data: string) => { + try { + ws.send(JSON.stringify({ type: 'output', data })); + } catch { + /* ws already closed */ + } +}; + +const sendExit = (ws: ServerWebSocket) => { + try { + ws.send(JSON.stringify({ type: 'exit' })); + } catch { + /* ws already closed */ + } +}; + +// Home-relative or leading-slash paths both mean "under the owner's home"; anything that escapes it after +// normalisation is rejected. The trailing separator matters: without it a sibling directory whose name +// merely starts with the home path would pass. +const resolveInHome = (file: string): string | null => { + const abs = normalize(resolve(ROOT_DIR, file.startsWith('/') ? `.${file}` : file)); + return abs === ROOT_DIR || abs.startsWith(ROOT_DIR + sep) ? abs : null; +}; + +const findCliamp = (): string | null => { + const which = Bun.which('cliamp'); + if (which) return which; + const candidates = [ + process.env.GOPATH ? `${process.env.GOPATH}/bin/cliamp` : null, + `${ROOT_DIR}/.local/go-path/bin/cliamp`, + `${ROOT_DIR}/go/bin/cliamp`, + ]; + for (const bin of candidates) { + if (!bin) continue; + try { + const stat = Bun.spawnSync({ cmd: ['test', '-x', bin], stdout: 'ignore', stderr: 'ignore' }); + if (stat.exitCode === 0) return bin; + } catch { + /* ignore */ + } + } + return null; +}; + +const shellEscape = (s: string) => `'${s.replace(/'/g, "'\\''")}'`; + +// Pump a byte stream into the socket until it ends; `frame` decides how it lands on the wire. +function pump( + ws: ServerWebSocket, + session: Session, + stream: ReadableStream, + frame: (ws: ServerWebSocket, chunk: Uint8Array) => void, + onEnd?: () => void, +): void { + const reader = stream.getReader(); + void (async () => { + try { + while (!session.closed) { + const { done, value } = await reader.read(); + if (done) break; + if (value && !session.closed) frame(ws, value); + } + } catch { + /* stream ended */ + } finally { + onEnd?.(); + } + })(); +} + +function openPlayer(ws: ServerWebSocket, files: string): void { + if (!files) return sendOutput(ws, '\r\n[Error] No files specified.\r\n'); + + const cliampPath = findCliamp(); + if (!cliampPath) return sendOutput(ws, '\r\n[Error] cliamp not found on host.\r\n'); + + const target = resolveInHome(files); + if (!target) return sendOutput(ws, '\r\n[Error] Invalid file path.\r\n'); + + // `script` fakes a PTY for cliamp, which avoids a node-pty native dependency here. + const cliampCmd = `${shellEscape(cliampPath)} ${shellEscape(target)}`; + console.log(`[music] cliamp spawning: ${cliampCmd}`); + let proc: Subprocess<'pipe', 'pipe', 'pipe'>; + try { + proc = spawn({ + cmd: ['script', '-qfc', cliampCmd, '/dev/null'], + stdin: 'pipe', + stdout: 'pipe', + stderr: 'pipe', + cwd: ROOT_DIR, + env: { ...process.env, TERM: 'xterm-256color', PULSE_SINK: VIRTUAL_SINK, ALSA_CONFIG_PATH: ASOUNDRC_PATH }, + }); + } catch (err) { + return sendOutput(ws, `\r\n[Error] ${err instanceof Error ? err.message : 'Failed to start cliamp'}\r\n`); + } + + const session: Session = { proc, closed: false }; + sessions.set(ws, session); + + const decoder = new TextDecoder(); + const asText = (sock: ServerWebSocket, chunk: Uint8Array) => sendOutput(sock, decoder.decode(chunk)); + const end = () => { + if (session.closed) return; + session.closed = true; + sendExit(ws); + }; + pump(ws, session, proc.stdout, asText, end); + pump(ws, session, proc.stderr, asText); // cliamp writes some output there + + void proc.exited.then((code) => { + console.log(`[music] cliamp exited code=${code}`); + end(); + sessions.delete(ws); + }); +} + +function openCapture(ws: ServerWebSocket): void { + const parecPath = Bun.which('parec'); + if (!parecPath) { + ws.close(4000, 'parec not found on host'); + return; + } + + let proc: Subprocess<'ignore', 'pipe', 'ignore'>; + try { + proc = spawn({ cmd: [parecPath, ...CAPTURE_ARGS], stdin: 'ignore', stdout: 'pipe', stderr: 'ignore' }); + } catch { + ws.close(4000, 'Failed to start audio capture'); + return; + } + + const session: Session = { proc, closed: false }; + sessions.set(ws, session); + console.log('[music] parec started, streaming PCM to the relay'); + + pump( + ws, + session, + proc.stdout, + (sock, chunk) => { + try { + sock.sendBinary(chunk); + } catch { + session.closed = true; + } + }, + () => { + if (session.closed) return; + session.closed = true; + try { + ws.close(); + } catch { + /* already closed */ + } + }, + ); +} + +export const musicWebsocket = { + open(ws: ServerWebSocket) { + if (ws.data.kind === 'player') openPlayer(ws, ws.data.files); + else openCapture(ws); + }, + + message(ws: ServerWebSocket, raw: string | Buffer) { + const session = sessions.get(ws); + if (!session || session.closed || ws.data.kind !== 'player') return; // capture is one-way + try { + const msg = JSON.parse(typeof raw === 'string' ? raw : raw.toString()); + if (msg.type === 'input' && msg.data) (session.proc as Subprocess<'pipe'>).stdin.write(msg.data); + } catch { + /* not a frame we understand */ + } + }, + + close(ws: ServerWebSocket) { + const session = sessions.get(ws); + if (!session) return; + session.closed = true; + try { + session.proc.kill(); + } catch { + /* already gone */ + } + sessions.delete(ws); + }, + + drain() {}, +}; + +/** Upgrade one of the two cliamp sockets, or return null if this request is not for them. */ +export function cliampUpgradeData(pathname: string, search: URLSearchParams): MusicWSData | null { + if (pathname === '/cliamp/ws') return { kind: 'player', files: search.get('files') ?? '' }; + if (pathname === '/cliamp/audio/ws') return { kind: 'capture' }; + return null; +} diff --git a/src/servers/sidecar/music/index.ts b/src/servers/sidecar/music/index.ts index 0c60c4b5..083d9270 100644 --- a/src/servers/sidecar/music/index.ts +++ b/src/servers/sidecar/music/index.ts @@ -3,6 +3,8 @@ import { join, basename } from 'node:path'; import type { SidecarCommand, SidecarEvent } from '../protocol'; import { createSidecarConnector } from '../connect'; import { streamAudioFile } from './stream-audio'; +import { cliampUpgradeData, musicWebsocket } from './cliamp-ws'; +import { ensurePulseAudio } from './pulse-audio'; import { startNightlyReindex, stopNightlyReindex } from './nightly-reindex'; import { startMusicWatcher, stopMusicWatcher } from './watcher'; import { @@ -137,6 +139,10 @@ startNightlyReindex(); // Recursive watcher on ~/Music → localized reindex on any change. startMusicWatcher(); +// PulseAudio daemon + the `virtual_out` null sink both cliamp halves depend on. Officer used to do this at +// its own boot, which meant every restart of a process with no audio responsibilities re-checked the sink. +ensurePulseAudio(); + const server = Bun.serve({ port, hostname: '127.0.0.1', @@ -145,6 +151,15 @@ const server = Bun.serve({ idleTimeout: 255, async fetch(req, server) { const url = new URL(req.url); + + // The two cliamp sockets. Officer has already authenticated the browser and is relaying frames; the + // player and the capture themselves live here (cliamp-ws.ts). + const wsData = cliampUpgradeData(url.pathname, url.searchParams); + if (wsData) { + if (server.upgrade(req, { data: wsData })) return undefined as unknown as Response; + return new Response('Expected a WebSocket upgrade', { status: 400 }); + } + // A from-scratch reindex can take many minutes with no bytes flowing on the triggering request. // Give the build endpoints a 30-min idle timeout so they aren't dropped (/manifest is a pure read now). if (url.pathname === '/reindex' || url.pathname === '/reindex/stream') { @@ -419,6 +434,7 @@ const server = Bun.serve({ return new Response('Not found', { status: 404 }); }, + websocket: musicWebsocket, }); console.log(`[music] audio server listening on http://127.0.0.1:${port}`); diff --git a/src/servers/sidecar/music/pulse-audio.ts b/src/servers/sidecar/music/pulse-audio.ts new file mode 100644 index 00000000..3c4f207c --- /dev/null +++ b/src/servers/sidecar/music/pulse-audio.ts @@ -0,0 +1,48 @@ +// Host audio plumbing for local playback: a PulseAudio daemon and a null sink named `virtual_out`. +// cliamp plays *into* that sink (PULSE_SINK) and the capture side reads `virtual_out.monitor`, so the +// sink has to exist before either of them starts — which is why this runs at sidecar startup rather +// than on first play. Both steps are idempotent and both failures are non-fatal: a host without +// pulseaudio simply has no browser playback, and everything else the music sidecar does still works. + +export const VIRTUAL_SINK = 'virtual_out'; + +export function ensurePulseAudio(): void { + const pulseaudio = Bun.which('pulseaudio'); + const pactl = Bun.which('pactl'); + if (!pulseaudio || !pactl) { + console.log('[music] pulseaudio not installed, skipping audio setup'); + return; + } + + const check = Bun.spawnSync({ cmd: [pulseaudio, '--check'], stdout: 'ignore', stderr: 'ignore' }); + if (check.exitCode !== 0) { + const start = Bun.spawnSync({ cmd: [pulseaudio, '--start', '-D'], stdout: 'ignore', stderr: 'ignore' }); + if (start.exitCode !== 0) { + console.error('[music] failed to start pulseaudio'); + return; + } + console.log('[music] pulseaudio started'); + } else { + console.log('[music] pulseaudio already running'); + } + + const sinks = Bun.spawnSync({ cmd: [pactl, 'list', 'short', 'sinks'], stdout: 'pipe', stderr: 'ignore' }); + if (sinks.stdout.toString().includes(VIRTUAL_SINK)) { + console.log(`[music] ${VIRTUAL_SINK} sink already exists`); + return; + } + + const load = Bun.spawnSync({ + cmd: [ + pactl, + 'load-module', + 'module-null-sink', + `sink_name=${VIRTUAL_SINK}`, + 'sink_properties=device.description=Virtual_Output', + ], + stdout: 'pipe', + stderr: 'pipe', + }); + if (load.exitCode !== 0) console.error('[music] failed to load null sink:', load.stderr.toString().trim()); + else console.log(`[music] ${VIRTUAL_SINK} null sink loaded`); +} 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 2386e42a..21d94757 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, 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 } @@ -51,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 @@ -93,6 +99,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 = { @@ -110,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 ── @@ -130,16 +141,15 @@ export type VncSessionInfo = { // ── PTY types ── +// What officer knows about a terminal, and nothing more. The shell, its arguments, the home directory and +// whether the shell is sandboxed are the sidecar's own decisions — they used to travel in here, which is +// how officer ended up reading the owner's SHELL and HOME and hardcoding `host: true`. export type PtyInitConfig = { sessionId: string; - shell?: { command: string; args?: string[] }; + /** The folder the panel was opened on. `~`, `~/x` and absolute paths only; resolved by the sidecar. */ cwd?: string; - homeDir?: string; - userLabel?: string; - host?: boolean; cols?: number; rows?: number; - env?: Record; }; // PTY commands (API → PTY sidecar) @@ -153,4 +163,7 @@ export type PtyCommand = export type PtyEvent = | { type: 'pty:ready'; id: string; sessionId: string } | { type: 'pty:output'; sessionId: string; data: string } + // Scrollback sent on re-attach, which the client may already be showing in part — distinct from + // `pty:output` so it can rebuild the screen rather than append a second copy of it. + | { type: 'pty:replay'; sessionId: string; data: string } | { type: 'pty:exit'; sessionId: string; exitCode: number; signal?: number }; diff --git a/src/servers/api/terminal/pty-sidecar.mjs b/src/servers/sidecar/pty/index.mjs similarity index 58% rename from src/servers/api/terminal/pty-sidecar.mjs rename to src/servers/sidecar/pty/index.mjs index c09d2999..72cc72d6 100644 --- a/src/servers/api/terminal/pty-sidecar.mjs +++ b/src/servers/sidecar/pty/index.mjs @@ -10,24 +10,10 @@ process.on('SIGINT', () => { }); process.on('SIGTERM', () => process.emit('SIGINT')); -import { existsSync } from 'node:fs'; -import { cp, mkdir } from 'node:fs/promises'; -import { join, dirname } from 'node:path'; -import { fileURLToPath } from 'node:url'; -import { execFile } from 'node:child_process'; +import { join } from 'node:path'; import WebSocket from 'ws'; import * as pty from 'node-pty'; -const run = (cmd, args, opts = {}) => - new Promise((resolve) => { - const proc = execFile(cmd, args, { stdio: 'ignore', ...opts }, () => resolve()); - proc.on('error', () => resolve()); - }); - -const __dirname = dirname(fileURLToPath(import.meta.url)); - -const templateDir = join(__dirname, 'templates'); - import 'dotenv/config'; const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${process.env.PORT ?? '5000'}`; @@ -36,14 +22,46 @@ const REGISTER_URL = `${API_URL}/api/sidecar/register`; const BUFFER_MAX = 50 * 1024; const RECONNECT_DELAYS = [200, 500, 1000, 2000, 4000, 8000, 15000]; -/** @type {Map} */ +// ── What kind of shell this process runs ── +// +// These used to arrive inside every pty:init, which meant officer chose the owner's shell and read the +// owner's HOME to do it. They are this process's business: it is the one that spawns the thing. +// +// HOME_DIR mirrors `data-path.ts:getOwnerHomeDir` — on a host where the owner's real login home differs +// from this process's HOME, the shell should open in the former, like every other host-executing surface. +const HOME_DIR = process.env.HOME_DIR ?? process.env.HOME ?? process.cwd(); +const SHELL = { command: process.env.SHELL ?? '/bin/zsh', args: ['-i'] }; +// A terminal is always a plain host shell: the owner is the only account and it is their own machine +// (`platform/CLAUDE.md` — do not add a jail without being asked). There used to be a second branch here +// for a bwrap sandbox, selected by `config.host`, which officer hardcoded to true. Nothing ever built the +// bwrap command it expected, on either side, so it could not have run — it is in git history if the +// decision is ever revisited, and the `ensureUserFiles` half of it duplicated +// `api/users/provision.ts:seedShellConfigs`, which is the live seeder of those templates. + +const resolveCwd = (cwd) => { + if (!cwd || cwd === '~') return HOME_DIR; + if (cwd.startsWith('~/')) return join(HOME_DIR, cwd.slice(2)); + if (cwd.startsWith('/')) return cwd; + // Relative paths have no meaning here — this process's cwd is the repo, not the user's folder. + return HOME_DIR; +}; + +/** @type {Map} */ const sessions = new Map(); // ── Helpers ── -const sendJson = (ws, msg) => { +// Always resolve the CURRENT registration socket, never one captured in a closure. +// +// `term.onData` used to close over the socket that was live when the session was created. Officer is a +// PM2 peer that restarts often, and each restart gives this process a brand new socket — so every +// pre-existing session went on writing to a closed one, where the readyState check below dropped it +// silently. The shell stayed alive and kept accepting input (that arrives on the new socket), but its +// output never came back: the terminal looked frozen until you closed the panel. Reading the module +// variable at send time is the whole fix. +const sendJson = (msg) => { try { - if (ws.readyState === WebSocket.OPEN) { + if (ws && ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify(msg)); } } catch { @@ -51,32 +69,6 @@ const sendJson = (ws, msg) => { } }; -const ensureUserFiles = async (homeDir) => { - await mkdir(homeDir, { recursive: true }); - await mkdir(join(homeDir, '.config'), { recursive: true }); - await mkdir(join(homeDir, '.local', 'bin'), { recursive: true }); - - const zshrcPath = join(homeDir, '.zshrc'); - if (!existsSync(zshrcPath)) { - await cp(join(templateDir, '.zshrc'), zshrcPath); - } - - const tmuxconfPath = join(homeDir, '.tmux.conf'); - if (!existsSync(tmuxconfPath)) { - await cp(join(templateDir, '.tmux.conf'), tmuxconfPath); - } - - const starshipPath = join(homeDir, '.config', 'starship-officer.toml'); - if (!existsSync(starshipPath)) { - await cp(join(templateDir, 'starship-officer.toml'), starshipPath); - } - - const ohMyZshPath = join(homeDir, '.oh-my-zsh'); - if (!existsSync(ohMyZshPath)) { - await run('git', ['clone', '--depth=1', 'https://github.com/ohmyzsh/ohmyzsh.git', ohMyZshPath]); - } -}; - const appendBuffer = (session, data) => { session.buffer += data; if (session.buffer.length > BUFFER_MAX) { @@ -86,7 +78,7 @@ const appendBuffer = (session, data) => { // ── Command handler ── -async function handleCommand(ws, msg) { +async function handleCommand(msg) { switch (msg.type) { case 'pty:init': { const { sessionId, config } = msg; @@ -96,9 +88,12 @@ async function handleCommand(ws, msg) { console.log(`[pty-sidecar] init sessionId=${sessionId} existing=${!!existing} total=${sessions.size}`); if (existing) { - // Replay buffer + // Re-attach. The scrollback goes out as `pty:replay`, not as ordinary output, because the client + // may already be showing some of it: after an officer restart the browser keeps its terminal and + // reconnects, so replaying blind appended a second copy of everything on screen. Marked as + // history, the client can reset and rebuild from it instead. if (existing.buffer.length > 0) { - sendJson(ws, { type: 'pty:output', sessionId, data: existing.buffer }); + sendJson({ type: 'pty:replay', sessionId, data: existing.buffer }); } // Resize PTY to new client dimensions @@ -114,81 +109,46 @@ async function handleCommand(ws, msg) { } } - sendJson(ws, { type: 'pty:ready', id: msg.id, sessionId }); + sendJson({ type: 'pty:ready', id: msg.id, sessionId }); return; } // New session — spawn PTY - const shell = config.shell ?? { command: '/bin/bash', args: ['-i'] }; - const cwd = config.cwd ?? process.cwd(); - const homeDir = config.homeDir ?? process.cwd(); - const userLabel = config.userLabel ?? 'officer'; + const cwd = resolveCwd(config.cwd); const cols = config.cols ?? 80; const rows = config.rows ?? 24; - const isHost = !!config.host; - - let spawnCommand; - let spawnArgs; - let ptyEnv; - - if (isHost) { - spawnCommand = shell.command; - spawnArgs = shell.args ?? []; - ptyEnv = { ...process.env, TERM: 'xterm-256color', ...(config.env ?? {}) }; - } else { - // Sandboxed mode: shell config contains the full bwrap command - spawnCommand = shell.command; - spawnArgs = shell.args ?? []; - - try { - await ensureUserFiles(homeDir); - } catch (err) { - console.error('[pty-sidecar] ensureUserFiles failed:', err); - } - - // bwrap sets env vars internally via --setenv, so use minimal host env - ptyEnv = { TERM: 'xterm-256color' }; - } - - const ptyCwd = isHost ? cwd : undefined; let term; try { - term = pty.spawn(spawnCommand, spawnArgs, { + term = pty.spawn(SHELL.command, SHELL.args, { name: 'xterm-256color', cols, rows, - cwd: ptyCwd, - env: ptyEnv, + cwd, + env: { ...process.env, TERM: 'xterm-256color' }, }); } catch (err) { const message = err instanceof Error ? err.message : 'Failed to start terminal'; - sendJson(ws, { type: 'pty:output', sessionId, data: `\r\n[Terminal error] ${message}\r\n` }); - sendJson(ws, { type: 'pty:exit', sessionId, exitCode: 1 }); + sendJson({ type: 'pty:output', sessionId, data: `\r\n[Terminal error] ${message}\r\n` }); + sendJson({ type: 'pty:exit', sessionId, exitCode: 1 }); return; } - const session = { - term, - buffer: '', - cols, - rows, - initConfig: { shell, cwd, homeDir, userLabel }, - }; + const session = { term, buffer: '', cols, rows }; sessions.set(sessionId, session); term.onData((output) => { appendBuffer(session, output); - sendJson(ws, { type: 'pty:output', sessionId, data: output }); + sendJson({ type: 'pty:output', sessionId, data: output }); }); term.onExit(({ exitCode, signal }) => { console.log(`[pty-sidecar] session ${sessionId} exited code=${exitCode} signal=${signal}`); - sendJson(ws, { type: 'pty:exit', sessionId, exitCode, signal }); + sendJson({ type: 'pty:exit', sessionId, exitCode, signal }); sessions.delete(sessionId); }); - sendJson(ws, { type: 'pty:ready', id: msg.id, sessionId }); + sendJson({ type: 'pty:ready', id: msg.id, sessionId }); return; } @@ -251,7 +211,7 @@ function connect() { ws.on('open', () => { reconnectAttempt = 0; console.log('[pty-sidecar] connected, sending registration...'); - sendJson(ws, { type: 'register', name: 'pty', capabilities: ['terminal'] }); + sendJson({ type: 'register', name: 'pty', capabilities: ['terminal'] }); }); ws.on('message', (data) => { @@ -263,7 +223,7 @@ function connect() { return; } - handleCommand(ws, msg); + handleCommand(msg); } catch { // skip malformed messages } diff --git a/src/servers/sidecar/pty/index.test.ts b/src/servers/sidecar/pty/index.test.ts new file mode 100644 index 00000000..8ed7af62 --- /dev/null +++ b/src/servers/sidecar/pty/index.test.ts @@ -0,0 +1,134 @@ +import { describe, test, expect, afterAll } from 'bun:test'; +import type { ServerWebSocket, Subprocess } from 'bun'; + +// Integration test for the one thing about the pty sidecar that cannot be reasoned about from the code +// alone: what happens to a live shell when officer goes away and comes back. It stands up a fake +// registration socket, runs the real sidecar against it, then kills the socket and rebinds the same port +// the way `pm2 restart officer` does. +// +// Nothing here touches the running officer — the sidecar dials API_URL, which is overridden per spawn. + +const SIDECAR = 'src/servers/sidecar/pty/index.mjs'; + +type Frame = Record; + +// One fake officer. `stop()` drops the socket; a new instance on the same port is the restart. +function fakeOfficer(port?: number) { + let socket: ServerWebSocket | null = null; + const frames: Frame[] = []; + const waiters: Array<{ match: (f: Frame) => boolean; resolve: (f: Frame) => void }> = []; + + const server = Bun.serve({ + port: port ?? 0, + fetch(req, srv) { + if (new URL(req.url).pathname === '/api/sidecar/register' && srv.upgrade(req)) return undefined; + return new Response('no', { status: 404 }); + }, + websocket: { + open(ws) { + socket = ws; + }, + message(_ws, raw) { + const frame = JSON.parse(String(raw)) as Frame; + frames.push(frame); + for (let i = waiters.length - 1; i >= 0; i--) { + if (waiters[i]!.match(frame)) waiters.splice(i, 1)[0]!.resolve(frame); + } + }, + close() { + socket = null; + }, + }, + }); + + return { + port: server.port, + send: (msg: Frame) => socket?.send(JSON.stringify(msg)), + /** Resolve on the first frame matching `match`, including ones already received. */ + await: (match: (f: Frame) => boolean, timeoutMs = 15_000) => + new Promise((resolve, reject) => { + const seen = frames.find(match); + if (seen) return resolve(seen); + const timer = setTimeout(() => reject(new Error(`timed out waiting for a frame`)), timeoutMs); + waiters.push({ + match, + resolve: (f) => { + clearTimeout(timer); + resolve(f); + }, + }); + }), + /** Output frames for one session, concatenated — the terminal's visible text. */ + outputFor: (sessionId: string) => + frames + .filter((f) => f.type === 'pty:output' && f.sessionId === sessionId) + .map((f) => f.data as string) + .join(''), + stop: () => server.stop(true), + }; +} + +const isOutput = (sessionId: string, needle: string) => (f: Frame) => + f.type === 'pty:output' && f.sessionId === sessionId && String(f.data).includes(needle); + +let child: Subprocess | null = null; +afterAll(() => child?.kill()); + +describe('pty sidecar', () => { + test('a shell keeps streaming output after officer restarts under it', async () => { + const sessionId = 'test-restart'; + let officer = fakeOfficer(); + const port = officer.port; + + child = Bun.spawn(['node', SIDECAR], { + env: { + ...process.env, + API_URL: `ws://127.0.0.1:${port}`, + // The sidecar now chooses the shell and the home itself, so the test pins both rather than + // spawning the owner's interactive zsh (which would read their rc files and their history). + SHELL: '/bin/sh', + HOME_DIR: '/tmp', + ENV: '/dev/null', + }, + stdout: 'ignore', + stderr: 'ignore', + }); + + await officer.await((f) => f.type === 'register' && f.capabilities?.includes('terminal')); + + // `~` is resolved by the sidecar against its own HOME_DIR, not by officer. + officer.send({ type: 'pty:init', id: 'i1', sessionId, config: { sessionId, cwd: '~', cols: 80, rows: 24 } }); + await officer.await((f) => f.type === 'pty:ready' && f.sessionId === sessionId); + + officer.send({ type: 'pty:input', id: 'in0', sessionId, data: 'pwd\n' }); + await officer.await(isOutput(sessionId, '/tmp')); + + officer.send({ type: 'pty:input', id: 'in1', sessionId, data: 'echo before-restart\n' }); + await officer.await(isOutput(sessionId, 'before-restart')); + + // ── the restart ── + officer.stop(); + officer = fakeOfficer(port); + await officer.await((f) => f.type === 'register'); + + // The shell is the same process; only officer changed. Before the sendJson fix this input was + // accepted and executed, but its output went to the socket captured at init time and vanished. + officer.send({ type: 'pty:input', id: 'in2', sessionId, data: 'echo after-restart\n' }); + await officer.await(isOutput(sessionId, 'after-restart')); + + // Re-attaching replays the scrollback, marked as history so the client can rebuild rather than + // append — and it contains what happened on both sides of the restart. + officer.send({ type: 'pty:init', id: 'i2', sessionId, config: { sessionId, cols: 80, rows: 24 } }); + const replay = await officer.await((f) => f.type === 'pty:replay' && f.sessionId === sessionId); + expect(String(replay.data)).toContain('before-restart'); + expect(String(replay.data)).toContain('after-restart'); + + // Re-attach must not spawn a second shell, and must not replay as ordinary output. + await officer.await((f) => f.type === 'pty:ready' && f.id === 'i2'); + expect(officer.outputFor(sessionId)).not.toContain('before-restart'); + + officer.send({ type: 'pty:close', id: 'c1', sessionId }); + await officer.await((f) => f.type === 'pty:exit' && f.sessionId === sessionId); + officer.stop(); + }, 30_000); +}); diff --git a/src/workspaces/officerdev/src/apps/Desktop/DesktopView.tsx b/src/workspaces/officerdev/src/apps/Desktop/DesktopView.tsx index 41793379..625030f7 100644 --- a/src/workspaces/officerdev/src/apps/Desktop/DesktopView.tsx +++ b/src/workspaces/officerdev/src/apps/Desktop/DesktopView.tsx @@ -41,7 +41,9 @@ export const DesktopView = ({ className, style }: DesktopViewProps) => { const rfbRef = useRef(null); const isMounted = useMounted(); const client = useClient(); - const [status, setStatus] = useState<'connecting' | 'connected' | 'disconnected' | 'error'>('connecting'); + const [status, setStatus] = useState<'connecting' | 'connected' | 'reconnecting' | 'disconnected' | 'error'>( + 'connecting', + ); const [errorMsg, setErrorMsg] = useState(''); useEffect(() => { @@ -50,20 +52,64 @@ export const DesktopView = ({ className, style }: DesktopViewProps) => { if (!container) return; let disposed = false; + let attempts = 0; + let retryTimer: ReturnType | null = null; + // Tearing an RFB down makes it fire its own `disconnect`, and a bad password fires `securityfailure` + // and then `disconnect` too. Both would otherwise be read as "officer went away, reattach". + let generation = 0; + let fatal = false; + const MAX_ATTEMPTS = 5; + const RETRY_DELAYS = [1000, 2000, 3000, 5000, 5000]; + + const detach = () => { + const rfb = rfbRef.current; + rfbRef.current = null; + generation++; + if (!rfb) return; + try { + rfb.disconnect(); + } catch { + /* ignore */ + } + }; + + // x11vnc mirrors :0 with `-forever`, so the desktop itself outlives this socket — losing it means + // officer restarted under us, not that the session ended. Reattach instead of parking on + // "Disconnected" until someone reopens the panel. + const retry = (reason: string) => { + if (disposed || fatal || retryTimer) return; + detach(); + if (attempts >= MAX_ATTEMPTS) { + setStatus('disconnected'); + setErrorMsg(reason); + return; + } + const delay = RETRY_DELAYS[attempts] ?? 5000; + attempts++; + setStatus('reconnecting'); + setErrorMsg(`${reason} — reconnecting (${attempts}/${MAX_ATTEMPTS})...`); + retryTimer = setTimeout(() => { + retryTimer = null; + void connect(); + }, delay); + }; const connect = async () => { + const gen = ++generation; + const isCurrent = () => !disposed && gen === generation; let password = ''; try { const res = await client.get<{ password: string }>('/desktop/vnc-password'); password = res.password; } catch { - if (disposed) return; - setStatus('error'); - setErrorMsg('Failed to fetch VNC password'); + if (!isCurrent()) return; + // Officer being down is the common case here, and it comes back — so this is a retry, not a + // dead end. A sidecar that is genuinely missing still ends up at "Disconnected" after five. + retry('Failed to fetch VNC password'); return; } - if (disposed) return; + if (!isCurrent()) return; let RFB: Awaited>['default']; try { @@ -71,13 +117,14 @@ export const DesktopView = ({ className, style }: DesktopViewProps) => { RFB = mod.default; } catch (err) { console.error('[desktop] Failed to load noVNC:', err); - if (disposed) return; + if (!isCurrent()) return; + fatal = true; setStatus('error'); setErrorMsg('Failed to load noVNC library'); return; } - if (disposed) return; + if (!isCurrent()) return; const wsUrl = buildWsUrl(); const rfb = new RFB(container, wsUrl, { @@ -90,15 +137,18 @@ export const DesktopView = ({ className, style }: DesktopViewProps) => { rfbRef.current = rfb; rfb.addEventListener('connect', () => { - if (!disposed) setStatus('connected'); + if (!isCurrent()) return; + attempts = 0; + setErrorMsg(''); + setStatus('connected'); }); - rfb.addEventListener('disconnect', (ev: CustomEvent) => { - if (disposed) return; - setStatus('disconnected'); - if (!ev.detail.clean) { - setErrorMsg('Connection lost'); - } + // Every disconnect we did not ask for is worth retrying, clean or not: officer closing its side + // tidily during a restart still reports `clean`, and the desktop behind it is still there. + rfb.addEventListener('disconnect', () => { + if (!isCurrent()) return; + rfbRef.current = null; + retry('Connection lost'); }); rfb.addEventListener('credentialsrequired', () => { @@ -106,25 +156,30 @@ export const DesktopView = ({ className, style }: DesktopViewProps) => { }); rfb.addEventListener('securityfailure', (ev: CustomEvent) => { - if (!disposed) { - setStatus('error'); - setErrorMsg(ev.detail.reason || 'Security failure'); - } + if (!isCurrent()) return; + fatal = true; + setStatus('error'); + setErrorMsg(ev.detail.reason || 'Security failure'); }); }; + // Coming back to the tab after the retries ran out should try once more rather than stay dead. + const handleVisibility = () => { + if (disposed || fatal || document.visibilityState !== 'visible') return; + if (rfbRef.current || retryTimer) return; + attempts = 0; + setStatus('connecting'); + void connect(); + }; + document.addEventListener('visibilitychange', handleVisibility); + void connect(); return () => { disposed = true; - if (rfbRef.current) { - try { - rfbRef.current.disconnect(); - } catch { - /* ignore */ - } - rfbRef.current = null; - } + document.removeEventListener('visibilitychange', handleVisibility); + if (retryTimer) clearTimeout(retryTimer); + detach(); }; }, [isMounted, client]); @@ -138,7 +193,7 @@ export const DesktopView = ({ className, style }: DesktopViewProps) => { Connecting to desktop... )} - {(status === 'disconnected' || status === 'error') && ( + {(status === 'reconnecting' || status === 'disconnected' || status === 'error') && (
{errorMsg || 'Disconnected from desktop'}
diff --git a/src/workspaces/officerdev/src/apps/Terminal/Terminal.tsx b/src/workspaces/officerdev/src/apps/Terminal/Terminal.tsx index 97af268c..8e4963f8 100644 --- a/src/workspaces/officerdev/src/apps/Terminal/Terminal.tsx +++ b/src/workspaces/officerdev/src/apps/Terminal/Terminal.tsx @@ -218,6 +218,14 @@ export const TerminalView = ({ onCommandDoneRef.current(exitCode, output); } } + } else if (msg.type === 'replay') { + // Scrollback for a session we are re-attaching to. On a page load this terminal is empty and + // the reset is a no-op; after an officer restart it still holds what it had before the socket + // dropped, and the replay overlaps it — so rebuild from the sidecar's copy rather than append + // a second one. Deliberately outside the `output` branch: replay must not re-trigger the + // command/initial-input logic above. + term.reset(); + term.write(msg.data); } else if (msg.type === 'exit') { processExited = true; term.write('\r\n[Process exited]\r\n'); 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;