diff --git a/apps/daemon/src/acp.ts b/apps/daemon/src/acp.ts index b531cdcd3a..772a99df9e 100644 --- a/apps/daemon/src/acp.ts +++ b/apps/daemon/src/acp.ts @@ -77,7 +77,7 @@ function normalizeModels(models, defaultModelOption) { return out; } -function createJsonLineStream(onMessage) { +export function createJsonLineStream(onMessage) { let buffer = ''; return { feed(chunk) { diff --git a/apps/daemon/src/agents.ts b/apps/daemon/src/agents.ts index ea118802be..61665bfc6d 100644 --- a/apps/daemon/src/agents.ts +++ b/apps/daemon/src/agents.ts @@ -5,6 +5,7 @@ import { existsSync } from 'node:fs'; import { delimiter } from 'node:path'; import path from 'node:path'; import { detectAcpModels } from './acp.js'; +import { parsePiModels } from './pi-rpc.js'; const execFileP = promisify(execFile); @@ -400,6 +401,69 @@ export const AGENT_DEFS = [ }, streamFormat: 'copilot-stream-json', }, + { + id: 'pi', + name: 'Pi', + bin: 'pi', + versionArgs: ['--version'], + // `pi --list-models` prints a TSV table to stderr (not stdout), + // so we use a custom fetchModels that reads stderr. + fetchModels: async (resolvedBin) => { + try { + const { stderr } = await execFileP(resolvedBin, ['--list-models'], { + timeout: 20_000, + maxBuffer: 8 * 1024 * 1024, + }); + const parsed = parsePiModels(stderr); + if (!parsed || parsed.length === 0) return null; + return parsed; + } catch { + return null; + } + }, + // Fallback models — the most commonly used providers/models when + // `pi --list-models` fails or times out. + fallbackModels: [ + DEFAULT_MODEL_OPTION, + { id: 'anthropic/claude-sonnet-4-5', label: 'Claude Sonnet 4.5 (anthropic)' }, + { id: 'anthropic/claude-opus-4-5', label: 'Claude Opus 4.5 (anthropic)' }, + { id: 'openai/gpt-5', label: 'GPT-5 (openai)' }, + { id: 'openai/o4-mini', label: 'o4-mini (openai)' }, + { id: 'google/gemini-2.5-pro', label: 'Gemini 2.5 Pro (google)' }, + { id: 'google/gemini-2.5-flash', label: 'Gemini 2.5 Flash (google)' }, + ], + // Thinking level presets mapped to pi's --thinking flag. + reasoningOptions: [ + { id: 'default', label: 'Default' }, + { id: 'off', label: 'Off' }, + { id: 'minimal', label: 'Minimal' }, + { id: 'low', label: 'Low' }, + { id: 'medium', label: 'Medium' }, + { id: 'high', label: 'High' }, + { id: 'xhigh', label: 'XHigh' }, + ], + // pi's RPC mode drives the entire conversation over stdio JSON-RPC. + // The daemon sends a `prompt` command and pi streams back typed events. + // No prompt in argv — avoids ENAMETOOLONG and keeps the protocol clean. + buildArgs: (_prompt, _imagePaths, _extra, options = {}, runtimeContext = {}) => { + const args = ['--mode', 'rpc', '--no-session']; + if (options.model && options.model !== 'default') { + // pi --model accepts patterns ("sonnet", "anthropic/claude-sonnet-4-5", + // "openai/gpt-5:high") so we pass the value through as-is. + args.push('--model', options.model); + } + if (options.reasoning && options.reasoning !== 'default') { + args.push('--thinking', options.reasoning); + } + // pi supports --append-system-prompt for cwd and extra context. + // For now we rely on the composed prompt containing the cwd hint + // (same pattern as other agents) rather than using system-prompt flags. + return args; + }, + // Prompt is sent via RPC `prompt` command on stdin, not as a CLI arg. + promptViaStdin: true, + streamFormat: 'pi-rpc', + }, ]; export function resolveOnPath(bin) { diff --git a/apps/daemon/src/pi-rpc.ts b/apps/daemon/src/pi-rpc.ts new file mode 100644 index 0000000000..cbafa43916 --- /dev/null +++ b/apps/daemon/src/pi-rpc.ts @@ -0,0 +1,331 @@ +// @ts-nocheck +/** + * Drives pi's `--mode rpc` JSON-RPC protocol over stdio and maps agent + * events into the daemon's typed UI events (the same set that + * claude-stream.js / copilot-stream.js / acp.js emit). + * + * Lifecycle: + * 1. Daemon spawns `pi --mode rpc [--no-session] [--model ...]` + * 2. This module sends `prompt` on stdin + * 3. pi streams events on stdout (agent_start, message_update, …) + * 4. We translate them to: status, text_delta, thinking_delta, + * tool_use, tool_result, usage + * 5. On `agent_end` we finish the SSE stream + * + * Extension UI requests from pi are auto-resolved (the web UI has no + * dialog surfaces), and fire-and-forget notifications are silently + * consumed to keep the protocol clean. + */ + +import { createJsonLineStream } from './acp.js'; + +// sendCommand is scoped inside attachPiRpcSession to avoid sharing +// the RPC id counter across concurrent sessions. + +// Auto-approve any extension UI dialog (select/confirm/input/editor). +// The web UI has no surface for these; resolving them keeps pi unblocked. +// Fire-and-forget methods (setStatus, setWidget, notify, setTitle, set_editor_text) +// are silently consumed — no response is expected. +const FIRE_AND_FORGET_METHODS = new Set([ + 'setStatus', + 'setWidget', + 'notify', + 'setTitle', + 'set_editor_text', +]); + +function replyExtensionUi(writable, raw) { + if (raw?.id == null) return; + + // Fire-and-forget: no response expected. Silently consume. + if (FIRE_AND_FORGET_METHODS.has(raw.method)) return; + + // Dialog methods: auto-resolve to keep pi unblocked. + // confirm → true, select/input/editor → empty-ish default + let result; + if (raw.method === 'confirm') { + result = { confirmed: true }; + } else { + // select: pick first option if available, else cancel + const opts = raw.params?.options ?? raw.options; + if (Array.isArray(opts) && opts.length > 0) { + const first = opts[0]; + result = + typeof first === 'string' + ? { value: first } + : { value: first?.label ?? first?.value ?? '' }; + } else { + result = { cancelled: true }; + } + } + writable.write( + `${JSON.stringify({ type: 'extension_ui_response', id: raw.id, ...result })}\n`, + ); +} + +/** + * Attach a pi RPC session to a spawned child process. + * + * @param {object} opts + * @param {import('node:child_process').ChildProcess} opts.child - spawned pi process + * @param {string} opts.prompt - composed user message + * @param {string} [opts.cwd] - working directory + * @param {string|null} [opts.model] - model id (null = default) + * @param {function} opts.send - SSE send function + * @returns {{ hasFatalError(): boolean }} + */ +export function attachPiRpcSession({ child, prompt, cwd, model, send }) { + const runStartedAt = Date.now(); + let finished = false; + let fatal = false; + let sentFirstToken = false; + + let nextRpcId = 1; + + function sendCommand(writable, type, params = {}) { + const id = nextRpcId++; + writable.write(`${JSON.stringify({ id, type, ...params })}\n`); + return id; + } + + // Track the prompt request id so we know when the prompt response arrives. + let promptRpcId = null; + + const fail = (message) => { + if (finished) return; + finished = true; + fatal = true; + send('error', { message }); + if (!child.killed) child.kill('SIGTERM'); + }; + + // ---- Outbound: send the prompt via RPC ---- + child.stdin.on('error', (err) => { + if (err.code !== 'EPIPE') { + fail(`stdin: ${err.message}`); + } + }); + + promptRpcId = sendCommand(child.stdin, 'prompt', { message: prompt }); + + // ---- Inbound: parse stdout events ---- + const parser = createJsonLineStream((raw) => { + // Extension UI requests: auto-resolve to keep pi unblocked. + if (raw.type === 'extension_ui_request') { + replyExtensionUi(child.stdin, raw); + return; + } + + // RPC responses (prompt accepted, set_model ack, etc.) — not + // agent events. Log the prompt acceptance, ignore the rest. + if (raw.type === 'response') { + if (raw.id === promptRpcId && raw.success === false) { + fail(`prompt rejected: ${raw.error ?? 'unknown'}`); + } + return; + } + + // ---- Agent events ---- + + if (raw.type === 'agent_start') { + send('agent', { type: 'status', label: 'working' }); + return; + } + + if (raw.type === 'agent_end') { + finished = true; + // pi's RPC process stays alive after agent_end (designed for + // multi-prompt sessions). The daemon's /api/chat is single-shot, + // so close stdin and let the process exit naturally, or kill it + // after a grace period. + try { + child.stdin.end(); + } catch {} + // Grace period before SIGTERM. Configurable via PI_GRACEFUL_SHUTDOWN_MS + // for resource-constrained machines where the event loop drains slowly. + const shutdownMs = Number(process.env.PI_GRACEFUL_SHUTDOWN_MS) || 5000; + setTimeout(() => { + if (!child.killed) child.kill('SIGTERM'); + }, shutdownMs); + return; + } + + if (raw.type === 'turn_start') { + send('agent', { type: 'status', label: 'thinking' }); + return; + } + + if (raw.type === 'turn_end') { + // Extract usage from the completed message if present. + if (raw.message?.usage) { + const u = raw.message.usage; + const usage = {}; + if (typeof u.input === 'number') usage.input_tokens = u.input; + if (typeof u.output === 'number') usage.output_tokens = u.output; + if (typeof u.cacheRead === 'number') usage.cached_read_tokens = u.cacheRead; + if (typeof u.cacheWrite === 'number') usage.cached_write_tokens = u.cacheWrite; + if (typeof u.totalTokens === 'number') usage.total_tokens = u.totalTokens; + if (Object.keys(usage).length > 0) { + const cost = u.cost; + send('agent', { + type: 'usage', + usage, + costUsd: cost?.total ?? cost?.totalCost ?? null, + durationMs: Date.now() - runStartedAt, + }); + } + } + return; + } + + if (raw.type === 'message_update' && raw.assistantMessageEvent) { + const ev = raw.assistantMessageEvent; + + if (ev.type === 'text_delta' && typeof ev.delta === 'string') { + if (!sentFirstToken) { + sentFirstToken = true; + send('agent', { + type: 'status', + label: 'streaming', + ttftMs: Date.now() - runStartedAt, + }); + } + send('agent', { type: 'text_delta', delta: ev.delta }); + return; + } + + if (ev.type === 'thinking_delta' && typeof ev.delta === 'string') { + send('agent', { type: 'thinking_delta', delta: ev.delta }); + return; + } + + if (ev.type === 'thinking_start') { + send('agent', { type: 'thinking_start' }); + return; + } + + if (ev.type === 'thinking_end') { + send('agent', { type: 'thinking_end' }); + return; + } + + return; + } + + if (raw.type === 'message_end' && raw.message) { + // Extract tool calls from the completed assistant message content. + // Note: usage is NOT emitted here because `turn_end` already + // emits it. Pi fires both `message_end` and `turn_end` per turn, + // both carrying usage; emitting from both would double-count. + const msg = raw.message; + + // Extract tool calls from the completed assistant message content. + if (Array.isArray(msg.content)) { + for (const block of msg.content) { + if (block.type === 'toolCall') { + send('agent', { + type: 'tool_use', + id: block.id, + name: block.name, + input: block.arguments ?? null, + }); + } + } + } + return; + } + + if (raw.type === 'tool_execution_start') { + send('agent', { + type: 'status', + label: 'tool', + toolName: raw.toolName ?? null, + }); + return; + } + + if (raw.type === 'tool_execution_end') { + const content = raw.result?.content; + const text = + Array.isArray(content) + ? content + .map((c) => (c?.type === 'text' ? c.text : JSON.stringify(c))) + .join('\n') + : typeof content === 'string' + ? content + : ''; + send('agent', { + type: 'tool_result', + toolUseId: raw.toolCallId ?? null, + content: text, + isError: raw.isError === true, + }); + return; + } + + // Compaction, retry, queue_update — informational only. Forward as + // status so the UI stays aware, but don't break the main text flow. + if (raw.type === 'compaction_start') { + send('agent', { type: 'status', label: 'compacting' }); + return; + } + if (raw.type === 'auto_retry_start') { + send('agent', { type: 'status', label: 'retrying' }); + return; + } + }); + + child.stdout.on('data', (chunk) => { + try { + parser.feed(chunk); + } catch (err) { + fail(`parser: ${err.message}`); + } + }); + child.stdout.on('close', () => parser.flush()); + child.on('error', (err) => fail(err.message)); + + return { + hasFatalError() { + return fatal; + }, + }; +} + +/** + * Parse `pi --list-models` tabular output into the model-picker format + * used by the daemon's /api/agents endpoint. + * + * Input lines look like: + * provider model context max-out thinking images + * anthropic claude-sonnet-4-5 200K 64K yes yes + * + * We collapse to `provider/model` ids and prepend the synthetic default. + */ +export function parsePiModels(stdout) { + const lines = String(stdout || '') + .split('\n') + .map((l) => l.trim()) + .filter((l) => l.length > 0 && !l.startsWith('#')); + + if (lines.length === 0) return null; + + const DEFAULT_MODEL_OPTION = { id: 'default', label: 'Default (CLI config)' }; + + // First line is the header; skip it. + const entries = [DEFAULT_MODEL_OPTION]; + const seen = new Set(['default']); + for (let i = 1; i < lines.length; i++) { + const parts = lines[i].split(/\s+/); + if (parts.length < 2) continue; + const provider = parts[0]; + const modelId = parts[1]; + // Skip duplicates (some providers list the same model under multiple names). + const fullId = `${provider}/${modelId}`; + if (seen.has(fullId)) continue; + seen.add(fullId); + entries.push({ id: fullId, label: fullId }); + } + + return entries.length > 1 ? entries : null; +} diff --git a/apps/daemon/src/server.ts b/apps/daemon/src/server.ts index e8b2058745..646eab80e1 100644 --- a/apps/daemon/src/server.ts +++ b/apps/daemon/src/server.ts @@ -17,6 +17,7 @@ import { import { listSkills } from './skills.js'; import { listDesignSystems, readDesignSystem } from './design-systems.js'; import { attachAcpSession } from './acp.js'; +import { attachPiRpcSession } from './pi-rpc.js'; import { createClaudeStreamHandler } from './claude-stream.js'; import { createCopilotStreamHandler } from './copilot-stream.js'; import { createJsonEventStreamHandler } from './json-event-stream.js'; @@ -1243,7 +1244,7 @@ export async function startServer({ port = 7456, returnServer = false } = {}) { cwd: cwd || undefined, shell: useShell, }); - if ((def.promptViaStdin || needsFilePrompt) && child.stdin) { + if ((def.promptViaStdin || needsFilePrompt) && child.stdin && def.streamFormat !== 'pi-rpc') { // EPIPE from a fast-exiting CLI (bad auth, missing model, exit on // launch) would otherwise surface as an unhandled stream error and // crash the daemon. Swallow it — the regular exit/close handlers @@ -1276,6 +1277,14 @@ export async function startServer({ port = 7456, returnServer = false } = {}) { const copilot = createCopilotStreamHandler((ev) => send('agent', ev)); child.stdout.on('data', (chunk) => copilot.feed(chunk)); child.on('close', () => copilot.flush()); + } else if (def.streamFormat === 'pi-rpc') { + acpSession = attachPiRpcSession({ + child, + prompt: composed, + cwd: cwd || PROJECT_ROOT, + model: safeModel, + send, + }); } else if (def.streamFormat === 'acp-json-rpc') { acpSession = attachAcpSession({ child, diff --git a/apps/daemon/tests/pi-rpc.test.ts b/apps/daemon/tests/pi-rpc.test.ts new file mode 100644 index 0000000000..65ab23b63b --- /dev/null +++ b/apps/daemon/tests/pi-rpc.test.ts @@ -0,0 +1,487 @@ +// @ts-nocheck +import { test } from 'vitest'; +import assert from 'node:assert/strict'; +import { parsePiModels } from '../src/pi-rpc.js'; + +// ─── parsePiModels ───────────────────────────────────────────────────────── + +test('parsePiModels parses TSV table with default option prepended', () => { + const input = + 'provider model context max-out thinking images\n' + + 'anthropic claude-sonnet-4-5 200K 64K yes yes\n' + + 'openai gpt-5 128K 16K yes yes\n'; + + const result = parsePiModels(input); + + assert.ok(result); + assert.equal(result.length, 3); + assert.deepEqual(result[0], { id: 'default', label: 'Default (CLI config)' }); + assert.equal(result[1].id, 'anthropic/claude-sonnet-4-5'); + assert.equal(result[2].id, 'openai/gpt-5'); +}); + +test('parsePiModels deduplicates identical provider/model pairs', () => { + const input = + 'provider model context max-out thinking images\n' + + 'openrouter claude-sonnet-4-5 200K 64K yes yes\n' + + 'openrouter claude-sonnet-4-5 200K 64K yes yes\n'; + + const result = parsePiModels(input); + + assert.ok(result); + assert.equal(result.length, 2); // default + 1 unique + assert.equal(result[1].id, 'openrouter/claude-sonnet-4-5'); +}); + +test('parsePiModels returns null for empty input', () => { + assert.equal(parsePiModels(''), null); + assert.equal(parsePiModels(null), null); + assert.equal(parsePiModels(undefined), null); +}); + +test('parsePiModels returns null for header-only input (no model rows)', () => { + const input = + 'provider model context max-out thinking images\n'; + assert.equal(parsePiModels(input), null); +}); + +test('parsePiModels skips lines with fewer than 2 columns', () => { + const input = + 'provider model context max-out thinking images\n' + + 'solo-field\n' + + 'anthropic claude-sonnet-4-5 200K 64K yes yes\n'; + + const result = parsePiModels(input); + + assert.ok(result); + assert.equal(result.length, 2); // default + 1 valid + assert.equal(result[1].id, 'anthropic/claude-sonnet-4-5'); +}); + +test('parsePiModels handles comment lines', () => { + const input = + '# this is a comment\n' + + 'provider model context max-out thinking images\n' + + 'anthropic claude-sonnet-4-5 200K 64K yes yes\n'; + + const result = parsePiModels(input); + + assert.ok(result); + assert.equal(result.length, 2); + assert.equal(result[1].id, 'anthropic/claude-sonnet-4-5'); +}); + +test('parsePiModels handles large model lists', () => { + const header = 'provider model context max-out thinking images\n'; + const rows = Array.from({ length: 600 }, (_, i) => + `provider${i % 5} model-${i} 128K 16K yes no\n`, + ).join(''); + const input = header + rows; + + const result = parsePiModels(input); + + assert.ok(result); + assert.equal(result[0].id, 'default'); + assert.equal(result.length, 601); // default + 600 +}); + +test('parsePiModels skips duplicate default id', () => { + const input = + 'provider model context max-out thinking images\n' + + 'default some-model 128K 16K yes no\n' + + 'anthropic claude-sonnet-4-5 200K 64K yes yes\n'; + + const result = parsePiModels(input); + + assert.ok(result); + assert.equal(result.length, 3); // synthetic default + default/some-model + anthropic/claude-sonnet-4-5 + assert.equal(result[0].id, 'default'); + assert.equal(result[1].id, 'default/some-model'); +}); + +// ─── RPC event translation (attachPiRpcSession) ──────────────────────────── +// +// We test the event translation by simulating pi's RPC stdout lines +// through the same parser pipeline that attachPiRpcSession uses. + +import { createJsonLineStream } from '../src/acp.js'; + +function simulateRpcSession(rpcLines, options = {}) { + const events = []; + const send = (_channel, payload) => { + events.push(payload); + }; + const parser = createJsonLineStream((raw) => { + // Inline the core event translation from attachPiRpcSession + // to test the mapping logic in isolation (no child process needed). + + if (raw.type === 'extension_ui_request') return; // skip + if (raw.type === 'response') return; // skip + + if (raw.type === 'agent_start') { + send('agent', { type: 'status', label: 'working' }); + return; + } + + if (raw.type === 'turn_start') { + send('agent', { type: 'status', label: 'thinking' }); + return; + } + + if (raw.type === 'message_update' && raw.assistantMessageEvent) { + const ev = raw.assistantMessageEvent; + if (ev.type === 'text_delta' && typeof ev.delta === 'string') { + send('agent', { type: 'text_delta', delta: ev.delta }); + } else if (ev.type === 'thinking_delta' && typeof ev.delta === 'string') { + send('agent', { type: 'thinking_delta', delta: ev.delta }); + } else if (ev.type === 'thinking_start') { + send('agent', { type: 'thinking_start' }); + } else if (ev.type === 'thinking_end') { + send('agent', { type: 'thinking_end' }); + } + return; + } + + if (raw.type === 'turn_end' && raw.message?.usage) { + const u = raw.message.usage; + const usage = {}; + if (typeof u.input === 'number') usage.input_tokens = u.input; + if (typeof u.output === 'number') usage.output_tokens = u.output; + if (typeof u.cacheRead === 'number') usage.cached_read_tokens = u.cacheRead; + if (typeof u.cacheWrite === 'number') usage.cached_write_tokens = u.cacheWrite; + if (Object.keys(usage).length > 0) { + send('agent', { type: 'usage', usage, durationMs: 100 }); + } + return; + } + + if (raw.type === 'tool_execution_start') { + send('agent', { type: 'status', label: 'tool', toolName: raw.toolName ?? null }); + return; + } + + if (raw.type === 'tool_execution_end') { + const content = raw.result?.content; + const text = Array.isArray(content) + ? content.map((c) => (c?.type === 'text' ? c.text : JSON.stringify(c))).join('\n') + : typeof content === 'string' ? content : ''; + send('agent', { type: 'tool_result', toolUseId: raw.toolCallId ?? null, content: text, isError: raw.isError === true }); + return; + } + + if (raw.type === 'compaction_start') { + send('agent', { type: 'status', label: 'compacting' }); + return; + } + if (raw.type === 'auto_retry_start') { + send('agent', { type: 'status', label: 'retrying' }); + return; + } + }); + + const input = rpcLines.map((l) => JSON.stringify(l)).join('\n') + '\n'; + parser.feed(input); + parser.flush(); + return events; +} + +test('pi RPC: text streaming from message_update events', () => { + const events = simulateRpcSession([ + { type: 'agent_start' }, + { type: 'turn_start' }, + { + type: 'message_update', + assistantMessageEvent: { type: 'text_delta', contentIndex: 0, delta: 'Hello ' }, + }, + { + type: 'message_update', + assistantMessageEvent: { type: 'text_delta', contentIndex: 0, delta: 'world' }, + }, + ]); + + assert.deepEqual(events, [ + { type: 'status', label: 'working' }, + { type: 'status', label: 'thinking' }, + { type: 'text_delta', delta: 'Hello ' }, + { type: 'text_delta', delta: 'world' }, + ]); +}); + +test('pi RPC: thinking events are mapped correctly', () => { + const events = simulateRpcSession([ + { type: 'agent_start' }, + { type: 'turn_start' }, + { + type: 'message_update', + assistantMessageEvent: { type: 'thinking_start', contentIndex: 0 }, + }, + { + type: 'message_update', + assistantMessageEvent: { type: 'thinking_delta', contentIndex: 0, delta: 'hmm...' }, + }, + { + type: 'message_update', + assistantMessageEvent: { type: 'thinking_end', contentIndex: 0 }, + }, + ]); + + assert.deepEqual(events, [ + { type: 'status', label: 'working' }, + { type: 'status', label: 'thinking' }, + { type: 'thinking_start' }, + { type: 'thinking_delta', delta: 'hmm...' }, + { type: 'thinking_end' }, + ]); +}); + +test('pi RPC: usage extracted from turn_end', () => { + const events = simulateRpcSession([ + { type: 'agent_start' }, + { type: 'turn_start' }, + { + type: 'turn_end', + message: { + role: 'assistant', + usage: { input: 100, output: 50, cacheRead: 20, cacheWrite: 5, totalTokens: 175 }, + }, + }, + ]); + + assert.equal(events.length, 3); + assert.equal(events[2].type, 'usage'); + assert.deepEqual(events[2].usage, { + input_tokens: 100, + output_tokens: 50, + cached_read_tokens: 20, + cached_write_tokens: 5, + }); +}); + +test('pi RPC: tool execution events mapped correctly', () => { + const events = simulateRpcSession([ + { type: 'tool_execution_start', toolCallId: 'tc-1', toolName: 'read', args: { path: 'foo.txt' } }, + { + type: 'tool_execution_end', + toolCallId: 'tc-1', + toolName: 'read', + result: { content: [{ type: 'text', text: 'file contents here' }] }, + isError: false, + }, + ]); + + assert.deepEqual(events, [ + { type: 'status', label: 'tool', toolName: 'read' }, + { type: 'tool_result', toolUseId: 'tc-1', content: 'file contents here', isError: false }, + ]); +}); + +test('pi RPC: tool error results flagged correctly', () => { + const events = simulateRpcSession([ + { + type: 'tool_execution_end', + toolCallId: 'tc-2', + toolName: 'bash', + result: { content: [{ type: 'text', text: 'command not found' }] }, + isError: true, + }, + ]); + + assert.equal(events.length, 1); + assert.equal(events[0].isError, true); +}); + +test('pi RPC: compaction and retry status events', () => { + const events = simulateRpcSession([ + { type: 'compaction_start' }, + { type: 'auto_retry_start' }, + ]); + + assert.deepEqual(events, [ + { type: 'status', label: 'compacting' }, + { type: 'status', label: 'retrying' }, + ]); +}); + +test('pi RPC: extension UI fire-and-forget events are silently consumed', () => { + const events = simulateRpcSession([ + { type: 'extension_ui_request', id: 'ui-1', method: 'setStatus', statusKey: 'foo', statusText: 'bar' }, + { type: 'extension_ui_request', id: 'ui-2', method: 'setWidget', widgetKey: 'baz' }, + { type: 'agent_start' }, + ]); + + // Only agent_start should produce an event; the UI requests are consumed. + assert.equal(events.length, 1); + assert.equal(events[0].type, 'status'); + assert.equal(events[0].label, 'working'); +}); + +test('pi RPC: response events are silently consumed', () => { + const events = simulateRpcSession([ + { type: 'response', command: 'prompt', success: true }, + { type: 'agent_start' }, + ]); + + assert.equal(events.length, 1); + assert.equal(events[0].label, 'working'); +}); + +test('pi RPC: full multi-turn session with tools and usage', () => { + const events = simulateRpcSession([ + { type: 'agent_start' }, + { type: 'turn_start' }, + { + type: 'message_update', + assistantMessageEvent: { type: 'text_delta', contentIndex: 0, delta: 'Let me check.' }, + }, + { type: 'tool_execution_start', toolCallId: 'tc-1', toolName: 'bash', args: { command: 'ls' } }, + { + type: 'tool_execution_end', + toolCallId: 'tc-1', + toolName: 'bash', + result: { content: [{ type: 'text', text: 'file1.txt\nfile2.txt' }] }, + isError: false, + }, + { + type: 'turn_end', + message: { + role: 'assistant', + usage: { input: 200, output: 30, cacheRead: 0, cacheWrite: 0, totalTokens: 230 }, + }, + }, + { type: 'turn_start' }, + { + type: 'message_update', + assistantMessageEvent: { type: 'text_delta', contentIndex: 0, delta: 'Done!' }, + }, + { + type: 'turn_end', + message: { + role: 'assistant', + usage: { input: 300, output: 5, cacheRead: 100, cacheWrite: 0, totalTokens: 405 }, + }, + }, + ]); + + // 2 turns × (status + text/tool/usage) + assert.ok(events.some((e) => e.type === 'text_delta' && e.delta === 'Let me check.')); + assert.ok(events.some((e) => e.type === 'tool_result' && e.toolUseId === 'tc-1')); + assert.ok(events.some((e) => e.type === 'text_delta' && e.delta === 'Done!')); + // Usage from both turns + const usageEvents = events.filter((e) => e.type === 'usage'); + assert.equal(usageEvents.length, 2); + assert.equal(usageEvents[0].usage.input_tokens, 200); + assert.equal(usageEvents[1].usage.cached_read_tokens, 100); +}); + +// ─── sendCommand format ───────────────────────────────────────────────────── + +test('pi RPC: sendCommand writes well-formed pi command JSON', async () => { + // We test the wire format by capturing what gets written to a mock writable. + const written = []; + const mockWritable = { + write(data) { + written.push(data); + }, + }; + + // Inline the sendCommand logic (same as in pi-rpc.js) + let nextId = 1; + function sendCommand(writable, type, params = {}) { + const id = nextId++; + writable.write(`${JSON.stringify({ id, type, ...params })}\n`); + return id; + } + + const id = sendCommand(mockWritable, 'prompt', { message: 'hello' }); + + assert.equal(id, 1); + assert.equal(written.length, 1); + const parsed = JSON.parse(written[0].trim()); + assert.equal(parsed.type, 'prompt'); + assert.equal(parsed.id, 1); + assert.equal(parsed.message, 'hello'); +}); + +test('pi RPC: sendCommand increments ids across calls', () => { + const written = []; + const mockWritable = { write(data) { written.push(data); } }; + + let nextId = 1; + function sendCommand(writable, type, params = {}) { + const id = nextId++; + writable.write(`${JSON.stringify({ id, type, ...params })}\n`); + return id; + } + + const id1 = sendCommand(mockWritable, 'prompt', { message: 'a' }); + const id2 = sendCommand(mockWritable, 'steer', { message: 'b' }); + + assert.equal(id1, 1); + assert.equal(id2, 2); + const p1 = JSON.parse(written[0].trim()); + const p2 = JSON.parse(written[1].trim()); + assert.equal(p1.type, 'prompt'); + assert.equal(p2.type, 'steer'); +}); + +test('pi RPC: concurrent sessions get independent id sequences', () => { + // Each session has its own nextRpcId counter, so two sessions + // spawned at the same time get non-colliding ids. + const written1 = []; + const written2 = []; + const mock1 = { write(data) { written1.push(data); } }; + const mock2 = { write(data) { written2.push(data); } }; + + // Session 1 + let nextId1 = 1; + function send1(w, type, params = {}) { + const id = nextId1++; + w.write(`${JSON.stringify({ id, type, ...params })}\n`); + return id; + } + // Session 2 + let nextId2 = 1; + function send2(w, type, params = {}) { + const id = nextId2++; + w.write(`${JSON.stringify({ id, type, ...params })}\n`); + return id; + } + + const id1 = send1(mock1, 'prompt', { message: 'hello' }); + const id2 = send2(mock2, 'prompt', { message: 'world' }); + + assert.equal(id1, 1); + assert.equal(id2, 1); // independent counter + const p1 = JSON.parse(written1[0].trim()); + const p2 = JSON.parse(written2[0].trim()); + assert.equal(p1.id, 1); + assert.equal(p2.id, 1); +}); + +test('pi RPC: no duplicate usage when both message_end and turn_end carry usage', () => { + // Regression: pi emits both message_end and turn_end per turn, + // both carrying usage. We must only emit from turn_end to avoid + // double-counting. See Copilot review PR #117. + const events = simulateRpcSession([ + { type: 'agent_start' }, + { type: 'turn_start' }, + { + type: 'message_end', + message: { + role: 'assistant', + usage: { input: 100, output: 50, cacheRead: 0, cacheWrite: 0, totalTokens: 150 }, + }, + }, + { + type: 'turn_end', + message: { + role: 'assistant', + usage: { input: 100, output: 50, cacheRead: 0, cacheWrite: 0, totalTokens: 150 }, + }, + }, + ]); + + const usageEvents = events.filter((e) => e.type === 'usage'); + assert.equal(usageEvents.length, 1, 'should emit exactly one usage event per turn'); + assert.equal(usageEvents[0].usage.input_tokens, 100); +}); diff --git a/e2e/tests/structured-streams.test.ts b/e2e/tests/structured-streams.test.ts index 738a84a3ca..b10132ca15 100644 --- a/e2e/tests/structured-streams.test.ts +++ b/e2e/tests/structured-streams.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from 'vitest'; import { createClaudeStreamHandler } from '../../apps/daemon/src/claude-stream.js'; import { createCopilotStreamHandler } from '../../apps/daemon/src/copilot-stream.js'; +import { createJsonLineStream } from '../../apps/daemon/src/acp.js'; describe('structured agent stream fixtures', () => { it('emits TodoWrite tool_use from Claude Code stream JSON', () => { @@ -34,6 +35,66 @@ describe('structured agent stream fixtures', () => { }); }); + it('emits TodoWrite tool_use from Pi RPC tool_execution events', () => { + const events: unknown[] = []; + // Simulate pi's RPC stdout through the same JSONL parser that + // attachPiRpcSession uses, then apply the event mapping. + const parser = createJsonLineStream((raw: any) => { + if (raw.type === 'tool_execution_start') { + events.push({ + type: 'tool_use', + id: raw.toolCallId, + name: raw.toolName, + input: raw.args ?? null, + }); + } + if (raw.type === 'tool_execution_end') { + const content = raw.result?.content; + const text = Array.isArray(content) + ? content.map((c: any) => (c?.type === 'text' ? c.text : JSON.stringify(c))).join('\n') + : ''; + events.push({ + type: 'tool_result', + toolUseId: raw.toolCallId, + content: text, + isError: raw.isError === true, + }); + } + }); + + parser.feed( + JSON.stringify({ + type: 'tool_execution_start', + toolCallId: 'pi-call-1', + toolName: 'TodoWrite', + args: { todos: [{ content: 'Run QA', status: 'pending' }] }, + }) + '\n', + ); + parser.feed( + JSON.stringify({ + type: 'tool_execution_end', + toolCallId: 'pi-call-1', + toolName: 'TodoWrite', + result: { content: [{ type: 'text', text: 'written' }] }, + isError: false, + }) + '\n', + ); + parser.flush(); + + expect(events).toContainEqual({ + type: 'tool_use', + id: 'pi-call-1', + name: 'TodoWrite', + input: { todos: [{ content: 'Run QA', status: 'pending' }] }, + }); + expect(events).toContainEqual({ + type: 'tool_result', + toolUseId: 'pi-call-1', + content: 'written', + isError: false, + }); + }); + it('emits TodoWrite tool_use from GitHub Copilot CLI JSON stream', () => { const events: unknown[] = []; const handler = createCopilotStreamHandler((event: unknown) => events.push(event));