/** * agent-runner.ts — spawn isolated sub-agents for analysis and fix phases. * * The production runner uses pi's `createAgentSession` SDK to spin up a fresh * in-memory agent session scoped to the target `cwd`, with the agent * definition's system prompt and tool allowlist applied. Because that path * needs live model credentials (unsuitable for CI), the runner is backed by an * injectable factory: tests swap it for a deterministic fake that executes a * tiny instruction protocol embedded in the task string. * * Instruction protocol (used by the fake runner, harmless to the real one): * task lines may begin with `!write ` — the fake writes the file * and reports it as a finding. Real sub-agents receive the whole task verbatim. * * @module pygienium/agent-runner */ import { mkdir, writeFile } from "node:fs/promises"; import { dirname, isAbsolute, join } from "node:path"; import type { AgentSession, AgentSessionEvent, ExtensionAPI, ExtensionCommandContext, } from "@earendil-works/pi-coding-agent"; import { loadAgents, extensionRoot, type AgentDef } from "./agents.js"; /** The invoking session's provider registry, as pi types it. */ type HostModelRegistry = ExtensionCommandContext["modelRegistry"]; /** * The two capabilities a sub-agent needs from the invoking session's * `ModelRegistry`: the registrations extensions made through * `pi.registerProvider`. Values are left in their host types — this module * relays them, it never interprets them. */ export interface ProviderRegistry { getRegisteredNativeProvider( providerId: string, ): ReturnType; getRegisteredProviderConfig( providerId: string, ): ReturnType; } /** * The inline extension that re-registers a provider in a sub-session. Spelled out * structurally — an `InlineExtension` in pi — so the generated omp port, which * names the type differently, still compiles. */ export interface InlineProviderExtension { name: string; hidden: boolean; factory: (pi: ExtensionAPI) => void; } /** * Narrow a host context's `modelRegistry` to a {@link ProviderRegistry}, * returning `undefined` when it cannot answer the two questions this module * asks. This is the decode boundary: everything downstream works with the * narrow interface, and a host whose registry does not know extension * registrations simply gets no provider forwarding. */ export function asProviderRegistry( candidate: unknown, ): ProviderRegistry | undefined { if (!candidate || typeof candidate !== "object") return undefined; const registry = candidate as Partial; if ( typeof registry.getRegisteredNativeProvider !== "function" || typeof registry.getRegisteredProviderConfig !== "function" ) { return undefined; } return registry as ProviderRegistry; } /** * Env var overriding the per-agent settle timeout (ms). Guards against a * sub-agent session that never settles (stalled provider stream, hung retry / * auto-compaction after the final tool call), which otherwise leaves the run * stuck mid-phase with no state save, no error, and no completion message. */ export const AGENT_TIMEOUT_ENV = "PYGIENIUM_AGENT_TIMEOUT_MS"; /** Default settle timeout per agent phase: 60 minutes. */ const DEFAULT_AGENT_TIMEOUT_MS = 60 * 60_000; /** Resolve the per-agent settle timeout, honouring the env override. */ export function agentTimeoutMs(): number { const raw = process.env[AGENT_TIMEOUT_ENV]; if (raw && /^\d+$/.test(raw.trim()) && Number(raw.trim()) > 0) { return Number(raw.trim()); } return DEFAULT_AGENT_TIMEOUT_MS; } export interface AgentTaskOptions { /** Absolute working directory for the sub-agent. */ cwd: string; /** Agent name to look up in `agents/*.md`. */ agentName: string; /** The task prompt handed to the sub-agent. */ task: string; /** Optional tool allowlist override (else uses the agent's `allowedTools`). */ allowedTools?: string[]; /** * The currently selected model from the invoking session. When omitted, * `createAgentSession` falls back to the settings default model. */ model?: ExtensionCommandContext["model"]; /** * The invoking session's provider registry, used to make the pinned * `model`'s provider resolvable inside the sub-session (see * {@link providerForwardingExtension}). */ modelRegistry?: ProviderRegistry; /** Optional explicit agent definition (skips `loadAgents`). */ agent?: AgentDef; /** * Live callback forwarding raw {@link AgentSessionEvent}s from the * sub-agent session. The `pygienium-stream` forwarder in `index.ts` * turns tool_execution_start/end + assistant turns into chat messages. */ onEvent?: (event: AgentSessionEvent) => void; /** * Maximum time the agent run may take before it is aborted and the phase * fails loudly (default {@link agentTimeoutMs}). A session that never * settles — stalled provider stream, hung retry/compaction after its last * tool call — would otherwise hang the check run silently with no state * update and no completion message. */ timeoutMs?: number; } /** * Race `promise` against a settle deadline. Returns `{ value }` on success or * `{ error }` when the deadline elapsed first (the caller aborts the work). * `promise` is still awaited-then-ignored afterwards so late rejections can * never surface as unhandled. */ export async function withSettleTimeout( promise: Promise, timeoutMs: number, label: string, ): Promise<{ value: T } | { error: string }> { if (timeoutMs <= 0) { try { return { value: await promise }; } catch (err) { return { error: err instanceof Error ? err.message : String(err) }; } } const settled = promise.then( (value) => ({ value }) as { value: T }, (err) => ({ error: err instanceof Error ? err.message : String(err) }) as { error: string; }, ); let timer: ReturnType | undefined; const deadline = new Promise<{ error: string }>((resolve) => { timer = setTimeout(() => { // ~`1m 30s` / `45s` for the message (timeoutMs is ms; tests use small values). const totalSec = Math.round(timeoutMs / 1000); const duration = totalSec >= 60 ? `${Math.floor(totalSec / 60)}m${totalSec % 60 ? ` ${totalSec % 60}s` : ""}` : `${totalSec}s`; resolve({ error: `${label} did not settle within ${duration}; aborted. Check model/provider connectivity, then resume with /pygienium-resume.`, }); }, timeoutMs); // Never hold the process open just because a deadline is pending. timer.unref?.(); }); try { return await Promise.race([settled, deadline]); } finally { if (timer) clearTimeout(timer); } } export interface AgentRunResult { /** Whether the sub-agent completed without throwing. */ ok: boolean; /** The final assistant text emitted by the sub-agent. */ text: string; /** Error message when `ok` is false. */ error?: string; } export type AgentRunner = (opts: AgentTaskOptions) => Promise; /** Module-level runner (defaults to the SDK-backed runner; tests override it). */ let currentRunner: AgentRunner = defaultAgentRunner; /** Entry point used by the check-runner. */ export async function runAgentTask( opts: AgentTaskOptions, ): Promise { // Backstop: even a third-party/custom runner must not be able to hang the // run forever. The default runner additionally aborts its session on // timeout (see defaultAgentRunner); this race covers every other runner. const settled = await withSettleTimeout( currentRunner(opts), opts.timeoutMs ?? agentTimeoutMs(), `sub-agent "${opts.agentName}"`, ); if ("error" in settled) { return { ok: false, text: "", error: settled.error }; } return settled.value; } /** Override the active agent runner (primarily for tests). */ export function setAgentRunner(runner: AgentRunner): void { currentRunner = runner; } /** Restore the default SDK-backed agent runner. */ export function resetAgentRunner(): void { currentRunner = defaultAgentRunner; } /** * Build the inline extension that re-registers a provider inside a sub-session. * * A provider that exists only because an extension called * `pi.registerProvider` (custom base URL, wrapped stream, its own auth wiring) * lives in the invoking session's registry and nowhere else — not in * `auth.json`, not in `models.json`. Sub-sessions are deliberately created with * `noExtensions: true`, so their own `ModelRuntime` cannot resolve credentials * for such a provider: pinning the invoking session's model then dies with * pi's `No API key found for ` before the first token, which reads as * a pygienium failure on a perfectly healthy login. * * Inline extension factories load regardless of `noExtensions`, so registering * the pinned model's provider there is the narrow cost of pinning: the * sub-agent authenticates as the host does, while every other extension stays * out of its scope. * * Returns `undefined` when there is nothing to forward — no registry, no pinned * model, or a provider pi already knows from its own catalogue (built-ins and * `models.json` providers need no registration). */ export function providerForwardingExtension( registry: ProviderRegistry | undefined, providerId: string | undefined, ): InlineProviderExtension | undefined { if (!registry || !providerId) return undefined; // A native provider carries its own stream/auth implementation, so it wins // over a config-only registration (which only augments a built-in one). const native = registry.getRegisteredNativeProvider(providerId); const config = native ? undefined : registry.getRegisteredProviderConfig(providerId); if (!native && !config) return undefined; return { name: "pygienium-provider", hidden: true, factory: (pi) => { if (native) pi.registerProvider(native); else if (config) pi.registerProvider(providerId, config); }, }; } /** * Real sub-agent runner: spins up an in-memory `AgentSession` scoped to `cwd`, * overrides the system prompt with the agent definition's body, restricts tools * to the agent's allowlist, and runs the task to completion. */ export async function defaultAgentRunner( opts: AgentTaskOptions, ): Promise { const agents = await loadAgents({ cwd: opts.cwd }); const agent = opts.agent ?? agents.get(opts.agentName); if (!agent) { const names = [...agents.keys()]; return { ok: false, text: "", error: `Unknown agent definition: "${opts.agentName}". ` + `Available agents: ${names.length > 0 ? names.join(", ") : "(none loaded — check agents/ directories exist)"}. ` + `Searched ${extensionRoot()}/agents/ (extension) and ${opts.cwd}/agents/ (project-local). ` + `Add agents/${opts.agentName}.md to either location.`, }; } // Lazily import the SDK so the rest of the module graph (and tests using the // fake runner) never resolve the heavy pi-coding-agent package. const { createAgentSession, DefaultResourceLoader, getAgentDir, SessionManager, } = await import("@earendil-works/pi-coding-agent"); // Re-register the pinned model's provider inside the sub-session: inline // extension factories load even under `noExtensions`, so this is the one // registration a pinned model needs in order to authenticate. const providerExtension = providerForwardingExtension( opts.modelRegistry, opts.model?.provider, ); const loader = new DefaultResourceLoader({ cwd: opts.cwd, agentDir: getAgentDir(), systemPromptOverride: () => agent.systemPrompt, // Keep the sub-agent isolated: no nested extensions/skills/themes/etc. noExtensions: true, noSkills: true, noThemes: true, noPromptTemplates: true, extensionFactories: providerExtension ? [providerExtension] : [], }); await loader.reload(); const tools = opts.allowedTools ?? agent.allowedTools ?? ["read", "bash", "grep", "find"]; const { session } = await createAgentSession({ cwd: opts.cwd, // Pin the sub-agent to the model the user has selected in the invoking // session rather than the settings default. Omitted when the caller // has no live session model (print/RPC modes), which keeps the // settings-default fallback. ...(opts.model ? { model: opts.model } : {}), tools, sessionManager: SessionManager.inMemory(opts.cwd), resourceLoader: loader, }); // A session that never settles (stalled provider stream, hung retry / // auto-compaction after its last tool call) must fail the phase loudly // instead of hanging the run mid-transition with zero diagnostics. On // timeout the session is disposed, which aborts the in-flight run. const settled = await withSettleTimeout( runSessionToCompletion(session, opts), opts.timeoutMs ?? agentTimeoutMs(), `sub-agent "${opts.agentName}"`, ); if ("error" in settled) { // Abort the still-running session so it can't keep burning provider // calls; the background settle path then finishes and disposes too. try { session.dispose(); } catch { /* ignore dispose errors */ } return { ok: false, text: "", error: settled.error }; } return settled.value; } /** * Run an in-memory agent session to completion, streaming tool events to the * chat forwarder, and return the agent's final text. Owns session cleanup. */ async function runSessionToCompletion( session: AgentSession, opts: AgentTaskOptions, ): Promise { const acc: SessionEventAccumulator = { text: "", sawMessage: false }; try { const unsubscribe = session.subscribe((event: AgentSessionEvent) => { applySessionEvent(acc, event, opts.onEvent); }); await session.prompt(opts.task, { expandPromptTemplates: false }); // Ensure the agent has fully settled (tool calls may still be in-flight // after prompt() resolves; piolium's runner calls this for the same // reason). await session.agent.waitForIdle(); unsubscribe(); // Surface session errors that didn't throw but left no useful output. // A session ending with stopReason "error" and no text means the model // call failed silently — treat that as a failed run, not ok:true. A // session that settled with NO text and NO observed events means the // model never produced anything at all (dead provider stream, failed // start) — reporting that as a successful analysis would let an empty // scan masquerade as a clean one, and the check's verify hook would // fail only later with a confusing "artifact missing" error. const sessionError = emptySessionError(acc); if (sessionError) { return { ok: false, text: acc.text, error: sessionError }; } return { ok: true, text: acc.text }; } catch (err) { return { ok: false, text: "", error: err instanceof Error ? err.message : String(err), }; } finally { try { session.dispose(); } catch { /* ignore dispose errors */ } } } /** Running capture state while a session streams assistant output. */ export interface SessionEventAccumulator { /** Joined assistant text seen so far (text_delta stream). */ text: string; /** Whether any assistant-message or tool event was observed at all. */ sawMessage: boolean; /** stopReason of the final assistant message, when reported. */ stopReason?: string; /** errorMessage of the final assistant message, when reported. */ errorMessage?: string; } /** * Classify a settled session's capture: `undefined` when the result is a * legitimate (possibly empty-text) outcome, else the error that should fail * the run. Kept pure so the decision is unit-testable without a session. */ export function emptySessionError( acc: SessionEventAccumulator, ): string | undefined { if (acc.errorMessage) return acc.errorMessage; if (!acc.text.trim()) { if (acc.stopReason === "error") { return "sub-agent session ended in error with no output."; } if (!acc.sawMessage) { return "sub-agent session settled with no output — no assistant message or tool activity was observed. Check model/provider connectivity, then resume."; } } return undefined; } /** * Interpret one session event into the running accumulator and forward the * stream-driving events to the chat. * * MUST never throw: it runs synchronously inside the SDK's event pipeline * (`session.subscribe` listeners are invoked from `processEvents`/`_emit`). * A throw there is not contained — the SDK's run-failure path re-emits * failure events through the same callback, so a callback that always throws * recurses until stack overflow and crashes the host, stranding run-state at * the phase boundary with no save and no completion. Guard every shape (the * runtime events are partial/streaming and weaker than their types) and never * let the cosmetic chat forwarder take the run down. */ export function applySessionEvent( acc: SessionEventAccumulator, event: AgentSessionEvent, forward?: (event: AgentSessionEvent) => void, ): void { if (!event) return; try { // Any message or tool event means the session actually ran — a settled // capture with none of these is a dead session, not an empty scan. if ( event.type === "message_update" || event.type === "message_end" || event.type === "tool_execution_start" || event.type === "tool_execution_end" ) { acc.sawMessage = true; } if ( event.type === "message_update" && event.assistantMessageEvent?.type === "text_delta" ) { acc.text += event.assistantMessageEvent.delta ?? ""; } if (event.type === "message_end") { // Capture the full assistant text from the finalized message — // models that don't stream text_delta (or truncate) still surface // their output here. Prefer the streamed text when non-empty. const message = event.message as | { role?: string; content?: unknown; stopReason?: string; errorMessage?: string; } | undefined; if (message) { if (message.stopReason) acc.stopReason = message.stopReason; if (message.errorMessage) acc.errorMessage = message.errorMessage; if (message.role === "assistant") { const full = extractAssistantText(message.content).trim(); if (full && !acc.text.trim()) acc.text = full; } } } } catch (err) { // A malformed event must not crash the SDK event pipeline; log and skip. console.error("pygienium: error processing sub-agent event", err); } // Forward the stream-driving events to the chat forwarder; it turns each // into its own `pygienium-stream` message (see index.ts). Cosmetic chat // UI: a broken forwarder must not fail or hang the agent run. if ( event.type === "tool_execution_start" || event.type === "tool_execution_end" || event.type === "message_end" ) { try { forward?.(event); } catch (err) { console.error("pygienium: error forwarding sub-agent event", err); } } } /** * Extract joined text from an assistant message's content blocks. * Mirrors piolium's `extractAssistantText`. */ function extractAssistantText(content: unknown): string { if (typeof content === "string") return content; if (!Array.isArray(content)) return ""; return content .flatMap((c) => c && typeof c === "object" && (c as { type?: string }).type === "text" ? [(c as { text?: string }).text ?? ""] : [], ) .join(""); } /** * Fake agent runner for tests: it understands a tiny instruction protocol * embedded in the task so a no-op check can produce deterministic findings * and write marker files without a model. Recognised instructions (one per * line, leading-whitespace tolerant): * * !write — write text to path (relative to cwd); recorded * !echo — appended to findings * * The agent's emitted findings text is the collected `!echo`/`!write` lines. */ export const fakeAgentRunner: AgentRunner = async (opts) => { const lines = opts.task.split(/\r?\n/); const findings: string[] = []; try { for (const line of lines) { const trimmed = line.trim(); const write = /^!write\s+(\S+)\s*(.*)$/.exec(trimmed); if (write) { const rel = write[1] as string; const content = (write[2] ?? "").replace(/^["']|["']$/g, ""); const full = isAbsolute(rel) ? rel : join(opts.cwd, rel); await mkdir(dirname(full), { recursive: true }); await writeFile(full, content + "\n", "utf8"); findings.push(`wrote ${rel}`); continue; } const echo = /^!echo\s+(.*)$/.exec(trimmed); if (echo) { findings.push((echo[1] ?? "").replace(/^["']|["']$/g, "")); } } return { ok: true, text: findings.join("\n") }; } catch (err) { return { ok: false, text: findings.join("\n"), error: err instanceof Error ? err.message : String(err), }; } };