Add /subagent-runs live monitoring panel
Extract run execution from index.ts into runner.ts, add an in-memory RunRegistry shared between the subagent tool and a new /subagent-runs command, give each run its own AbortController for per-run cancellation, and add the runs-panel.ts monitor with a live-refresh timer. Also adds TypeScript type-checking (tsconfig.json + typecheck script) so the refactor has an automated safety net. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
EislM0203
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
1921bf1a82
commit
b8dbb25c17
+73
-367
@@ -10,31 +10,34 @@
|
||||
* - Chain: { chain: [{ agent: "name", task: "... {previous} ..." }, ...] }
|
||||
*
|
||||
* Uses JSON mode to capture structured output from subagents.
|
||||
*
|
||||
* Run execution lives in runner.ts; the live-run registry in runs.ts; the
|
||||
* /subagent-runs monitor panel in runs-panel.ts. This file wires those together
|
||||
* and owns the tool registration and result rendering.
|
||||
*/
|
||||
|
||||
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 { AgentToolResult } from "@earendil-works/pi-agent-core";
|
||||
import type { Message } from "@earendil-works/pi-ai";
|
||||
import { StringEnum } from "@earendil-works/pi-ai";
|
||||
import { type ExtensionAPI, getMarkdownTheme, withFileMutationQueue } from "@earendil-works/pi-coding-agent";
|
||||
import { type ExtensionAPI, getMarkdownTheme } from "@earendil-works/pi-coding-agent";
|
||||
import { Container, Markdown, Spacer, Text } from "@earendil-works/pi-tui";
|
||||
import { Type } from "typebox";
|
||||
import { type AgentConfig, type AgentScope, discoverAgents } from "./agents.js";
|
||||
import {
|
||||
formatTokens,
|
||||
getFinalOutput,
|
||||
MAX_PARALLEL_TASKS,
|
||||
runChain,
|
||||
runParallel,
|
||||
runSingle,
|
||||
type SingleResult,
|
||||
type SubagentDetails,
|
||||
} from "./runner.js";
|
||||
import { openRunsPanel } from "./runs-panel.js";
|
||||
import { createRunRegistry } from "./runs.js";
|
||||
|
||||
const MAX_PARALLEL_TASKS = 8;
|
||||
const MAX_CONCURRENCY = 4;
|
||||
const COLLAPSED_ITEM_COUNT = 10;
|
||||
|
||||
function formatTokens(count: number): string {
|
||||
if (count < 1000) return count.toString();
|
||||
if (count < 10000) return `${(count / 1000).toFixed(1)}k`;
|
||||
if (count < 1000000) return `${Math.round(count / 1000)}k`;
|
||||
return `${(count / 1000000).toFixed(1)}M`;
|
||||
}
|
||||
|
||||
function formatUsageStats(
|
||||
usage: {
|
||||
input: number;
|
||||
@@ -129,49 +132,6 @@ function formatToolCall(
|
||||
}
|
||||
}
|
||||
|
||||
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[];
|
||||
}
|
||||
|
||||
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 "";
|
||||
}
|
||||
|
||||
type DisplayItem = { type: "text"; text: string } | { type: "toolCall"; name: string; args: Record<string, any> };
|
||||
|
||||
function getDisplayItems(messages: Message[]): DisplayItem[] {
|
||||
@@ -187,218 +147,6 @@ function getDisplayItems(messages: Message[]): DisplayItem[] {
|
||||
return items;
|
||||
}
|
||||
|
||||
async function mapWithConcurrencyLimit<TIn, TOut>(
|
||||
items: TIn[],
|
||||
concurrency: number,
|
||||
fn: (item: TIn, index: number) => Promise<TOut>,
|
||||
): Promise<TOut[]> {
|
||||
if (items.length === 0) return [];
|
||||
const limit = Math.max(1, Math.min(concurrency, items.length));
|
||||
const results: TOut[] = new Array(items.length);
|
||||
let nextIndex = 0;
|
||||
const workers = new Array(limit).fill(null).map(async () => {
|
||||
while (true) {
|
||||
const current = nextIndex++;
|
||||
if (current >= items.length) return;
|
||||
results[current] = await fn(items[current], current);
|
||||
}
|
||||
});
|
||||
await Promise.all(workers);
|
||||
return results;
|
||||
}
|
||||
|
||||
async function writePromptToTempFile(agentName: string, prompt: string): Promise<{ dir: string; filePath: string }> {
|
||||
const tmpDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "pi-subagent-"));
|
||||
const safeName = agentName.replace(/[^\w.-]+/g, "_");
|
||||
const filePath = path.join(tmpDir, `prompt-${safeName}.md`);
|
||||
await withFileMutationQueue(filePath, async () => {
|
||||
await fs.promises.writeFile(filePath, prompt, { encoding: "utf-8", mode: 0o600 });
|
||||
});
|
||||
return { dir: tmpDir, filePath };
|
||||
}
|
||||
|
||||
function getPiInvocation(args: string[]): { command: string; args: string[] } {
|
||||
const currentScript = process.argv[1];
|
||||
const isBunVirtualScript = currentScript?.startsWith("/$bunfs/root/");
|
||||
if (currentScript && !isBunVirtualScript && fs.existsSync(currentScript)) {
|
||||
return { command: process.execPath, args: [currentScript, ...args] };
|
||||
}
|
||||
|
||||
const execName = path.basename(process.execPath).toLowerCase();
|
||||
const isGenericRuntime = /^(node|bun)(\.exe)?$/.test(execName);
|
||||
if (!isGenericRuntime) {
|
||||
return { command: process.execPath, args };
|
||||
}
|
||||
|
||||
return { command: "pi", args };
|
||||
}
|
||||
|
||||
type OnUpdateCallback = (partial: AgentToolResult<SubagentDetails>) => void;
|
||||
|
||||
async function runSingleAgent(
|
||||
defaultCwd: string,
|
||||
agents: AgentConfig[],
|
||||
agentName: string,
|
||||
task: string,
|
||||
cwd: string | undefined,
|
||||
step: number | undefined,
|
||||
signal: AbortSignal | undefined,
|
||||
onUpdate: OnUpdateCallback | undefined,
|
||||
makeDetails: (results: SingleResult[]) => SubagentDetails,
|
||||
): Promise<SingleResult> {
|
||||
const agent = agents.find((a) => a.name === agentName);
|
||||
|
||||
if (!agent) {
|
||||
const available = agents.map((a) => `"${a.name}"`).join(", ") || "none";
|
||||
return {
|
||||
agent: agentName,
|
||||
agentSource: "unknown",
|
||||
task,
|
||||
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 },
|
||||
step,
|
||||
};
|
||||
}
|
||||
|
||||
const args: string[] = ["--mode", "json", "-p", "--no-session"];
|
||||
if (agent.model) args.push("--model", agent.model);
|
||||
if (agent.tools && agent.tools.length > 0) args.push("--tools", agent.tools.join(","));
|
||||
|
||||
let tmpPromptDir: string | null = null;
|
||||
let tmpPromptPath: string | null = null;
|
||||
|
||||
const currentResult: SingleResult = {
|
||||
agent: agentName,
|
||||
agentSource: agent.source,
|
||||
task,
|
||||
exitCode: 0,
|
||||
messages: [],
|
||||
stderr: "",
|
||||
usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 },
|
||||
model: agent.model,
|
||||
step,
|
||||
};
|
||||
|
||||
const emitUpdate = () => {
|
||||
if (onUpdate) {
|
||||
onUpdate({
|
||||
content: [{ type: "text", text: getFinalOutput(currentResult.messages) || "(running...)" }],
|
||||
details: makeDetails([currentResult]),
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
try {
|
||||
if (agent.systemPrompt.trim()) {
|
||||
const tmp = await writePromptToTempFile(agent.name, agent.systemPrompt);
|
||||
tmpPromptDir = tmp.dir;
|
||||
tmpPromptPath = tmp.filePath;
|
||||
args.push("--append-system-prompt", tmpPromptPath);
|
||||
}
|
||||
|
||||
args.push(`Task: ${task}`);
|
||||
let wasAborted = false;
|
||||
|
||||
const exitCode = await new Promise<number>((resolve) => {
|
||||
const invocation = getPiInvocation(args);
|
||||
const proc = spawn(invocation.command, invocation.args, {
|
||||
cwd: cwd ?? defaultCwd,
|
||||
shell: false,
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
let buffer = "";
|
||||
|
||||
const processLine = (line: string) => {
|
||||
if (!line.trim()) return;
|
||||
let event: any;
|
||||
try {
|
||||
event = JSON.parse(line);
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.type === "message_end" && event.message) {
|
||||
const msg = event.message as 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;
|
||||
}
|
||||
emitUpdate();
|
||||
}
|
||||
|
||||
if (event.type === "tool_result_end" && event.message) {
|
||||
currentResult.messages.push(event.message as Message);
|
||||
emitUpdate();
|
||||
}
|
||||
};
|
||||
|
||||
proc.stdout.on("data", (data) => {
|
||||
buffer += data.toString();
|
||||
const lines = buffer.split("\n");
|
||||
buffer = lines.pop() || "";
|
||||
for (const line of lines) processLine(line);
|
||||
});
|
||||
|
||||
proc.stderr.on("data", (data) => {
|
||||
currentResult.stderr += data.toString();
|
||||
});
|
||||
|
||||
proc.on("close", (code) => {
|
||||
if (buffer.trim()) processLine(buffer);
|
||||
resolve(code ?? 0);
|
||||
});
|
||||
|
||||
proc.on("error", () => {
|
||||
resolve(1);
|
||||
});
|
||||
|
||||
if (signal) {
|
||||
const killProc = () => {
|
||||
wasAborted = true;
|
||||
proc.kill("SIGTERM");
|
||||
setTimeout(() => {
|
||||
if (!proc.killed) proc.kill("SIGKILL");
|
||||
}, 5000);
|
||||
};
|
||||
if (signal.aborted) killProc();
|
||||
else signal.addEventListener("abort", killProc, { once: true });
|
||||
}
|
||||
});
|
||||
|
||||
currentResult.exitCode = exitCode;
|
||||
if (wasAborted) throw new Error("Subagent was aborted");
|
||||
return currentResult;
|
||||
} finally {
|
||||
if (tmpPromptPath)
|
||||
try {
|
||||
fs.unlinkSync(tmpPromptPath);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
if (tmpPromptDir)
|
||||
try {
|
||||
fs.rmdirSync(tmpPromptDir);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const TaskItem = Type.Object({
|
||||
agent: Type.String({ description: "Name of the agent to invoke" }),
|
||||
task: Type.String({ description: "Task to delegate to the agent" }),
|
||||
@@ -429,6 +177,25 @@ const SubagentParams = Type.Object({
|
||||
});
|
||||
|
||||
export default function (pi: ExtensionAPI) {
|
||||
// Shared by closure between the subagent tool (records runs) and the
|
||||
// /subagent-runs command (displays them). Lives for the pi session.
|
||||
const registry = createRunRegistry();
|
||||
|
||||
pi.registerCommand("subagent-runs", {
|
||||
description: "Monitor subagent runs — live status, usage, and per-run cancel",
|
||||
handler: async (_args, ctx) => {
|
||||
if (!ctx.hasUI) {
|
||||
ctx.ui.notify("/subagent-runs needs an interactive session", "warning");
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await openRunsPanel(ctx, registry);
|
||||
} catch (err) {
|
||||
ctx.ui.notify(`/subagent-runs error: ${(err as Error).message}`, "error");
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
pi.registerTool({
|
||||
name: "subagent",
|
||||
label: "Subagent",
|
||||
@@ -499,56 +266,34 @@ export default function (pi: ExtensionAPI) {
|
||||
}
|
||||
|
||||
if (params.chain && params.chain.length > 0) {
|
||||
const results: SingleResult[] = [];
|
||||
let previousOutput = "";
|
||||
const results = await runChain({
|
||||
defaultCwd: ctx.cwd,
|
||||
agents,
|
||||
steps: params.chain,
|
||||
signal,
|
||||
onUpdate,
|
||||
makeDetails: makeDetails("chain"),
|
||||
tracker: registry,
|
||||
});
|
||||
|
||||
for (let i = 0; i < params.chain.length; i++) {
|
||||
const step = params.chain[i];
|
||||
const taskWithContext = step.task.replace(/\{previous\}/g, previousOutput);
|
||||
|
||||
// Create update callback that includes all previous results
|
||||
const chainUpdate: OnUpdateCallback | undefined = onUpdate
|
||||
? (partial) => {
|
||||
// Combine completed results with current streaming result
|
||||
const currentResult = partial.details?.results[0];
|
||||
if (currentResult) {
|
||||
const allResults = [...results, currentResult];
|
||||
onUpdate({
|
||||
content: partial.content,
|
||||
details: makeDetails("chain")(allResults),
|
||||
});
|
||||
}
|
||||
}
|
||||
: undefined;
|
||||
|
||||
const result = await runSingleAgent(
|
||||
ctx.cwd,
|
||||
agents,
|
||||
step.agent,
|
||||
taskWithContext,
|
||||
step.cwd,
|
||||
i + 1,
|
||||
signal,
|
||||
chainUpdate,
|
||||
makeDetails("chain"),
|
||||
);
|
||||
results.push(result);
|
||||
|
||||
const isError =
|
||||
result.exitCode !== 0 || result.stopReason === "error" || result.stopReason === "aborted";
|
||||
if (isError) {
|
||||
const errorMsg =
|
||||
result.errorMessage || result.stderr || getFinalOutput(result.messages) || "(no output)";
|
||||
return {
|
||||
content: [{ type: "text", text: `Chain stopped at step ${i + 1} (${step.agent}): ${errorMsg}` }],
|
||||
details: makeDetails("chain")(results),
|
||||
isError: true,
|
||||
};
|
||||
}
|
||||
previousOutput = getFinalOutput(result.messages);
|
||||
const last = results[results.length - 1];
|
||||
const lastIsError =
|
||||
last.exitCode !== 0 || last.stopReason === "error" || last.stopReason === "aborted";
|
||||
if (lastIsError) {
|
||||
const errorMsg =
|
||||
last.errorMessage || last.stderr || getFinalOutput(last.messages) || "(no output)";
|
||||
return {
|
||||
content: [
|
||||
{ type: "text", text: `Chain stopped at step ${last.step} (${last.agent}): ${errorMsg}` },
|
||||
],
|
||||
details: makeDetails("chain")(results),
|
||||
isError: true,
|
||||
};
|
||||
}
|
||||
return {
|
||||
content: [{ type: "text", text: getFinalOutput(results[results.length - 1].messages) || "(no output)" }],
|
||||
content: [
|
||||
{ type: "text", text: getFinalOutput(results[results.length - 1].messages) || "(no output)" },
|
||||
],
|
||||
details: makeDetails("chain")(results),
|
||||
};
|
||||
}
|
||||
@@ -565,56 +310,14 @@ export default function (pi: ExtensionAPI) {
|
||||
details: makeDetails("parallel")([]),
|
||||
};
|
||||
|
||||
// Track all results for streaming updates
|
||||
const allResults: SingleResult[] = new Array(params.tasks.length);
|
||||
|
||||
// Initialize placeholder results
|
||||
for (let i = 0; i < params.tasks.length; i++) {
|
||||
allResults[i] = {
|
||||
agent: params.tasks[i].agent,
|
||||
agentSource: "unknown",
|
||||
task: params.tasks[i].task,
|
||||
exitCode: -1, // -1 = still running
|
||||
messages: [],
|
||||
stderr: "",
|
||||
usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 },
|
||||
};
|
||||
}
|
||||
|
||||
const emitParallelUpdate = () => {
|
||||
if (onUpdate) {
|
||||
const running = allResults.filter((r) => r.exitCode === -1).length;
|
||||
const done = allResults.filter((r) => r.exitCode !== -1).length;
|
||||
onUpdate({
|
||||
content: [
|
||||
{ type: "text", text: `Parallel: ${done}/${allResults.length} done, ${running} running...` },
|
||||
],
|
||||
details: makeDetails("parallel")([...allResults]),
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
const results = await mapWithConcurrencyLimit(params.tasks, MAX_CONCURRENCY, async (t, index) => {
|
||||
const result = await runSingleAgent(
|
||||
ctx.cwd,
|
||||
agents,
|
||||
t.agent,
|
||||
t.task,
|
||||
t.cwd,
|
||||
undefined,
|
||||
signal,
|
||||
// Per-task update callback
|
||||
(partial) => {
|
||||
if (partial.details?.results[0]) {
|
||||
allResults[index] = partial.details.results[0];
|
||||
emitParallelUpdate();
|
||||
}
|
||||
},
|
||||
makeDetails("parallel"),
|
||||
);
|
||||
allResults[index] = result;
|
||||
emitParallelUpdate();
|
||||
return result;
|
||||
const results = await runParallel({
|
||||
defaultCwd: ctx.cwd,
|
||||
agents,
|
||||
tasks: params.tasks,
|
||||
signal,
|
||||
onUpdate,
|
||||
makeDetails: makeDetails("parallel"),
|
||||
tracker: registry,
|
||||
});
|
||||
|
||||
const successCount = results.filter((r) => r.exitCode === 0).length;
|
||||
@@ -635,7 +338,7 @@ export default function (pi: ExtensionAPI) {
|
||||
}
|
||||
|
||||
if (params.agent && params.task) {
|
||||
const result = await runSingleAgent(
|
||||
const result = await runSingle(
|
||||
ctx.cwd,
|
||||
agents,
|
||||
params.agent,
|
||||
@@ -645,6 +348,9 @@ export default function (pi: ExtensionAPI) {
|
||||
signal,
|
||||
onUpdate,
|
||||
makeDetails("single"),
|
||||
registry,
|
||||
"single",
|
||||
undefined,
|
||||
);
|
||||
const isError = result.exitCode !== 0 || result.stopReason === "error" || result.stopReason === "aborted";
|
||||
if (isError) {
|
||||
|
||||
@@ -0,0 +1,447 @@
|
||||
/**
|
||||
* Subagent run execution — spawns child `pi` processes and orchestrates
|
||||
* single / parallel / chain runs. Extracted from index.ts so the run registry
|
||||
* has a clean home and index.ts stays focused on tool I/O and rendering.
|
||||
*/
|
||||
|
||||
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 { AgentToolResult } from "@earendil-works/pi-agent-core";
|
||||
import type { Message } from "@earendil-works/pi-ai";
|
||||
import { withFileMutationQueue } from "@earendil-works/pi-coding-agent";
|
||||
import type { AgentConfig, AgentScope } from "./agents.js";
|
||||
import type { RunRegistry } from "./runs.js";
|
||||
|
||||
export const MAX_PARALLEL_TASKS = 8;
|
||||
export const MAX_CONCURRENCY = 4;
|
||||
|
||||
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 OnUpdateCallback = (partial: AgentToolResult<SubagentDetails>) => void;
|
||||
|
||||
/** Compact token count: 1234 -> "1.2k", 2_000_000 -> "2.0M". */
|
||||
export function formatTokens(count: number): string {
|
||||
if (count < 1000) return count.toString();
|
||||
if (count < 10000) return `${(count / 1000).toFixed(1)}k`;
|
||||
if (count < 1000000) return `${Math.round(count / 1000)}k`;
|
||||
return `${(count / 1000000).toFixed(1)}M`;
|
||||
}
|
||||
|
||||
export 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 "";
|
||||
}
|
||||
|
||||
async function mapWithConcurrencyLimit<TIn, TOut>(
|
||||
items: TIn[],
|
||||
concurrency: number,
|
||||
fn: (item: TIn, index: number) => Promise<TOut>,
|
||||
): Promise<TOut[]> {
|
||||
if (items.length === 0) return [];
|
||||
const limit = Math.max(1, Math.min(concurrency, items.length));
|
||||
const results: TOut[] = new Array(items.length);
|
||||
let nextIndex = 0;
|
||||
const workers = new Array(limit).fill(null).map(async () => {
|
||||
while (true) {
|
||||
const current = nextIndex++;
|
||||
if (current >= items.length) return;
|
||||
results[current] = await fn(items[current], current);
|
||||
}
|
||||
});
|
||||
await Promise.all(workers);
|
||||
return results;
|
||||
}
|
||||
|
||||
async function writePromptToTempFile(agentName: string, prompt: string): Promise<{ dir: string; filePath: string }> {
|
||||
const tmpDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "pi-subagent-"));
|
||||
const safeName = agentName.replace(/[^\w.-]+/g, "_");
|
||||
const filePath = path.join(tmpDir, `prompt-${safeName}.md`);
|
||||
await withFileMutationQueue(filePath, async () => {
|
||||
await fs.promises.writeFile(filePath, prompt, { encoding: "utf-8", mode: 0o600 });
|
||||
});
|
||||
return { dir: tmpDir, filePath };
|
||||
}
|
||||
|
||||
function getPiInvocation(args: string[]): { command: string; args: string[] } {
|
||||
const currentScript = process.argv[1];
|
||||
const isBunVirtualScript = currentScript?.startsWith("/$bunfs/root/");
|
||||
if (currentScript && !isBunVirtualScript && fs.existsSync(currentScript)) {
|
||||
return { command: process.execPath, args: [currentScript, ...args] };
|
||||
}
|
||||
|
||||
const execName = path.basename(process.execPath).toLowerCase();
|
||||
const isGenericRuntime = /^(node|bun)(\.exe)?$/.test(execName);
|
||||
if (!isGenericRuntime) {
|
||||
return { command: process.execPath, args };
|
||||
}
|
||||
|
||||
return { command: "pi", args };
|
||||
}
|
||||
|
||||
export async function runSingle(
|
||||
defaultCwd: string,
|
||||
agents: AgentConfig[],
|
||||
agentName: string,
|
||||
task: string,
|
||||
cwd: string | undefined,
|
||||
step: number | undefined,
|
||||
signal: AbortSignal | undefined,
|
||||
onUpdate: OnUpdateCallback | undefined,
|
||||
makeDetails: (results: SingleResult[]) => SubagentDetails,
|
||||
tracker: RunRegistry,
|
||||
mode: "single" | "parallel" | "chain",
|
||||
stepCount: number | undefined,
|
||||
): Promise<SingleResult> {
|
||||
const agent = agents.find((a) => a.name === agentName);
|
||||
|
||||
if (!agent) {
|
||||
const available = agents.map((a) => `"${a.name}"`).join(", ") || "none";
|
||||
return {
|
||||
agent: agentName,
|
||||
agentSource: "unknown",
|
||||
task,
|
||||
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 },
|
||||
step,
|
||||
};
|
||||
}
|
||||
|
||||
const args: string[] = ["--mode", "json", "-p", "--no-session"];
|
||||
if (agent.model) args.push("--model", agent.model);
|
||||
if (agent.tools && agent.tools.length > 0) args.push("--tools", agent.tools.join(","));
|
||||
|
||||
let tmpPromptDir: string | null = null;
|
||||
let tmpPromptPath: string | null = null;
|
||||
|
||||
const currentResult: SingleResult = {
|
||||
agent: agentName,
|
||||
agentSource: agent.source,
|
||||
task,
|
||||
exitCode: 0,
|
||||
messages: [],
|
||||
stderr: "",
|
||||
usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 },
|
||||
model: agent.model,
|
||||
step,
|
||||
};
|
||||
|
||||
// Per-run cancellation: this run can be killed either by the session-wide
|
||||
// `signal` (whole-tool abort) or by its own controller (panel "cancel").
|
||||
const runController = new AbortController();
|
||||
const record = tracker.register({
|
||||
agent: agentName,
|
||||
agentSource: agent.source,
|
||||
task,
|
||||
mode,
|
||||
step,
|
||||
stepCount,
|
||||
abort: () => runController.abort(),
|
||||
});
|
||||
|
||||
const emitUpdate = () => {
|
||||
tracker.update(record.id, {
|
||||
usage: currentResult.usage,
|
||||
outputSoFar: getFinalOutput(currentResult.messages),
|
||||
stderr: currentResult.stderr,
|
||||
});
|
||||
if (onUpdate) {
|
||||
onUpdate({
|
||||
content: [{ type: "text", text: getFinalOutput(currentResult.messages) || "(running...)" }],
|
||||
details: makeDetails([currentResult]),
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
try {
|
||||
if (agent.systemPrompt.trim()) {
|
||||
const tmp = await writePromptToTempFile(agent.name, agent.systemPrompt);
|
||||
tmpPromptDir = tmp.dir;
|
||||
tmpPromptPath = tmp.filePath;
|
||||
args.push("--append-system-prompt", tmpPromptPath);
|
||||
}
|
||||
|
||||
args.push(`Task: ${task}`);
|
||||
let wasAborted = false;
|
||||
|
||||
const exitCode = await new Promise<number>((resolve) => {
|
||||
const invocation = getPiInvocation(args);
|
||||
const proc = spawn(invocation.command, invocation.args, {
|
||||
cwd: cwd ?? defaultCwd,
|
||||
shell: false,
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
let buffer = "";
|
||||
|
||||
const processLine = (line: string) => {
|
||||
if (!line.trim()) return;
|
||||
let event: any;
|
||||
try {
|
||||
event = JSON.parse(line);
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.type === "message_end" && event.message) {
|
||||
const msg = event.message as 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;
|
||||
}
|
||||
emitUpdate();
|
||||
}
|
||||
|
||||
if (event.type === "tool_result_end" && event.message) {
|
||||
currentResult.messages.push(event.message as Message);
|
||||
emitUpdate();
|
||||
}
|
||||
};
|
||||
|
||||
proc.stdout.on("data", (data) => {
|
||||
buffer += data.toString();
|
||||
const lines = buffer.split("\n");
|
||||
buffer = lines.pop() || "";
|
||||
for (const line of lines) processLine(line);
|
||||
});
|
||||
|
||||
proc.stderr.on("data", (data) => {
|
||||
currentResult.stderr += data.toString();
|
||||
});
|
||||
|
||||
proc.on("close", (code) => {
|
||||
if (buffer.trim()) processLine(buffer);
|
||||
resolve(code ?? 0);
|
||||
});
|
||||
|
||||
proc.on("error", () => {
|
||||
resolve(1);
|
||||
});
|
||||
|
||||
const killProc = () => {
|
||||
wasAborted = true;
|
||||
proc.kill("SIGTERM");
|
||||
setTimeout(() => {
|
||||
if (!proc.killed) proc.kill("SIGKILL");
|
||||
}, 5000);
|
||||
};
|
||||
|
||||
if (signal) {
|
||||
if (signal.aborted) killProc();
|
||||
else signal.addEventListener("abort", killProc, { once: true });
|
||||
}
|
||||
if (runController.signal.aborted) killProc();
|
||||
else runController.signal.addEventListener("abort", killProc, { once: true });
|
||||
});
|
||||
|
||||
currentResult.exitCode = exitCode;
|
||||
tracker.update(record.id, {
|
||||
usage: currentResult.usage,
|
||||
outputSoFar: getFinalOutput(currentResult.messages),
|
||||
stderr: currentResult.stderr,
|
||||
});
|
||||
|
||||
if (wasAborted) {
|
||||
tracker.finalize(record.id, { status: "aborted", exitCode, errorMessage: "Subagent was aborted" });
|
||||
throw new Error("Subagent was aborted");
|
||||
}
|
||||
|
||||
const isError =
|
||||
exitCode !== 0 || currentResult.stopReason === "error" || currentResult.stopReason === "aborted";
|
||||
tracker.finalize(record.id, {
|
||||
status: isError ? "error" : "done",
|
||||
exitCode,
|
||||
errorMessage: currentResult.errorMessage,
|
||||
});
|
||||
return currentResult;
|
||||
} finally {
|
||||
if (tmpPromptPath)
|
||||
try {
|
||||
fs.unlinkSync(tmpPromptPath);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
if (tmpPromptDir)
|
||||
try {
|
||||
fs.rmdirSync(tmpPromptDir);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export interface RunChainOptions {
|
||||
defaultCwd: string;
|
||||
agents: AgentConfig[];
|
||||
steps: { agent: string; task: string; cwd?: string }[];
|
||||
signal: AbortSignal | undefined;
|
||||
onUpdate: OnUpdateCallback | undefined;
|
||||
makeDetails: (results: SingleResult[]) => SubagentDetails;
|
||||
tracker: RunRegistry;
|
||||
}
|
||||
|
||||
/**
|
||||
* Run chain steps sequentially, substituting `{previous}` with the prior
|
||||
* step's final output. Stops early on the first failing step. Returns the
|
||||
* results of every step that ran (the last one is the failure, if any).
|
||||
*/
|
||||
export async function runChain(opts: RunChainOptions): Promise<SingleResult[]> {
|
||||
const { defaultCwd, agents, steps, signal, onUpdate, makeDetails, tracker } = opts;
|
||||
const results: SingleResult[] = [];
|
||||
let previousOutput = "";
|
||||
|
||||
for (let i = 0; i < steps.length; i++) {
|
||||
const step = steps[i];
|
||||
const taskWithContext = step.task.replace(/\{previous\}/g, previousOutput);
|
||||
|
||||
const chainUpdate: OnUpdateCallback | undefined = onUpdate
|
||||
? (partial) => {
|
||||
const currentResult = partial.details?.results[0];
|
||||
if (currentResult) {
|
||||
const allResults = [...results, currentResult];
|
||||
onUpdate({
|
||||
content: partial.content,
|
||||
details: makeDetails(allResults),
|
||||
});
|
||||
}
|
||||
}
|
||||
: undefined;
|
||||
|
||||
const result = await runSingle(
|
||||
defaultCwd,
|
||||
agents,
|
||||
step.agent,
|
||||
taskWithContext,
|
||||
step.cwd,
|
||||
i + 1,
|
||||
signal,
|
||||
chainUpdate,
|
||||
makeDetails,
|
||||
tracker,
|
||||
"chain",
|
||||
steps.length,
|
||||
);
|
||||
results.push(result);
|
||||
|
||||
const isError =
|
||||
result.exitCode !== 0 || result.stopReason === "error" || result.stopReason === "aborted";
|
||||
if (isError) break;
|
||||
previousOutput = getFinalOutput(result.messages);
|
||||
}
|
||||
return results;
|
||||
}
|
||||
|
||||
export interface RunParallelOptions {
|
||||
defaultCwd: string;
|
||||
agents: AgentConfig[];
|
||||
tasks: { agent: string; task: string; cwd?: string }[];
|
||||
signal: AbortSignal | undefined;
|
||||
onUpdate: OnUpdateCallback | undefined;
|
||||
makeDetails: (results: SingleResult[]) => SubagentDetails;
|
||||
tracker: RunRegistry;
|
||||
}
|
||||
|
||||
/** Run tasks concurrently (capped at MAX_CONCURRENCY), reporting live progress. */
|
||||
export async function runParallel(opts: RunParallelOptions): Promise<SingleResult[]> {
|
||||
const { defaultCwd, agents, tasks, signal, onUpdate, makeDetails, tracker } = opts;
|
||||
|
||||
const allResults: SingleResult[] = new Array(tasks.length);
|
||||
for (let i = 0; i < tasks.length; i++) {
|
||||
allResults[i] = {
|
||||
agent: tasks[i].agent,
|
||||
agentSource: "unknown",
|
||||
task: tasks[i].task,
|
||||
exitCode: -1, // -1 = still running
|
||||
messages: [],
|
||||
stderr: "",
|
||||
usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 },
|
||||
};
|
||||
}
|
||||
|
||||
const emitParallelUpdate = () => {
|
||||
if (onUpdate) {
|
||||
const running = allResults.filter((r) => r.exitCode === -1).length;
|
||||
const done = allResults.filter((r) => r.exitCode !== -1).length;
|
||||
onUpdate({
|
||||
content: [{ type: "text", text: `Parallel: ${done}/${allResults.length} done, ${running} running...` }],
|
||||
details: makeDetails([...allResults]),
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
return mapWithConcurrencyLimit(tasks, MAX_CONCURRENCY, async (t, index) => {
|
||||
const result = await runSingle(
|
||||
defaultCwd,
|
||||
agents,
|
||||
t.agent,
|
||||
t.task,
|
||||
t.cwd,
|
||||
index + 1,
|
||||
signal,
|
||||
(partial) => {
|
||||
if (partial.details?.results[0]) {
|
||||
allResults[index] = partial.details.results[0];
|
||||
emitParallelUpdate();
|
||||
}
|
||||
},
|
||||
makeDetails,
|
||||
tracker,
|
||||
"parallel",
|
||||
tasks.length,
|
||||
);
|
||||
allResults[index] = result;
|
||||
emitParallelUpdate();
|
||||
return result;
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,225 @@
|
||||
/**
|
||||
* /subagent-runs — interactive panel listing subagent runs from the registry.
|
||||
* Refreshes live while open; supports per-run cancel and clearing finished
|
||||
* runs. Modeled on the /local-models panel.
|
||||
*/
|
||||
|
||||
import type { ExtensionCommandContext } from "@earendil-works/pi-coding-agent";
|
||||
import { Key, matchesKey, truncateToWidth, visibleWidth } from "@earendil-works/pi-tui";
|
||||
import { formatTokens } from "./runner.js";
|
||||
import type { RunRecord, RunRegistry, RunStatus } from "./runs.js";
|
||||
|
||||
const REFRESH_MS = 400;
|
||||
|
||||
function statusIcon(status: RunStatus): string {
|
||||
switch (status) {
|
||||
case "running":
|
||||
return "⟳";
|
||||
case "done":
|
||||
return "✓";
|
||||
case "error":
|
||||
return "✗";
|
||||
case "aborted":
|
||||
return "⊘";
|
||||
}
|
||||
}
|
||||
|
||||
function formatElapsed(record: RunRecord): string {
|
||||
const end = record.finishedAt ?? Date.now();
|
||||
const secs = Math.max(0, Math.floor((end - record.startedAt) / 1000));
|
||||
const m = Math.floor(secs / 60);
|
||||
const s = secs % 60;
|
||||
return `${m}:${s.toString().padStart(2, "0")}`;
|
||||
}
|
||||
|
||||
function formatUsage(record: RunRecord): string {
|
||||
const u = record.usage;
|
||||
const parts: string[] = [];
|
||||
if (u.input) parts.push(`↑${formatTokens(u.input)}`);
|
||||
if (u.output) parts.push(`↓${formatTokens(u.output)}`);
|
||||
if (u.cost) parts.push(`$${u.cost.toFixed(2)}`);
|
||||
return parts.join(" ");
|
||||
}
|
||||
|
||||
function modeLabel(record: RunRecord): string {
|
||||
if (record.step && record.stepCount) return `${record.mode} ${record.step}/${record.stepCount}`;
|
||||
if (record.step) return `${record.mode} ${record.step}`;
|
||||
return record.mode;
|
||||
}
|
||||
|
||||
function oneLine(s: string): string {
|
||||
return s.replace(/\s+/g, " ").trim();
|
||||
}
|
||||
|
||||
/** What the list component asks openRunsPanel to do after it closes. */
|
||||
type PanelAction = { kind: "close" } | { kind: "detail"; id: string };
|
||||
|
||||
/** Show the live run list. Resolves with the chosen PanelAction. */
|
||||
function showRunsList(ctx: ExtensionCommandContext, registry: RunRegistry): Promise<PanelAction> {
|
||||
return ctx.ui.custom<PanelAction>((tui, theme, _kb, done) => {
|
||||
let index = 0;
|
||||
const timer = setInterval(() => tui.requestRender(), REFRESH_MS);
|
||||
const cleanup = () => clearInterval(timer);
|
||||
|
||||
function handleInput(data: string): void {
|
||||
const runs = registry.list();
|
||||
if (matchesKey(data, Key.up)) {
|
||||
index = Math.max(0, index - 1);
|
||||
tui.requestRender();
|
||||
return;
|
||||
}
|
||||
if (matchesKey(data, Key.down)) {
|
||||
index = Math.min(Math.max(0, runs.length - 1), index + 1);
|
||||
tui.requestRender();
|
||||
return;
|
||||
}
|
||||
if (matchesKey(data, Key.escape)) {
|
||||
cleanup();
|
||||
done({ kind: "close" });
|
||||
return;
|
||||
}
|
||||
if (matchesKey(data, Key.enter)) {
|
||||
const run = runs[index];
|
||||
if (run) {
|
||||
cleanup();
|
||||
done({ kind: "detail", id: run.id });
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (data === "c") {
|
||||
const run = runs[index];
|
||||
if (!run) return;
|
||||
if (run.status === "running") {
|
||||
run.abort();
|
||||
ctx.ui.notify(`Cancelling "${run.agent}"...`, "info");
|
||||
} else {
|
||||
ctx.ui.notify("Run already finished", "warning");
|
||||
}
|
||||
tui.requestRender();
|
||||
return;
|
||||
}
|
||||
if (data === "x") {
|
||||
registry.clearFinished();
|
||||
index = 0;
|
||||
tui.requestRender();
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
function render(width: number): string[] {
|
||||
const runs = registry.list();
|
||||
if (index >= runs.length) index = Math.max(0, runs.length - 1);
|
||||
const lines: string[] = [];
|
||||
const add = (s: string) => lines.push(truncateToWidth(s, width));
|
||||
|
||||
add(theme.fg("accent", "─".repeat(width)));
|
||||
add(theme.fg("text", " Subagent Runs"));
|
||||
lines.push("");
|
||||
|
||||
if (runs.length === 0) {
|
||||
add(theme.fg("muted", " (no subagent runs yet)"));
|
||||
} else {
|
||||
for (let i = 0; i < runs.length; i++) {
|
||||
const r = runs[i];
|
||||
const line =
|
||||
`${statusIcon(r.status)} ${r.status.padEnd(8)} ${r.agent.padEnd(12)} ` +
|
||||
`${modeLabel(r).padEnd(14)} ${formatElapsed(r).padEnd(6)} ${formatUsage(r).padEnd(20)} ${oneLine(r.task)}`;
|
||||
if (i === index) add(theme.fg("accent", `> ${line}`));
|
||||
else add(` ${theme.fg("text", line)}`);
|
||||
}
|
||||
}
|
||||
|
||||
lines.push("");
|
||||
// Pack key hints onto as many lines as the width needs, so none
|
||||
// get truncated off on a narrow terminal.
|
||||
const hints = ["↑↓ navigate", "Enter detail", "c cancel", "x clear finished", "Esc close"];
|
||||
let hintLine = "";
|
||||
for (const part of hints) {
|
||||
const candidate = hintLine === "" ? part : `${hintLine} • ${part}`;
|
||||
if (hintLine !== "" && visibleWidth(` ${candidate}`) > width) {
|
||||
add(theme.fg("dim", ` ${hintLine}`));
|
||||
hintLine = part;
|
||||
} else {
|
||||
hintLine = candidate;
|
||||
}
|
||||
}
|
||||
if (hintLine !== "") add(theme.fg("dim", ` ${hintLine}`));
|
||||
add(theme.fg("accent", "─".repeat(width)));
|
||||
return lines;
|
||||
}
|
||||
|
||||
return { render, handleInput, invalidate: () => {}, dispose: cleanup };
|
||||
});
|
||||
}
|
||||
|
||||
/** Show a read-only detail view for one run. Resolves when the user goes back. */
|
||||
function showRunDetail(ctx: ExtensionCommandContext, registry: RunRegistry, id: string): Promise<void> {
|
||||
return ctx.ui.custom<void>((tui, theme, _kb, done) => {
|
||||
const timer = setInterval(() => tui.requestRender(), REFRESH_MS);
|
||||
const cleanup = () => clearInterval(timer);
|
||||
|
||||
function handleInput(data: string): void {
|
||||
if (matchesKey(data, Key.escape) || matchesKey(data, Key.enter)) {
|
||||
cleanup();
|
||||
done();
|
||||
}
|
||||
}
|
||||
|
||||
function render(width: number): string[] {
|
||||
const lines: string[] = [];
|
||||
const add = (s: string) => lines.push(truncateToWidth(s, width));
|
||||
add(theme.fg("accent", "─".repeat(width)));
|
||||
|
||||
const r = registry.get(id);
|
||||
if (!r) {
|
||||
add(theme.fg("muted", " (run no longer available)"));
|
||||
add(theme.fg("dim", " Esc back"));
|
||||
add(theme.fg("accent", "─".repeat(width)));
|
||||
return lines;
|
||||
}
|
||||
|
||||
add(theme.fg("text", ` ${statusIcon(r.status)} ${r.agent} (${r.agentSource}) — ${r.status}`));
|
||||
lines.push("");
|
||||
add(theme.fg("muted", " Mode: ") + theme.fg("text", modeLabel(r)));
|
||||
add(theme.fg("muted", " Elapsed: ") + theme.fg("text", formatElapsed(r)));
|
||||
const usage = formatUsage(r);
|
||||
if (usage) add(theme.fg("muted", " Usage: ") + theme.fg("text", usage));
|
||||
if (r.exitCode !== undefined) add(theme.fg("muted", " Exit code: ") + theme.fg("text", String(r.exitCode)));
|
||||
if (r.errorMessage) add(theme.fg("error", ` Error: ${r.errorMessage}`));
|
||||
|
||||
lines.push("");
|
||||
add(theme.fg("muted", " ─── Task ───"));
|
||||
for (const ln of r.task.split("\n")) add(` ${theme.fg("dim", ln)}`);
|
||||
|
||||
lines.push("");
|
||||
add(theme.fg("muted", " ─── Output so far ───"));
|
||||
const output = r.outputSoFar || "(no output yet)";
|
||||
for (const ln of output.split("\n")) add(` ${theme.fg("toolOutput", ln)}`);
|
||||
|
||||
if (r.stderr.trim()) {
|
||||
lines.push("");
|
||||
add(theme.fg("muted", " ─── stderr ───"));
|
||||
for (const ln of r.stderr.split("\n")) add(` ${theme.fg("error", ln)}`);
|
||||
}
|
||||
|
||||
lines.push("");
|
||||
add(theme.fg("dim", " Esc back"));
|
||||
add(theme.fg("accent", "─".repeat(width)));
|
||||
return lines;
|
||||
}
|
||||
|
||||
return { render, handleInput, invalidate: () => {}, dispose: cleanup };
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Open the /subagent-runs panel. Loops: show the list; if the user opens a
|
||||
* detail view, show it then return to the list — until the user presses Esc.
|
||||
*/
|
||||
export async function openRunsPanel(ctx: ExtensionCommandContext, registry: RunRegistry): Promise<void> {
|
||||
for (;;) {
|
||||
const action = await showRunsList(ctx, registry);
|
||||
if (action.kind === "close") break;
|
||||
if (action.kind === "detail") await showRunDetail(ctx, registry, action.id);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { createRunRegistry } from "./runs.js";
|
||||
|
||||
const baseInit = {
|
||||
agent: "scout",
|
||||
agentSource: "user" as const,
|
||||
task: "explore the auth module",
|
||||
mode: "single" as const,
|
||||
abort: () => {},
|
||||
};
|
||||
|
||||
describe("createRunRegistry", () => {
|
||||
it("register creates a running record with a unique id", () => {
|
||||
const registry = createRunRegistry();
|
||||
const a = registry.register(baseInit);
|
||||
const b = registry.register(baseInit);
|
||||
|
||||
expect(a.status).toBe("running");
|
||||
expect(a.startedAt).toBeGreaterThan(0);
|
||||
expect(a.id).not.toBe(b.id);
|
||||
expect(registry.get(a.id)).toBe(a);
|
||||
});
|
||||
|
||||
it("update merges live fields into an existing record", () => {
|
||||
const registry = createRunRegistry();
|
||||
const record = registry.register(baseInit);
|
||||
|
||||
registry.update(record.id, {
|
||||
usage: { input: 10, output: 5, cacheRead: 0, cacheWrite: 0, cost: 0.01, contextTokens: 15, turns: 1 },
|
||||
outputSoFar: "partial output",
|
||||
stderr: "a warning",
|
||||
});
|
||||
|
||||
expect(record.usage.input).toBe(10);
|
||||
expect(record.outputSoFar).toBe("partial output");
|
||||
expect(record.stderr).toBe("a warning");
|
||||
expect(record.status).toBe("running");
|
||||
});
|
||||
|
||||
it("update is a no-op for an unknown id", () => {
|
||||
const registry = createRunRegistry();
|
||||
expect(() => registry.update("nope", { outputSoFar: "x" })).not.toThrow();
|
||||
});
|
||||
|
||||
it("finalize sets terminal status, finishedAt, and exit info", () => {
|
||||
const registry = createRunRegistry();
|
||||
const record = registry.register(baseInit);
|
||||
|
||||
registry.finalize(record.id, { status: "error", exitCode: 1, errorMessage: "boom" });
|
||||
|
||||
expect(record.status).toBe("error");
|
||||
expect(record.exitCode).toBe(1);
|
||||
expect(record.errorMessage).toBe("boom");
|
||||
expect(record.finishedAt).toBeGreaterThanOrEqual(record.startedAt);
|
||||
});
|
||||
|
||||
it("list returns records newest-first", () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const registry = createRunRegistry();
|
||||
vi.setSystemTime(1000);
|
||||
const first = registry.register(baseInit);
|
||||
vi.setSystemTime(2000);
|
||||
const second = registry.register(baseInit);
|
||||
|
||||
expect(registry.list().map((r) => r.id)).toEqual([second.id, first.id]);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("get returns undefined for an unknown id", () => {
|
||||
const registry = createRunRegistry();
|
||||
expect(registry.get("missing")).toBeUndefined();
|
||||
});
|
||||
|
||||
it("clearFinished drops finished records but keeps running ones", () => {
|
||||
const registry = createRunRegistry();
|
||||
const running = registry.register(baseInit);
|
||||
const done = registry.register(baseInit);
|
||||
const aborted = registry.register(baseInit);
|
||||
registry.finalize(done.id, { status: "done", exitCode: 0 });
|
||||
registry.finalize(aborted.id, { status: "aborted", exitCode: 1 });
|
||||
|
||||
registry.clearFinished();
|
||||
|
||||
expect(registry.list().map((r) => r.id)).toEqual([running.id]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,129 @@
|
||||
/**
|
||||
* In-memory registry of subagent runs. Created once in the extension factory
|
||||
* and shared by closure between the `subagent` tool (which records runs) and
|
||||
* the `/subagent-runs` command (which displays them). Lives for the pi
|
||||
* session; nothing is persisted to disk.
|
||||
*/
|
||||
|
||||
import { randomUUID } from "node:crypto";
|
||||
import type { UsageStats } from "./runner.js";
|
||||
|
||||
export type RunStatus = "running" | "done" | "error" | "aborted";
|
||||
|
||||
export interface RunRecord {
|
||||
id: string;
|
||||
agent: string;
|
||||
agentSource: "user" | "project" | "unknown";
|
||||
task: string;
|
||||
mode: "single" | "parallel" | "chain";
|
||||
/** 1-based position within a parallel batch or chain; unset for single. */
|
||||
step?: number;
|
||||
/** Total runs in this parallel batch or chain; unset for single. */
|
||||
stepCount?: number;
|
||||
status: RunStatus;
|
||||
startedAt: number;
|
||||
finishedAt?: number;
|
||||
usage: UsageStats;
|
||||
outputSoFar: string;
|
||||
stderr: string;
|
||||
exitCode?: number;
|
||||
errorMessage?: string;
|
||||
/** Cancels just this run. Safe to call after the run has finished. */
|
||||
abort: () => void;
|
||||
}
|
||||
|
||||
/** Fields supplied when a run is first registered. */
|
||||
export interface RunInit {
|
||||
agent: string;
|
||||
agentSource: "user" | "project" | "unknown";
|
||||
task: string;
|
||||
mode: "single" | "parallel" | "chain";
|
||||
step?: number;
|
||||
stepCount?: number;
|
||||
abort: () => void;
|
||||
}
|
||||
|
||||
/** Live fields updated while a run is in flight. */
|
||||
export interface RunUpdate {
|
||||
usage?: UsageStats;
|
||||
outputSoFar?: string;
|
||||
stderr?: string;
|
||||
}
|
||||
|
||||
/** Terminal fields recorded when a run ends. */
|
||||
export interface RunFinal {
|
||||
status: Exclude<RunStatus, "running">;
|
||||
exitCode?: number;
|
||||
errorMessage?: string;
|
||||
}
|
||||
|
||||
export interface RunRegistry {
|
||||
/** Create a new record (status "running") and return it. */
|
||||
register(init: RunInit): RunRecord;
|
||||
/** Merge live fields into an existing record. No-op if the id is unknown. */
|
||||
update(id: string, patch: RunUpdate): void;
|
||||
/** Mark a record terminal and stamp finishedAt. No-op if the id is unknown. */
|
||||
finalize(id: string, final: RunFinal): void;
|
||||
/** All records, newest-first. */
|
||||
list(): RunRecord[];
|
||||
get(id: string): RunRecord | undefined;
|
||||
/** Drop every record whose status is not "running". */
|
||||
clearFinished(): void;
|
||||
}
|
||||
|
||||
export function createRunRegistry(): RunRegistry {
|
||||
const records = new Map<string, RunRecord>();
|
||||
|
||||
return {
|
||||
register(init) {
|
||||
const record: RunRecord = {
|
||||
id: randomUUID(),
|
||||
agent: init.agent,
|
||||
agentSource: init.agentSource,
|
||||
task: init.task,
|
||||
mode: init.mode,
|
||||
step: init.step,
|
||||
stepCount: init.stepCount,
|
||||
status: "running",
|
||||
startedAt: Date.now(),
|
||||
usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 },
|
||||
outputSoFar: "",
|
||||
stderr: "",
|
||||
abort: init.abort,
|
||||
};
|
||||
records.set(record.id, record);
|
||||
return record;
|
||||
},
|
||||
|
||||
update(id, patch) {
|
||||
const record = records.get(id);
|
||||
if (!record) return;
|
||||
if (patch.usage !== undefined) record.usage = patch.usage;
|
||||
if (patch.outputSoFar !== undefined) record.outputSoFar = patch.outputSoFar;
|
||||
if (patch.stderr !== undefined) record.stderr = patch.stderr;
|
||||
},
|
||||
|
||||
finalize(id, final) {
|
||||
const record = records.get(id);
|
||||
if (!record) return;
|
||||
record.status = final.status;
|
||||
record.finishedAt = Date.now();
|
||||
record.exitCode = final.exitCode;
|
||||
record.errorMessage = final.errorMessage;
|
||||
},
|
||||
|
||||
list() {
|
||||
return Array.from(records.values()).sort((a, b) => b.startedAt - a.startedAt);
|
||||
},
|
||||
|
||||
get(id) {
|
||||
return records.get(id);
|
||||
},
|
||||
|
||||
clearFinished() {
|
||||
for (const [id, record] of records) {
|
||||
if (record.status !== "running") records.delete(id);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
Generated
+15
@@ -8,6 +8,7 @@
|
||||
"name": "pi-customizations",
|
||||
"version": "0.1.0",
|
||||
"devDependencies": {
|
||||
"typescript": "^6.0.3",
|
||||
"vitest": "^3.2.4"
|
||||
},
|
||||
"peerDependencies": {
|
||||
@@ -4202,6 +4203,20 @@
|
||||
"license": "MIT",
|
||||
"peer": true
|
||||
},
|
||||
"node_modules/typescript": {
|
||||
"version": "6.0.3",
|
||||
"resolved": "https://registry.npmjs.org/typescript/-/typescript-6.0.3.tgz",
|
||||
"integrity": "sha512-y2TvuxSZPDyQakkFRPZHKFm+KKVqIisdg9/CZwm9ftvKXLP8NRWj38/ODjNbr43SsoXqNuAisEf1GdCxqWcdBw==",
|
||||
"dev": true,
|
||||
"license": "Apache-2.0",
|
||||
"bin": {
|
||||
"tsc": "bin/tsc",
|
||||
"tsserver": "bin/tsserver"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=14.17"
|
||||
}
|
||||
},
|
||||
"node_modules/uint8array-extras": {
|
||||
"version": "1.5.0",
|
||||
"resolved": "https://registry.npmjs.org/uint8array-extras/-/uint8array-extras-1.5.0.tgz",
|
||||
|
||||
+13
-5
@@ -3,22 +3,30 @@
|
||||
"version": "0.1.0",
|
||||
"description": "Personal pi coding agent customizations: subagent + local-models extensions, agents, prompt templates",
|
||||
"private": true,
|
||||
"keywords": ["pi-package"],
|
||||
"keywords": [
|
||||
"pi-package"
|
||||
],
|
||||
"pi": {
|
||||
"extensions": ["./extensions"],
|
||||
"prompts": ["./prompts"]
|
||||
"extensions": [
|
||||
"./extensions"
|
||||
],
|
||||
"prompts": [
|
||||
"./prompts"
|
||||
]
|
||||
},
|
||||
"scripts": {
|
||||
"test": "vitest --run"
|
||||
"test": "vitest --run",
|
||||
"typecheck": "tsc --noEmit"
|
||||
},
|
||||
"peerDependencies": {
|
||||
"@earendil-works/pi-ai": "*",
|
||||
"@earendil-works/pi-agent-core": "*",
|
||||
"@earendil-works/pi-ai": "*",
|
||||
"@earendil-works/pi-coding-agent": "*",
|
||||
"@earendil-works/pi-tui": "*",
|
||||
"typebox": "*"
|
||||
},
|
||||
"devDependencies": {
|
||||
"typescript": "^6.0.3",
|
||||
"vitest": "^3.2.4"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
{
|
||||
"compilerOptions": {
|
||||
"target": "ES2022",
|
||||
"module": "ESNext",
|
||||
"moduleResolution": "Bundler",
|
||||
"lib": ["ES2023"],
|
||||
"types": ["node"],
|
||||
"strict": true,
|
||||
"noEmit": true,
|
||||
"skipLibCheck": true,
|
||||
"esModuleInterop": true,
|
||||
"forceConsistentCasingInFileNames": true
|
||||
},
|
||||
"include": ["extensions/**/*.ts"]
|
||||
}
|
||||
Reference in New Issue
Block a user