/** * @pi-unipi/subagents — Extension entry * * Tools: spawn_helper, get_helper_result * Features: renderCall/renderResult, message renderer, conversation viewer * ESC propagation: all children abort on parent ESC */ import { defineTool, type ExtensionAPI } from "@earendil-works/pi-coding-agent"; import { Text } from "@earendil-works/pi-tui"; import { Type } from "typebox"; import { existsSync, readdirSync, writeFileSync } from "node:fs"; import { join } from "node:path"; import { homedir } from "node:os"; import { MODULES, UNIPI_EVENTS, emitEvent, type UnipiBadgeGenerateRequestEvent } from "@pi-unipi/core"; import { boundHelperOutput, withHerdrBlocked } from "./core-compat.js"; import { AgentManager } from "./agent-manager.js"; import { initConfig, loadRawGlobalConfig, loadRawWorkspaceConfig } from "./config.js"; import { type AgentActivity, type NotificationDetails, BUILTIN_TYPES } from "./types.js"; import { loadBuiltinFileAgents } from "./custom-agents.js"; import { ConversationViewer } from "./conversation-viewer.js"; import { AgentWidget, SPINNER, TOOL_DISPLAY, formatMs, formatTurns, describeActivity } from "./widget.js"; import { handleSpawnHelper, type HandlerDeps } from "./tool-handler.js"; import { SpawnHelperParams, GetHelperResultParams } from "./schemas.js"; import { runAsyncSubagent, createAsyncRunDir, writeStatus, readStatus } from "./async-runner.js"; import { createResultWatcher, cleanupAsyncRetention } from "./result-watcher.js"; import { writeAsyncResultFile, readAsyncResultFile } from "./result-files.js"; import { writePendingSubscription } from "./result-watcher.js"; import { RESULTS_DIR, ASYNC_DIR, ensureDirs } from "./parity-types.js"; import { createForkContextResolver } from "./fork-context.js"; import { createWorktrees, cleanupWorktrees, diffWorktrees, type WorktreeSetup } from "./worktree.js"; import { FleetView } from "./fleet-view.js"; import { registerSlashCommands } from "./slash-commands.js"; import { parseDetachShortcut, formatDetachHint, matchesDetachInput, } from "./foreground-detach.js"; import { coerceThinkingLevel } from "./agent-runner.js"; import { childExtensionsArg } from "./pi-args.js"; /** Get info registry from global */ function getInfoRegistry() { return globalThis.__unipi_info_registry; } // ---- Formatting helpers (shared between renderers and inline text) ---- /** Tool name → human-readable action. */ function formatTokens(count: number): string { if (count >= 1_000_000) return `${(count / 1_000_000).toFixed(1)}M token`; if (count >= 1_000) return `${(count / 1_000).toFixed(1)}k token`; return `${count} token`; } /** Format tokens safely from session. */ function safeFormatTokens(session: any): string { if (!session) return ""; try { const stats = session.getSessionStats(); const total = stats.tokens?.total ?? 0; return formatTokens(total); } catch { return ""; } } /** Get raw token count from session. */ function safeTokenCount(session: any): number { if (!session) return 0; try { return session.getSessionStats().tokens?.total ?? 0; } catch { return 0; } } /** Build result text */ function textResult(msg: string, details?: any) { return { content: [{ type: "text" as const, text: msg }], details }; } /** Escape XML for structured notifications. */ function escapeXml(s: string): string { return s.replace(/&/g, "&").replace(//g, ">"); } /** Human-readable status label. */ function getStatusLabel(status: string, error?: string): string { switch (status) { case "error": return `Error: ${error ?? "unknown"}`; case "aborted": return "Aborted (max turns exceeded)"; case "stopped": return "Stopped"; default: return "Done"; } } export default function (pi: ExtensionAPI) { // Initialize config const config = initConfig(process.cwd()); if (!config.enabled) return; // Compute paths at factory time const homeDir = homedir(); const cwd = process.cwd(); const globalAgentsDir = join(homeDir, ".unipi", "config", "agents"); const workspaceAgentsDir = join(cwd, ".unipi", "config", "agents"); // Activity tracking for widget const agentActivity = new Map(); /** * Set once `session_shutdown` fires — after which `pi` must not be touched. * * Pi disposes the session as soon as the shutdown handlers resolve, and * `AgentSession.dispose()` invalidates the extension runtime. Every * `assertActive`-gated method (`sendMessage`, `setSessionName`, * `appendEntry`, `setModel`, …) then throws "This extension ctx is stale * after session replacement or reload". * * Background agents outlive that moment: `abortAll()` only signals their * AbortController, so the in-flight promise settles a microtask *later* and * fires this completion callback against a dead runtime — an unhandled * throw that crashed the process on exit. * * Scoped to the extension factory rather than module scope on purpose: * `session_shutdown` also fires for `/new`, `/fork` and `/resume` (reasons * "new" / "fork" / "resume"), and pi re-invokes the extension factory for * the replacement session. A fresh closure therefore starts with * `sessionEnded = false`, so the guard can never latch permanently. * Verified: `/new` emits `shutdown reason=new` then re-runs the factory. * * `pi.events` is NOT gated, so cross-module events still fire. */ let sessionEnded = false; // Create manager with completion callback const manager = new AgentManager( (record) => { agentActivity.delete(record.id); // After shutdown the UI is gone and the runtime is stale — nothing here // is deliverable, and touching `pi` would throw. if (sessionEnded) return; widget.markFinished(record.id); widget.update(); // Build notification details const details = buildNotificationDetails(record, agentActivity.get(record.id)); // Badge generation: extract name from agent result and set directly. // Mark resultConsumed BEFORE the notification check so the main agent // never sees this subagent. if (record.description === "Generate session name" && record.result && record.status === "completed") { const name = record.result.split("\n")[0]?.trim().slice(0, 50) ?? ""; if (name && !name.startsWith("Error") && !name.includes("error")) { try { pi.setSessionName(name); } catch { /* best effort */ } } record.resultConsumed = true; } // Send styled notification via message renderer const status = getStatusLabel(record.status, record.error); const durationMs = record.completedAt ? record.completedAt - record.startedAt : 0; const resultPreview = record.result ? record.result.length > 500 ? record.result.slice(0, 500) + "…" : record.result : "No output."; const notificationXml = [ ``, `${record.id}`, `${escapeXml(status)}`, `Agent "${escapeXml(record.description)}" ${record.status}`, `${escapeXml(resultPreview)}`, `${details.totalTokens}${record.toolUses}${durationMs}`, ``, ].join("\n"); if (!record.resultConsumed) { // Defence in depth: `sessionEnded` covers the ordinary shutdown path, // but a session can also be replaced mid-flight. Delivering a // notification is best-effort — it must never take the process down. try { pi.sendMessage( { customType: "subagent-notification", content: notificationXml, display: true, details, }, { deliverAs: "followUp", triggerTurn: true }, ); } catch { // Runtime went stale between the guard and here — nothing to notify. } } pi.events.emit("subagents:completed", { id: record.id, type: record.type, description: record.description, status: record.status, result: record.result, error: record.error, }); }, config.maxConcurrent, (record) => { pi.events.emit("subagents:started", { id: record.id, type: record.type, description: record.description, }); }, config.types, process.cwd(), { user: loadRawGlobalConfig()?.subagents, project: loadRawWorkspaceConfig(process.cwd())?.subagents, }, ); // ---- Async process runner (Phase 3) ---- ensureDirs(); const activeAsyncRuns = new Map(); const asyncSessionId = `unipi-${process.pid}`; const runAsyncDep: NonNullable = async (launch) => { const agent = manager.getAgentConfig(manager.resolveAlias(launch.agentName)); if (!agent) throw new Error(`Unknown agent "${launch.agentName}".`); if (launch.resumeSessionFile && !existsSync(launch.resumeSessionFile)) { throw new Error(`Resume session file is missing: ${launch.resumeSessionFile}`); } // Worktree isolation: managed worktree per child; cleaned up after the run // (diffs preserved to the run dir as handoff artifacts). let worktreeSetup: WorktreeSetup | undefined; let childCwd = process.cwd(); if (launch.worktree === true) { try { worktreeSetup = createWorktrees(process.cwd(), `async-${Date.now().toString(36)}`, 1); childCwd = worktreeSetup.worktrees[0]!.agentCwd; } catch (worktreeError) { throw new Error( `Worktree isolation failed: ${worktreeError instanceof Error ? worktreeError.message : String(worktreeError)}`, ); } } const runDir = createAsyncRunDir(launch.agentName); const runId = runDir.split("/").pop()!; const controller = new AbortController(); activeAsyncRuns.set(runId, controller); // Fork context: branch a child session from the parent conversation // (sanitized thinking blocks, thinking forced off when sanitized). let forkSessionFile: string | undefined; let forceThinkingOff: boolean | undefined; if (launch.context === "fork") { try { const sessionManager = (sessionCtx as unknown as { sessionManager?: { getSessionFile(): string | undefined; getLeafId(): string | null; openSession?: Parameters[0]["openSession"]; }; })?.sessionManager; if (!sessionManager) { throw new Error("Forked context requires a persisted parent session (session manager unavailable)."); } const resolver = createForkContextResolver(sessionManager, "fork"); forkSessionFile = resolver.sessionFileForIndex(0); forceThinkingOff = resolver.thinkingOverrideForIndex(0) === "off"; if (!forkSessionFile) { throw new Error("Forked context failed to produce a branched session file."); } writeStatus(runDir, { context: "fork", forkSessionFile }); } catch (forkError) { // Reference rule: explicit fork never silently downgrades. activeAsyncRuns.delete(runId); throw forkError; } } // Fire-and-track: the promise writes the durable result file + notifies on // completion; the tool call returns immediately with the run id. void (async () => { try { const result = await runAsyncSubagent( { agent, task: launch.task, cwd: childCwd, model: launch.model, thinking: launch.thinking, tools: agent.builtinToolNames, extensions: childExtensionsArg(agent.extensions), timeoutMs: launch.timeoutMs, parentSessionId: asyncSessionId, config, ...(launch.resumeSessionFile ? { sessionFile: launch.resumeSessionFile } : {}), ...(forkSessionFile ? { forkSessionFile } : {}), ...(forceThinkingOff ? { forceThinkingOff } : {}), }, runDir, controller.signal, ); writeAsyncResultFile(RESULTS_DIR, { runId, sessionId: asyncSessionId, ...(result.output !== undefined ? { output: result.output } : {}), ...(result.error ? { error: result.error } : {}), success: result.status === "completed", state: result.status, timedOut: result.status === "timedOut", durationMs: result.durationMs, }, { asyncDir: runDir }); } catch (error) { writeStatus(runDir, { status: "failed", error: error instanceof Error ? error.message : String(error), }); writeAsyncResultFile(RESULTS_DIR, { runId, sessionId: asyncSessionId, error: error instanceof Error ? error.message : String(error), success: false, state: "failed", }, { asyncDir: runDir }); } finally { activeAsyncRuns.delete(runId); if (worktreeSetup) { try { const diffs = diffWorktrees(worktreeSetup, [launch.agentName], runDir); writeFileSync( join(runDir, "handoff.json"), JSON.stringify({ patches: diffs.map((d) => d.patchPath) }), { mode: 0o600 }, ); cleanupWorktrees(worktreeSetup, { kind: "preserve", capturedDiffs: diffs, handoffManifestPath: join(runDir, "handoff.json"), }); } catch { // Worktree cleanup is best-effort; preserved trees surface in the report. } } } })(); return { runId, status: "running" }; }; // Result watcher: deliver async completions as follow-up notifications. const watcher = createResultWatcher({ resultsDir: RESULTS_DIR, sessionId: asyncSessionId, resultScanLogging: config.resultScanLogging ?? "activity", notifier: (notification) => { if (sessionEnded) return; try { pi.sendMessage( { customType: "unipi-response", content: `\n` + `${notification.runId}\n` + `${notification.success ? "completed" : notification.state ?? "failed"}\n` + `Background agent "${notification.agent ?? "agent"}" ${notification.state ?? (notification.success ? "completed" : "failed")}\n` + (notification.error ? `${notification.error}\n` : `${(notification.output ?? "").slice(0, 2000)}\n`) + ``, display: false, }, { deliverAs: "followUp", triggerTurn: true }, ); } catch { // Runtime went stale — nothing to notify. } }, }); // Periodic retention cleanup (hourly, unref'd). const retentionTimer = setInterval(() => { cleanupAsyncRetention(ASYNC_DIR, RESULTS_DIR); }, 60 * 60 * 1000); retentionTimer.unref?.(); // Scheduled runs: poll for due schedules every minute when enabled. let schedulePoller: { stopPolling(): void } | undefined; if (config.scheduledRuns?.enabled !== false && deps_runAsyncAvailable()) { import("./scheduled-runs.js").then(({ ScheduledRunManager }) => { const manager = new ScheduledRunManager(process.cwd(), { storeRoot: config.scheduledRuns?.storeRoot, maxPending: config.scheduledRuns?.maxPending, launch: async (record) => { const result = await runAsyncDep({ agentName: record.agent, task: record.task, description: `schedule: ${record.name}`, context: "fresh", timeoutMs: record.timeoutMs, }); return result.runId; }, }); manager.startPolling(); schedulePoller = manager; }).catch(() => {}); } function deps_runAsyncAvailable(): boolean { return true; // runAsyncDep is defined below in the closure } // ---- Parity handler wiring (spawn_helper surface) ---- // Session-wide cumulative spawn accounting (maxSubagentSpawnsPerSession) let sessionSpawnsUsed = 0; const sessionSpawnCap = config.maxSubagentSpawnsPerSession && config.maxSubagentSpawnsPerSession > 0 ? config.maxSubagentSpawnsPerSession : undefined; const handlerDeps: HandlerDeps = { pi, manager, config, spawnAccounting: { used: () => sessionSpawnsUsed, cap: () => sessionSpawnCap, consume: (count) => { sessionSpawnsUsed += count; }, }, runAsync: runAsyncDep, spawnBackground: (spawnCtx, agentName, childPrompt, options) => { const { state: bgState, callbacks: bgCallbacks } = createActivityTracker(options.maxTurns); const origOnSession = bgCallbacks.onSessionCreated; bgCallbacks.onSessionCreated = (session: any) => { origOnSession(session); bgState.tokens = safeFormatTokens(session); widget.update(); }; const id = manager.spawn(pi, spawnCtx, agentName, childPrompt, { description: options.description ?? `${agentName} task`, maxTurns: options.maxTurns, modelInput: options.modelInput, modelRegistry: spawnCtx.modelRegistry, thinkingLevel: coerceThinkingLevel(options.thinkingLevel as never), isBackground: true, ...bgCallbacks, }); agentActivity.set(id, bgState); widget.ensureTimer(); widget.update(); return id; }, spawnForeground: async (spawnCtx, agentName, childPrompt, options) => { // Stream progress via the widget — reuse the activity tracker. let spinnerFrame = 0; const startedAt = Date.now(); let fgId: string | undefined; const { state: fgState, callbacks: fgCallbacks } = createActivityTracker(options.maxTurns); const streamUpdate = () => { onUpdateForForeground({ status: "running", toolUses: fgState.toolUses, tokens: fgState.tokens, turnCount: fgState.turnCount, maxTurns: fgState.maxTurns, durationMs: Date.now() - startedAt, activity: describeActivity(fgState.activeTools, fgState.responseText), spinnerFrame: spinnerFrame % SPINNER.length, }); }; const origOnSession = fgCallbacks.onSessionCreated; fgCallbacks.onSessionCreated = (session: any) => { origOnSession(session); fgState.tokens = safeFormatTokens(session); for (const a of manager.listAgents()) { if (a.session === session) { fgId = a.id; agentActivity.set(a.id, fgState); widget.ensureTimer(); break; } } }; const spinnerInterval = setInterval(() => { spinnerFrame++; streamUpdate(); }, 80); // Detach: stop waiting without killing the child. The completion // notification arrives later via the normal onComplete path. activeForegroundDetach = () => { if (!fgId) return false; const record = manager.getRecord(fgId); if (!record || record.status !== "running") return false; record.resultConsumed = true; // suppress the duplicate notification return true; }; try { const record = await manager.spawnAndWait(pi, spawnCtx, agentName, childPrompt, { description: options.description ?? `${agentName} task`, maxTurns: options.maxTurns, modelInput: options.modelInput, modelRegistry: spawnCtx.modelRegistry, thinkingLevel: coerceThinkingLevel(options.thinkingLevel as never), ...fgCallbacks, }); const durationMs = (record.completedAt ?? Date.now()) - record.startedAt; const tokenText = safeFormatTokens(fgState.session); if (record.status === "error") { return { ok: false, output: record.error ?? "failed", error: record.error, toolUses: record.toolUses, durationMs }; } const output = boundHelperOutput(record.result?.trim() || "No output."); record.resultArtifactPath = output.artifactPath; return { ok: true, output: output.text, toolUses: record.toolUses, durationMs }; } finally { clearInterval(spinnerInterval); activeForegroundDetach = undefined; if (fgId) { agentActivity.delete(fgId); widget.markFinished(fgId); widget.update(); } } }, }; // onUpdate stream for the current foreground execution (set per execute() call) let onUpdateForForeground: (details: Record) => void = () => {}; // ---- Foreground detach (foregroundDetachShortcut) ---- const detachParts = parseDetachShortcut(config.foregroundDetachShortcut); let activeForegroundDetach: (() => boolean) | undefined; // Build notification details for the message renderer function buildNotificationDetails(record: any, activity?: AgentActivity): NotificationDetails { return { id: record.id, description: record.description, status: record.status, toolUses: record.toolUses, turnCount: activity?.turnCount ?? 0, maxTurns: activity?.maxTurns, totalTokens: safeTokenCount(record.session), durationMs: record.completedAt ? record.completedAt - record.startedAt : 0, error: record.error, resultPreview: record.result ? record.result.length > 200 ? record.result.slice(0, 200) + "…" : record.result : "No output.", }; } // ---- Register custom notification renderer ---- pi.registerMessageRenderer( "subagent-notification", (message, { expanded }, theme) => { const d = message.details; if (!d) return undefined; function renderOne(d: NotificationDetails): string { const isError = d.status === "error" || d.status === "stopped" || d.status === "aborted"; const icon = isError ? theme.fg("error", "✗") : theme.fg("success", "✓"); const statusText = isError ? d.status : d.status === "steered" ? "completed (steered)" : "completed"; // Line 1: icon + agent description + status let line = `${icon} ${theme.bold(d.description)} ${theme.fg("dim", statusText)}`; // Line 2: stats const parts: string[] = []; if (d.turnCount > 0) parts.push(formatTurns(d.turnCount, d.maxTurns)); if (d.toolUses > 0) parts.push(`${d.toolUses} tool use${d.toolUses === 1 ? "" : "s"}`); if (d.totalTokens > 0) parts.push(formatTokens(d.totalTokens)); if (d.durationMs > 0) parts.push(formatMs(d.durationMs)); if (parts.length) { line += "\n " + parts.map((p) => theme.fg("dim", p)).join(" " + theme.fg("dim", "·") + " "); } // Line 3: result preview (collapsed) or full (expanded) if (expanded) { const lines = d.resultPreview.split("\n").slice(0, 30); for (const l of lines) line += "\n" + theme.fg("dim", ` ${l}`); } else { const preview = d.resultPreview.split("\n")[0]?.slice(0, 80) ?? ""; line += "\n " + theme.fg("dim", `⎿ ${preview}`); } return line; } const all = [d, ...(d.others ?? [])]; return new Text(all.map(renderOne).join("\n"), 0, 0); }, ); // Create widget const widget = new AgentWidget(manager, agentActivity); // ---- FleetView (persistent fleet panel; our AgentWidget slot system) ---- const fleetView = new FleetView(manager, agentActivity, ASYNC_DIR, { placement: config.fleetView === false ? undefined : (config.fleetViewPlacement ?? "belowEditor"), openInspector: async (entry) => { if (!sessionCtx?.ui) return; if (entry.source === "inprocess") { const record = manager.getRecord(entry.key.replace("inprocess:", "")); if (record?.session) { await withHerdrBlocked( pi, "fleet inspector", () => sessionCtx!.ui.custom( (tui, theme, _keybindings, done) => new ConversationViewer(tui, record.session!, { type: record.type, description: record.description, status: record.status, toolUses: record.toolUses, startedAt: record.startedAt, completedAt: record.completedAt, }, agentActivity.get(record.id), theme, done), { overlay: true, overlayOptions: { anchor: "center", width: "90%" } }, ), ); } return; } // Async run: show the transcript tail from the result payload / output.txt. const runId = entry.key.replace("async:", ""); const payload = readAsyncResultFile(RESULTS_DIR, runId); const text = payload ? `${payload.success ? "completed" : payload.state ?? "failed"}\n\n${payload.output ?? payload.error ?? "(no output)"}` : "(run still active — transcript available after completion)"; await withHerdrBlocked( pi, "fleet inspector", () => sessionCtx!.ui.custom( (_tui, theme, _keybindings, done) => ({ render: (width: number): string[] => [ ...text.split("\n").slice(-30).map((line) => ` ${line}`), "", ` ${theme.fg("dim", "esc/q close")}`, ], handleInput: (data: string): void => { if (data === "\x1b" || data === "q") done(undefined); }, invalidate: (): void => {}, }), { overlay: true, overlayOptions: { anchor: "center", width: "90%" } }, ), ); }, }); // ---- Slash commands (/unipi:subagents-*) ---- registerSlashCommands(pi, () => sessionCtx ?? undefined, { manager, config, asyncDirRoot: ASYNC_DIR, }); // Register info group at factory time (not session_start) const registry = getInfoRegistry(); if (registry) { registry.registerGroup({ id: "subagents", name: "Subagents", icon: "🤖", priority: 80, config: { showByDefault: true, stats: [ { id: "maxConcurrent", label: "Max Concurrent", show: true }, { id: "activeCount", label: "Active Agents", show: true }, { id: "enabled", label: "Enabled", show: true }, { id: "types", label: "Available Types", show: true }, ], }, dataProvider: async () => { const types = config.types || {}; const codeBuiltins: string[] = [...BUILTIN_TYPES]; const fileBuiltins = [...loadBuiltinFileAgents().keys()]; const customTypes: string[] = []; for (const dir of [globalAgentsDir, workspaceAgentsDir]) { try { if (existsSync(dir)) { for (const file of readdirSync(dir)) { if (file.endsWith(".md") && !customTypes.includes(file.replace(".md", ""))) { customTypes.push(file.replace(".md", "")); } } } } catch { /* ignore */ } } const allTypes = manager.getKnownTypes(); const typeList = allTypes.map((t) => { const isEnabled = manager.isTypeEnabled(t); const scope = codeBuiltins.includes(t) || fileBuiltins.includes(t) ? "builtin" : customTypes.includes(t) ? "custom" : "config"; return `${t}(${scope})${isEnabled ? "" : " [disabled]"}`; }).join(", "); const activeAgents = manager.listAgents().filter((a) => a.status === "running").length; return { maxConcurrent: { value: String(manager.getMaxConcurrent()) }, activeCount: { value: String(activeAgents) }, enabled: { value: config.enabled ? "yes" : "no" }, types: { value: allTypes.length > 0 ? allTypes[0] : "none", detail: allTypes.length > 1 ? typeList : undefined, }, }; }, }); } // Store session context for badge generation let sessionCtx: import("@earendil-works/pi-coding-agent").ExtensionContext | null = null; // Session start: emit MODULE_READY + capture context pi.on("session_start", async (_event, ctx) => { sessionCtx = ctx; // FleetView keyboard activation (↓/← to inspect active work). try { if (typeof (ctx.ui as unknown as { onTerminalInput?: unknown }).onTerminalInput === "function") { (ctx.ui as unknown as { onTerminalInput(handler: (data: string) => { consume?: boolean } | undefined): () => void; }).onTerminalInput((data) => { // Detach shortcut: detach the active foreground run. if (detachParts && matchesDetachInput(data, detachParts) && activeForegroundDetach) { if (activeForegroundDetach()) { try { ctx.ui.notify("Foreground run detached — it continues in the background.", "info"); } catch { /* best effort */ } return { consume: true }; } } return fleetView.handleKey(data, () => { try { return ctx.ui.getEditorText() !== ""; } catch { return false; } }); }); } } catch { // Terminal input hooking is optional. } fleetView.setUICtx(ctx.ui); emitEvent(pi, UNIPI_EVENTS.MODULE_READY, { name: MODULES.SUBAGENTS || "subagents", version: "0.2.0", commands: [], tools: ["spawn_helper", "get_helper_result"], }); }); // Listen for badge generation requests — spawn background agent pi.events.on(UNIPI_EVENTS.BADGE_GENERATE_REQUEST, async (data) => { const event = data as UnipiBadgeGenerateRequestEvent; if (!sessionCtx) return; const summary = event?.conversationSummary ?? ""; const prompt = summary ? `Based on this conversation, generate a concise session title (MAX 5 WORDS). Reply with ONLY the title. No quotes, no explanation, no punctuation.\n\nConversation:\n${summary}` : `Generate a concise session title (MAX 5 WORDS) for this session. Reply with ONLY the title. No quotes, no explanation, no punctuation.`; // Try with configured model, fallback to inherit let modelInput: string | undefined = undefined; try { const fs = await import("node:fs"); const path = await import("node:path"); const configPath = path.resolve(process.cwd(), ".unipi/config/badge.json"); if (fs.existsSync(configPath)) { const parsed = JSON.parse(fs.readFileSync(configPath, "utf-8")); if (typeof parsed.generationModel === "string" && parsed.generationModel !== "inherit") { modelInput = parsed.generationModel; } } } catch { /* ignore — inherit parent model */ } let resolvedModel: any = undefined; // Check if model is available if (modelInput && sessionCtx.modelRegistry) { const { resolveModel } = await import("./model-resolver.js"); const result = resolveModel(modelInput, sessionCtx.modelRegistry); if (typeof result !== "string") { resolvedModel = result; } // If result is a string (error), resolvedModel stays undefined → inherit parent } manager.spawn(pi, sessionCtx, "name-gen", prompt, { description: "Generate session name", model: resolvedModel, isBackground: true, isolated: true, maxTurns: 1, }); }); // ESC propagation: abort all agents on session shutdown. // Set the guard FIRST: abortAll() settles in-flight promises, whose // completion callbacks would otherwise reach a runtime that pi is about to // invalidate. pi.on("session_shutdown", async () => { watcher.stop(); schedulePoller?.stopPolling(); sessionEnded = true; manager.abortAll(); manager.dispose(); }); // Wire UI context for widget + age finished agents on new turn pi.on("tool_execution_start", async (_event, ctx) => { widget.setUICtx(ctx.ui); widget.onTurnStart(); }); // Create activity tracker function createActivityTracker(maxTurns?: number, onStreamUpdate?: () => void) { const state: AgentActivity = { activeTools: new Map(), toolUses: 0, turnCount: 1, maxTurns, tokens: "", responseText: "", }; const callbacks = { onToolActivity: (activity: { type: "start" | "end"; toolName: string }) => { if (activity.type === "start") { state.activeTools.set(activity.toolName + "_" + Date.now(), activity.toolName); } else { for (const [key, name] of state.activeTools) { if (name === activity.toolName) { state.activeTools.delete(key); break; } } state.toolUses++; } state.tokens = safeFormatTokens(state.session); onStreamUpdate?.(); }, onTextDelta: (_delta: string, fullText: string) => { state.responseText = fullText; onStreamUpdate?.(); }, onTurnEnd: (turnCount: number) => { state.turnCount = turnCount; onStreamUpdate?.(); }, onSessionCreated: (session: any) => { state.session = session; }, }; return { state, callbacks }; } // ---- Agent tool ---- const builtinTypes = BUILTIN_TYPES.join(", "); const enabledTypes = manager.getKnownTypes().filter((type) => manager.isTypeEnabled(type)); const availableTypes = enabledTypes.join(", ") || "none"; pi.registerTool( defineTool({ name: "spawn_helper", label: "Spawn Helper", description: `Launch a sub-agent for parallel work. Available agent types: ${availableTypes} Custom types can be defined in: - ~/.unipi/config/agents/.md (global) - /.unipi/config/agents/.md (project) Guidelines: - Use "explore" or "scout" for parallel file reads / fast codebase recon - Use "work" or "worker" for parallel file writes (transparent locking) - Use "reviewer" for code review of diffs, plans, or solutions - Use "researcher" for web research (needs web-api tools) - Use "oracle" for a second opinion on decisions before acting - Use "delegate" for a lightweight general delegate - Use run_in_background for work you don't need immediately - ESC kills all running agents immediately - Agents inherit the parent model by default`, parameters: SpawnHelperParams, // ---- Rich inline rendering ---- renderCall(args, theme) { const displayName = args.type ? args.type : "Agent"; const desc = args.description ?? ""; return new Text( "▸ " + theme.fg("toolTitle", theme.bold(displayName)) + (desc ? " " + theme.fg("muted", desc) : ""), 0, 0, ); }, renderResult(result, { expanded, isPartial }, theme) { const details = result.details as any; if (!details) { const text = result.content[0]?.type === "text" ? result.content[0].text : ""; return new Text(text, 0, 0); } // inlineToolDisplay: "summary" keeps one stable row per state. if (config.inlineToolDisplay === "summary") { const glyph = isPartial || details.status === "running" ? theme.fg("accent", SPINNER[details.spinnerFrame ?? 0]) : details.status === "completed" ? theme.fg("success", "✓") : details.status === "background" ? theme.fg("dim", "■") : theme.fg("error", "✗"); const label = details.status === "completed" ? "completed" : details.status === "background" ? "background" : details.status === "running" ? "running" : details.status; return new Text(theme.fg("muted", `${glyph} ${label}`), 0, 0); } // Stats helper const stats = (d: any) => { const parts: string[] = []; if (d.turnCount != null && d.turnCount > 0) parts.push(formatTurns(d.turnCount, d.maxTurns)); if (d.toolUses > 0) parts.push(`${d.toolUses} tool use${d.toolUses === 1 ? "" : "s"}`); if (d.tokens) parts.push(d.tokens); return parts.map((p) => theme.fg("dim", p)).join(" " + theme.fg("dim", "·") + " "); }; // Running if (isPartial || details.status === "running") { const frame = SPINNER[details.spinnerFrame ?? 0]; const s = stats(details); let line = theme.fg("accent", frame) + (s ? " " + s : ""); line += "\n" + theme.fg("dim", ` ⎿ ${details.activity ?? "thinking…"}`); return new Text(line, 0, 0); } // Background launched if (details.status === "background") { return new Text(theme.fg("dim", ` ⎿ Running in background (ID: ${details.agentId})`), 0, 0); } // Completed if (details.status === "completed") { const duration = formatMs(details.durationMs); const s = stats(details); let line = theme.fg("success", "✓") + (s ? " " + s : ""); line += " " + theme.fg("dim", "·") + " " + theme.fg("dim", duration); if (expanded) { const resultText = result.content[0]?.type === "text" ? result.content[0].text : ""; if (resultText) { const rlines = resultText.split("\n").slice(0, 50); for (const l of rlines) { line += "\n" + theme.fg("dim", ` ${l}`); } } } else { line += "\n" + theme.fg("dim", " ⎿ Done"); } return new Text(line, 0, 0); } // Error / Aborted / Stopped const isError = details.status === "error"; const isStopped = details.status === "stopped"; const s = stats(details); let line = (isStopped ? theme.fg("dim", "■") : theme.fg("error", "✗")) + (s ? " " + s : ""); if (isError) { line += "\n" + theme.fg("error", ` ⎿ Error: ${details.error ?? "unknown"}`); } else if (isStopped) { line += "\n" + theme.fg("dim", " ⎿ Stopped"); } else { line += "\n" + theme.fg("warning", " ⎿ Aborted (max turns exceeded)"); } return new Text(line, 0, 0); }, // ---- Execute ---- execute: async (toolCallId, params, signal, onUpdate, ctx) => { widget.setUICtx(ctx.ui); fleetView.setUICtx(ctx.ui); // Route through the parity handler (actions, workflowScript, legacy // single-child) — widget/notify plumbing lives in the deps adapters. onUpdateForForeground = (details) => { onUpdate?.({ content: [{ type: "text", text: `${details.toolUses ?? 0} tool uses...` }], details, }); }; try { return (await handleSpawnHelper(handlerDeps, ctx, params as Record, signal)) as never; } finally { onUpdateForForeground = () => {}; } }, }), ); // ---- get_helper_result tool ---- pi.registerTool( defineTool({ name: "get_helper_result", label: "Get Helper Result", description: "Check status and retrieve results from a background agent or async run. Use view: true to open a live conversation overlay; nonBlocking: true to subscribe and be woken on completion.", parameters: GetHelperResultParams, execute: async (_toolCallId, rawParams, _signal, _onUpdate, ctx) => { const params = rawParams as Record; const id = (params.id ?? params.agent_id) as string | undefined; const nonBlocking = params.nonBlocking === true; const waitAll = params.all === true; const timeoutMs = typeof params.timeoutMs === "number" ? params.timeoutMs : undefined; // ---- Async process runs ---- const asyncPayload = id ? readAsyncResultFile(RESULTS_DIR, id) : undefined; if (asyncPayload && !nonBlocking) { const duration = asyncPayload.durationMs ? `${(asyncPayload.durationMs / 1000).toFixed(1)}s` : ""; return textResult( `Run: ${asyncPayload.runId}\nAgent: ${asyncPayload.agent ?? "unknown"} | Status: ${asyncPayload.state ?? (asyncPayload.success ? "completed" : "failed")}\n\n${asyncPayload.output ?? asyncPayload.error ?? "(no output)"}`, { status: asyncPayload.success ? "completed" : "error", runId: asyncPayload.runId }, ); } if (nonBlocking && id) { // Persist a wake subscription: when the result file appears, the // watcher delivers a followUp notification that wakes this session. if (readAsyncResultFile(RESULTS_DIR, id)) { return textResult(`Run ${id} already completed.`, { status: "completed", runId: id }); } writePendingSubscription(RESULTS_DIR, asyncSessionId, id); return textResult( `Subscribed to run ${id}. This session will be woken on completion or failure.`, { status: "subscribed", runId: id }, ); } // ---- In-process records ---- const record = manager.getRecord(id as string); if (!record) { if (waitAll) { // Wait for ALL active in-process agents. const active = manager.listAgents().filter((a) => a.status === "running" || a.status === "queued"); if (active.length === 0) return textResult("No active agents."); await Promise.race([ Promise.allSettled(active.map((a) => a.promise).filter(Boolean)), new Promise((resolve) => setTimeout(resolve, timeoutMs ?? 1_800_000)), ]); return textResult(`All ${active.length} agent(s) settled (or wait timed out).`); } return textResult(`Helper not found: "${id}". It may have been cleaned up.`); } // Open conversation viewer overlay if requested if (params.view && record.session) { const activity = agentActivity.get(record.id); await withHerdrBlocked( pi, "helper viewer", () => ctx.ui.custom( (tui, theme, _keybindings, done) => { return new ConversationViewer( tui, record.session!, { type: record.type, description: record.description, status: record.status, toolUses: record.toolUses, startedAt: record.startedAt, completedAt: record.completedAt, }, activity, theme, done, ); }, { overlay: true, overlayOptions: { anchor: "center", width: "90%" }, }, ), ); } if (params.wait && record.status === "running" && record.promise) { record.resultConsumed = true; await record.promise; } const duration = record.completedAt ? `${((record.completedAt - record.startedAt) / 1000).toFixed(1)}s` : "running"; let output = `Agent: ${record.id}\n` + `Type: ${record.type} | Status: ${record.status} | Tool uses: ${record.toolUses} | Duration: ${duration}\n` + `Description: ${record.description}\n\n`; if (record.status === "running") { output += "Agent is still running. Use wait: true or check back later."; } else if (record.status === "error") { output += `Error: ${record.error}`; } else { const bounded = boundHelperOutput( record.result?.trim() || "No output.", 64 * 1024, record.resultArtifactPath, ); record.resultArtifactPath = bounded.artifactPath; output += bounded.text; } if (record.status !== "running" && record.status !== "queued") { record.resultConsumed = true; } return textResult(output); }, }), ); }