mirror of
https://github.com/nexu-io/open-design.git
synced 2026-10-27 09:51:12 +01:00
feat(daemon): add pi coding agent adapter (#117)
* feat(daemon): add pi coding agent adapter Add pi (https://pi.dev) as a supported coding agent, using its --mode rpc JSON-RPC protocol over stdio for structured event streaming. Changes: - apps/daemon/pi-rpc.js: new RPC session handler that drives pi's --mode rpc protocol, translating typed agent events (text_delta, thinking_delta, tool_use, tool_result, usage, status) into the daemon's UI event format. Auto-resolves extension UI requests (fire-and-forget consumed, dialogs auto-approved) so pi stays unblocked in the headless web UI. Kills the process after agent_end since pi's RPC process is designed for multi-prompt sessions. - apps/daemon/agents.js: add pi agent definition with custom fetchModels (pi --list-models outputs to stderr, not stdout), 575+ models from 20+ providers, reasoning/thinking level support via --thinking flag, and streamFormat 'pi-rpc'. - apps/daemon/server.js: wire pi-rpc stream format to attachPiRpcSession; skip stdin.end() for pi-rpc since the RPC session manages stdin bidirectionally. - apps/daemon/acp.js: export createJsonLineStream for reuse by pi-rpc.js. - apps/daemon/pi-rpc.test.mjs: 19 unit tests covering model list parsing (TSV, dedup, edge cases), RPC event translation (text, thinking, tools, usage, compaction, retry), sendCommand wire format, extension UI auto-resolution. - e2e/tests/structured-streams.test.ts: add pi RPC tool_use/tool_result event mapping test alongside existing Claude/Copilot fixtures. Verified end-to-end: daemon /api/chat → pi RPC → SSE stream with status, text_delta, usage, and tool events. Live E2E test passes (OD_E2E_RUNTIMES=pi). All 59 project tests green. * refactor(daemon): migrate pi-rpc to TypeScript Follow upstream #118 TypeScript migration convention: rename pi-rpc.js → pi-rpc.ts and pi-rpc.test.mjs → pi-rpc.test.ts with @ts-nocheck header (same as all other daemon modules). Import paths remain ./pi-rpc.js per NodeNext module resolution. * fix(daemon): avoid duplicate usage events in pi-rpc handler Pi emits both message_end and turn_end per turn, both carrying usage data. Emitting from both handlers caused double-counting in the UI and any consumer that aggregates usage. Remove usage emission from the message_end branch since turn_end is the canonical per-turn usage source. Keep tool call extraction in message_end (unique data not available in turn_end). Add regression test confirming exactly one usage event is emitted when both message_end and turn_end carry usage for the same turn. Addresses Copilot P2 review on PR #117. * fix(daemon): scope pi RPC id counter per session, bump graceful shutdown Move nextRpcId and sendCommand inside attachPiRpcSession as local state, matching the pattern in acp.ts where nextId is scoped per session. Prevents RPC id collisions across concurrent /api/chat requests. Bump post-agent_end SIGTERM grace period from 2s to 5s and make it configurable via PI_GRACEFUL_SHUTDOWN_MS env var for resource- constrained machines. Add test confirming concurrent sessions get independent id sequences. * fix(daemon): wrap parser.feed in try-catch in pi-rpc Catch errors from parser.feed and route them through the existing fail() handler instead of letting them propagate as unhandled exceptions.
This commit is contained in:
Tom
@@ -77,7 +77,7 @@ function normalizeModels(models, defaultModelOption) {
|
||||
return out;
|
||||
}
|
||||
|
||||
function createJsonLineStream(onMessage) {
|
||||
export function createJsonLineStream(onMessage) {
|
||||
let buffer = '';
|
||||
return {
|
||||
feed(chunk) {
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
@@ -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));
|
||||
|
||||
Reference in New Issue
Block a user