dotfiles: add interactive pi planner workflow

This commit is contained in:
Rasyidan Akbar F. 2026-03-12 16:54:57 +07:00
commit 5e01e2e44f
15 changed files with 1472 additions and 108 deletions

View file

@ -5,6 +5,8 @@ Installed roles:
- `reviewer` — review changes, bugs, risks, maintainability
- `librarian` — explore and map a codebase
- `uiux-designer` — UI/UX design direction, flows, states, and implementation guidance
- `executor` — implement ordered coding tasks from a defined todo list
- `planner` — interactive planning specialist that can ask clarifying questions and return an ordered executor-ready todo list
## Commands
@ -16,6 +18,16 @@ Installed roles:
- `Use the subagent tool with reviewer to inspect the auth refactor.`
- `Use subagent in parallel: librarian maps the settings flow, uiux-designer proposes an improved UX direction, reviewer lists product and implementation risks.`
- `Chain librarian -> uiux-designer to redesign onboarding. Use {previous} in the second task.`
- `Use executor to implement this ordered todo list exactly as written, then summarize what was completed and validated.`
- `Use planner to clarify ambiguity only when needed and output an ordered implementation todo list for executor handoff.`
## Interactive planner + ask_user
`planner` is marked `interactive: true` and runs through an RPC-backed child agent path.
- Interactive child agents use a helper extension tool: `ask_user`
- `ask_user` supports `input`, `confirm`, and `select` modes (plus `editor`)
- Child tool activation is applied on `session_start` via `PI_SUBAGENT_TOOLS` so custom tools like `ask_user` are reliably active
## Notes
@ -26,3 +38,11 @@ Installed roles:
- `reviewer` uses `openai-codex/gpt-5.4:xhigh`
- `librarian` uses `anthropic/claude-sonnet-4-6`
- `uiux-designer` uses `google-antigravity/gemini-3.1-pro-high`
- `executor` uses `openai-codex/gpt-5.3-codex:high`
- `planner` uses `anthropic/claude-sonnet-4-6`
## Current interactive limitations
- Interactive subagents are currently supported only in **single mode** (`{ agent, task }`)
- Interactive subagents are currently rejected in **parallel** and **chain** modes
- Interactive subagents require `ctx.hasUI` (interactive TUI or RPC mode with extension UI handling)

View file

