import * as fs from "node:fs"; import * as path from "node:path"; import { formatDuration, formatModelThinking, formatTokens, shortenPath } from "../../shared/formatters.ts"; import { formatActivityLabel, formatParallelOutcome } from "../../shared/status-format.ts"; import { type ActivityState, type AsyncJobStep, type AsyncParallelGroupStatus, type AsyncStatus, type CostSummary, type NestedRunSummary, type SteeringStatus, type SubagentRunMode, type TokenUsage, type TurnBudgetState } from "../../shared/types.ts"; import type { ResolvedSubagentCapabilityCeiling, SubagentCapabilityAudit } from "../shared/capability-ceiling.ts"; import { readStatus } from "../../shared/utils.ts"; import { attachRootChildrenToSteps, buildNestedRouteIndex, type NestedRoute, projectNestedEvents } from "../shared/nested-events.ts"; import { formatNestedRunStatusLines } from "../shared/nested-render.ts"; import { flatToLogicalStepIndex, normalizeParallelGroups } from "./parallel-groups.ts"; import { contextModeLabel, summarizeContextModes, type ContextMode, type ContextSummary } from "../shared/context-mode.ts"; import { reconcileAsyncRun, reconcileNestedAsyncDescendants } from "./stale-run-reconciler.ts"; import { readProcessTerminal, sanitizeProcessTerminal } from "./process-terminal.ts"; interface AsyncRunStepSummary { index: number; agent: string; context?: ContextMode; label?: string; phase?: string; outputName?: string; structured?: boolean; status: AsyncJobStep["status"]; activityState?: ActivityState; lastActivityAt?: number; currentTool?: string; currentToolArgs?: string; currentToolStartedAt?: number; currentPath?: string; recentTools?: Array<{ tool: string; args: string; endMs: number }>; recentOutput?: string[]; turnCount?: number; toolCount?: number; steering?: SteeringStatus; durationMs?: number; tokens?: TokenUsage; totalCost?: CostSummary; skills?: string[]; model?: string; thinking?: string; attemptedModels?: string[]; sessionFile?: string; transcriptPath?: string; error?: string; timedOut?: boolean; stopped?: boolean; turnBudget?: TurnBudgetState; turnBudgetExceeded?: boolean; wrapUpRequested?: boolean; acceptance?: AsyncJobStep["acceptance"]; agentContract?: AsyncJobStep["agentContract"]; execution?: AsyncJobStep["execution"]; review?: AsyncJobStep["review"]; effects?: AsyncJobStep["effects"]; processTerminal?: AsyncJobStep["processTerminal"]; capabilityCeiling?: ResolvedSubagentCapabilityCeiling; capabilityAudit?: SubagentCapabilityAudit; children?: NestedRunSummary[]; } export interface AsyncRunSummary { id: string; asyncDir: string; sessionId?: string; state: "queued" | "running" | "complete" | "failed" | "paused" | "stopped"; error?: string; activityState?: ActivityState; lastActivityAt?: number; currentTool?: string; currentToolStartedAt?: number; currentPath?: string; turnCount?: number; toolCount?: number; steering?: SteeringStatus; mode: SubagentRunMode; context?: ContextSummary; cwd?: string; startedAt: number; lastUpdate?: number; endedAt?: number; timeoutMs?: number; deadlineAt?: number; timedOut?: boolean; stopped?: boolean; turnBudget?: TurnBudgetState; turnBudgetExceeded?: boolean; wrapUpRequested?: boolean; currentStep?: number; chainStepCount?: number; pendingAppends?: number; parallelGroups?: AsyncParallelGroupStatus[]; steps: AsyncRunStepSummary[]; sessionDir?: string; outputFile?: string; totalTokens?: TokenUsage; totalCost?: CostSummary; sessionFile?: string; nestedChildren?: NestedRunSummary[]; nestedWarnings?: string[]; processTerminal?: AsyncStatus["processTerminal"]; capabilityCeiling?: ResolvedSubagentCapabilityCeiling; capabilityAudit?: SubagentCapabilityAudit; } interface AsyncRunListOptions { states?: Array; sessionId?: string; limit?: number; resultsDir?: string; kill?: (pid: number, signal?: NodeJS.Signals | 0) => boolean; now?: () => number; reconcile?: boolean; runId?: string; } function getErrorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } function isNotFoundError(error: unknown): boolean { return typeof error === "object" && error !== null && "code" in error && (error as NodeJS.ErrnoException).code === "ENOENT"; } function isAsyncRunDir(root: string, entry: string): boolean { const entryPath = path.join(root, entry); try { return fs.statSync(entryPath).isDirectory(); } catch (error) { if (isNotFoundError(error)) return false; throw new Error(`Failed to inspect async run path '${entryPath}': ${getErrorMessage(error)}`, { cause: error instanceof Error ? error : undefined, }); } } type TargetedAsyncRunResolution = | { kind: "exact"; id: string } | { kind: "scan" } | { kind: "reject" }; /** * Resolve an exact targeted run without following a run-directory symlink or * accepting a path whose canonical location escaped the async root. */ export function resolveTargetedAsyncRun(asyncDirRoot: string, id: string, sessionId?: string): TargetedAsyncRunResolution { if (!id || id === "." || id === ".." || path.basename(id) !== id) return { kind: "reject" }; const asyncDir = path.join(asyncDirRoot, id); let entryStat: fs.Stats; try { entryStat = fs.lstatSync(asyncDir); } catch (error) { if (isNotFoundError(error)) return { kind: "scan" }; throw new Error(`Failed to inspect async run path '${asyncDir}': ${getErrorMessage(error)}`, { cause: error instanceof Error ? error : undefined, }); } if (!entryStat.isDirectory() || entryStat.isSymbolicLink()) return { kind: "reject" }; try { const canonicalRoot = fs.realpathSync(asyncDirRoot); const canonicalDir = fs.realpathSync(asyncDir); if (canonicalDir !== canonicalRoot && !canonicalDir.startsWith(`${canonicalRoot}${path.sep}`)) return { kind: "reject" }; } catch (error) { if (isNotFoundError(error)) return { kind: "reject" }; throw new Error(`Failed to resolve async run path '${asyncDir}': ${getErrorMessage(error)}`, { cause: error instanceof Error ? error : undefined, }); } if (sessionId !== undefined) { const status = readStatus(asyncDir); if (status?.sessionId !== sessionId) return { kind: "scan" }; } return { kind: "exact", id }; } function outputFileMtime(outputFile: string | undefined): number | undefined { if (!outputFile) return undefined; try { return fs.statSync(outputFile).mtimeMs; } catch (error) { if (isNotFoundError(error)) return undefined; throw new Error(`Failed to inspect async output file '${outputFile}': ${getErrorMessage(error)}`, { cause: error instanceof Error ? error : undefined, }); } } function deriveAsyncActivityState(asyncDir: string, status: AsyncStatus): { activityState?: ActivityState; lastActivityAt?: number } { if (status.state !== "running") return { activityState: status.activityState, lastActivityAt: status.lastActivityAt }; const outputPath = status.outputFile ? (path.isAbsolute(status.outputFile) ? status.outputFile : path.join(asyncDir, status.outputFile)) : undefined; const currentStep = typeof status.currentStep === "number" ? status.steps?.[status.currentStep] : undefined; return { activityState: status.activityState, lastActivityAt: status.lastActivityAt ?? outputFileMtime(outputPath) ?? currentStep?.lastActivityAt ?? currentStep?.startedAt ?? status.startedAt, }; } function statusToSummary(asyncDir: string, status: AsyncStatus & { cwd?: string }, nestedWarnings: string[] = [], nestedRoute?: NestedRoute): AsyncRunSummary { if (status.sessionId !== undefined && typeof status.sessionId !== "string") { throw new Error(`Invalid async status '${path.join(asyncDir, "status.json")}': sessionId must be a string.`); } const { activityState, lastActivityAt } = deriveAsyncActivityState(asyncDir, status); const processTerminal = readProcessTerminal(asyncDir, { runId: status.runId, runnerProcessInstanceId: status.processTerminal?.runnerProcessInstanceId }) ?? sanitizeProcessTerminal(status.processTerminal, { runId: status.runId, runnerProcessInstanceId: status.processTerminal?.runnerProcessInstanceId }, path.join(asyncDir, "status.json")); const steps = status.steps ?? []; const chainStepCount = status.chainStepCount ?? steps.length; const parallelGroups = normalizeParallelGroups(status.parallelGroups, steps.length, chainStepCount); let nestedChildren: NestedRunSummary[] = []; if (nestedWarnings.length === 0 && nestedRoute) { try { // The route is resolved by the caller via buildNestedRouteIndex, so this // avoids a fresh scan of the nested-events directory per run. nestedChildren = projectNestedEvents(nestedRoute)?.children ?? []; } catch (error) { nestedWarnings.push(`Nested status unavailable: ${getErrorMessage(error)}`); } } const summarizedSteps = steps.map((step, index) => { const stepActivityState = step.activityState; const stepLastActivityAt = step.lastActivityAt; return { index, agent: step.agent, ...(step.context ? { context: step.context } : {}), ...(step.label ? { label: step.label } : {}), ...(step.phase ? { phase: step.phase } : {}), ...(step.outputName ? { outputName: step.outputName } : {}), ...(step.structured ? { structured: step.structured } : {}), status: step.status, ...(stepActivityState ? { activityState: stepActivityState } : {}), ...(stepLastActivityAt ? { lastActivityAt: stepLastActivityAt } : {}), ...(step.currentTool ? { currentTool: step.currentTool } : {}), ...(step.currentToolArgs ? { currentToolArgs: step.currentToolArgs } : {}), ...(step.currentToolStartedAt ? { currentToolStartedAt: step.currentToolStartedAt } : {}), ...(step.currentPath ? { currentPath: step.currentPath } : {}), ...(step.recentTools ? { recentTools: step.recentTools.map((tool) => ({ ...tool })) } : {}), ...(step.recentOutput ? { recentOutput: [...step.recentOutput] } : {}), ...(step.turnCount !== undefined ? { turnCount: step.turnCount } : {}), ...(step.toolCount !== undefined ? { toolCount: step.toolCount } : {}), ...(step.steering ? { steering: step.steering } : {}), ...(step.durationMs !== undefined ? { durationMs: step.durationMs } : {}), ...(step.tokens ? { tokens: step.tokens } : {}), ...(step.totalCost ? { totalCost: step.totalCost } : {}), ...(step.skills ? { skills: step.skills } : {}), ...(step.model ? { model: step.model } : {}), ...(step.thinking ? { thinking: step.thinking } : {}), ...(step.attemptedModels ? { attemptedModels: step.attemptedModels } : {}), ...(step.sessionFile ? { sessionFile: step.sessionFile } : {}), ...(step.transcriptPath ? { transcriptPath: step.transcriptPath } : {}), ...(step.error ? { error: step.error } : {}), ...(step.timedOut !== undefined ? { timedOut: step.timedOut } : {}), ...(step.stopped !== undefined ? { stopped: step.stopped } : {}), ...(step.turnBudget ? { turnBudget: step.turnBudget } : {}), ...(step.turnBudgetExceeded !== undefined ? { turnBudgetExceeded: step.turnBudgetExceeded } : {}), ...(step.wrapUpRequested !== undefined ? { wrapUpRequested: step.wrapUpRequested } : {}), ...(step.acceptance ? { acceptance: step.acceptance } : {}), ...(step.agentContract ? { agentContract: step.agentContract } : {}), ...(step.launchContractDigest ? { launchContractDigest: step.launchContractDigest } : {}), ...(step.execution ? { execution: step.execution } : {}), ...(step.review ? { review: step.review } : {}), ...(step.effects ? { effects: step.effects } : {}), ...(step.processTerminal ? { processTerminal: sanitizeProcessTerminal(step.processTerminal, { runId: status.runId, runnerProcessInstanceId: step.processTerminal.runnerProcessInstanceId }, `${path.join(asyncDir, "status.json")} step ${index}`) } : {}), ...(step.capabilityCeiling ? { capabilityCeiling: step.capabilityCeiling } : {}), ...(step.capabilityAudit ? { capabilityAudit: step.capabilityAudit } : {}), ...(step.children?.length ? { children: step.children } : {}), }; }); attachRootChildrenToSteps(status.runId || path.basename(asyncDir), summarizedSteps, nestedChildren); return { id: status.runId || path.basename(asyncDir), asyncDir, ...(status.sessionId ? { sessionId: status.sessionId } : {}), state: status.state, ...(status.error ? { error: status.error } : {}), activityState, lastActivityAt, currentTool: status.currentTool, currentToolStartedAt: status.currentToolStartedAt, currentPath: status.currentPath, turnCount: status.turnCount, toolCount: status.toolCount, steering: status.steering, mode: status.mode, ...(summarizeContextModes(summarizedSteps.map((step) => step.context)) ? { context: summarizeContextModes(summarizedSteps.map((step) => step.context)) } : {}), cwd: status.cwd, startedAt: status.startedAt, lastUpdate: status.lastUpdate, endedAt: status.endedAt, ...(status.timeoutMs !== undefined ? { timeoutMs: status.timeoutMs } : {}), ...(status.deadlineAt !== undefined ? { deadlineAt: status.deadlineAt } : {}), ...(status.timedOut !== undefined ? { timedOut: status.timedOut } : {}), ...(status.stopped !== undefined ? { stopped: status.stopped } : {}), ...(status.turnBudget ? { turnBudget: status.turnBudget } : {}), ...(status.turnBudgetExceeded !== undefined ? { turnBudgetExceeded: status.turnBudgetExceeded } : {}), ...(status.wrapUpRequested !== undefined ? { wrapUpRequested: status.wrapUpRequested } : {}), currentStep: status.currentStep, ...(status.chainStepCount !== undefined ? { chainStepCount: status.chainStepCount } : {}), ...(status.pendingAppends !== undefined ? { pendingAppends: status.pendingAppends } : {}), ...(parallelGroups.length ? { parallelGroups } : {}), steps: summarizedSteps, ...(nestedChildren.length ? { nestedChildren } : {}), ...(nestedWarnings.length ? { nestedWarnings } : {}), ...(processTerminal ? { processTerminal } : {}), ...(status.launchContractDigest ? { launchContractDigest: status.launchContractDigest } : {}), ...(status.capabilityCeiling ? { capabilityCeiling: status.capabilityCeiling } : {}), ...(status.capabilityAudit ? { capabilityAudit: status.capabilityAudit } : {}), ...(status.sessionDir ? { sessionDir: status.sessionDir } : {}), ...(status.outputFile ? { outputFile: status.outputFile } : {}), ...(status.totalTokens ? { totalTokens: status.totalTokens } : {}), ...(status.totalCost ? { totalCost: status.totalCost } : {}), ...(status.sessionFile ? { sessionFile: status.sessionFile } : {}), }; } export function summarizeAsyncStatus(asyncDir: string, status: AsyncStatus & { cwd?: string }): AsyncRunSummary { return statusToSummary(asyncDir, status); } function sortRuns(runs: AsyncRunSummary[]): AsyncRunSummary[] { const rank = (state: AsyncRunSummary["state"]): number => { switch (state) { case "running": return 0; case "queued": return 1; case "failed": return 2; case "stopped": return 2; case "paused": return 2; case "complete": return 3; } }; return [...runs].sort((a, b) => { const byState = rank(a.state) - rank(b.state); if (byState !== 0) return byState; const aTime = a.lastUpdate ?? a.endedAt ?? a.startedAt; const bTime = b.lastUpdate ?? b.endedAt ?? b.startedAt; return bTime - aTime; }); } export function listAsyncRuns(asyncDirRoot: string, options: AsyncRunListOptions = {}): AsyncRunSummary[] { let entries: string[]; try { if (options.runId !== undefined) { const resolution = resolveTargetedAsyncRun(asyncDirRoot, options.runId, options.sessionId); entries = resolution.kind === "exact" ? [resolution.id] : resolution.kind === "scan" ? fs.readdirSync(asyncDirRoot).filter((entry) => (entry === options.runId || entry.startsWith(options.runId!)) && resolveTargetedAsyncRun(asyncDirRoot, entry, options.sessionId).kind === "exact" ) : []; } else { entries = fs.readdirSync(asyncDirRoot).filter((entry) => isAsyncRunDir(asyncDirRoot, entry)); } } catch (error) { if (isNotFoundError(error)) return []; throw new Error(`Failed to list async runs in '${asyncDirRoot}': ${getErrorMessage(error)}`, { cause: error instanceof Error ? error : undefined, }); } const allowedStates = options.states ? new Set(options.states) : undefined; const runs: AsyncRunSummary[] = []; // Route resolution for every run shares a single index built from the // nested-events directory, so the per-run lookup is O(1) instead of scanning // the directory once per run. The index is built lazily on first use, so // load-time restoration (which only wants queued/running runs) skips it // entirely when no active runs match. let nestedRouteIndex: Map | undefined; const resolveNestedRoute = (rootRunId: string): NestedRoute | undefined => { if (!nestedRouteIndex) nestedRouteIndex = buildNestedRouteIndex(); return nestedRouteIndex.get(rootRunId); }; for (const entry of entries) { const asyncDir = path.join(asyncDirRoot, entry); const reconciliation = options.reconcile === false ? undefined : reconcileAsyncRun(asyncDir, { resultsDir: options.resultsDir, kill: options.kill, now: options.now }); const status = (reconciliation?.status ?? readStatus(asyncDir)) as (AsyncStatus & { cwd?: string }) | null; if (!status) continue; // Filter before the nested-route lookup: the lookup builds an index over // the nested-events directory, so deferring it for filtered-out runs keeps // restoration at load from scanning that directory when no active runs // match. if (allowedStates && !allowedStates.has(status.state)) continue; if (options.sessionId && status.sessionId !== options.sessionId) continue; const nestedWarnings: string[] = []; let nestedRoute: NestedRoute | undefined; if (options.reconcile !== false) { try { nestedRoute = resolveNestedRoute(status.runId || path.basename(asyncDir)); if (nestedRoute) reconcileNestedAsyncDescendants(nestedRoute, { resultsDir: options.resultsDir, kill: options.kill, now: options.now }); } catch (error) { nestedWarnings.push(`Nested status unavailable: ${getErrorMessage(error)}`); } } const summary = statusToSummary(asyncDir, status, nestedWarnings, nestedRoute); runs.push(summary); } const sorted = sortRuns(runs); return options.limit !== undefined ? sorted.slice(0, options.limit) : sorted; } function formatActivityFacts(input: { activityState?: ActivityState; lastActivityAt?: number; currentTool?: string; currentToolStartedAt?: number; currentPath?: string; turnCount?: number; toolCount?: number; steering?: SteeringStatus; turnBudget?: TurnBudgetState; turnBudgetExceeded?: boolean; wrapUpRequested?: boolean }): string | undefined { const facts: string[] = []; if (input.currentTool && input.currentToolStartedAt !== undefined) facts.push(`tool ${input.currentTool} ${formatDuration(Math.max(0, Date.now() - input.currentToolStartedAt))}`); else if (input.currentTool) facts.push(`tool ${input.currentTool}`); if (input.currentPath) facts.push(shortenPath(input.currentPath)); if (input.turnCount !== undefined) facts.push(`${input.turnCount} turns`); if (input.turnBudgetExceeded && input.turnBudget) facts.push(`turn budget exceeded ${input.turnBudget.turnCount}/${input.turnBudget.maxTurns}+${input.turnBudget.graceTurns}`); else if (input.turnBudget?.outcome === "termination-deferred") facts.push(`turn-budget termination deferred ${input.turnBudget.turnCount}/${input.turnBudget.maxTurns}+${input.turnBudget.graceTurns}`); else if (input.wrapUpRequested && input.turnBudget) facts.push(`wrap-up requested ${input.turnBudget.turnCount}/${input.turnBudget.maxTurns}`); else if (input.turnBudget) facts.push(`turn budget ${input.turnBudget.turnCount}/${input.turnBudget.maxTurns}+${input.turnBudget.graceTurns}`); if (input.toolCount !== undefined) facts.push(`${input.toolCount} tools`); if (input.steering) facts.push(`steering ${input.steering.scheduled} scheduled, ${input.steering.pending} pending, ${input.steering.delivered} delivered, ${input.steering.failed} failed, ${input.steering.recovered} recovered`); const activity = formatActivityLabel(input.lastActivityAt, input.activityState); return activity || facts.length ? [activity, ...facts].filter(Boolean).join(" | ") : undefined; } function formatStepLine(step: AsyncRunStepSummary): string { const display = step.label ? `${step.label} (${step.agent})` : step.agent; const context = contextModeLabel(step.context); const phase = step.phase ? `[${step.phase}] ` : ""; const parts = [`${step.index + 1}. ${phase}${display}${context ? ` ${context}` : ""}`, step.status]; const activity = formatActivityFacts(step); if (activity) parts.push(activity); const modelThinking = formatModelThinking(step.model, step.thinking); if (modelThinking) parts.push(modelThinking); if (step.durationMs !== undefined) parts.push(formatDuration(step.durationMs)); if (step.tokens) parts.push(`${formatTokens(step.tokens.total)} tok`); return parts.join(" | "); } export function formatAsyncRunOutputPath(run: Pick): string | undefined { if (!run.outputFile) return undefined; return path.isAbsolute(run.outputFile) ? run.outputFile : path.join(run.asyncDir, run.outputFile); } export function formatAsyncRunProgressLabel(run: Pick): string { const stepCount = run.steps.length || 1; const chainStepCount = run.chainStepCount ?? stepCount; const groups = normalizeParallelGroups(run.parallelGroups, run.steps.length, chainStepCount); const activeGroup = run.currentStep !== undefined ? groups.find((group) => run.currentStep! >= group.start && run.currentStep! < group.start + group.count) : undefined; if (activeGroup) { const groupSteps = run.steps.slice(activeGroup.start, activeGroup.start + activeGroup.count); const groupLabel = formatParallelOutcome(groupSteps, activeGroup.count, { showRunning: run.state === "running" }); if (run.mode === "parallel") return groupLabel; return `step ${activeGroup.stepIndex + 1}/${chainStepCount} ยท parallel group: ${groupLabel}`; } if (run.mode === "parallel") return formatParallelOutcome(run.steps, stepCount, { showRunning: run.state === "running" }); if (run.mode === "chain" && run.currentStep !== undefined && groups.length > 0) { const logicalStep = flatToLogicalStepIndex(run.currentStep, chainStepCount, groups); return `step ${logicalStep + 1}/${chainStepCount}`; } return run.currentStep !== undefined ? `step ${run.currentStep + 1}/${stepCount}` : `steps ${stepCount}`; } function formatRunHeader(run: AsyncRunSummary): string { const stepLabel = formatAsyncRunProgressLabel(run); const cwd = run.cwd ? shortenPath(run.cwd) : shortenPath(run.asyncDir); const activity = formatActivityFacts(run); const pending = run.pendingAppends ? ` | ${run.pendingAppends} pending append${run.pendingAppends === 1 ? "" : "s"}` : ""; const context = contextModeLabel(run.context); return `${run.id} | ${run.state}${activity ? ` | ${activity}` : ""} | ${run.mode}${context ? ` ${context}` : ""} | ${stepLabel}${pending} | ${cwd}`; } export function formatAsyncRunList(runs: AsyncRunSummary[], heading = "Active async runs"): string { if (runs.length === 0) return `No ${heading.toLowerCase()}.`; const lines = [`${heading}: ${runs.length}`, ""]; for (const run of runs) { lines.push(`- ${formatRunHeader(run)}`); for (const step of run.steps) { lines.push(` ${formatStepLine(step)}`); lines.push(...formatNestedRunStatusLines(step.children, { indent: " ", maxLines: 12 })); } const attached = new Set(run.steps.flatMap((step) => step.children?.map((child) => child.id) ?? [])); const unattached = run.nestedChildren?.filter((child) => !attached.has(child.id)) ?? []; lines.push(...formatNestedRunStatusLines(unattached, { indent: " ", maxLines: 12 })); if (run.error) lines.push(` Error: ${run.error}`); for (const warning of run.nestedWarnings ?? []) lines.push(` Warning: ${warning}`); const outputPath = formatAsyncRunOutputPath(run); if (outputPath) lines.push(` output: ${shortenPath(outputPath)}`); if (run.sessionFile) lines.push(` session: ${shortenPath(run.sessionFile)}`); lines.push(""); } return lines.join("\n").trimEnd(); }