diff --git a/server/index.js b/server/index.js index c866ebd..3208943 100755 --- a/server/index.js +++ b/server/index.js @@ -63,6 +63,10 @@ import { spawnOmp, abortOmpSession, } from './omp-cli.js'; +import { + spawnOmo, + abortOmoSession, +} from './omo-cli.js'; import { stripAnsiSequences, normalizeDetectedUrl, @@ -170,6 +174,7 @@ const wss = createWebSocketServer(server, { opencode: spawnOpenCode, gjc: spawnGjc, omp: spawnOmp, + omo: spawnOmo, }, abortFns: { claude: abortClaudeSDKSession, @@ -178,6 +183,7 @@ const wss = createWebSocketServer(server, { opencode: abortOpenCodeSession, gjc: abortGjcSession, omp: abortOmpSession, + omo: abortOmoSession, }, resolveToolApproval: resolveProviderToolApproval, getPendingApprovalsForSession: getPendingProviderApprovalsForSession, diff --git a/server/modules/providers/services/provider-capabilities.service.ts b/server/modules/providers/services/provider-capabilities.service.ts index 89b60c6..09a2e84 100644 --- a/server/modules/providers/services/provider-capabilities.service.ts +++ b/server/modules/providers/services/provider-capabilities.service.ts @@ -113,9 +113,9 @@ const PROVIDER_CAPABILITIES: Record = { provider: 'omo', permissionModes: ['default'], defaultPermissionMode: 'default', - // omo is wired for discovery and transcript reading only. There is no omo - // send runtime (no `server/omo-cli.ts`), so nothing forwards attachments or - // can cancel an in-flight run yet. + // Mirrors the Oh My Pi posture: buildPiCliArgs forwards image paths as + // @ and createPiCliRuntime terminates the tracked child with SIGTERM + // (`server/pi-cli.ts`), but this matrix still hides both UI controls. supportsImages: false, supportsAbort: false, // omo's TUI renders "↑↓ navigate • enter select • esc close" and has no diff --git a/server/omo-cli.test.ts b/server/omo-cli.test.ts new file mode 100644 index 0000000..685c670 --- /dev/null +++ b/server/omo-cli.test.ts @@ -0,0 +1,100 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; + +import { buildOmoArgs, normalizeOmoEvent } from './omo-cli.js'; + +test('buildOmoArgs preserves resume, model, thinking, images, and prompt as distinct argv', () => { + assert.deepEqual( + buildOmoArgs('Explain this image', { + sessionId: '019ff9fa-abab-78a3-83b0-67c261374f42', + model: 'anthropic/claude-opus-5', + effort: 'high', + images: [{ path: '/tmp/shot.png' }, { path: ' ' }, { path: 42 }], + }), + [ + '--mode', 'json', '--print', + '--resume', '019ff9fa-abab-78a3-83b0-67c261374f42', + '--model', 'anthropic/claude-opus-5', + '--thinking', 'high', + '@/tmp/shot.png', + 'Explain this image', + ], + ); +}); + +test('buildOmoArgs omits placeholder model and effort selections', () => { + assert.deepEqual( + buildOmoArgs('hi', { model: 'default', effort: 'default' }), + ['--mode', 'json', '--print', 'hi'], + ); +}); + +// Event shapes below are verbatim from `omo --mode json --print`. +test('normalizeOmoEvent captures the native session id and streams assistant text', () => { + assert.deepEqual( + normalizeOmoEvent( + { type: 'session', version: 3, id: '019ff9fa-abab-78a3-83b0-67c261374f42', cwd: '/tmp' }, + null, + ).providerSessionId, + '019ff9fa-abab-78a3-83b0-67c261374f42', + ); + + const streamed = normalizeOmoEvent({ + type: 'message_update', + assistantMessageEvent: { type: 'text_delta', contentIndex: 0, delta: 'ok' }, + }, 'session-1'); + assert.equal(streamed.messages.length, 1); + assert.equal(streamed.messages[0].kind, 'stream_delta'); + assert.equal(streamed.messages[0].content, 'ok'); + assert.equal(streamed.messages[0].provider, 'omo'); +}); + +test('normalizeOmoEvent maps the native tool lifecycle to shared tool messages', () => { + const started = normalizeOmoEvent({ + type: 'tool_execution_start', + toolCallId: 'toolu_01FqdHoP72XmbXtPo5yrEdkM', + toolName: 'read', + args: { path: '/tmp/omo-json-probe/sample.txt' }, + }, 'session-1'); + assert.equal(started.messages[0].kind, 'tool_use'); + assert.equal(started.messages[0].toolName, 'read'); + assert.equal(started.messages[0].toolId, 'toolu_01FqdHoP72XmbXtPo5yrEdkM'); + assert.deepEqual(started.messages[0].toolInput, { path: '/tmp/omo-json-probe/sample.txt' }); + + const ended = normalizeOmoEvent({ + type: 'tool_execution_end', + toolCallId: 'toolu_01FqdHoP72XmbXtPo5yrEdkM', + toolName: 'read', + result: { content: [{ type: 'text', text: 'probe file\n' }] }, + isError: false, + }, 'session-1'); + assert.equal(ended.messages[0].kind, 'tool_result'); + assert.equal(ended.messages[0].toolId, 'toolu_01FqdHoP72XmbXtPo5yrEdkM'); + assert.equal(ended.messages[0].content, 'probe file\n'); + assert.equal(ended.messages[0].isError, false); +}); + +test('normalizeOmoEvent ignores the non-message envelopes the CLI also emits', () => { + for (const event of [ + { type: 'agent_start' }, + { type: 'turn_start' }, + { type: 'turn_end', message: { role: 'assistant', content: [] } }, + { type: 'agent_settled' }, + { type: 'entry_appended', entry: { type: 'custom', customType: 'omo-memory:accepted-turns' } }, + { type: 'tool_hook_status', hookName: 'PreToolUse', toolName: 'read' }, + { type: 'message_update', assistantMessageEvent: { type: 'text_start', contentIndex: 0 } }, + ]) { + assert.deepEqual(normalizeOmoEvent(event, 'session-1').messages, [], event.type); + } +}); + +test('normalizeOmoEvent surfaces errors with an omo-labelled fallback', () => { + assert.equal( + normalizeOmoEvent({ type: 'error', error: { message: 'boom' } }, 'session-1').messages[0].content, + 'boom', + ); + assert.equal( + normalizeOmoEvent({ type: 'error' }, 'session-1').messages[0].content, + 'omo failed.', + ); +}); diff --git a/server/omo-cli.ts b/server/omo-cli.ts new file mode 100644 index 0000000..1646453 --- /dev/null +++ b/server/omo-cli.ts @@ -0,0 +1,40 @@ +import { + buildPiCliArgs, + createPiCliRuntime, + normalizePiCliEvent, + type PiCliDescriptor, + type PiCliRunOptions, + type PiCliWriter, +} from './pi-cli.js'; +import type { NormalizedMessage } from './shared/types.js'; + +const OMO: PiCliDescriptor = { + provider: 'omo', + binary: 'omo', + label: 'omo', +}; + +const runtime = createPiCliRuntime(OMO); + +export function buildOmoArgs(command: string, options: PiCliRunOptions): string[] { + return buildPiCliArgs(command, options); +} + +export function normalizeOmoEvent( + eventValue: unknown, + sessionId: string | null, +): { providerSessionId?: string; messages: NormalizedMessage[] } { + return normalizePiCliEvent(eventValue, sessionId, OMO); +} + +export function spawnOmo( + command: string, + options: PiCliRunOptions = {}, + writer: PiCliWriter, +): Promise { + return runtime.spawn(command, options, writer); +} + +export function abortOmoSession(sessionId: string): boolean { + return runtime.abort(sessionId); +} diff --git a/server/omp-cli.ts b/server/omp-cli.ts index 1e815ba..15cd8de 100644 --- a/server/omp-cli.ts +++ b/server/omp-cli.ts @@ -1,257 +1,40 @@ -import { randomUUID } from 'node:crypto'; -import { spawn } from 'node:child_process'; -import { createInterface } from 'node:readline'; - -import { createCompleteMessage, createNormalizedMessage } from './shared/utils.js'; -import type { AnyRecord, NormalizedMessage } from './shared/types.js'; - -type OmpWriter = { - send(value: unknown): void; - setSessionId?(id: string): void; - getAppSessionId?(): string | undefined; -}; - -type OmpRunOptions = { - sessionId?: string; - cwd?: string; - projectPath?: string; - model?: string; - effort?: string; - images?: Array<{ path?: unknown }>; +import { + buildPiCliArgs, + createPiCliRuntime, + normalizePiCliEvent, + type PiCliDescriptor, + type PiCliRunOptions, + type PiCliWriter, +} from './pi-cli.js'; +import type { NormalizedMessage } from './shared/types.js'; + +const OMP: PiCliDescriptor = { + provider: 'omp', + binary: 'omp', + label: 'Oh My Pi', }; -type ActiveOmpProcess = ReturnType & { aborted?: boolean }; - -const activeOmpProcesses = new Map(); - -function readRecord(value: unknown): AnyRecord | null { - return value !== null && typeof value === 'object' && !Array.isArray(value) - ? value as AnyRecord - : null; -} - -function readContentText(value: unknown): string { - if (typeof value === 'string') return value; - if (!Array.isArray(value)) return value == null ? '' : JSON.stringify(value); - return value - .map((part) => { - const record = readRecord(part); - return typeof record?.text === 'string' ? record.text : ''; - }) - .filter(Boolean) - .join('\n'); -} +const runtime = createPiCliRuntime(OMP); -export function buildOmpArgs(command: string, options: OmpRunOptions): string[] { - const args = ['--mode', 'json', '--print']; - if (options.sessionId) args.push('--resume', options.sessionId); - if (options.model && options.model !== 'default') args.push('--model', options.model); - if (options.effort && options.effort !== 'default') args.push('--thinking', options.effort); - - for (const image of options.images ?? []) { - if (typeof image.path === 'string' && image.path.trim()) { - args.push(`@${image.path}`); - } - } - args.push(command); - return args; +export function buildOmpArgs(command: string, options: PiCliRunOptions): string[] { + return buildPiCliArgs(command, options); } export function normalizeOmpEvent( eventValue: unknown, sessionId: string | null, ): { providerSessionId?: string; messages: NormalizedMessage[] } { - const event = readRecord(eventValue); - if (!event) return { messages: [] }; - - if (event.type === 'session' && typeof event.id === 'string' && event.id.trim()) { - return { providerSessionId: event.id, messages: [] }; - } - - if (event.type === 'message_update') { - const update = readRecord(event.assistantMessageEvent); - if (update?.type === 'text_delta' && typeof update.delta === 'string' && update.delta) { - return { - messages: [createNormalizedMessage({ - kind: 'stream_delta', - content: update.delta, - sessionId, - provider: 'omp', - })], - }; - } - if (update?.type === 'thinking_delta' && typeof update.delta === 'string' && update.delta) { - return { - messages: [createNormalizedMessage({ - kind: 'thinking', - content: update.delta, - sessionId, - provider: 'omp', - })], - }; - } - } - - if (event.type === 'tool_execution_start') { - return { - messages: [createNormalizedMessage({ - kind: 'tool_use', - toolName: typeof event.toolName === 'string' ? event.toolName : 'Unknown', - toolInput: readRecord(event.args) ?? {}, - toolId: typeof event.toolCallId === 'string' ? event.toolCallId : randomUUID(), - sessionId, - provider: 'omp', - })], - }; - } - - if (event.type === 'tool_execution_end') { - const result = readRecord(event.result); - return { - messages: [createNormalizedMessage({ - kind: 'tool_result', - toolId: typeof event.toolCallId === 'string' ? event.toolCallId : '', - content: readContentText(result?.content), - isError: Boolean(event.isError), - sessionId, - provider: 'omp', - })], - }; - } - - if (event.type === 'error') { - const error = readRecord(event.error); - const content = typeof event.message === 'string' - ? event.message - : typeof error?.message === 'string' - ? error.message - : 'Oh My Pi failed.'; - return { - messages: [createNormalizedMessage({ - kind: 'error', - content, - sessionId, - provider: 'omp', - })], - }; - } - - return { messages: [] }; + return normalizePiCliEvent(eventValue, sessionId, OMP); } -export function spawnOmp(command: string, options: OmpRunOptions = {}, writer: OmpWriter): Promise { - const workingDir = options.cwd || options.projectPath || process.cwd(); - const processKey = writer.getAppSessionId?.() || options.sessionId || randomUUID(); - let capturedSessionId = options.sessionId ?? null; - let child: ActiveOmpProcess | null = null; - let settled = false; - - const run = new Promise((resolve, reject) => { - const finish = (error?: Error): void => { - if (settled) return; - settled = true; - if (error) reject(error); - else resolve(); - }; - - const registerSession = (providerSessionId: string): void => { - if (!providerSessionId || capturedSessionId === providerSessionId) return; - const previousId = capturedSessionId; - capturedSessionId = providerSessionId; - writer.setSessionId?.(providerSessionId); - if (child) { - activeOmpProcesses.set(providerSessionId, child); - if (previousId) activeOmpProcesses.delete(previousId); - } - if (!options.sessionId) { - writer.send(createNormalizedMessage({ - kind: 'session_created', - newSessionId: providerSessionId, - sessionId: providerSessionId, - provider: 'omp', - })); - } - }; - - try { - child = spawn('omp', buildOmpArgs(command, options), { - cwd: workingDir, - env: process.env, - stdio: ['ignore', 'pipe', 'pipe'], - windowsHide: true, - }) as ActiveOmpProcess; - if (!child.stdout || !child.stderr) { - throw new Error('Oh My Pi did not expose its output streams.'); - } - const { stdout, stderr } = child; - activeOmpProcesses.set(processKey, child); - if (capturedSessionId) activeOmpProcesses.set(capturedSessionId, child); - - const lines = createInterface({ input: stdout }); - lines.on('line', (line) => { - if (!line.trim()) return; - let event: unknown; - try { - event = JSON.parse(line); - } catch { - return; - } - const normalized = normalizeOmpEvent(event, capturedSessionId); - if (normalized.providerSessionId) registerSession(normalized.providerSessionId); - for (const message of normalized.messages) writer.send(message); - }); - - stderr.on('data', (chunk) => { - const content = String(chunk).trim(); - if (!content) return; - writer.send(createNormalizedMessage({ - kind: 'error', - content, - sessionId: capturedSessionId, - provider: 'omp', - })); - }); - - child.on('error', (error) => { - activeOmpProcesses.delete(processKey); - if (capturedSessionId) activeOmpProcesses.delete(capturedSessionId); - writer.send(createNormalizedMessage({ - kind: 'error', - content: error.message, - sessionId: capturedSessionId, - provider: 'omp', - })); - if (!child?.aborted) { - writer.send(createCompleteMessage({ provider: 'omp', sessionId: capturedSessionId, exitCode: 1 })); - } - finish(error); - }); - - child.on('close', (code) => { - activeOmpProcesses.delete(processKey); - if (capturedSessionId) activeOmpProcesses.delete(capturedSessionId); - if (!child?.aborted) { - writer.send(createCompleteMessage({ provider: 'omp', sessionId: capturedSessionId, exitCode: code })); - } - if (code === 0 || child?.aborted) finish(); - else finish(new Error(`Oh My Pi exited with code ${code ?? 'unknown'}`)); - }); - } catch (error) { - finish(error instanceof Error ? error : new Error(String(error))); - } - }); - - Object.assign(run, { abortHandle: processKey }); - return run; +export function spawnOmp( + command: string, + options: PiCliRunOptions = {}, + writer: PiCliWriter, +): Promise { + return runtime.spawn(command, options, writer); } export function abortOmpSession(sessionId: string): boolean { - const child = activeOmpProcesses.get(sessionId); - if (!child) return false; - child.aborted = true; - child.kill('SIGTERM'); - for (const [key, value] of activeOmpProcesses.entries()) { - if (value === child) activeOmpProcesses.delete(key); - } - return true; + return runtime.abort(sessionId); } diff --git a/server/pi-cli.ts b/server/pi-cli.ts new file mode 100644 index 0000000..c115edb --- /dev/null +++ b/server/pi-cli.ts @@ -0,0 +1,296 @@ +import { randomUUID } from 'node:crypto'; +import { spawn } from 'node:child_process'; +import { createInterface } from 'node:readline'; + +import { createCompleteMessage, createNormalizedMessage } from './shared/utils.js'; +import type { AnyRecord, NormalizedMessage } from './shared/types.js'; + +/** + * Shared non-interactive runtime for the pi-derived CLIs. + * + * Oh My Pi and omo accept the same flags (`--mode json --print`, `--resume`, + * `--model`, `--thinking`, `@`) and emit the same JSON event stream + * (`session`, `message_update`, `tool_execution_start|end`, `error`), so one + * runtime serves both and a third pi CLI needs only a new descriptor. + * + * Both write human-readable log lines to stdout alongside the JSON, so every + * line that fails to parse is skipped rather than treated as a protocol error. + */ +export type PiCliProvider = 'omp' | 'omo'; + +export type PiCliDescriptor = { + provider: PiCliProvider; + /** Executable resolved from PATH. */ + binary: string; + /** Human-readable name used in error text surfaced to the client. */ + label: string; +}; + +export type PiCliWriter = { + send(value: unknown): void; + setSessionId?(id: string): void; + getAppSessionId?(): string | undefined; +}; + +export type PiCliRunOptions = { + sessionId?: string; + cwd?: string; + projectPath?: string; + model?: string; + effort?: string; + images?: Array<{ path?: unknown }>; +}; + +type ActivePiProcess = ReturnType & { aborted?: boolean }; + +function readRecord(value: unknown): AnyRecord | null { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as AnyRecord + : null; +} + +function readContentText(value: unknown): string { + if (typeof value === 'string') return value; + if (!Array.isArray(value)) return value == null ? '' : JSON.stringify(value); + return value + .map((part) => { + const record = readRecord(part); + return typeof record?.text === 'string' ? record.text : ''; + }) + .filter(Boolean) + .join('\n'); +} + +export function buildPiCliArgs(command: string, options: PiCliRunOptions): string[] { + const args = ['--mode', 'json', '--print']; + if (options.sessionId) args.push('--resume', options.sessionId); + if (options.model && options.model !== 'default') args.push('--model', options.model); + if (options.effort && options.effort !== 'default') args.push('--thinking', options.effort); + + for (const image of options.images ?? []) { + if (typeof image.path === 'string' && image.path.trim()) { + args.push(`@${image.path}`); + } + } + args.push(command); + return args; +} + +export function normalizePiCliEvent( + eventValue: unknown, + sessionId: string | null, + descriptor: PiCliDescriptor, +): { providerSessionId?: string; messages: NormalizedMessage[] } { + const event = readRecord(eventValue); + if (!event) return { messages: [] }; + const provider = descriptor.provider; + + if (event.type === 'session' && typeof event.id === 'string' && event.id.trim()) { + return { providerSessionId: event.id, messages: [] }; + } + + if (event.type === 'message_update') { + const update = readRecord(event.assistantMessageEvent); + if (update?.type === 'text_delta' && typeof update.delta === 'string' && update.delta) { + return { + messages: [createNormalizedMessage({ + kind: 'stream_delta', + content: update.delta, + sessionId, + provider, + })], + }; + } + if (update?.type === 'thinking_delta' && typeof update.delta === 'string' && update.delta) { + return { + messages: [createNormalizedMessage({ + kind: 'thinking', + content: update.delta, + sessionId, + provider, + })], + }; + } + } + + if (event.type === 'tool_execution_start') { + return { + messages: [createNormalizedMessage({ + kind: 'tool_use', + toolName: typeof event.toolName === 'string' ? event.toolName : 'Unknown', + toolInput: readRecord(event.args) ?? {}, + toolId: typeof event.toolCallId === 'string' ? event.toolCallId : randomUUID(), + sessionId, + provider, + })], + }; + } + + if (event.type === 'tool_execution_end') { + const result = readRecord(event.result); + return { + messages: [createNormalizedMessage({ + kind: 'tool_result', + toolId: typeof event.toolCallId === 'string' ? event.toolCallId : '', + content: readContentText(result?.content), + isError: Boolean(event.isError), + sessionId, + provider, + })], + }; + } + + if (event.type === 'error') { + const error = readRecord(event.error); + const content = typeof event.message === 'string' + ? event.message + : typeof error?.message === 'string' + ? error.message + : `${descriptor.label} failed.`; + return { + messages: [createNormalizedMessage({ + kind: 'error', + content, + sessionId, + provider, + })], + }; + } + + return { messages: [] }; +} + +export type PiCliRuntime = { + spawn(command: string, options: PiCliRunOptions | undefined, writer: PiCliWriter): Promise; + abort(sessionId: string): boolean; +}; + +export function createPiCliRuntime(descriptor: PiCliDescriptor): PiCliRuntime { + const activeProcesses = new Map(); + const { provider, binary, label } = descriptor; + + const spawnRun = ( + command: string, + options: PiCliRunOptions = {}, + writer: PiCliWriter, + ): Promise => { + const workingDir = options.cwd || options.projectPath || process.cwd(); + const processKey = writer.getAppSessionId?.() || options.sessionId || randomUUID(); + let capturedSessionId = options.sessionId ?? null; + let child: ActivePiProcess | null = null; + let settled = false; + + const run = new Promise((resolve, reject) => { + const finish = (error?: Error): void => { + if (settled) return; + settled = true; + if (error) reject(error); + else resolve(); + }; + + const registerSession = (providerSessionId: string): void => { + if (!providerSessionId || capturedSessionId === providerSessionId) return; + const previousId = capturedSessionId; + capturedSessionId = providerSessionId; + writer.setSessionId?.(providerSessionId); + if (child) { + activeProcesses.set(providerSessionId, child); + if (previousId) activeProcesses.delete(previousId); + } + if (!options.sessionId) { + writer.send(createNormalizedMessage({ + kind: 'session_created', + newSessionId: providerSessionId, + sessionId: providerSessionId, + provider, + })); + } + }; + + try { + // stdin must be closed: with an inherited stdin these CLIs wait for + // interactive input and never emit their first event. + child = spawn(binary, buildPiCliArgs(command, options), { + cwd: workingDir, + env: process.env, + stdio: ['ignore', 'pipe', 'pipe'], + windowsHide: true, + }) as ActivePiProcess; + if (!child.stdout || !child.stderr) { + throw new Error(`${label} did not expose its output streams.`); + } + const { stdout, stderr } = child; + activeProcesses.set(processKey, child); + if (capturedSessionId) activeProcesses.set(capturedSessionId, child); + + const lines = createInterface({ input: stdout }); + lines.on('line', (line) => { + if (!line.trim()) return; + let event: unknown; + try { + event = JSON.parse(line); + } catch { + return; + } + const normalized = normalizePiCliEvent(event, capturedSessionId, descriptor); + if (normalized.providerSessionId) registerSession(normalized.providerSessionId); + for (const message of normalized.messages) writer.send(message); + }); + + stderr.on('data', (chunk) => { + const content = String(chunk).trim(); + if (!content) return; + writer.send(createNormalizedMessage({ + kind: 'error', + content, + sessionId: capturedSessionId, + provider, + })); + }); + + child.on('error', (error) => { + activeProcesses.delete(processKey); + if (capturedSessionId) activeProcesses.delete(capturedSessionId); + writer.send(createNormalizedMessage({ + kind: 'error', + content: error.message, + sessionId: capturedSessionId, + provider, + })); + if (!child?.aborted) { + writer.send(createCompleteMessage({ provider, sessionId: capturedSessionId, exitCode: 1 })); + } + finish(error); + }); + + child.on('close', (code) => { + activeProcesses.delete(processKey); + if (capturedSessionId) activeProcesses.delete(capturedSessionId); + if (!child?.aborted) { + writer.send(createCompleteMessage({ provider, sessionId: capturedSessionId, exitCode: code })); + } + if (code === 0 || child?.aborted) finish(); + else finish(new Error(`${label} exited with code ${code ?? 'unknown'}`)); + }); + } catch (error) { + finish(error instanceof Error ? error : new Error(String(error))); + } + }); + + Object.assign(run, { abortHandle: processKey }); + return run; + }; + + const abort = (sessionId: string): boolean => { + const child = activeProcesses.get(sessionId); + if (!child) return false; + child.aborted = true; + child.kill('SIGTERM'); + for (const [key, value] of activeProcesses.entries()) { + if (value === child) activeProcesses.delete(key); + } + return true; + }; + + return { spawn: spawnRun, abort }; +}