@ -9,6 +9,7 @@ export interface AgentConfig {
description: string;
tools?: string[];
model?: string;
interactive?: boolean;
systemPrompt: string;
source: "user" | "project";
filePath: string;
@ -19,6 +20,55 @@ export interface AgentDiscoveryResult {
projectAgentsDir: string | null;
}
function normalizeString(value: unknown): string | undefined {
if (typeof value === "string") {
const trimmed = value.trim();
return trimmed.length > 0 ? trimmed : undefined;
}
if (typeof value === "number" || typeof value === "boolean") return String(value);
return undefined;
}
function normalizeTools(value: unknown): string[] | undefined {
if (Array.isArray(value)) {
const tools = value
.flatMap((entry) => {
if (typeof entry !== "string") return [];
return entry
.split(",")
.map((part) => part.trim())
.filter(Boolean);
})
.filter(Boolean);
return tools.length > 0 ? tools : undefined;
}
if (typeof value === "string") {
const tools = value
.split(",")
.map((tool) => tool.trim())
.filter(Boolean);
return tools.length > 0 ? tools : undefined;
}
return undefined;
}
function normalizeBoolean(value: unknown): boolean | undefined {
if (typeof value === "boolean") return value;
if (typeof value === "number") {
if (value === 1) return true;
if (value === 0) return false;
return undefined;
}
if (typeof value === "string") {
const normalized = value.trim().toLowerCase();
if (["true", "1", "yes", "y", "on"].includes(normalized)) return true;
if (["false", "0", "no", "n", "off"].includes(normalized)) return false;
}
return undefined;
}
function loadAgentsFromDir(dir: string, source: "user" | "project"): AgentConfig[] {
const agents: AgentConfig[] = [];
if (!fs.existsSync(dir)) return agents;
@ -42,19 +92,17 @@ function loadAgentsFromDir(dir: string, source: "user" | "project"): AgentConfig
continue;
}
const { frontmatter, body } = parseFrontmatter<Record<string, string>>(content);
if (!frontmatter.name || !frontmatter.description) continue;
const tools = frontmatter.tools
?.split(",")
.map((t: string) => t.trim())
.filter(Boolean);
const { frontmatter, body } = parseFrontmatter<Record<string, unknown>>(content);
const name = normalizeString(frontmatter.name);
const description = normalizeString(frontmatter.description);
if (!name || !description) continue;
agents.push({
name: frontmatter.name,
description: frontmatter.description,
tools: tools && tools.length > 0 ? tools : undefined,
model: frontmatter.model,
name,
description,
tools: normalizeTools(frontmatter.tools),
model: normalizeString(frontmatter.model),
interactive: normalizeBoolean(frontmatter.interactive),
systemPrompt: body,
source,
filePath,
@ -109,7 +157,7 @@ export function formatAgentList(agents: AgentConfig[], maxItems: number): { text
const listed = agents.slice(0, maxItems);
const remaining = agents.length - listed.length;
return {
text: listed.map((a) => `${a.name} (${a.source}): ${a.description}`).join("; "),
text: listed.map((agent) => `${agent.name} (${agent.source}): ${agent.description}`).join("; "),
remaining,
};
}

View file

@ -2,53 +2,27 @@ import { spawn } from "node:child_process";
import * as fs from "node:fs";
import * as os from "node:os";
import * as path from "node:path";
import type { Message } from "@mariozechner/pi-ai";
import { StringEnum } from "@mariozechner/pi-ai";
import { type ExtensionAPI, getMarkdownTheme } from "@mariozechner/pi-coding-agent";
import { Container, Markdown, Spacer, Text } from "@mariozechner/pi-tui";
import { Type } from "@sinclair/typebox";
import { type AgentConfig, type AgentScope, discoverAgents, formatAgentList } from "./agents.js";
import { runInteractiveAgentRpc } from "./rpc-runner.js";
import {
applyAssistantUsage,
createEmptyUsageStats,
getDisplayItems,
getFinalOutput,
type DisplayItem,
type OnUpdateCallback,
type SingleResult,
type SubagentDetails,
} from "./subagent-types.js";
const MAX_PARALLEL_TASKS = 8;
const MAX_CONCURRENCY = 4;
const COLLAPSED_ITEM_COUNT = 10;
interface UsageStats {
input: number;
output: number;
cacheRead: number;
cacheWrite: number;
cost: number;
contextTokens: number;
turns: number;
}
interface SingleResult {
agent: string;
agentSource: "user" | "project" | "unknown";
task: string;
exitCode: number;
messages: Message[];
stderr: string;
usage: UsageStats;
model?: string;
stopReason?: string;
errorMessage?: string;
step?: number;
}
interface SubagentDetails {
mode: "single" | "parallel" | "chain";
agentScope: AgentScope;
projectAgentsDir: string | null;
results: SingleResult[];
}
type ToolContent = Array<{ type: "text"; text: string }>;
type ToolUpdate = { content: ToolContent; details?: SubagentDetails; isError?: boolean };
type OnUpdateCallback = (partial: ToolUpdate) => void;
type DisplayItem = { type: "text"; text: string } | { type: "toolCall"; name: string; args: Record<string, any> };
function formatTokens(count: number): string {
if (count < 1000) return count.toString();
if (count < 10000) return `${(count / 1000).toFixed(1)}k`;
@ -144,31 +118,6 @@ function formatToolCall(
}
}
function getFinalOutput(messages: Message[]): string {
for (let i = messages.length - 1; i >= 0; i--) {
const msg = messages[i];
if (msg.role === "assistant") {
for (const part of msg.content) {
if (part.type === "text") return part.text;
}
}
}
return "";
}
function getDisplayItems(messages: Message[]): DisplayItem[] {
const items: DisplayItem[] = [];
for (const msg of messages) {
if (msg.role === "assistant") {
for (const part of msg.content) {
if (part.type === "text") items.push({ type: "text", text: part.text });
else if (part.type === "toolCall") items.push({ type: "toolCall", name: part.name, args: part.arguments });
}
}
}
return items;
}
async function mapWithConcurrencyLimit<TIn, TOut>(
items: TIn[],
concurrency: number,
@ -218,7 +167,7 @@ async function runSingleAgent(
exitCode: 1,
messages: [],
stderr: `Unknown agent: "${agentName}". Available agents: ${available}.`,
usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 },
usage: createEmptyUsageStats(),
step,
};
}
@ -237,7 +186,7 @@ async function runSingleAgent(
exitCode: 0,
messages: [],
stderr: "",
usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 },
usage: createEmptyUsageStats(),
model: agent.model,
step,
};
@ -275,28 +224,14 @@ async function runSingleAgent(
}
if (event.type === "message_end" && event.message) {
const msg = event.message as Message;
const msg = event.message;
currentResult.messages.push(msg);
if (msg.role === "assistant") {
currentResult.usage.turns++;
const usage = msg.usage;
if (usage) {
currentResult.usage.input += usage.input || 0;
currentResult.usage.output += usage.output || 0;
currentResult.usage.cacheRead += usage.cacheRead || 0;
currentResult.usage.cacheWrite += usage.cacheWrite || 0;
currentResult.usage.cost += usage.cost?.total || 0;
currentResult.usage.contextTokens = usage.totalTokens || 0;
}
if (!currentResult.model && msg.model) currentResult.model = msg.model;
if (msg.stopReason) currentResult.stopReason = msg.stopReason;
if (msg.errorMessage) currentResult.errorMessage = msg.errorMessage;
}
applyAssistantUsage(currentResult, msg);
emitUpdate();
}
if (event.type === "tool_result_end" && event.message) {
currentResult.messages.push(event.message as Message);
currentResult.messages.push(event.message);
emitUpdate();
}
};
@ -349,6 +284,10 @@ async function runSingleAgent(
}
}
function findAgentConfig(agents: AgentConfig[], name: string): AgentConfig | undefined {
return agents.find((agent) => agent.name === name);
}
const TaskItem = Type.Object({
agent: Type.String({ description: "Name of the agent to invoke" }),
task: Type.String({ description: "Task to delegate to the agent" }),
@ -396,7 +335,7 @@ export default function (pi: ExtensionAPI) {
description: [
"Delegate work to specialized subagents with isolated context.",
"Supports single, parallel, and chained execution.",
"Useful built-in roles: reviewer, librarian, uiux-designer.",
"Useful built-in roles: reviewer, librarian, uiux-designer, executor.",
'Default scope is "user" (~/.pi/agent/agents). Set agentScope to "both" or "project" to include repo-local agents.',
].join(" "),
parameters: SubagentParams,
@ -429,6 +368,33 @@ export default function (pi: ExtensionAPI) {
};
}
if (hasTasks && params.tasks?.some((task) => findAgentConfig(agents, task.agent)?.interactive === true)) {
return {
content: [{ type: "text", text: "Interactive subagents are currently supported only in single mode." }],
details: makeDetails("parallel")([]),
isError: true,
};
}
if (hasChain && params.chain?.some((step) => findAgentConfig(agents, step.agent)?.interactive === true)) {
return {
content: [{ type: "text", text: "Interactive subagents are currently supported only in single mode." }],
details: makeDetails("chain")([]),
isError: true,
};
}
if (hasSingle) {
const selectedAgent = params.agent ? findAgentConfig(agents, params.agent) : undefined;
if (selectedAgent?.interactive === true && !ctx.hasUI) {
return {
content: [{ type: "text", text: "Interactive subagents require a UI-enabled session." }],
details: makeDetails("single")([]),
isError: true,
};
}
}
if ((agentScope === "project" || agentScope === "both") && confirmProjectAgents && ctx.hasUI) {
const requestedAgentNames = new Set<string>();
if (params.chain) for (const step of params.chain) requestedAgentNames.add(step.agent);
@ -521,7 +487,7 @@ export default function (pi: ExtensionAPI) {
exitCode: -1,
messages: [],
stderr: "",
usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 },
usage: createEmptyUsageStats(),
};
}
@ -571,17 +537,31 @@ export default function (pi: ExtensionAPI) {
}
if (params.agent && params.task) {
const result = await runSingleAgent(
ctx.cwd,
agents,
params.agent,
params.task,
params.cwd,
undefined,
signal,
onUpdate as OnUpdateCallback | undefined,
makeDetails("single"),
);
const selectedAgent = findAgentConfig(agents, params.agent);
const result = selectedAgent?.interactive
? await runInteractiveAgentRpc(
ctx.cwd,
agents,
params.agent,
params.task,
params.cwd,
undefined,
signal,
onUpdate as OnUpdateCallback | undefined,
makeDetails("single"),
ctx,
)
: await runSingleAgent(
ctx.cwd,
agents,
params.agent,
params.task,
params.cwd,
undefined,
signal,
onUpdate as OnUpdateCallback | undefined,
makeDetails("single"),
);
const isError = result.exitCode !== 0 || result.stopReason === "error" || result.stopReason === "aborted";
if (isError) {
const errorMsg = result.errorMessage || result.stderr || getFinalOutput(result.messages) || "(no output)";

View file

@ -0,0 +1,130 @@
import { StringEnum } from "@mariozechner/pi-ai";
import type { ExtensionAPI } from "@mariozechner/pi-coding-agent";
import { Type } from "@sinclair/typebox";
const AskUserParams = Type.Object({
mode: Type.Optional(
StringEnum(["input", "confirm", "select", "editor"] as const, {
description: "Question mode",
default: "input",
}),
),
question: Type.String({ description: "Question title to show the user" }),
message: Type.Optional(Type.String({ description: "Optional body text (confirm mode)" })),
placeholder: Type.Optional(Type.String({ description: "Placeholder text (input mode)" })),
options: Type.Optional(Type.Array(Type.String(), { description: "Options (select mode)" })),
prefill: Type.Optional(Type.String({ description: "Prefilled text (editor mode)" })),
});
function parseToolNames(raw: string | undefined): string[] {
if (!raw) return [];
const value = raw.trim();
if (!value) return [];
if (value.startsWith("[")) {
try {
const parsed = JSON.parse(value);
if (Array.isArray(parsed)) {
return parsed.map((entry) => String(entry).trim()).filter(Boolean);
}
} catch {
// ignore invalid JSON and fall back to comma-separated parsing
}
}
return value
.split(",")
.map((tool) => tool.trim())
.filter(Boolean);
}
export default function interactiveTools(pi: ExtensionAPI) {
pi.registerTool({
name: "ask_user",
label: "Ask User",
description:
"Ask the user an interactive clarification question. Supports input, confirm, select, and editor modes.",
parameters: AskUserParams,
promptGuidelines: [
"Use ask_user only when required information is missing or ambiguous.",
"Ask one focused question at a time and continue once answered.",
],
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
if (!ctx.hasUI) {
return {
content: [{ type: "text", text: "Error: ask_user requires an interactive UI." }],
isError: true,
details: { mode: params.mode ?? "input", cancelled: true },
};
}
const mode = params.mode ?? "input";
if (mode === "confirm") {
const confirmed = await ctx.ui.confirm(params.question, params.message ?? "Please confirm.");
return {
content: [{ type: "text", text: `User confirmation: ${confirmed ? "yes" : "no"}` }],
details: { mode, confirmed },
};
}
if (mode === "select") {
const options = (params.options ?? []).map((option) => option.trim()).filter(Boolean);
if (options.length === 0) {
return {
content: [{ type: "text", text: "Error: select mode requires a non-empty options array." }],
isError: true,
details: { mode, cancelled: true },
};
}
const value = await ctx.ui.select(params.question, options);
if (value === undefined) {
return {
content: [{ type: "text", text: "User cancelled the selection." }],
details: { mode, cancelled: true },
};
}
return {
content: [{ type: "text", text: `User selected: ${value}` }],
details: { mode, value, cancelled: false },
};
}
if (mode === "editor") {
const value = await ctx.ui.editor(params.question, params.prefill ?? "");
if (value === undefined) {
return {
content: [{ type: "text", text: "User cancelled editor input." }],
details: { mode, cancelled: true },
};
}
return {
content: [{ type: "text", text: `User input: ${value}` }],
details: { mode, value, cancelled: false },
};
}
const value = await ctx.ui.input(params.question, params.placeholder);
if (value === undefined) {
return {
content: [{ type: "text", text: "User cancelled input." }],
details: { mode: "input", cancelled: true },
};
}
return {
content: [{ type: "text", text: `User input: ${value}` }],
details: { mode: "input", value, cancelled: false },
};
},
});
pi.on("session_start", () => {
const requestedTools = parseToolNames(process.env.PI_SUBAGENT_TOOLS);
if (requestedTools.length === 0) return;
const available = new Set(pi.getAllTools().map((tool) => tool.name));
const validTools = requestedTools.filter((tool) => available.has(tool));
if (validTools.length > 0) {
pi.setActiveTools(validTools);
}
});
}

View file

@ -0,0 +1,40 @@
import { StringDecoder } from "node:string_decoder";
export function serializeJsonLine(value: unknown): string {
return `${JSON.stringify(value)}\n`;
}
export function attachJsonlLineReader(stream: NodeJS.ReadableStream, onLine: (line: string) => void): () => void {
const decoder = new StringDecoder("utf8");
let buffer = "";
const emitLine = (line: string) => {
onLine(line.endsWith("\r") ? line.slice(0, -1) : line);
};
const onData = (chunk: Buffer | string) => {
buffer += typeof chunk === "string" ? chunk : decoder.write(chunk);
while (true) {
const newlineIndex = buffer.indexOf("\n");
if (newlineIndex === -1) return;
emitLine(buffer.slice(0, newlineIndex));
buffer = buffer.slice(newlineIndex + 1);
}
};
const onEnd = () => {
buffer += decoder.end();
if (buffer.length > 0) {
emitLine(buffer);
buffer = "";
}
};
stream.on("data", onData);
stream.on("end", onEnd);
return () => {
stream.off("data", onData);
stream.off("end", onEnd);
};
}

View file

@ -0,0 +1,336 @@
import { spawn } from "node:child_process";
import * as fs from "node:fs";
import * as os from "node:os";
import * as path from "node:path";
import { fileURLToPath } from "node:url";
import type { ExtensionContext } from "@mariozechner/pi-coding-agent";
import type { AgentConfig } from "./agents.js";
import { attachJsonlLineReader, serializeJsonLine } from "./jsonl.js";
import {
applyAssistantUsage,
createEmptyUsageStats,
getFinalOutput,
type OnUpdateCallback,
type SingleResult,
type SubagentDetails,
} from "./subagent-types.js";
interface RpcResponse {
type: "response";
id?: string;
command: string;
success: boolean;
error?: string;
}
interface PendingRequest {
resolve: (response: RpcResponse) => void;
reject: (error: Error) => void;
timeout: NodeJS.Timeout;
}
type RpcExtensionUIRequest =
| { type: "extension_ui_request"; id: string; method: "select"; title: string; options: string[]; timeout?: number }
| { type: "extension_ui_request"; id: string; method: "confirm"; title: string; message: string; timeout?: number }
| { type: "extension_ui_request"; id: string; method: "input"; title: string; placeholder?: string; timeout?: number }
| { type: "extension_ui_request"; id: string; method: "editor"; title: string; prefill?: string }
| { type: "extension_ui_request"; id: string; method: "notify"; message: string; notifyType?: "info" | "warning" | "error" }
| { type: "extension_ui_request"; id: string; method: "setStatus"; statusKey: string; statusText?: string }
| {
type: "extension_ui_request";
id: string;
method: "setWidget";
widgetKey: string;
widgetLines?: string[];
widgetPlacement?: "aboveEditor" | "belowEditor";
}
| { type: "extension_ui_request"; id: string; method: "setTitle"; title: string }
| { type: "extension_ui_request"; id: string; method: "set_editor_text"; text: string };
function writePromptToTempFile(agentName: string, prompt: string): { dir: string; filePath: string } {
const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "pi-subagent-rpc-"));
const safeName = agentName.replace(/[^\w.-]+/g, "_");
const filePath = path.join(tmpDir, `prompt-${safeName}.md`);
fs.writeFileSync(filePath, prompt, { encoding: "utf-8", mode: 0o600 });
return { dir: tmpDir, filePath };
}
function terminateProcess(proc: ReturnType<typeof spawn>): void {
if (proc.exitCode !== null) return;
proc.kill("SIGTERM");
setTimeout(() => {
if (proc.exitCode === null) proc.kill("SIGKILL");
}, 3000);
}
function safeJsonParse(line: string): any | undefined {
if (!line.trim()) return undefined;
try {
return JSON.parse(line);
} catch {
return undefined;
}
}
async function bridgeExtensionUiRequest(
request: RpcExtensionUIRequest,
ctx: ExtensionContext,
signal: AbortSignal | undefined,
sendJson: (obj: unknown) => void,
): Promise<void> {
const sendCancelled = () => sendJson({ type: "extension_ui_response", id: request.id, cancelled: true });
try {
switch (request.method) {
case "select": {
const value = await ctx.ui.select(request.title, request.options, { signal, timeout: request.timeout });
if (value === undefined) sendCancelled();
else sendJson({ type: "extension_ui_response", id: request.id, value });
return;
}
case "confirm": {
const confirmed = await ctx.ui.confirm(request.title, request.message, { signal, timeout: request.timeout });
sendJson({ type: "extension_ui_response", id: request.id, confirmed });
return;
}
case "input": {
const value = await ctx.ui.input(request.title, request.placeholder, { signal, timeout: request.timeout });
if (value === undefined) sendCancelled();
else sendJson({ type: "extension_ui_response", id: request.id, value });
return;
}
case "editor": {
const value = await ctx.ui.editor(request.title, request.prefill ?? "");
if (value === undefined) sendCancelled();
else sendJson({ type: "extension_ui_response", id: request.id, value });
return;
}
case "notify":
ctx.ui.notify(request.message, request.notifyType);
return;
case "setStatus":
ctx.ui.setStatus(request.statusKey, request.statusText);
return;
case "setWidget":
if (request.widgetLines === undefined || Array.isArray(request.widgetLines)) {
ctx.ui.setWidget(request.widgetKey, request.widgetLines, { placement: request.widgetPlacement });
}
return;
case "setTitle":
ctx.ui.setTitle(request.title);
return;
case "set_editor_text":
ctx.ui.setEditorText(request.text);
return;
}
} catch {
if (request.method === "select" || request.method === "confirm" || request.method === "input" || request.method === "editor") {
sendCancelled();
}
}
}
export async function runInteractiveAgentRpc(
defaultCwd: string,
agents: AgentConfig[],
agentName: string,
task: string,
cwd: string | undefined,
step: number | undefined,
signal: AbortSignal | undefined,
onUpdate: OnUpdateCallback | undefined,
makeDetails: (results: SingleResult[]) => SubagentDetails,
ctx: ExtensionContext,
): Promise<SingleResult> {
const agent = agents.find((entry) => entry.name === agentName);
if (!agent) {
const available = agents.map((entry) => `"${entry.name}"`).join(", ") || "none";
return {
agent: agentName,
agentSource: "unknown",
task,
exitCode: 1,
messages: [],
stderr: `Unknown agent: "${agentName}". Available agents: ${available}.`,
usage: createEmptyUsageStats(),
step,
};
}
const args: string[] = ["--mode", "rpc", "--no-session"];
if (agent.model) args.push("--model", agent.model);
let tmpPromptDir: string | null = null;
let tmpPromptPath: string | null = null;
if (agent.systemPrompt.trim()) {
const tmp = writePromptToTempFile(agent.name, agent.systemPrompt);
tmpPromptDir = tmp.dir;
tmpPromptPath = tmp.filePath;
args.push("--append-system-prompt", tmpPromptPath);
}
const extensionDir = path.dirname(fileURLToPath(import.meta.url));
const helperExtensionPath = path.join(extensionDir, "interactive-tools.ts");
args.push("--extension", helperExtensionPath);
const env = { ...process.env };
if (agent.tools && agent.tools.length > 0) {
env.PI_SUBAGENT_TOOLS = JSON.stringify(agent.tools);
}
const currentResult: SingleResult = {
agent: agentName,
agentSource: agent.source,
task,
exitCode: 0,
messages: [],
stderr: "",
usage: createEmptyUsageStats(),
model: agent.model,
step,
};
const emitUpdate = () => {
if (!onUpdate) return;
onUpdate({
content: [{ type: "text", text: getFinalOutput(currentResult.messages) || "(running...)" }],
details: makeDetails([currentResult]),
});
};
const proc = spawn("pi", args, { cwd: cwd ?? defaultCwd, shell: false, stdio: ["pipe", "pipe", "pipe"], env });
const pendingRequests = new Map<string, PendingRequest>();
let requestCount = 0;
let agentEnded = false;
let resolveAgentEnd: (() => void) | undefined;
const agentEndPromise = new Promise<void>((resolve) => {
resolveAgentEnd = resolve;
});
const processExitPromise = new Promise<number>((resolve) => {
proc.on("close", (code) => {
if (!agentEnded) resolveAgentEnd?.();
for (const [id, pending] of pendingRequests.entries()) {
clearTimeout(pending.timeout);
pending.reject(new Error(`RPC process exited before response: ${id}`));
}
pendingRequests.clear();
resolve(code ?? 0);
});
});
let wasAborted = false;
const sendJson = (obj: unknown) => {
if (proc.stdin.destroyed || proc.stdin.writableEnded) return;
proc.stdin.write(serializeJsonLine(obj));
};
const sendCommand = (command: Record<string, unknown>): Promise<RpcResponse> => {
const id = `subagent_${++requestCount}`;
return new Promise<RpcResponse>((resolve, reject) => {
const timeout = setTimeout(() => {
pendingRequests.delete(id);
reject(new Error(`RPC command timed out: ${command.type}`));
}, 30000);
pendingRequests.set(id, { resolve, reject, timeout });
sendJson({ ...command, id });
});
};
const detachStdout = attachJsonlLineReader(proc.stdout, (line) => {
const event = safeJsonParse(line);
if (!event) {
currentResult.stderr += `[rpc] Ignored non-JSON output: ${line}\n`;
return;
}
if (event.type === "response" && event.id && pendingRequests.has(event.id)) {
const pending = pendingRequests.get(event.id)!;
pendingRequests.delete(event.id);
clearTimeout(pending.timeout);
pending.resolve(event as RpcResponse);
return;
}
if (event.type === "extension_ui_request") {
void bridgeExtensionUiRequest(event as RpcExtensionUIRequest, ctx, signal, sendJson);
return;
}
if (event.type === "message_end" && event.message) {
const msg = event.message;
currentResult.messages.push(msg);
applyAssistantUsage(currentResult, msg);
emitUpdate();
return;
}
if (event.type === "agent_end") {
agentEnded = true;
resolveAgentEnd?.();
return;
}
if (event.type === "extension_error") {
const extensionPath = typeof event.extensionPath === "string" ? event.extensionPath : "unknown";
const error = typeof event.error === "string" ? event.error : "unknown extension error";
currentResult.stderr += `[extension:${extensionPath}] ${error}\n`;
}
});
proc.stderr.on("data", (data) => {
currentResult.stderr += data.toString();
});
proc.on("error", (error) => {
currentResult.stderr += `${error.message}\n`;
});
const abortHandler = () => {
wasAborted = true;
try {
sendJson({ type: "abort" });
} catch {}
terminateProcess(proc);
};
if (signal) {
if (signal.aborted) abortHandler();
else signal.addEventListener("abort", abortHandler, { once: true });
}
try {
const promptResponse = await sendCommand({ type: "prompt", message: `Task: ${task}` });
if (!promptResponse.success) {
currentResult.exitCode = 1;
currentResult.stderr += `Prompt failed: ${promptResponse.error || "unknown error"}`;
return currentResult;
}
await agentEndPromise;
if (currentResult.stopReason === "error") currentResult.exitCode = 1;
if (wasAborted) throw new Error("Subagent was aborted");
return currentResult;
} finally {
detachStdout();
for (const [id, pending] of pendingRequests.entries()) {
clearTimeout(pending.timeout);
pending.reject(new Error(`RPC command interrupted: ${id}`));
}
pendingRequests.clear();
if (signal) signal.removeEventListener("abort", abortHandler);
terminateProcess(proc);
const processCode = await processExitPromise;
if (!wasAborted && currentResult.exitCode === 0 && processCode !== 0 && !agentEnded) {
currentResult.exitCode = processCode;
}
if (tmpPromptPath) {
try {
fs.unlinkSync(tmpPromptPath);
} catch {}
}
if (tmpPromptDir) {
try {
fs.rmdirSync(tmpPromptDir);
} catch {}
}
}
}

