/** * Workflow manager for background execution, pause/resume, and run management. */ import { EventEmitter } from "node:events"; import type { ModelRegistry, ToolDefinition } from "@earendil-works/pi-coding-agent"; import { preview, type WorkflowAgentSnapshot, type WorkflowSnapshot } from "./display.js"; import { isProviderUsageLimit, WorkflowError, WorkflowErrorCode } from "./errors.js"; import type { ExecutorRegistry, WorkflowExecutor } from "./executor.js"; import { createRunPersistence, generateRunId, type PersistedRunState, type RunLease, type RunPersistence, type RunStatus, } from "./run-persistence.js"; import { type JournalEntry, parseWorkflowScript, runWorkflow, type WorkflowAgentRunner, type WorkflowRunResult, type WorkflowRuntimeEvent, } from "./workflow.js"; export interface ManagedRun { runId: string; status: RunStatus; snapshot: WorkflowSnapshot; result?: WorkflowRunResult; error?: WorkflowError; /** Current execution promise; resume waits for a paused attempt to drain telemetry. */ executionPromise?: Promise; controller: AbortController; startedAt: Date; /** The real script, kept so the run can be resumed. */ script: string; args?: unknown; /** Immediate source run for an immutable completed-run revision. */ forkedFromRunId?: string; /** First run in the revision lineage. */ rootRunId?: string; /** Accumulated agent results for resume (deterministic call index -> result). */ journal: JournalEntry[]; /** Cross-process execution lease for this run, when it is actively executing. */ lease?: RunLease; /** * True when the run was started in the background (or resumed) and the caller is * not awaiting its result inline. Only background runs deliver their result back * into the conversation; a foreground sync run already returns it as the tool * result, so re-delivering would duplicate it. */ background: boolean; /** * Auto-resume eligibility for this run (see ExecOptions.autoResume). Set once * at creation and carried through resume() so it survives pause/resume cycles. * Undefined means eligible (default-on); false opts out. */ autoResume?: boolean; /** * The run's resolved hard token budget (per-run value, else the manager * default), fixed at run start and carried through resume() — a resumed run * must keep the budget it started with, not re-resolve against the current * default (an explicit `null` opt-out would otherwise regain a budget). */ tokenBudget?: number | null; /** * Named toolset tag for this run (see WorkflowManagerOptions.toolsets). * ToolDefinitions are functions and can't be persisted, so the tag is what * survives on disk — resume() re-resolves it so e.g. a resumed * `/deep-research` run keeps its web tools instead of silently degrading to * the default coding tools. */ toolset?: string; /** * Real per-agent start/end timestamps, captured at onAgentStart/onAgentEnd * (never fabricated), keyed by the agent's snapshot id. A running agent has * an entry with no endedAt; persistRun() reads from here instead of stamping * every agent with the run's startedAt / "now". */ agentTimestamps: Map; /** * Live snapshot-agent lookup keyed by the agent CALL's unique id (see * WorkflowRunOptions.onAgentStart/onAgentEnd/onAgentHistory's `id` field in * workflow.ts — unique per call, never per label). onAgentEnd/onAgentHistory * must resolve the snapshot entry to update through this map, never by * scanning managed.snapshot.agents for a label match: two concurrent agents * routinely share a label (e.g. parallel()'s default `"${phase} agent N"` * labeling, or an author-supplied label reused across a fan-out), and a * label+status scan would update whichever same-label entry it happens to * find first — misattributing one agent's end/history event to a different, * still-running sibling. */ agentsById: Map; /** * The run's cap on total agents (per-run value, else left undefined so * runWorkflow applies its own MAX_AGENTS_PER_RUN default), fixed at run * start/resume and carried through resume() — mirrors ManagedRun.tokenBudget * exactly: a resumed run must keep the cap it started with, not silently * regain the (much larger) default because ExecOptions.maxAgents isn't * threaded through resume()'s executeRun() call. */ maxAgents?: number; /** * The run's resolved per-agent timeout (per-run value, else the manager * default at the time), fixed at run start/resume — same rationale as * tokenBudget/maxAgents: resume() must not re-resolve against the manager's * CURRENT defaultAgentTimeoutMs. */ agentTimeoutMs?: number | null; /** * The run's resolved concurrency (per-run value, else the manager's * concurrency at the time), fixed at run start/resume for the same reason * as tokenBudget. */ concurrency?: number; /** * The run's resolved agent-retry count (per-run value, else the manager * default at the time), fixed at run start/resume for the same reason as * tokenBudget. */ agentRetries?: number; /** Run-level executor default, frozen at start and carried through resume. */ defaultExecutor: WorkflowExecutor; } /** Per-execution options shared by sync, background, and resume runs. */ export interface ExecOptions { /** * Replay these journaled agent/checkpoint results for the unchanged prefix * (resume), keyed by `${runId}:${index}` — see * WorkflowRunOptions.resumeJournal in workflow.ts. */ resumeJournal?: Map; /** Cap on total agents for this run. */ maxAgents?: number; /** Per-agent timeout in milliseconds. null/omitted means no hard timeout. */ agentTimeoutMs?: number | null; /** Host signal (e.g. tool/Esc) that should abort this run when fired. */ externalSignal?: AbortSignal; /** Called with the live snapshot on every progress event. */ onProgress?: (snapshot: WorkflowSnapshot) => void; /** Hard token budget for this run; once spent reaches it, agent() throws. */ tokenBudget?: number | null; /** * Tool set for this run's subagents, replacing the default coding tools — * e.g. built-in `/deep-research` appends web tools. Omit for the default. * Not persistable (functions): pair with `toolset` so a resumed run can * re-resolve the same tools. */ tools?: ToolDefinition[]; /** * Named toolset tag, resolved via WorkflowManagerOptions.toolsets. Persisted * with the run and re-resolved on resume(). When both `tools` and `toolset` * are given, `tools` wins for this execution and `toolset` is what resumes use. */ toolset?: string; /** Max concurrent agents for this execution. */ concurrency?: number; /** Retry attempts after recoverable agent failures for this execution. */ agentRetries?: number; /** Run-level executor default; call-site agent({ executor }) still wins. */ defaultExecutor?: WorkflowExecutor; /** Optional registry for external executor adapters. */ executorRegistry?: ExecutorRegistry; /** Resolve a checkpoint() question with a human reply (only for UI-bearing runs). */ confirm?: (promptText: string, options: unknown) => Promise; /** Forward workflow runtime instrumentation to manager listeners. */ onRuntimeEvent?: (event: WorkflowRuntimeEvent) => void; /** * Whether this run is eligible for auto-resume when it pauses on a provider * usage limit. Default-on: omit or pass true to stay eligible, pass false to * opt out. Persisted on the run so a cold-start UsageLimitScheduler respects * it too. See usage-limit-scheduler.ts. */ autoResume?: boolean; /** * Seed for the execution's cumulative token counters — passed through to * runWorkflow's WorkflowRunOptions.initialTokenUsage. Only resume() sets * this (from the persisted run's tokenUsage-at-pause), so the resumed * execution's fresh SharedRuntime starts counting from the already-spent * total instead of zero (see A2 in workflow-manager's resume()). */ initialTokenUsage?: { input: number; output: number; total: number; cost: number; cacheRead: number; cacheWrite: number; }; } /** Serializable resume overrides plus host UI hooks for the new execution. */ export interface WorkflowResumeOptions { script?: string; args?: unknown; /** * Resume creates a fresh execution, so an interactive host must reattach its * checkpoint resolver instead of relying on the callback from the old one. */ confirm?: ExecOptions["confirm"]; } export interface WorkflowManagerOptions { cwd?: string; concurrency?: number; /** Resolve a saved-workflow name to its script, enabling nested `workflow('name')`. */ loadSavedWorkflow?: (name: string) => string | undefined; /** Inject a custom agent runner (tests); defaults to a real subagent session. */ agent?: WorkflowAgentRunner; /** Default executor for agent calls that do not specify one. */ defaultExecutor?: WorkflowExecutor; /** Registry used by WorkflowAgent to dispatch external executors. */ executorRegistry?: ExecutorRegistry; /** The session's main model (provider/id), for auto-tiering explore agents. */ mainModel?: string; /** * The host Pi session's model registry. When provided, workflow subagents * resolve models against the same registry as the main session, including * extension-registered providers such as ollama-cloud. */ modelRegistry?: ModelRegistry; /** The pi session id to tag runs with (see setSessionId). */ sessionId?: string; /** Default per-agent timeout when a run does not pass agentTimeoutMs. null means no hard timeout. */ defaultAgentTimeoutMs?: number | null; /** Default retry attempts after recoverable agent failures. */ defaultAgentRetries?: number; /** Default hard token budget when a run does not pass tokenBudget. null/omitted means no budget. */ defaultTokenBudget?: number | null; /** * Named toolsets resolvable by ExecOptions.toolset — e.g. * `{ "web-research": () => [...createCodingTools(cwd), ...createWebTools()] }`. * Called lazily per execution (including on resume). An unknown tag resolves * to the default coding tools. */ toolsets?: Record ToolDefinition[]>; /** * Extra tool NAMES to deny in every subagent session, on top of the always-on * `workflow`/`workflow_control` defaults (see DEFAULT_EXCLUDED_SUBAGENT_TOOLS). * Host wiring passes settings.excludeSubagentTools here so users can also block * other recursive-orchestration tools (#107). */ excludeSubagentTools?: string[]; /** * Persist each subagent transcript as a real pi session file under the * standard sessions directory. Default false (in-memory, discarded). */ persistAgentSessions?: boolean; /** * How many terminal (completed/failed/aborted) runs to retain full * in-memory state for before the oldest is evicted from `runs` (see the * class-level doc comment on that field). Defaults to * DEFAULT_MAX_TERMINAL_RUNS_IN_MEMORY; exposed mainly for tests that want * to observe eviction without creating dozens of runs. */ maxTerminalRunsInMemory?: number; /** * Keep a paused run's lease until its aborted executor settles, and make an * immediate resume wait for that cleanup. Defaults on for every runner so * late provider usage is persisted before resume seeds its cumulative budget. * A custom runner that cannot cooperate with AbortSignal may explicitly opt * out. */ abortCleanupFence?: boolean; } /** Options that a fresh extension generation may safely refresh on a live * manager handed across `/reload`. Execution identity (`cwd`, persistence, * injected agent, and in-memory runs) is intentionally excluded. */ export type WorkflowManagerReloadOptions = Pick< WorkflowManagerOptions, | "concurrency" | "loadSavedWorkflow" | "defaultAgentTimeoutMs" | "defaultAgentRetries" | "defaultTokenBudget" | "defaultExecutor" | "toolsets" | "excludeSubagentTools" | "persistAgentSessions" >; /** * Statuses in which a run's execution has genuinely settled — no promise is * still pending, no lease is still held, nothing will asynchronously mutate * this ManagedRun again. "paused" is deliberately excluded: both a manual * pause() and a usage-limit checkpoint leave the run resumable and, from the * in-memory-retention question's point of view, still "the run the user is * looking at" — only completed/failed/aborted runs are eviction candidates. * See the `runs` field doc comment for the full eviction lifecycle contract. */ const IN_MEMORY_TERMINAL_STATUSES: ReadonlySet = new Set(["completed", "failed", "aborted"]); /** * How many terminal (completed/failed/aborted) runs' full in-memory state * (agents array, journal, snapshot, agentTimestamps) to retain in `runs` * before the oldest is evicted. Kept small: a terminal run's data is fully * on disk (run-persistence.ts) by the time it's eviction-eligible, so the * in-memory copy exists only to serve a `getRun()`/`getSnapshot()` caller * that wants the LIVE object (vs. listRuns()'s persisted view) for a run * that *just* finished — a handful is enough for that; unbounded retention * is exactly the leak this bounds (run-level analog of the subagent * memory-retention mitigation in agent.ts). */ const DEFAULT_MAX_TERMINAL_RUNS_IN_MEMORY = 20; function remapForkJournalRunId(entryRunId: string | undefined, sourceRunId: string, forkRunId: string): string { if (!entryRunId || entryRunId === sourceRunId) return forkRunId; if (entryRunId.startsWith(`${sourceRunId}-nested`)) { return `${forkRunId}${entryRunId.slice(sourceRunId.length)}`; } // Unknown namespaces deliberately remain unmatched rather than replaying into // the wrong frame. They will cache-miss safely if the fork reaches them. return entryRunId; } export class WorkflowManager extends EventEmitter { /** * Lifecycle contract for `runs`: * * - An entry is added when a run starts (startInBackground/runSync) or is * resumed (resume()), always with a live AbortController and (usually) * an active RunLease. * - While status is "running" or "paused", the entry is NEVER evicted — * its execution could still settle (a pending executeRun() promise) or * it is mid-usage-limit-checkpoint/manually-paused and still considered * "the current state of this run" by callers. Eviction only ever * considers an entry AFTER executeRun() has fully settled it to * "completed" | "failed" | "aborted" (see IN_MEMORY_TERMINAL_STATUSES) * and persisted + released its lease — i.e. strictly after the same * isCurrent()-gated persistRun()/releaseRunLease() calls in * executeRun()'s success/catch tails. * - Once terminal, an entry becomes eviction-ELIGIBLE (recordTerminalRun()) * but is not necessarily evicted immediately: up to * maxTerminalRunsInMemory terminal entries are kept, oldest evicted * first, so a `getRun()` call immediately after completion (e.g. the * "complete" event's own synchronous listeners — task-panel's result * delivery, `/workflows watch`) still sees the live object. Once * evicted, the entry is simply removed from `runs`; nothing else reads * or writes it again. * - Every caller of getRun()/getSnapshot() must treat "undefined"/null as * "no live in-memory copy right now" and fall back to listRuns() (backed * by run-persistence.ts, which is what's authoritative for a run once * the in-memory copy is gone) — this mirrors how those callers already * treat any run this process never had in memory (e.g. one started by a * different process and only ever seen via listRuns()). resume() never * depends on `runs` for a run's state either: it always reloads from * persistence, so an evicted runId resumes exactly like one from a * prior process. * - isCurrent(managed) composes with eviction the same way it composes * with resume()/deleteRun() replacing or removing an entry: eviction * removes the map entry outright, so a stale execution's later settle * (isCurrent() check) sees `this.runs.get(runId) !== managed` (in fact * undefined) and correctly no-ops, exactly as it would after * resume()/deleteRun(). */ private runs = new Map(); /** * FIFO of runIds that reached IN_MEMORY_TERMINAL_STATUSES, oldest first — * the eviction order for `runs` (see its doc comment). A runId can appear * more than once (e.g. resumed after eviction, then terminates again); * evicting is idempotent (recordTerminalRun() re-checks the CURRENT status * of the current map entry for that id before deleting), so duplicates * are harmless. */ private terminalRunQueue: string[] = []; private maxTerminalRunsInMemory: number; private persistence: RunPersistence; private cwd: string; private concurrency: number; private loadSavedWorkflow?: (name: string) => string | undefined; private agent?: WorkflowAgentRunner; private defaultExecutor: WorkflowExecutor; private executorRegistry?: ExecutorRegistry; /** The session's main model (provider/id), for auto-tiering explore agents. */ private mainModel?: string; /** The host Pi session's model registry, shared with subagents. */ private modelRegistry?: ModelRegistry; /** The current pi session id; runs are stamped with it and listRuns() filters by it. */ private sessionId?: string; private defaultAgentTimeoutMs: number | null; private defaultAgentRetries: number; private defaultTokenBudget: number | null; private toolsets?: Record ToolDefinition[]>; private excludeSubagentTools?: string[]; private persistAgentSessions: boolean; private abortCleanupFence: boolean; constructor(options: WorkflowManagerOptions = {}) { super(); this.cwd = options.cwd ?? process.cwd(); this.concurrency = options.concurrency ?? 8; this.loadSavedWorkflow = options.loadSavedWorkflow; this.agent = options.agent; this.defaultExecutor = options.defaultExecutor ?? "pi"; this.executorRegistry = options.executorRegistry; this.mainModel = options.mainModel; this.modelRegistry = options.modelRegistry; this.sessionId = options.sessionId; this.defaultAgentTimeoutMs = options.defaultAgentTimeoutMs ?? null; this.defaultAgentRetries = options.defaultAgentRetries ?? 0; this.defaultTokenBudget = options.defaultTokenBudget ?? null; this.toolsets = options.toolsets; this.excludeSubagentTools = options.excludeSubagentTools; this.persistAgentSessions = options.persistAgentSessions ?? false; this.abortCleanupFence = options.abortCleanupFence ?? true; this.maxTerminalRunsInMemory = options.maxTerminalRunsInMemory ?? DEFAULT_MAX_TERMINAL_RUNS_IN_MEMORY; this.persistence = createRunPersistence(this.cwd); this.recoverStaleRuns(); } /** Bind the manager to the current pi session, so new runs are tagged with it and * the navigator/task-panel show only this session's runs (set on session_start). */ setSessionId(id: string | undefined): void { this.sessionId = id; } /** * On startup, any persisted run still marked "running" belongs to a process * that died mid-run (this fresh manager has it nowhere in memory). Reconcile it * to "paused" — never "failed" — so its journal is preserved and resume() can * replay the completed prefix and finish the rest. */ private recoverStaleRuns(): void { try { for (const p of this.listAllRuns()) { if (p.status === "running" && !this.runs.has(p.runId)) { const lease = this.persistence.acquireRunLease(p.runId); if (!lease) continue; try { this.persistence.save({ ...p, status: "paused" }); } finally { this.persistence.releaseRunLease(lease); } } } } catch { // Recovery is best-effort; never let it block manager construction. } } /** * Refresh host configuration after Pi reloads the extension while retaining * this manager's live runs, controllers, leases, and event listeners. * Existing executions keep the options they captured at start; subsequent * runs and resumes use these refreshed defaults. */ reconfigureAfterReload(options: WorkflowManagerReloadOptions): void { this.concurrency = options.concurrency ?? 8; this.loadSavedWorkflow = options.loadSavedWorkflow; this.defaultAgentTimeoutMs = options.defaultAgentTimeoutMs ?? null; this.defaultAgentRetries = options.defaultAgentRetries ?? 0; this.defaultTokenBudget = options.defaultTokenBudget ?? null; this.defaultExecutor = options.defaultExecutor ?? "pi"; this.toolsets = options.toolsets; this.excludeSubagentTools = options.excludeSubagentTools; this.persistAgentSessions = options.persistAgentSessions ?? false; } /** Set the session's main model (provider/id). Used to auto-tier explore agents. */ setMainModel(spec: string | undefined): void { this.mainModel = spec; } /** Set the host session's model registry so subagents resolve models consistently. */ setModelRegistry(registry: ModelRegistry): void { this.modelRegistry = registry; } /** * Expose the host session's model registry to integrations sharing this * manager. Workflow execution reads the same registry internally. */ getModelRegistry(): ModelRegistry | undefined { return this.modelRegistry; } /** * Start a workflow in the background. * Returns immediately with a run ID; the workflow executes asynchronously. */ startInBackground( script: string, args?: unknown, exec: ExecOptions = {}, ): { runId: string; promise: Promise } { const parsed = parseWorkflowScript(script); const slug = parsed.meta.name ? parsed.meta.name .toLowerCase() .replace(/[^a-z0-9]+/g, "-") .replace(/^-+|-+$/g, "") .slice(0, 40) || "workflow" : ""; const runId = slug ? `${slug}-${generateRunId()}` : generateRunId(); const controller = new AbortController(); const lease = this.persistence.acquireRunLease(runId); if (!lease) throw new Error(`Could not acquire workflow run lease for ${runId}`); const managed: ManagedRun = { runId, status: "running", snapshot: { name: parsed.meta.name, description: parsed.meta.description, executor: exec.defaultExecutor ?? this.defaultExecutor, phases: parsed.meta.phases?.map((p) => p.title) ?? [], logs: [], agents: [], agentCount: 0, runningCount: 0, doneCount: 0, errorCount: 0, }, controller, startedAt: new Date(), script, args, journal: [], background: true, lease, autoResume: exec.autoResume, // Resolve the budget once at start and freeze it on the run (see // ManagedRun.tokenBudget) so resume keeps start-time semantics. tokenBudget: exec.tokenBudget !== undefined ? exec.tokenBudget : this.defaultTokenBudget, toolset: exec.toolset, // Same freeze-at-start pattern as tokenBudget, for the same reason: a // resumed run must keep these values, not re-resolve against the // manager's current defaults (see ManagedRun doc comments). maxAgents: exec.maxAgents, agentTimeoutMs: exec.agentTimeoutMs !== undefined ? exec.agentTimeoutMs : this.defaultAgentTimeoutMs, concurrency: exec.concurrency !== undefined ? exec.concurrency : this.concurrency, agentRetries: exec.agentRetries !== undefined ? exec.agentRetries : this.defaultAgentRetries, defaultExecutor: exec.defaultExecutor ?? this.defaultExecutor, agentTimestamps: new Map(), agentsById: new Map(), }; this.runs.set(runId, managed); try { // Persist initial state this.persistence.save({ runId, workflowName: parsed.meta.name, script, args, sessionId: this.sessionId, status: "running", phases: managed.snapshot.phases, agents: [], logs: [], startedAt: managed.startedAt.toISOString(), updatedAt: managed.startedAt.toISOString(), autoResume: managed.autoResume, tokenBudget: managed.tokenBudget, toolset: managed.toolset, maxAgents: managed.maxAgents, agentTimeoutMs: managed.agentTimeoutMs, concurrency: managed.concurrency, agentRetries: managed.agentRetries, defaultExecutor: managed.defaultExecutor, }); } catch (err) { this.releaseRunLease(managed); this.runs.delete(runId); throw err; } // Run workflow asynchronously. // Attach a side-channel catch to prevent Node.js unhandled-rejection crashes // when a workflow is aborted/paused/stopped — executeRun()'s catch block // already records status/event/persist, but the promise still rejects. // The original promise is returned so callers can await it in try/catch. const promise = this.executeRun(managed, script, args, exec); managed.executionPromise = promise; promise.catch(() => {}); return { runId, promise }; } /** * Execute a workflow synchronously (blocking) while still tracking it like a * background run, so the `/workflows` navigator and the live task panel see it. * `onProgress` fires on every progress event with the current snapshot, letting * a caller (e.g. the workflow tool) drive its own inline display. */ async runSync(script: string, args?: unknown, exec: ExecOptions = {}): Promise { const managed = this.createManaged(script, args); const lease = this.persistence.acquireRunLease(managed.runId); if (!lease) throw new Error(`Could not acquire workflow run lease for ${managed.runId}`); managed.lease = lease; managed.autoResume = exec.autoResume; managed.tokenBudget = exec.tokenBudget !== undefined ? exec.tokenBudget : this.defaultTokenBudget; managed.toolset = exec.toolset; // Same freeze-at-start pattern as tokenBudget (see startInBackground/ManagedRun). managed.maxAgents = exec.maxAgents; managed.agentTimeoutMs = exec.agentTimeoutMs !== undefined ? exec.agentTimeoutMs : this.defaultAgentTimeoutMs; managed.concurrency = exec.concurrency !== undefined ? exec.concurrency : this.concurrency; managed.agentRetries = exec.agentRetries !== undefined ? exec.agentRetries : this.defaultAgentRetries; managed.defaultExecutor = exec.defaultExecutor ?? this.defaultExecutor; managed.snapshot.executor = managed.defaultExecutor; this.runs.set(managed.runId, managed); // Persist the initial state immediately so listRuns()/the task panel can see // the run the moment it starts, not only after the first agent journals. this.persistRun(managed); const promise = this.executeRun(managed, script, args, exec); managed.executionPromise = promise; return promise; } /** Build a fresh managed run with an empty snapshot. */ private createManaged(script: string, args?: unknown): ManagedRun { const parsed = parseWorkflowScript(script); const slug = parsed.meta.name ? parsed.meta.name .toLowerCase() .replace(/[^a-z0-9]+/g, "-") .replace(/^-+|-+$/g, "") .slice(0, 40) || "workflow" : ""; const runId = slug ? `${slug}-${generateRunId()}` : generateRunId(); return { runId, status: "running", snapshot: { name: parsed.meta.name, description: parsed.meta.description, phases: parsed.meta.phases?.map((p) => p.title) ?? [], logs: [], agents: [], agentCount: 0, runningCount: 0, doneCount: 0, errorCount: 0, }, controller: new AbortController(), startedAt: new Date(), script, args, journal: [], background: false, defaultExecutor: this.defaultExecutor, agentTimestamps: new Map(), agentsById: new Map(), }; } private async executeRun( managed: ManagedRun, script: string, args?: unknown, exec: ExecOptions = {}, ): Promise { const { resumeJournal, maxAgents, agentTimeoutMs, externalSignal, onProgress, tokenBudget, concurrency, agentRetries, confirm, tools, initialTokenUsage, onRuntimeEvent, executorRegistry, } = exec; // maxAgents/agentTimeoutMs/concurrency/agentRetries were resolved (per-run // value, else the manager default at the time) and frozen on the managed // run at start/resume (see ManagedRun doc comments) — read them from there // first, exactly like resolvedTokenBudget below, so a resumed run keeps the // values it started with instead of re-resolving against the manager's // CURRENT defaults. The exec.* fallbacks are a safety net for direct // executeRun callers that skipped the start paths (same rationale as // resolvedTokenBudget's tokenBudget fallback). const resolvedMaxAgents = managed.maxAgents !== undefined ? managed.maxAgents : maxAgents; const resolvedAgentTimeoutMs = managed.agentTimeoutMs !== undefined ? managed.agentTimeoutMs : agentTimeoutMs !== undefined ? agentTimeoutMs : this.defaultAgentTimeoutMs; const resolvedConcurrency = managed.concurrency !== undefined ? managed.concurrency : (concurrency ?? this.concurrency); const resolvedAgentRetries = managed.agentRetries !== undefined ? managed.agentRetries : (agentRetries ?? this.defaultAgentRetries); // The budget was resolved (per-run value, else defaultTokenBudget) and frozen // on the managed run at start/resume — read it from there so a resumed run // keeps the budget it started with. exec.tokenBudget is a safety net for // direct executeRun callers that skipped the start paths. const resolvedTokenBudget = managed.tokenBudget !== undefined ? managed.tokenBudget : (tokenBudget ?? null); // Explicit tools win for this execution; else re-resolve the run's persisted // toolset tag (how a resumed /deep-research keeps its web tools); else the // agent layer's default coding tools. const resolvedTools = tools ?? (managed.toolset ? this.toolsets?.[managed.toolset]?.() : undefined); // Gated the same way as this.emitLive() below (see isCurrent()) — a stale // execution's progress callback would otherwise keep driving live UI // (task panel, etc.) for a run that's been superseded or deleted. const progress = () => { if (this.isCurrent(managed)) onProgress?.(managed.snapshot); }; // Let a host abort (e.g. Esc during a blocking tool call) cancel this run. if (externalSignal) { if (externalSignal.aborted) managed.controller.abort(); else externalSignal.addEventListener("abort", () => managed.controller.abort(), { once: true }); } try { const result = await runWorkflow(script, { cwd: this.cwd, args, // Use the managed run's persisted id as the workflow runId so the value // returned in result.runId matches the id that listRuns()/resume() use. // Otherwise runWorkflow mints an ephemeral `run-` id and the sync // path would surface a non-resumable id to the model. runId: managed.runId, agent: this.agent, defaultExecutor: managed.defaultExecutor ?? this.defaultExecutor, executorRegistry: executorRegistry ?? this.executorRegistry, mainModel: this.mainModel, modelRegistry: this.modelRegistry, persistAgentSessions: this.persistAgentSessions, signal: managed.controller.signal, concurrency: resolvedConcurrency, agentRetries: resolvedAgentRetries, maxAgents: resolvedMaxAgents, agentTimeoutMs: resolvedAgentTimeoutMs, tokenBudget: resolvedTokenBudget, tools: resolvedTools, excludeTools: this.excludeSubagentTools, confirm, loadSavedWorkflow: this.loadSavedWorkflow, resumeJournal, resumeFromRunId: resumeJournal ? managed.runId : undefined, // Seed the fresh SharedRuntime's spend counter from the persisted total // (resume()) so the hard tokenBudget cap holds cumulatively across a // pause/resume cycle instead of resetting to zero each time (see A2 — // runWorkflow only applies this on the fresh-SharedRuntime branch, never // overriding an inherited options.sharedRuntime from a nested workflow()). initialTokenUsage, // Retried-attempt spend (see WorkflowRunOptions.onRetrySpend and A2): // recordTokens() in workflow.ts already folded this into // shared.spent/tokenUsage, but onAgentEnd never sees a retried // (non-final) attempt — fold it into the same persisted aggregate here // so a run paused after a retry doesn't under-count against the budget. onRetrySpend: ({ tokens, tokenUsage }) => { this.accumulateTokenUsage(managed, tokens, tokenUsage); }, onAgentJournal: (entry) => { // Append (crash-safe-ish): keep the latest entry per (runId, index) // pair, then persist. Matching on index ALONE would let a nested // workflow()'s callIndex-0 entry evict the parent's own // callIndex-0 entry (and vice versa) — they're only distinguished // by runId (see JournalEntry.runId). This is the high-frequency // progress persist (fires once per completed agent, can burst // under concurrency) — throttled (trailing edge). Every // lifecycle-critical persist below (status transitions, run end, // pause/resume/stop) still calls persistRun() directly and flushes this. managed.journal = managed.journal.filter((e) => !(e.index === entry.index && e.runId === entry.runId)); managed.journal.push(entry); this.schedulePersist(managed); }, onLog: (message) => { managed.snapshot.logs.push(message); this.emitLive(managed, "log", { runId: managed.runId, message }); progress(); }, onPhase: (title) => { managed.snapshot.currentPhase = title; if (!managed.snapshot.phases.includes(title)) { managed.snapshot.phases.push(title); } this.emitLive(managed, "phase", { runId: managed.runId, title }); progress(); }, onRuntimeEvent: (event) => { onRuntimeEvent?.(event); this.emitLive(managed, "runtimeEvent", { runId: managed.runId, ...event }); progress(); }, onAgentStart: (event) => { const id = managed.snapshot.agents.length + 1; const agentSnapshot: WorkflowAgentSnapshot = { id, callId: event.id, label: event.label, phase: event.phase, prompt: event.prompt, status: "running", model: event.model, executor: event.executor, required: event.required, }; managed.snapshot.agents.push(agentSnapshot); // Index by the call's unique id (never label — see agentsById's doc // comment) so onAgentEnd/onAgentHistory can resolve back to exactly // THIS entry even when a concurrent sibling shares its label. managed.agentsById.set(event.id, agentSnapshot); // Real per-agent start time, captured the moment the agent actually // starts (not the run's startedAt) — see agentTimestamps. managed.agentTimestamps.set(id, { startedAt: new Date().toISOString() }); this.emitLive(managed, "agentStart", { runId: managed.runId, ...event }); progress(); }, onAgentEnd: (event) => { const agent = managed.agentsById.get(event.id); if (agent) { agent.status = event.result === null ? "error" : "done"; // Keep the full value for the interactive pager; compact surfaces // continue to use resultPreview. agent.result = event.result; agent.resultPreview = preview(event.result); agent.error = event.error; agent.errorCode = event.errorCode; agent.recoverable = event.recoverable; agent.tokens = event.tokens; if (event.tokenUsage) agent.tokenUsage = event.tokenUsage; if (event.model) agent.model = event.model; if (event.executor) agent.executor = event.executor; if (event.required !== undefined) agent.required = event.required; // Real per-agent end time — only terminal agents get one; a still- // running agent's entry keeps endedAt undefined. const ts = managed.agentTimestamps.get(agent.id); if (ts) ts.endedAt = new Date().toISOString(); } // Progressive run-wide token aggregate (A2): workflow.ts's onTokenUsage // callback below fires exactly once, only when the whole script finishes // successfully (a deliberate, tested contract — see // "agent() accumulates usage across multiple agents" in agent.test.ts, // which asserts one final event, not one per agent). A run that // pauses/aborts/fails mid-flight never reaches it, so without tracking // it here too, a paused run's persisted tokenUsage would stay whatever // it was (usually unset) — starving resume()'s spend-seeding of the // very data it needs. Accumulate additively from every onAgentEnd // instead: a cache-hit replay reports tokens: 0 (see agent()'s replay // branch in workflow.ts), so replaying the unchanged prefix on resume // is a no-op add here, matching the "already historically spent, don't // double-count" semantics of journal replay. this.accumulateTokenUsage(managed, event.tokens ?? 0, event.tokenUsage); this.emitLive(managed, "agentEnd", { runId: managed.runId, ...event }); progress(); }, onAgentHistory: (event) => { const agent = managed.agentsById.get(event.id); if (agent) { agent.history = event.history; } this.emitLive(managed, "agentHistory", { runId: managed.runId, agentId: agent?.id, ...event }); progress(); }, onTokenUsage: (usage) => { managed.snapshot.tokenUsage = usage; this.emitLive(managed, "tokenUsage", { runId: managed.runId, usage }); progress(); }, }); result.forkedFromRunId = managed.forkedFromRunId; result.rootRunId = managed.rootRunId; managed.status = "completed"; managed.result = result; // Gated the same way as disk/lease below (see emitLive()): a stale // execution's "complete" would otherwise still deliver a result for a // run that's been superseded or deleted (e.g. background result // delivery into the conversation) even though it's no longer current. this.emitLive(managed, "complete", { runId: managed.runId, result }); // Persist final state. persistRun()/writeRunToDisk() already no-op if // `managed` has been superseded (resume()/deleteRun() took over this // runId) — see isCurrent(). Guard the lease release the same way: a // stale execution settling after resume() has already acquired a NEW // lease for this runId must not touch that newer lease's bookkeeping. this.persistRun(managed); if (this.isCurrent(managed)) { this.releaseRunLease(managed); // Now (and only now — after the run's data is safely on disk and its // lease released) does this run become eviction-eligible; see the // `runs` field doc comment. this.recordTerminalRun(managed.runId); } return result; } catch (error) { const workflowError = error instanceof WorkflowError ? error : new WorkflowError( error instanceof Error ? error.message : String(error), WorkflowErrorCode.WORKFLOW_ABORTED, { recoverable: true }, ); const usageLimitPaused = !managed.controller.signal.aborted && isProviderUsageLimit(workflowError); const lifecycleAbort = managed.controller.signal.aborted && (managed.status === "paused" || managed.status === "aborted"); if (managed.controller.signal.aborted) { // Intentional abort (pause/stop/Esc) — preserve status set by pause()/stop() if (managed.status === "running") { managed.status = "aborted"; } } else if (usageLimitPaused) { // Provider quota/usage limit: NOT a failure. Checkpoint the run as paused so // the persisted journal (completed agent results) is replayed by resume() // once the budget refills — instead of the user starting from scratch. managed.status = "paused"; } else { managed.status = "failed"; } managed.error = workflowError; // Both branches gated via emitLive() (see its doc comment) — a stale // execution's "paused"/"error" is equally misleading once superseded. if (usageLimitPaused) { this.emitLive(managed, "paused", { runId: managed.runId, reason: "usage_limit", error: workflowError, resetHint: workflowError.resetHint, }); } else if (!lifecycleAbort && this.listenerCount("error") > 0) { // Manual pause/stop is an intentional lifecycle transition, not a run // failure notification. External host aborts still emit the error. // Guarded: EventEmitter throws on an unlistened "error" emit, which // would abort this catch block mid-way — skipping the final persist, // the lease release, and the real error rethrow below. this.emitLive(managed, "error", { runId: managed.runId, error: workflowError }); } // Persist final state (see the success-path comment above for the // isCurrent() rationale — same guard, same reason). this.persistRun(managed); if (this.isCurrent(managed)) { this.releaseRunLease(managed); // "paused" (manual pause() or a usage-limit checkpoint) is // deliberately NOT eviction-eligible — only a genuinely settled // terminal status is (see IN_MEMORY_TERMINAL_STATUSES / the `runs` // field doc comment). recordTerminalRun() itself re-checks this too, // but skip the call entirely here so a paused run never even enters // the eviction queue. if (IN_MEMORY_TERMINAL_STATUSES.has(managed.status)) this.recordTerminalRun(managed.runId); } throw workflowError; } } /** * True when `managed` is still the live, current entry for its runId in * `this.runs` — false once resume() has replaced it with a new ManagedRun * object for the same runId, or deleteRun() has removed it entirely. A * superseded ManagedRun's async completion (executeRun's promise settling * well after something else already took over or tore down that runId) * must not write to disk or touch lease state on the newer execution's * behalf — see writeRunToDisk() and executeRun()'s post-await persist calls. */ private isCurrent(managed: ManagedRun): boolean { return this.runs.get(managed.runId) === managed; } /** * Emit an event on behalf of `managed`, but only while it's still the * current entry for its runId (see isCurrent()) — mirrors the disk/lease * guard for the observer-facing side of the same problem. A superseded * execution's progress/terminal events (log, phase, agentStart/End, * tokenUsage, complete, error, paused) are not just stale-but-harmless: * "complete" in particular can drive background result delivery into the * conversation, so letting a deleted/superseded run's stale settle still * fire it would deliver a result for a run that, from the caller's POV, no * longer exists (or has since been superseded by a newer execution whose * own events already tell the true story). No event in this set has a * legitimate reason to still reach listeners once superseded — unlike * disk writes there's no "expected race, harmless no-op" nuance here, it's * simply wrong to notify twice (or for a run that's gone). Events emitted * directly by pause()/stop()/resume()/deleteRun() themselves are NOT routed * through this helper — those methods own the transition and ARE current * at the moment they fire, same precedent as their persist/lease calls. */ private emitLive(managed: ManagedRun, event: string, payload: unknown): void { if (this.isCurrent(managed)) this.emit(event, payload); } /** * Mark `runId` as eviction-eligible now that its execution has genuinely * settled to a terminal status (completed/failed/aborted — see * IN_MEMORY_TERMINAL_STATUSES), and evict the oldest eligible entries * beyond maxTerminalRunsInMemory. Callers must only invoke this after the * same isCurrent()-gated persistRun()/releaseRunLease() sequence executeRun() * already uses (see the `runs` field doc comment for the full contract) — * this method itself re-validates the CURRENT entry's status before * deleting anything, so it never evicts a run that isn't (or is no longer) * genuinely terminal, including one resumed back to "running" after being * queued here but before its turn to be evicted came up. */ private recordTerminalRun(runId: string): void { this.terminalRunQueue.push(runId); while (this.terminalRunQueue.length > this.maxTerminalRunsInMemory) { const oldest = this.terminalRunQueue.shift(); if (oldest === undefined) break; const current = this.runs.get(oldest); // Re-check the CURRENT entry for this id (not the ManagedRun object // that was terminal when queued) — resume() may have since replaced // it with a fresh, live execution, which must never be evicted here. if (current && IN_MEMORY_TERMINAL_STATUSES.has(current.status)) { this.runs.delete(oldest); } } } /** * Additively fold one agent-call's token cost into the run-wide persisted * aggregate (managed.snapshot.tokenUsage), seeded (on resume) from the * persisted total-at-pause — see A2. Shared by onAgentEnd (a completed or * finally-failed agent call) and onRetrySpend (a failed attempt that WILL * be retried, whose cost recordTokens() already folded into * shared.spent/tokenUsage in workflow.ts, but which onAgentEnd never sees — * see WorkflowRunOptions.onRetrySpend for why that needs its own channel). */ private accumulateTokenUsage( managed: ManagedRun, tokens: number, tokenUsage?: { input: number; output: number; cost: number; cacheRead: number; cacheWrite: number }, ): void { const prior = managed.snapshot.tokenUsage; const usage = { input: prior?.input ?? 0, output: prior?.output ?? 0, total: prior?.total ?? 0, cost: prior?.cost ?? 0, cacheRead: prior?.cacheRead ?? 0, cacheWrite: prior?.cacheWrite ?? 0, }; usage.total += tokens; if (tokenUsage) { usage.input += tokenUsage.input; usage.output += tokenUsage.output; usage.cost += tokenUsage.cost; usage.cacheRead += tokenUsage.cacheRead; usage.cacheWrite += tokenUsage.cacheWrite; } managed.snapshot.tokenUsage = usage; } private releaseRunLease(managed: ManagedRun): void { if (!managed.lease) return; this.persistence.releaseRunLease(managed.lease); managed.lease = undefined; } /** Trailing-edge throttle window for high-frequency progress persists (see schedulePersist). */ private static readonly PERSIST_THROTTLE_MS = 400; /** Pending trailing-edge persist timers for high-frequency progress events, keyed by runId. */ private persistTimers = new Map>(); /** * Coalesce rapid progress persists (currently: onAgentJournal, which fires * once per completed agent and can burst under concurrency) to at most one * disk write per PERSIST_THROTTLE_MS (trailing edge) instead of one write * per tick — persistRun() does a full JSON.stringify of the run plus up to * 3 sync writes, so firing it once per agent in a long run is O(N^2). * * Lifecycle-critical writes (status transitions, run end, pause/resume/stop) * must NOT use this — call persistRun() directly, which flushes (and cancels) * any pending timer first so a stale trailing write can never fire after, and * resurrect, a terminal state. */ private schedulePersist(managed: ManagedRun): void { if (this.persistTimers.has(managed.runId)) return; // already scheduled; the trailing write reads live state const timer = setTimeout(() => { this.persistTimers.delete(managed.runId); this.writeRunToDisk(managed); }, WorkflowManager.PERSIST_THROTTLE_MS); // A pending progress persist should never keep the process alive on its own. timer.unref?.(); this.persistTimers.set(managed.runId, timer); } /** * Persist immediately and synchronously. Cancels any pending throttled write * for this run first, so the write that lands is always the caller's current * (final) state — never superseded by a stale deferred write. Use this for * every lifecycle-critical persist: run start, status transitions, run end, * pause()/resume()/stop(). */ private persistRun(managed: ManagedRun, options: { throwOnError?: boolean } = {}): boolean { // A superseded execution's persist call must not touch the CURRENT // execution's pending-timer bookkeeping for this runId (see isCurrent()). // writeRunToDisk() below re-checks this too (it's the sole choke point // schedulePersist()'s deferred timer also funnels through), so this is a // belt-and-suspenders early-out specifically for the timer-clearing side // effect, which writeRunToDisk() alone wouldn't prevent. if (!this.isCurrent(managed)) return false; const timer = this.persistTimers.get(managed.runId); if (timer) { clearTimeout(timer); this.persistTimers.delete(managed.runId); } return this.writeRunToDisk(managed, options); } private writeRunToDisk(managed: ManagedRun, options: { throwOnError?: boolean } = {}): boolean { // The sole choke point for every disk write (both persistRun()'s direct // calls and schedulePersist()'s deferred timer funnel through here) — skip // silently when `managed` is no longer the current entry for its runId // (see isCurrent()). This is an expected race outcome (resume() replaced // it, or deleteRun() removed it), not an error: writing anyway would // resurrect a torn-down run's file, or clobber a newer execution's // in-progress/completed state with this stale one's. // // This check is redundant with persistRun()'s own early-return for every // CURRENT call site — it earns its keep solely for schedulePersist()'s // deferred setTimeout callback, the one path into this method that skips // persistRun() entirely. That callback only fires from onAgentJournal, and // onAgentJournal only fires for a call that got PAST agent()'s // throwIfAborted() check (see workflow.ts) — which, since run-fatal abort // (SharedRuntime.runFatalController) now seals every top-level run's // shared runtime the instant any error escapes it uncaught, means a // genuinely superseded-but-never-aborted execution (the only kind that // could previously still journal a stray call after resume() replaced it) // is structurally impossible to construct anymore — see the "unreachable // defense-in-depth (#2)" test in workflow-manager.test.ts for the worked // example and its own note. This check is KEPT anyway: it costs nothing, // and removing it would silently reopen a stale-write path the moment any // future change (e.g. a new way to journal without throwIfAborted()'s // gate) reintroduces a producer for it. if (!this.isCurrent(managed)) return false; try { // Paused/failed runs resume in place and completed runs may be forked into // immutable revisions, so all three retain their replay journal. Aborted // runs remain terminal/non-replayable. Persist exactly one full copy of each // successful agent result in the journal rather than duplicating agents[].result. const keepsResumeJournal = managed.status !== "aborted"; this.persistence.save({ runId: managed.runId, workflowName: managed.snapshot.name, description: managed.snapshot.description, // Persist the real script + journal so the run can be resumed. Runs live // in workflow run storage — protect via directory permissions, not blanking. script: managed.script, args: managed.args, sessionId: this.sessionId, forkedFromRunId: managed.forkedFromRunId, rootRunId: managed.rootRunId, journal: keepsResumeJournal ? managed.journal : undefined, status: managed.status, error: managed.error ? { message: managed.error.message, code: managed.error.code, recoverable: managed.error.recoverable, } : undefined, // Persisted every write (not just at pause) so a stale read during the // "paused" event race (see UsageLimitScheduler) is still correct — this // is fixed at run-start and doesn't change over the run's lifetime. autoResume: managed.autoResume, // Start-time execution context, re-read by resume() (see ManagedRun). tokenBudget: managed.tokenBudget, toolset: managed.toolset, maxAgents: managed.maxAgents, agentTimeoutMs: managed.agentTimeoutMs, concurrency: managed.concurrency, agentRetries: managed.agentRetries, defaultExecutor: managed.defaultExecutor, // Why a usage-limit pause happened, so the navigator / a future cold start // can show it and (eventually) re-arm resume after the budget refills. pauseReason: managed.status === "paused" && isProviderUsageLimit(managed.error) ? "usage_limit" : undefined, resetHint: managed.status === "paused" && isProviderUsageLimit(managed.error) ? managed.error.resetHint : undefined, phases: managed.snapshot.phases, currentPhase: managed.snapshot.currentPhase, // Real per-agent timestamps only (see agentTimestamps) — never the run's // own startedAt or "now" stamped onto every agent on every write. A // still-running agent is persisted with no endedAt. agents: managed.snapshot.agents.map((a) => { const { result, ...summary } = a; const ts = managed.agentTimestamps.get(a.id); return { ...summary, // Live runs keep the rich value in memory. Cold resumable runs use // the journal and retain resultPreview until replay reconstructs it. ...(keepsResumeJournal || result === undefined ? {} : { result }), startedAt: ts?.startedAt, endedAt: ts?.endedAt, }; }), logs: managed.snapshot.logs, result: managed.result?.result, tokenUsage: managed.snapshot.tokenUsage ? { input: managed.snapshot.tokenUsage.input, output: managed.snapshot.tokenUsage.output, total: managed.snapshot.tokenUsage.total, cost: managed.snapshot.tokenUsage.cost, cacheRead: managed.snapshot.tokenUsage.cacheRead, cacheWrite: managed.snapshot.tokenUsage.cacheWrite, } : undefined, startedAt: managed.startedAt.toISOString(), updatedAt: new Date().toISOString(), completedAt: managed.status === "completed" ? new Date().toISOString() : undefined, durationMs: managed.result?.durationMs, }); return true; } catch (err) { if (options.throwOnError) throw err; // Persistence is best-effort: the run is still healthy in memory. // Log so an operator debugging state-loss has a lead, but never crash // the workflow over a disk-full situation. console.warn("[workflow-manager] Persist run failed:", err); return false; } } /** * Pause a running workflow. */ pause(runId: string): boolean { const managed = this.runs.get(runId); if (managed?.status !== "running") return false; managed.controller.abort(); managed.status = "paused"; this.emit("paused", { runId }); this.persistRun(managed); // Built-in/opted-in runners keep the lease until executeRun's catch tail has // actually drained the aborted executor. Releasing it here would let an // immediate resume overlap with a child process that is still terminating. if (!this.abortCleanupFence) this.releaseRunLease(managed); return true; } /** * Resume an interrupted run: replay journaled results for the unchanged prefix * and run the rest live. Returns false if there is nothing resumable. * * `opts.script` lets the orchestrating model resume with an EDITED script * (cached-prefix reuse / iteration): unchanged agent() calls whose content * hash still matches the journal entry at their positional callIndex replay * from cache, while the first changed or newly inserted call — and everything * after it — re-runs live. When `opts.script` is omitted, resume behaves * exactly as before and uses the persisted script (auto-resume, TUI resume); * this keeps the existing single-arg `resume(runId)` callers (e.g. the * UsageLimitScheduler) unchanged. `opts.args` overrides the persisted args * only when provided; otherwise the persisted args are kept. */ async resume(runId: string, opts?: WorkflowResumeOptions): Promise { // Guard: refuse to resume a run that is already running, or one that was // intentionally aborted (pause/stop/Esc). Paused and failed runs can restart. const active = this.runs.get(runId); if (active?.status === "running") return false; if (active?.status === "aborted") return false; // A manual pause flips status synchronously while the aborted execution may // still be draining its provider process and reporting partial usage. Do not // replace the ManagedRun or seed a new budget until that telemetry is safely // persisted and executeRun releases the original lease. Custom/injected // runners can explicitly disable this fence when they do not cooperate with // AbortSignal; built-in CLI executors always keep it enabled. if (this.abortCleanupFence && active?.status === "paused") { if (active.executionPromise) { await active.executionPromise.catch(() => {}); } else if (active.lease) { // Compatibility fallback for a live run created before executionPromise // was tracked. Current start/resume/fork paths always take the branch // above. const released = await this.waitForAbortCleanup(active); if (!released) return false; } } const persisted = this.persistence.load(runId); if (!persisted?.script || persisted.status === "completed" || persisted.status === "aborted") return false; const lease = this.persistence.acquireRunLease(runId); if (!lease) return false; // Use the edited script when supplied, else the persisted one (backward-compat). const script = opts?.script ?? persisted.script; const args = opts?.args !== undefined ? opts.args : persisted.args; // Normalize the persisted total-at-pause once: PersistedRunState.tokenUsage // has optional cost/cacheRead/cacheWrite (legacy runs may lack them), but // both the seeded snapshot and initialTokenUsage need concrete numbers. const priorTokenUsage = persisted.tokenUsage ? { input: persisted.tokenUsage.input, output: persisted.tokenUsage.output, total: persisted.tokenUsage.total, cost: persisted.tokenUsage.cost ?? 0, cacheRead: persisted.tokenUsage.cacheRead ?? 0, cacheWrite: persisted.tokenUsage.cacheWrite ?? 0, } : undefined; const controller = new AbortController(); const managed: ManagedRun = { runId, status: "running", snapshot: { name: persisted.workflowName, description: persisted.description, executor: persisted.defaultExecutor ?? "pi", phases: persisted.phases ?? [], logs: persisted.logs ?? [], agents: [], agentCount: 0, runningCount: 0, doneCount: 0, errorCount: 0, // Seed the live snapshot's aggregate from the persisted total-at-pause // (see A2) so a pause that lands before this resume's first agent // completes doesn't lose the prior spend — onAgentEnd accumulates on // top of this rather than starting from scratch. tokenUsage: priorTokenUsage, }, controller, startedAt: new Date(), // The (possibly edited) script + args become the run's own — persistRun() // writes them below, so a later resume of this run sees the edited script. script, args, forkedFromRunId: persisted.forkedFromRunId, rootRunId: persisted.rootRunId, journal: persisted.journal ?? [], background: true, lease, // Carry the original opt-out forward across resumes; it's fixed at // run-start and persistRun() re-persists it on every subsequent write. autoResume: persisted.autoResume, // Restore start-time execution context: the budget the run started with // (legacy runs without one resume unbudgeted — never re-apply the current // default to a run that predates it) and the toolset tag executeRun // re-resolves so e.g. a resumed /deep-research keeps its web tools. tokenBudget: persisted.tokenBudget !== undefined ? persisted.tokenBudget : null, toolset: persisted.toolset, // Restore the same start-time execution context for the other four // per-run knobs (see ManagedRun doc comments) — same rationale as // tokenBudget: never re-resolve against the manager's CURRENT defaults. // maxAgents: legacy/never-set runs resume with no cap carried forward // (runWorkflow's own MAX_AGENTS_PER_RUN default applies), exactly as if // maxAgents had never been passed at all. maxAgents: persisted.maxAgents, // agentTimeoutMs: unlike tokenBudget, a legacy run's real timeout at // start was never "no timeout" by omission — it was always // this.defaultAgentTimeoutMs, because pre-A1 resume() never threaded // agentTimeoutMs through at all and unconditionally fell back to the // manager default (see executeRun's resolvedAgentTimeoutMs fallback // chain). Falling back to null here would change what a legacy run's // resume actually does versus both its original start AND pre-fix // resume behavior. So — deliberately unlike tokenBudget's null // fallback — legacy runs resume with the manager's CURRENT default, // matching the only semantics such a run ever had. agentTimeoutMs: persisted.agentTimeoutMs !== undefined ? persisted.agentTimeoutMs : this.defaultAgentTimeoutMs, // concurrency/agentRetries have no "explicit opt-out sentinel" the way // tokenBudget's null does — a legacy run without a persisted value falls // back to the manager's current values, matching how this execution // resolved unset concurrency/agentRetries before this fix ever existed. concurrency: persisted.concurrency !== undefined ? persisted.concurrency : this.concurrency, agentRetries: persisted.agentRetries !== undefined ? persisted.agentRetries : this.defaultAgentRetries, defaultExecutor: persisted.defaultExecutor ?? "pi", // Fresh per-resume: agents (and any prior timing) are rebuilt live as // onAgentStart/onAgentEnd fire again for this attempt (see `agents: []` // above); the journal, not this map, is what makes replayed agents cheap. agentTimestamps: new Map(), agentsById: new Map(), }; this.runs.set(runId, managed); // Persist before notifying renderers: listRuns() is their source of truth for // lifecycle status, while getRun() supplies the live in-memory snapshot. this.persistRun(managed); // Namespace by (runId, index) exactly like the live onAgentJournal dedup // above and like SharedStore's deltaKey — see JournalEntry.runId. A // legacy entry persisted before namespacing existed has no `runId`; it is // assumed to belong to this run's own top-level runId (the only frame // that existed before nested workflow() journaling was namespaced), so it // still resume-hits for a top-level call and safely cache-misses (re-runs // live, does not misapply) for what was actually a nested-run entry. const resumeJournal = new Map((persisted.journal ?? []).map((e) => [`${e.runId ?? runId}:${e.index}`, e] as const)); this.emit("resumed", { runId }); // Run in the background; executeRun records status/errors on the managed run. // initialTokenUsage seeds the resumed execution's fresh SharedRuntime.spent // (A2) from the persisted total-at-pause, so the tokenBudget cap holds // cumulatively instead of resetting to zero. Note: shared.agentCount is // deliberately NOT seeded the same way — it doesn't need to be. Unlike // token spend (whose cache-hit replay branch skips recordTokens() to avoid // double-counting already-spent tokens), agent()'s shared.agentCount++ // fires unconditionally for EVERY call, cache-hit or live, before the // replay check runs (see workflow.ts). Because resume() always replays the // whole script from callIndex 0, that replay alone reconstructs the // correct cumulative count inside this fresh SharedRuntime by the time any // new live agent runs — so maxAgents (via A1) is already a genuine // cumulative cap across resume with no extra seeding required. const promise = this.executeRun(managed, script, args, { resumeJournal, initialTokenUsage: priorTokenUsage, confirm: opts?.confirm, }); managed.executionPromise = promise; void promise.catch(() => {}); return true; } private async waitForAbortCleanup(managed: ManagedRun, timeoutMs = 5_000): Promise { const deadline = Date.now() + timeoutMs; while (managed.lease && this.isCurrent(managed) && Date.now() < deadline) { await new Promise((resolveWait) => setTimeout(resolveWait, 20)); } return !managed.lease; } /** * Fork a completed run into a new immutable revision. The source run is never * replaced or mutated; its replay journal seeds a fresh run ID and budget. */ startFork( sourceRunId: string, opts: { script?: string; args?: unknown } = {}, ): { runId: string; promise: Promise } { const source = this.persistence.load(sourceRunId); if (!source) throw new Error(`Cannot fork workflow run "${sourceRunId}": run not found.`); if (source.status !== "completed") { throw new Error( `Cannot fork workflow run "${sourceRunId}": expected completed status, found ${source.status}. ` + "Use resumeFromRunId for paused or failed runs.", ); } if (!source.script) throw new Error(`Cannot fork workflow run "${sourceRunId}": persisted script is missing.`); if (!Array.isArray(source.journal)) { throw new Error( `Cannot fork workflow run "${sourceRunId}": this legacy completed run has no replay journal. ` + "Run the workflow again before creating a revision.", ); } const script = opts.script ?? source.script; const args = opts.args !== undefined ? opts.args : source.args; const managed = this.createManaged(script, args); const runId = managed.runId; const lease = this.persistence.acquireRunLease(runId); if (!lease) throw new Error(`Could not acquire workflow run lease for ${runId}`); managed.background = true; managed.lease = lease; managed.forkedFromRunId = sourceRunId; managed.rootRunId = source.rootRunId ?? sourceRunId; managed.snapshot.forkedFromRunId = sourceRunId; managed.snapshot.rootRunId = managed.rootRunId; managed.autoResume = source.autoResume; managed.tokenBudget = source.tokenBudget !== undefined ? source.tokenBudget : null; managed.toolset = source.toolset; managed.maxAgents = source.maxAgents; managed.agentTimeoutMs = source.agentTimeoutMs !== undefined ? source.agentTimeoutMs : this.defaultAgentTimeoutMs; managed.concurrency = source.concurrency !== undefined ? source.concurrency : this.concurrency; managed.agentRetries = source.agentRetries !== undefined ? source.agentRetries : this.defaultAgentRetries; managed.defaultExecutor = source.defaultExecutor ?? "pi"; managed.snapshot.executor = managed.defaultExecutor; managed.journal = source.journal.map((entry) => ({ ...entry, runId: remapForkJournalRunId(entry.runId, sourceRunId, runId), storeDelta: entry.storeDelta ? { ...entry.storeDelta } : undefined, })); this.runs.set(runId, managed); try { this.persistRun(managed, { throwOnError: true }); } catch (error) { this.releaseRunLease(managed); this.runs.delete(runId); throw error; } const resumeJournal = new Map( managed.journal.map((entry) => [`${entry.runId ?? runId}:${entry.index}`, entry] as const), ); this.emit("forked", { runId, forkedFromRunId: sourceRunId, rootRunId: managed.rootRunId }); const promise = this.executeRun(managed, script, args, { resumeJournal }); managed.executionPromise = promise; promise.catch(() => {}); return { runId, promise }; } /** * Stop a running workflow. * * Fast path: the run is live in this process (`this.runs`) — abort its * controller and persist "aborted" as before. Fallback: the run is not in * memory but is persisted as "running" or "paused" — e.g. it belongs to a * prior pi session that this process's recoverStaleRuns() flipped to * "paused" on disk without repopulating this.runs (see workflow-control-tool's * findRun(), which resolves candidates from disk via listRuns()). There is no * live controller to abort in that case — the run simply isn't executing in * this process — so mark it aborted on disk directly, mirroring resume()'s * persisted-fallback lease handling. */ stop(runId: string): boolean { const managed = this.runs.get(runId); if (managed) { if (managed.status !== "running" && managed.status !== "paused") return false; // Whether this run's OWN executeRun() promise has already fully settled // matters for whether stop() itself must be the one to call // recordTerminalRun(): a usage-limit checkpoint runs executeRun()'s // catch tail to completion before "paused" is ever observable (it // deliberately skipped recordTerminalRun() then, since "paused" isn't // terminal) — so there is no FUTURE tail left that will ever call it // for this managed object. A manual pause() sets "paused" while its // cooperative abort may still be settling; in that narrow window the // tail later settles this object to "aborted" (terminal) and records a // SECOND time — a tolerated duplicate: recordTerminalRun() is // idempotent-safe under duplicates (re-validates the current entry), // the lease was already cleared here, and the worst case is the // stopped run leaving memory earlier than FIFO order (persistence // fallback covers every consumer). A "running" run, by contrast, // always still has that tail pending; // it (not stop()) is what calls recordTerminalRun() once it actually // settles to "aborted" — see the `runs` field doc comment's rule that // eviction eligibility must wait for the real settle, not a request to // abort. Without this, stopping an already-paused run left it in // `runs` forever (no future tail to mark it eviction-eligible) — a // small leak in exactly the class this manager otherwise bounds. const hadNoPendingSettle = managed.status === "paused"; managed.controller.abort(); managed.status = "aborted"; this.emit("stopped", { runId }); this.persistRun(managed); this.releaseRunLease(managed); if (hadNoPendingSettle) this.recordTerminalRun(runId); return true; } const persisted = this.persistence.load(runId); if (!persisted || (persisted.status !== "running" && persisted.status !== "paused")) return false; const lease = this.persistence.acquireRunLease(runId); if (!lease) return false; try { this.persistence.save({ ...persisted, status: "aborted", updatedAt: new Date().toISOString() }); } finally { this.persistence.releaseRunLease(lease); } this.emit("stopped", { runId }); return true; } /** * Get status of a specific run. */ getRun(runId: string): ManagedRun | undefined { return this.runs.get(runId); } /** * List all runs (active + persisted). */ /** * Runs for the navigator/task panel. Once bound to a session (setSessionId), only * that session's runs are returned — runs from other sessions stay on disk and * reappear when you switch back. Unbound (tests/legacy) returns everything. */ listRuns(): PersistedRunState[] { const all = this.persistence.list(); return this.sessionId ? all.filter((r) => r.sessionId === this.sessionId) : all; } /** All persisted runs regardless of session (used by cross-session recovery). */ listAllRuns(): PersistedRunState[] { return this.persistence.list(); } /** * Get snapshot of a run. */ getSnapshot(runId: string): WorkflowSnapshot | null { return this.runs.get(runId)?.snapshot ?? null; } /** * Delete a persisted run. * * If `runId` is still live in this process (running or paused-in-memory), * abort its controller FIRST, before any teardown below — a live run left * un-aborted would otherwise keep executing in the background indefinitely * (burning API calls/tokens/holding a worktree) after its record is gone. * Aborting first, while `managed` is still `this.runs.get(runId)`, costs * nothing extra: the abort signal is fire-and-forget (cooperative — the * execution winds down on its own schedule), so the exact instant we flip * `this.runs`/release the lease/delete files relative to it doesn't matter * for correctness. What DOES matter is that once this method returns, the * aborted execution's eventual settle (executeRun's success/catch path, * asynchronously, possibly much later) must be a harmless no-op rather than * a resurrection — that's what isCurrent() guarantees: `this.runs.delete()` * below means executeRun's later persistRun()/releaseRunLease() calls on * this same `managed` object find `this.runs.get(runId) !== managed` (in * fact `undefined`, since the entry is gone) and skip writing/releasing. */ deleteRun(runId: string): boolean { const managed = this.runs.get(runId); if (managed) { if (!managed.controller.signal.aborted) managed.controller.abort(); this.releaseRunLease(managed); } this.runs.delete(runId); // Cancel any pending throttled write so a deferred persist can't fire after // deletion and resurrect the run's file on disk. const timer = this.persistTimers.get(runId); if (timer) { clearTimeout(timer); this.persistTimers.delete(runId); } return this.persistence.delete(runId); } /** * Get the persistence layer (for saving workflows). */ getPersistence(): RunPersistence { return this.persistence; } }