Something went wrong. Try again.
Pi extension for async subagent delegation with truncation, artifacts, and session sharing
Something went wrong. Try again.
23 kB · 625 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626/** * General utility functions for the subagent extension */
import * as fs from "node:fs";import * as os from "node:os";import * as path from "node:path";import type { Message } from "@earendil-works/pi-ai";import { previewDisplayText, sanitizeDisplayText, truncateDisplayText } from "./display-text.ts";import { formatToolCall } from "./formatters.ts";import type { AgentProgress, AsyncStatus, Details, DisplayItem, ErrorInfo, NestedRunSummary, SingleResult, ToolCallSummary, Usage } from "./types.ts";
// ============================================================================// File System Utilities// ============================================================================
const DEFAULT_CONFIG_DIR_NAME = ".pi";const PI_CODING_AGENT_PACKAGE_NAME = "@earendil-works/pi-coding-agent";export const PI_CODING_AGENT_PACKAGE_ROOT_ENV = "PI_SUBAGENTS_PI_CODING_AGENT_PACKAGE_ROOT";export const PROMPT_REDACTED = "[prompt redacted]";
export function resolveWatchPath( watchPath: string, nativeRealpath: (filePath: string) => string = fs.realpathSync.native,): string { // libuv's Windows watcher cannot mix 8.3 registration paths with long event paths. try { return nativeRealpath(watchPath); } catch { return watchPath; }}
function validConfigDirName(value: unknown): string | undefined { return typeof value === "string" && value.trim() ? value : undefined;}
function readConfigDirNameFromPackageRoot(packageRoot: string | undefined): string | undefined { if (!packageRoot) return undefined; try { const pkg = JSON.parse(fs.readFileSync(path.join(packageRoot, "package.json"), "utf-8")) as { name?: unknown; piConfig?: { configDir?: unknown }; }; if (pkg.name !== PI_CODING_AGENT_PACKAGE_NAME) return undefined; return validConfigDirName(pkg.piConfig?.configDir); } catch { return undefined; }}
function resolveConfigDirNameFromPackageJson(entryPoint = process.argv[1], packageRoot = process.env[PI_CODING_AGENT_PACKAGE_ROOT_ENV]): string | undefined { const packageRootValue = readConfigDirNameFromPackageRoot(packageRoot); if (packageRootValue) return packageRootValue; if (!entryPoint) return undefined; try { let dir = path.dirname(fs.realpathSync(entryPoint)); while (dir !== path.dirname(dir)) { const value = readConfigDirNameFromPackageRoot(dir); if (value) return value; dir = path.dirname(dir); } } catch { // Package metadata lookup is best-effort; detached runners must not fail here. } return undefined;}
export function resolveConfigDirName(codingAgentModule?: unknown, entryPoint?: string, packageRoot?: string): string { const moduleValue = codingAgentModule && typeof codingAgentModule === "object" ? validConfigDirName((codingAgentModule as { CONFIG_DIR_NAME?: unknown }).CONFIG_DIR_NAME) : undefined; return moduleValue ?? resolveConfigDirNameFromPackageJson(entryPoint, packageRoot) ?? DEFAULT_CONFIG_DIR_NAME;}
let cachedConfigDirName: { entryPoint: string | undefined; packageRoot: string | undefined; value: string } | undefined;
export function getConfigDirName(): string { const entryPoint = process.argv[1]; const packageRoot = process.env[PI_CODING_AGENT_PACKAGE_ROOT_ENV]; if (cachedConfigDirName && cachedConfigDirName.entryPoint === entryPoint && cachedConfigDirName.packageRoot === packageRoot) { return cachedConfigDirName.value; } const value = resolveConfigDirName(undefined, entryPoint, packageRoot); cachedConfigDirName = { entryPoint, packageRoot, value }; return value;}
export function getProjectConfigDir(projectRoot: string): string { return path.join(projectRoot, getConfigDirName());}
export function getAgentDir(): string { const configured = process.env.PI_CODING_AGENT_DIR; if (configured === "~") return os.homedir(); if (configured?.startsWith("~/")) return path.join(os.homedir(), configured.slice(2)); return configured || path.join(os.homedir(), getConfigDirName(), "agent");}
const statusCache = new Map<string, { mtime: number; ctime: number; size: number; ino: number; status: AsyncStatus }>();
export function pruneStatusCacheForAsyncRoot(asyncDirRoot: string, runIds: Iterable<string>): number { const root = path.resolve(asyncDirRoot); const currentStatusPaths = new Set( Array.from(runIds, (runId) => path.resolve(root, runId, "status.json")), ); let removed = 0; for (const statusPath of statusCache.keys()) { const resolved = path.resolve(statusPath); const relative = path.relative(root, resolved); if (relative && !relative.startsWith("..") && !path.isAbsolute(relative) && !currentStatusPaths.has(resolved)) { statusCache.delete(statusPath); removed++; } } return removed;}
function getErrorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error);}
export function resolveChildCwd(baseCwd: string, childCwd: string | undefined): string { if (!childCwd) return baseCwd; return path.isAbsolute(childCwd) ? childCwd : path.resolve(baseCwd, childCwd);}
function isNotFoundError(error: unknown): boolean { return typeof error === "object" && error !== null && "code" in error && (error as NodeJS.ErrnoException).code === "ENOENT";}
/** * Read async job status from disk (with mtime-based caching) */export function readStatus(asyncDir: string): AsyncStatus | null { const statusPath = path.join(asyncDir, "status.json");
let stat: fs.Stats; try { stat = fs.statSync(statusPath); } catch (error) { if (isNotFoundError(error)) { statusCache.delete(statusPath); return null; } throw new Error(`Failed to inspect async status file '${statusPath}': ${getErrorMessage(error)}`, { cause: error instanceof Error ? error : undefined, }); }
const cached = statusCache.get(statusPath); if ( cached && cached.mtime === stat.mtimeMs && cached.ctime === stat.ctimeMs && cached.size === stat.size && cached.ino === stat.ino ) { return cached.status; }
let content: string; try { content = fs.readFileSync(statusPath, "utf-8"); } catch (error) { if (isNotFoundError(error)) { statusCache.delete(statusPath); return null; } throw new Error(`Failed to read async status file '${statusPath}': ${getErrorMessage(error)}`, { cause: error instanceof Error ? error : undefined, }); }
let status: AsyncStatus; try { status = JSON.parse(content) as AsyncStatus; } catch (error) { throw new Error(`Failed to parse async status file '${statusPath}': ${getErrorMessage(error)}`, { cause: error instanceof Error ? error : undefined, }); }
statusCache.set(statusPath, { mtime: stat.mtimeMs, ctime: stat.ctimeMs, size: stat.size, ino: stat.ino, status, }); return status;}
const outputTailCache = new Map<string, { mtime: number; size: number; lines: string[] }>();
/** * Get the last N lines from an output file (with mtime/size-based caching) */function getOutputTail(outputFile: string | undefined, maxLines: number = 3): string[] { if (!outputFile) return []; let fd: number | null = null; try { const stat = fs.statSync(outputFile); if (stat.size === 0) return [];
const cached = outputTailCache.get(outputFile); if (cached && cached.mtime === stat.mtimeMs && cached.size === stat.size) { return cached.lines; }
const tailBytes = 4096; const start = Math.max(0, stat.size - tailBytes); fd = fs.openSync(outputFile, "r"); const buffer = Buffer.alloc(Math.min(tailBytes, stat.size)); fs.readSync(fd, buffer, 0, buffer.length, start); const content = buffer.toString("utf-8"); const allLines = content.split("\n").filter((l) => l.trim()); const lines = allLines.slice(-maxLines).map((l) => l.slice(0, 120) + (l.length > 120 ? "..." : ""));
outputTailCache.set(outputFile, { mtime: stat.mtimeMs, size: stat.size, lines }); if (outputTailCache.size > 20) { const firstKey = outputTailCache.keys().next().value; if (firstKey) outputTailCache.delete(firstKey); }
return lines; } catch { // Output tails are UI-only hints; unreadable or missing files should render as no tail. return []; } finally { if (fd !== null) { try { fs.closeSync(fd); } catch { // Closing the best-effort tail file handle should not surface over the main status view. } } }}
/** * Get human-readable last activity time for a file */ export function getLastActivity(outputFile: string | undefined): string { if (!outputFile) return ""; try { const stat = fs.statSync(outputFile); const ago = Date.now() - stat.mtimeMs; if (ago < 1000) return "active now"; if (ago < 60000) return `active ${Math.floor(ago / 1000)}s ago`; return `active ${Math.floor(ago / 60000)}m ago`; } catch { // Last-activity text is best effort; missing files should simply omit the hint. return ""; }}
/** * Find the latest session file in a directory */export function findLatestSessionFile(sessionDir: string): string | null { if (!fs.existsSync(sessionDir)) return null; const files = fs.readdirSync(sessionDir) .filter((f) => f.endsWith(".jsonl")) .map((f) => { const filePath = path.join(sessionDir, f); return { path: filePath, mtime: fs.statSync(filePath).mtimeMs, }; }) .sort((a, b) => b.mtime - a.mtime); const latest = files[0]; return latest ? latest.path : null;}
/** * Write a prompt to a temporary file */function writePrompt(agent: string, prompt: string): { dir: string; path: string } { const dir = fs.mkdtempSync(path.join(os.tmpdir(), "pi-subagent-")); const p = path.join(dir, `${agent.replace(/[^\w.-]/g, "_")}.md`); fs.writeFileSync(p, prompt, { mode: 0o600 }); return { dir, path: p };}
// ============================================================================// Message Parsing Utilities// ============================================================================
/** * Get the final text output from a list of messages */export function getFinalOutput(messages: Message[]): string { const validTextParts: string[] = []; for (let i = messages.length - 1; i >= 0; i--) { const msg = messages[i]; if (!msg || msg.role !== "assistant") continue; const hasAssistantError = ("errorMessage" in msg && typeof msg.errorMessage === "string" && msg.errorMessage.length > 0) || ("stopReason" in msg && msg.stopReason === "error"); if (hasAssistantError) continue; const messageText = msg.content .filter((part) => part.type === "text" && part.text.trim().length > 0) .map((part) => part.type === "text" ? part.text : "") .join("\n"); for (let j = msg.content.length - 1; j >= 0; j--) { const part = msg.content[j]; if (!part || part.type !== "text" || part.text.trim().length === 0) continue; validTextParts.push(part.text); if (/```acceptance[-_]report\s*\n[\s\S]*?```/i.test(part.text)) return messageText; for (const match of part.text.matchAll(/```(?:json|jsonc|json5)\s*\n([\s\S]*?)```/gi)) { const body = match[1] ?? ""; if (/"(?:criteriaSatisfied|criteria_satisfied)"/.test(body) && /"(?:changedFiles|changed_files|testsAddedOrUpdated|tests_added_or_updated|commandsRun|commands_run|validationOutput|validation_output|residualRisks|residual_risks|noStagedFiles|no_staged_files|diffSummary|diff_summary|reviewFindings|review_findings|manualNotes|manual_notes)"/.test(body)) { return messageText; } } if (/ACCEPTANCE_REPORT\s*:/i.test(part.text)) return messageText; } } return validTextParts[0] ?? "";}
export function getSingleResultOutput(result: Pick<SingleResult, "finalOutput" | "messages">): string { return result.finalOutput ?? getFinalOutput(result.messages ?? []);}
/** * Extract display items (text and tool calls) from messages */export function getDisplayItems(messages: Message[] | undefined): DisplayItem[] { if (!messages || messages.length === 0) return []; 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: "tool", name: part.name, args: part.arguments }); } } } return items;}
function compactCompletedProgress(progress: AgentProgress): AgentProgress { if (progress.status === "running") return progress; return { index: progress.index, agent: progress.agent, status: progress.status, activityState: progress.activityState, task: "[prompt redacted]", skills: progress.skills, toolCount: progress.toolCount, tokens: progress.tokens, durationMs: progress.durationMs, error: progress.error, failedTool: progress.failedTool, recentTools: [], recentOutput: [], };}
function extractToolCallSummaries(messages: Message[] | undefined): ToolCallSummary[] { if (!messages?.length) return []; const summaries: ToolCallSummary[] = []; for (const msg of messages) { if (msg.role !== "assistant") continue; for (const part of msg.content) { if (part.type !== "toolCall") continue; const args = typeof part.arguments === "object" && part.arguments !== null && !Array.isArray(part.arguments) ? part.arguments : {}; summaries.push({ text: formatToolCall(part.name, args), expandedText: formatToolCall(part.name, args, true), }); } } return summaries;}
export function sumResultsUsage(results: SingleResult[]): Usage { const usage: Usage = { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, turns: 0 }; for (const result of results) { usage.input += result.usage.input; usage.output += result.usage.output; usage.cacheRead += result.usage.cacheRead; usage.cacheWrite += result.usage.cacheWrite; usage.cost += result.usage.cost; usage.turns += result.usage.turns; } return usage;}
function addNestedCost(total: NonNullable<Details["totalCost"]>, children: NestedRunSummary[] | undefined): void { for (const child of children ?? []) { if (child.totalCost) { total.inputTokens += child.totalCost.inputTokens; total.outputTokens += child.totalCost.outputTokens; total.costUsd += child.totalCost.costUsd; continue; } addNestedCost(total, child.children); for (const step of child.steps ?? []) addNestedCost(total, step.children); }}
/** Sum input tokens, output tokens, and cost across a set of SingleResults. */export function sumResultsCost(results: SingleResult[]): NonNullable<Details["totalCost"]> { const total = { inputTokens: 0, outputTokens: 0, costUsd: 0 }; for (const result of results) { total.inputTokens += result.usage.input; total.outputTokens += result.usage.output; total.costUsd += result.usage.cost; addNestedCost(total, result.children); } return total;}
export function compactForegroundResult(result: SingleResult): SingleResult { if (result.progress?.status === "running") return result; const toolCalls = result.toolCalls?.length ? result.toolCalls : extractToolCallSummaries(result.messages); return { ...result, task: "[prompt redacted]", messages: undefined, progress: undefined, toolCalls: toolCalls.length ? toolCalls : undefined, };}
export function compactForegroundDetails(details: Details): Details { return { ...details, results: details.results.map(compactForegroundResult), progress: details.progress ? details.progress.map(compactCompletedProgress) : undefined, };}
/** * Streaming counterparts to compactForegroundResult / compactCompletedProgress. * * The completed-compaction helpers above bail out while a child is still * `running`, so a long or deeply nested fan-out streams full, unbounded progress on * every tick. Pi serializes each streamed `tool_execution_update` as a single * child-stdout line, which the parent reads under `MAX_CHILD_PENDING_LINE_BYTES`; * an unbounded running snapshot can cross that cap and kill the child with * `protocol_output_limit`. * * These bound the STREAMED snapshot only. The final returned result keeps the full * live progress and message transcript, and every live-display consumer already * reads just the last few entries (`recentTools.slice(-3)`, `recentOutput.slice(-5)`). */export const MAX_STREAMED_RECENT_TOOLS = 32;export const MAX_STREAMED_TOOL_CALLS = 64;export const MAX_STREAMED_OUTPUT_LINE_CHARS = 2000;
/** Keep only the most recent tool-history entries in a streamed snapshot. */export function boundStreamedRecentTools(recentTools: AgentProgress["recentTools"]): AgentProgress["recentTools"] { return recentTools.slice(-MAX_STREAMED_RECENT_TOOLS).map((tool) => ({ ...tool }));}
/** Cap per-line length of recent output so one long line can't inflate a snapshot. */export function boundStreamedRecentOutput(recentOutput: string[]): string[] { return recentOutput.map((line) => line.length > MAX_STREAMED_OUTPUT_LINE_CHARS ? `${line.slice(0, MAX_STREAMED_OUTPUT_LINE_CHARS)}… [truncated]` : line, );}
/** * Compact tool-call summaries for a streamed snapshot, standing in for the * unbounded `messages` transcript. Prefers an existing `toolCalls` summary, else * derives one from `messages`; bounded to the most recent calls. */export function boundStreamedToolCalls(result: Pick<SingleResult, "toolCalls" | "messages">): ToolCallSummary[] | undefined { const summaries = result.toolCalls?.length ? result.toolCalls : extractToolCallSummaries(result.messages); if (!summaries.length) return undefined; return summaries.slice(-MAX_STREAMED_TOOL_CALLS).map((summary) => ({ ...summary }));}
export function hasEmptyTerminalAssistantResponse(messages: Message[]): boolean { const lastAssistant = messages.findLast((message) => message.role === "assistant"); return lastAssistant?.role === "assistant" && Array.isArray(lastAssistant.content) && lastAssistant.content.length === 0 && lastAssistant.usage.output === 0;}
/** * Detect errors in subagent execution from messages (only errors with no subsequent success) */export function detectSubagentError(messages: Message[]): ErrorInfo { let lastAssistantTextIndex = -1; for (let i = messages.length - 1; i >= 0; i--) { const msg = messages[i]; if (msg?.role === "assistant") { const hasText = Array.isArray(msg.content) && msg.content.some( (c) => c.type === "text" && "text" in c && typeof c.text === "string" && c.text.trim().length > 0, ); if (hasText) { lastAssistantTextIndex = i; break; } } }
const scanStart = lastAssistantTextIndex >= 0 ? lastAssistantTextIndex + 1 : 0;
for (let i = messages.length - 1; i >= scanStart; i--) { const msg = messages[i]; if (!msg || msg.role !== "toolResult") continue; const toolName = "toolName" in msg && typeof msg.toolName === "string" ? msg.toolName : undefined; const isError = "isError" in msg && msg.isError === true;
if (!isError) continue;
const text = msg.content.find((c) => c.type === "text"); const details = text && "text" in text ? text.text : undefined; const exitMatch = details?.match(/exit(?:ed)?\s*(?:with\s*)?(?:code|status)?\s*[:\s]?\s*(\d+)/i); const exitCodeText = exitMatch?.[1]; return { hasError: true, exitCode: exitCodeText ? parseInt(exitCodeText, 10) : 1, errorType: toolName || "tool", details: details?.slice(0, 200), }; }
return { hasError: false };}
/** * Extract a preview of tool arguments for display */export function extractToolArgsPreview(args: Record<string, unknown>): string { const stringifyPreviewValue = (value: unknown): string | undefined => { if (typeof value === "string" && value.trim().length > 0) return sanitizeDisplayText(value); if (typeof value === "number" || typeof value === "boolean") return String(value); return undefined; };
const previewArray = (value: unknown): string | undefined => { if (!Array.isArray(value) || value.length === 0) return undefined; const first = stringifyPreviewValue(value[0]); if (!first) return undefined; const suffix = value.length > 1 ? ` (+${value.length - 1} more)` : ""; return `${first}${suffix}`; };
if (typeof args.tool === "string") { const server = typeof args.server === "string" ? `${sanitizeDisplayText(args.server)}/` : ""; const toolArgs = typeof args.args === "string" ? ` ${truncateDisplayText(sanitizeDisplayText(args.args), 40)}` : ""; return sanitizeDisplayText(`${server}${args.tool}${toolArgs}`); }
const queriesPreview = previewArray(args.queries); if (queriesPreview) return previewDisplayText(queriesPreview, 60); if (typeof args.query === "string" && args.query.trim().length > 0) return previewDisplayText(args.query, 60); if (typeof args.workflow === "string" && args.workflow.trim().length > 0) return `workflow=${previewDisplayText(args.workflow, 48)}`;
if (typeof args.url === "string" && args.url.trim().length > 0) return previewDisplayText(args.url, 60); const urlsPreview = previewArray(args.urls); if (urlsPreview) return previewDisplayText(urlsPreview, 60); if (typeof args.prompt === "string" && args.prompt.trim().length > 0) return previewDisplayText(args.prompt, 60);
const previewKeys = ["command", "path", "file_path", "pattern", "query", "url", "task", "describe", "search"]; for (const key of previewKeys) { if (typeof args[key] === "string") return previewDisplayText(args[key], 60); }
for (const [key, value] of Object.entries(args)) { const displayKey = sanitizeDisplayText(key); const arrayPreview = previewArray(value); if (arrayPreview) return `${displayKey}=${previewDisplayText(arrayPreview, 50)}`; if (typeof value === "string" && value.length > 0) return `${displayKey}=${previewDisplayText(value, 50)}`; } return "";}
/** * Extract text content from various message content formats */export function extractTextFromContent(content: unknown): string { if (!content) return ""; // Handle string content directly if (typeof content === "string") return content; // Handle array content if (!Array.isArray(content)) return ""; const texts: string[] = []; for (const part of content) { if (part && typeof part === "object") { // Handle { type: "text", text: "..." } if ("type" in part && part.type === "text" && "text" in part) { texts.push(String(part.text)); } // Handle { type: "tool_result", content: "..." } else if ("type" in part && part.type === "tool_result" && "content" in part) { const inner = extractTextFromContent(part.content); if (inner) texts.push(inner); } // Handle { text: "..." } without type else if ("text" in part) { texts.push(String(part.text)); } } } return texts.join("\n");}
// ============================================================================// Concurrency Utilities// ============================================================================
export { mapConcurrent } from "../runs/shared/parallel-utils.ts";