View file

@ -0,0 +1,90 @@
import type { Message } from "@mariozechner/pi-ai";
import type { AgentScope } from "./agents.js";
export interface UsageStats {
input: number;
output: number;
cacheRead: number;
cacheWrite: number;
cost: number;
contextTokens: number;
turns: number;
}
export interface SingleResult {
agent: string;
agentSource: "user" | "project" | "unknown";
task: string;
exitCode: number;
messages: Message[];
stderr: string;
usage: UsageStats;
model?: string;
stopReason?: string;
errorMessage?: string;
step?: number;
}
export interface SubagentDetails {
mode: "single" | "parallel" | "chain";
agentScope: AgentScope;
projectAgentsDir: string | null;
results: SingleResult[];
}
export type ToolContent = Array<{ type: "text"; text: string }>;
export type ToolUpdate = { content: ToolContent; details?: SubagentDetails; isError?: boolean };
export type OnUpdateCallback = (partial: ToolUpdate) => void;
export type DisplayItem = { type: "text"; text: string } | { type: "toolCall"; name: string; args: Record<string, any> };
export function createEmptyUsageStats(): UsageStats {
return {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
cost: 0,
contextTokens: 0,
turns: 0,
};
}
export function applyAssistantUsage(result: SingleResult, msg: Message): void {
if (msg.role !== "assistant") return;
result.usage.turns++;
const usage = msg.usage;
if (usage) {
result.usage.input += usage.input || 0;
result.usage.output += usage.output || 0;
result.usage.cacheRead += usage.cacheRead || 0;
result.usage.cacheWrite += usage.cacheWrite || 0;
result.usage.cost += usage.cost?.total || 0;
result.usage.contextTokens = usage.totalTokens || 0;
}
if (!result.model && msg.model) result.model = msg.model;
if (msg.stopReason) result.stopReason = msg.stopReason;
if (msg.errorMessage) result.errorMessage = msg.errorMessage;
}
export function getFinalOutput(messages: Message[]): string {
for (let i = messages.length - 1; i >= 0; i--) {
const msg = messages[i];
if (msg.role !== "assistant") continue;
for (const part of msg.content) {
if (part.type === "text") return part.text;
}
}
return "";
}
export function getDisplayItems(messages: Message[]): DisplayItem[] {
const items: DisplayItem[] = [];
for (const msg of messages) {
if (msg.role !== "assistant") continue;
for (const part of msg.content) {
if (part.type === "text") items.push({ type: "text", text: part.text });
else if (part.type === "toolCall") items.push({ type: "toolCall", name: part.name, args: part.arguments });
}
}
return items;
}