/** * @fileoverview Workflow (ultracode) Run Watcher * * Emits events powering the master-detail "working agents" view (tasks/phases on * the LEFT, per-agent tokens/tool-calls on the RIGHT) AND the floating run * windows, from TWO disk sources per run: * * 1. COMPLETION artifact — `…/workflows/wf_.json`. The Workflow runtime * writes this with the FULL run state (phases, per-agent tokens/tool-calls, * result), but — as of the mid-2026 runtime — only when the run FINISHES * (always a terminal status). It is the authoritative, detailed record. * 2. LIVE transcript dir — `…/subagents/workflows/wf_/` (agent-*.jsonl + * journal.jsonl). This appears WHILE a run is in flight, before any * `wf_.json` exists. From it we synthesize a minimal ACTIVE run * (status 'running', agent slots keyed by agentId, lastActivityAt from file * mtimes) so the floating window pops DURING the run instead of only after. * * Precedence: when a completion `wf_.json` exists it ALWAYS supersedes the * synthesized live record (same runId), so a finished run shows full detail and * the normal finish→auto-close flow runs. Without source 2 the floating-window * feature is dead for live runs (the completion file only lands at the end, so * the watcher would never see a run while it is active). * * Still STANDALONE: it never imports from or touches subagent-watcher.ts. It * independently reads the same `subagents/workflows/` tree subagent-watcher uses, * but as a separate singleton with no shared mutable state. * * Discovery is dual: a periodic poll (catches new run dirs + removals) plus a * per-dir chokidar watcher (live updates). A per-source mtime skip keeps the hot * path cheap — the completion JSON / live dir is re-read only when its mtime moves. * * @module workflow-run-watcher */ import { EventEmitter } from 'node:events'; import { readdir, readFile, stat } from 'node:fs/promises'; import { homedir } from 'node:os'; import { join } from 'node:path'; import { watch as chokidarWatch, type FSWatcher as ChokidarWatcher } from 'chokidar'; import type { WorkflowRunInfo, WorkflowRunSummary, WorkflowAgentInfo, WorkflowRunPhase } from './types/workflow-run.js'; import { LRUMap } from './utils/lru-map.js'; import { WORKFLOW_RUN_POLL_INTERVAL_MS, MAX_CACHED_WORKFLOW_RUNS, WORKFLOW_RUN_RECENT_WINDOW_MIN, } from './config/workflow-config.js'; const WORKFLOWS_SUBDIR = 'workflows'; const SUBAGENTS_SUBDIR = 'subagents'; const RUN_FILE_PREFIX = 'wf_'; const RUN_FILE_SUFFIX = '.json'; const LIVE_JOURNAL_FILE = 'journal.jsonl'; const LIVE_AGENT_PREFIX = 'agent-'; const LIVE_TRANSCRIPT_SUFFIX = '.jsonl'; const SCRIPTS_SUBDIR = 'scripts'; /** Hard caps on the largest per-agent strings so a 28-agent run stays compact. */ const PROMPT_PREVIEW_MAX = 200; const RESULT_PREVIEW_MAX = 240; /** Cap a live script read so a runaway/huge embedded script can't blow memory. */ const SCRIPT_READ_MAX_BYTES = 512 * 1024; /** LRU cap on cached per-agent transcript stats across all live runs. */ const MAX_CACHED_AGENT_STATS = 2000; function truncate(value: string | undefined, max: number): string | undefined { if (typeof value !== 'string') return undefined; return value.length > max ? `${value.slice(0, max)}…` : value; } /** First text from a transcript message's `content` (string or content-block array). */ function extractMessageText(content: unknown): string | undefined { if (typeof content === 'string') return content.trim() || undefined; if (Array.isArray(content)) { for (const block of content) { if (block && typeof block === 'object' && (block as { type?: string }).type === 'text') { const t = (block as { text?: string }).text; if (typeof t === 'string' && t.trim()) return t.trim(); } } } return undefined; } /** * Isolate the `meta = { … }` object literal from a workflow script so that later * declarations (e.g. JSON-Schema objects with their own `description:`/`title:` * fields, or prose containing `phases: [`) can't be mistaken for meta. Brace-matches * naively — a `{`/`}` inside a string value can still confuse it, in which case we * return undefined and the caller falls back to scanning the whole body. Best-effort. */ function extractMetaBlock(body: string): string | undefined { const m = /(?:^|[^A-Za-z0-9_])(?:export\s+const|const|let|var)\s+meta\s*=\s*\{/.exec(body); if (!m) return undefined; const open = body.indexOf('{', m.index); if (open < 0) return undefined; let depth = 0; for (let i = open; i < body.length; i++) { if (body[i] === '{') depth++; else if (body[i] === '}' && --depth === 0) return body.slice(open, i + 1); } return undefined; // unterminated (truncated script) — caller falls back to whole body } /** * Best-effort extraction of a single/double-quoted `key: '...'` value from a workflow * `meta` literal (the script is JS we must not eval). The key is anchored to a key * position (`{`/`,`/whitespace before it) so e.g. `description` can't match the tail * of `..._description`; pass the scoped meta block so later same-named schema fields * can't win. Returns undefined on no match. */ function extractMetaString(body: string, key: string): string | undefined { const re = new RegExp(`(?:^|[{,\\s])${key}\\s*:\\s*(['"])((?:\\\\.|(?!\\1).)*)\\1`); const m = re.exec(body); return m ? m[2].replace(/\\(['"\\`])/g, '$1') : undefined; } /** Best-effort `meta.phases: [{title, detail}, …]` extraction (pass the scoped meta block). */ function extractMetaPhases(body: string): WorkflowRunPhase[] { const keyMatch = /(?:^|[^A-Za-z0-9_])phases\s*:\s*\[/.exec(body); if (!keyMatch) return []; const open = body.indexOf('[', keyMatch.index); if (open < 0) return []; // Brace-match to the array's closing ] so nested arrays don't end it early. let depth = 0; let end = -1; for (let i = open; i < body.length; i++) { if (body[i] === '[') depth++; else if (body[i] === ']' && --depth === 0) { end = i; break; } } if (end < 0) return []; const block = body.slice(open, end + 1); const phases: WorkflowRunPhase[] = []; const re = /title\s*:\s*(['"])((?:\\.|(?!\1).)*)\1(?:\s*,\s*detail\s*:\s*(['"])((?:\\.|(?!\3).)*)\3)?/g; let m: RegExpExecArray | null; while ((m = re.exec(block)) !== null && phases.length < 20) { phases.push({ title: m[2].replace(/\\(['"\\`])/g, '$1'), detail: (m[4] || '').replace(/\\(['"\\`])/g, '$1'), }); } return phases; } /** Drop the heavy `agents[]` for list/snapshot use. */ export function summarizeRun(info: WorkflowRunInfo): WorkflowRunSummary { const { agents: _agents, ...summary } = info; void _agents; return summary; } interface DiscoveredRun { filePath: string; projectHash: string; sessionUuid: string; runId: string; } /** An in-flight run discovered from its `subagents/workflows/wf_/` transcript dir. */ interface DiscoveredLiveRun { dirPath: string; projectHash: string; sessionUuid: string; runId: string; } /** Per-agent stats parsed from one `agent-.jsonl` transcript (mtime-cached). */ interface AgentTranscriptStats { /** Tokens for the agent's LAST usage-bearing message (in+out+cache) ≈ the completion-JSON `tokens`. */ tokens: number; /** Count of `tool_use` blocks across the transcript (matches the completion-JSON `toolCalls`). */ toolCalls: number; /** Most-recent model id seen. */ model: string; /** Name of the most-recent `tool_use` block. */ lastToolName?: string; /** First user-message text, truncated — a hint of what the agent was asked. */ promptPreview?: string; } /** Workflow meta derived live from `workflows/scripts/-.js`. */ interface LiveWorkflowMeta { workflowName?: string; summary?: string; phases: WorkflowRunPhase[]; } /** A live agent file on disk (transcript and/or meta), keyed by its `` stem. */ interface LiveAgentFile { agentId: string; transcriptPath?: string; transcriptMtime?: number; } /** A `workflowProgress[]` entry as it appears on disk (loosely typed for defensive parsing). */ interface RawProgressEntry { type?: string; index?: number; label?: string; phaseIndex?: number; phaseTitle?: string; model?: string; state?: string; queuedAt?: number; lastProgressAt?: number; promptPreview?: string; agentId?: string; startedAt?: number; attempt?: number; tokens?: number; toolCalls?: number; lastToolName?: string; lastToolSummary?: string; durationMs?: number; resultPreview?: string; } export class WorkflowRunWatcher extends EventEmitter { private projectsDir: string; private pollTimer: NodeJS.Timeout | null = null; private _isRunning = false; /** runId -> latest parsed run info (LRU-bounded). */ private runs = new LRUMap({ maxSize: MAX_CACHED_WORKFLOW_RUNS }); /** absolute run-file path -> last seen mtimeMs (skip unchanged files). */ private fileMtimes = new Map(); /** runId -> absolute run-file path (for mtime cleanup on removal). */ private runIdToPath = new Map(); /** absolute live transcript-dir path -> newest member mtimeMs (skip unchanged live runs). */ private liveDirMtimes = new Map(); /** runId -> absolute live transcript-dir path (for mtime cleanup on removal). */ private runIdToLiveDir = new Map(); /** transcript abs path -> { mtimeMs, stats }; re-parse a transcript only when its mtime moves. */ private agentStatCache = new LRUMap({ maxSize: MAX_CACHED_AGENT_STATS, }); /** runId -> derived script meta (immutable per run; re-derived only until a name is found). */ private liveMetaCache = new Map(); /** journal abs path -> { mtimeMs, parsed }; re-parse the journal only when it grows. */ private journalCache = new Map< string, { mtimeMs: number; parsed: { startedOrder: string[]; doneIds: Set } } >(); /** watched-dir absolute path -> chokidar watcher (workflows/ + subagents/workflows/). */ private dirWatchers = new Map(); constructor(projectsDir?: string) { super(); this.projectsDir = projectsDir || join(homedir(), '.claude', 'projects'); this.setMaxListeners(50); } // ========== Public API ========== isRunning(): boolean { return this._isRunning; } start(): void { if (this._isRunning) return; this._isRunning = true; this.poll(); this.pollTimer = setInterval(() => this.poll(), WORKFLOW_RUN_POLL_INTERVAL_MS); } stop(): void { this._isRunning = false; if (this.pollTimer) { clearInterval(this.pollTimer); this.pollTimer = null; } for (const watcher of this.dirWatchers.values()) { watcher.close().catch(() => {}); // best-effort teardown } this.dirWatchers.clear(); this.runs.clear(); this.fileMtimes.clear(); this.runIdToPath.clear(); this.liveDirMtimes.clear(); this.runIdToLiveDir.clear(); this.agentStatCache.clear(); this.liveMetaCache.clear(); this.journalCache.clear(); } /** All cached runs (no recency filter), most-recently-active first. */ getAllRuns(): WorkflowRunInfo[] { return Array.from(this.runs.values()).sort((a, b) => b.lastActivityAt - a.lastActivityAt); } /** Runs active within the last `minutes`, most-recently-active first. */ getRecentRuns(minutes: number = WORKFLOW_RUN_RECENT_WINDOW_MIN): WorkflowRunInfo[] { const cutoff = Date.now() - minutes * 60_000; return this.getAllRuns().filter((r) => r.lastActivityAt >= cutoff); } /** * Lightweight summaries (no agents[]) of ALL cached runs, most-recently-active * first. This is the LEFT-pane list + getLightState snapshot source: the cache * is LRU-bounded (MAX_CACHED_WORKFLOW_RUNS), so it's already size-capped, and a * run-browser should show past runs — NOT hide everything older than a window. */ getAllRunSummaries(): WorkflowRunSummary[] { return this.getAllRuns().map(summarizeRun); } /** Summaries filtered to the last `minutes` of activity (opt-in via ?minutes). */ getRecentRunSummaries(minutes: number = WORKFLOW_RUN_RECENT_WINDOW_MIN): WorkflowRunSummary[] { return this.getRecentRuns(minutes).map(summarizeRun); } getRun(runId: string): WorkflowRunInfo | undefined { return this.runs.get(runId); } getStats(): { runCount: number; running: number; agentCount: number } { let running = 0; let agentCount = 0; for (const run of this.runs.values()) { if (run.status === 'running') running++; agentCount += run.agents.length; } return { runCount: this.runs.size, running, agentCount }; } // ========== Private ========== private poll(): void { this.pollAsync().catch(() => { // Filesystem may be transiently unavailable; the next poll retries. }); } private async pollAsync(): Promise { const { files, liveDirs, watchDirs } = await this.discover(); // Install a live watcher for each watched dir; tear down watchers for dirs that vanished. for (const dir of watchDirs) this.ensureDirWatcher(dir); for (const dir of Array.from(this.dirWatchers.keys())) { if (!watchDirs.has(dir)) this.removeDirWatcher(dir); } const seenRunIds = new Set(); const realRunIds = new Set(); for (const file of files) { seenRunIds.add(file.runId); realRunIds.add(file.runId); await this.maybeParse(file); } // In-flight runs: synthesize from the transcript tree ONLY while no completion // wf_*.json exists yet — the real file (full detail + terminal status) supersedes. for (const live of liveDirs) { if (realRunIds.has(live.runId)) continue; seenRunIds.add(live.runId); await this.maybeParseLive(live); } // Removal by set-diff: a cached run discoverable from neither source. for (const runId of Array.from(this.runs.keys())) { if (!seenRunIds.has(runId)) { this.runs.delete(runId); const path = this.runIdToPath.get(runId); if (path) this.fileMtimes.delete(path); this.runIdToPath.delete(runId); const liveDir = this.runIdToLiveDir.get(runId); if (liveDir) { this.liveDirMtimes.delete(liveDir); // Drop cached transcript stats + journal parse under this run's live dir. const prefix = liveDir.endsWith('/') ? liveDir : liveDir + '/'; for (const key of Array.from(this.agentStatCache.keys())) { if (key.startsWith(prefix)) this.agentStatCache.delete(key); } for (const key of Array.from(this.journalCache.keys())) { if (key.startsWith(prefix)) this.journalCache.delete(key); } } this.liveMetaCache.delete(runId); this.runIdToLiveDir.delete(runId); this.emit('run_removed', { runId }); } } } /** * Walk projects/// for both run sources: * - completion files: `workflows/wf_*.json` * - in-flight runs: `subagents/workflows/wf_/` (transcript dirs) * Returns the dirs to chokidar-watch (so a new run/file is caught sub-poll). */ private async discover(): Promise<{ files: DiscoveredRun[]; liveDirs: DiscoveredLiveRun[]; watchDirs: Set; }> { const files: DiscoveredRun[] = []; const liveDirs: DiscoveredLiveRun[] = []; const watchDirs = new Set(); let projectHashes: string[]; try { projectHashes = await readdir(this.projectsDir); } catch { return { files, liveDirs, watchDirs }; } for (const projectHash of projectHashes) { let sessions: string[]; try { sessions = await readdir(join(this.projectsDir, projectHash)); } catch { continue; } for (const sessionUuid of sessions) { const sessionDir = join(this.projectsDir, projectHash, sessionUuid); // (1) Completion artifacts: workflows/wf_*.json const workflowsDir = join(sessionDir, WORKFLOWS_SUBDIR); try { const names = await readdir(workflowsDir); let hasRun = false; for (const name of names) { if (!name.startsWith(RUN_FILE_PREFIX) || !name.endsWith(RUN_FILE_SUFFIX)) continue; hasRun = true; files.push({ filePath: join(workflowsDir, name), projectHash, sessionUuid, runId: name.slice(0, -RUN_FILE_SUFFIX.length), }); } if (hasRun) watchDirs.add(workflowsDir); } catch { // no workflows dir for this session — normal } // (2) In-flight runs: subagents/workflows/wf_*/ const liveParent = join(sessionDir, SUBAGENTS_SUBDIR, WORKFLOWS_SUBDIR); try { const names = await readdir(liveParent); let hasLive = false; for (const name of names) { if (!name.startsWith(RUN_FILE_PREFIX) || name.endsWith(RUN_FILE_SUFFIX)) continue; // wf_ dir, not a file hasLive = true; liveDirs.push({ dirPath: join(liveParent, name), projectHash, sessionUuid, runId: name, }); } if (hasLive) watchDirs.add(liveParent); } catch { // no subagents/workflows dir for this session — normal } } } return { files, liveDirs, watchDirs }; } private async maybeParse(file: DiscoveredRun): Promise { let mtime: number; try { mtime = (await stat(file.filePath)).mtimeMs; } catch { return; // vanished between discover and stat } if (this.fileMtimes.get(file.filePath) === mtime) return; this.fileMtimes.set(file.filePath, mtime); const info = await this.parseFile(file); if (!info) return; const existed = this.runs.has(info.runId); this.runs.set(info.runId, info); this.runIdToPath.set(info.runId, file.filePath); this.emit(existed ? 'run_updated' : 'run_discovered', info); } /** * Re-synthesize an in-flight run from its transcript dir. Cheap pass first: stat * the dir members and skip entirely when the newest mtime is unchanged, so an idle * poll never reads a single (large) transcript. Only on a real change do we parse — * and even then per-transcript stats come from an mtime-keyed cache, so only the * transcripts that actually grew are re-read. */ private async maybeParseLive(live: DiscoveredLiveRun): Promise { let entries: string[]; try { entries = await readdir(live.dirPath); } catch { return; // vanished between discover and read } // Cheap pass: newest member mtime + the agent-file set (no transcript reads yet). let newestMtime = 0; let journalPath: string | null = null; let journalMtime = 0; const agentFiles = new Map(); for (const name of entries) { if (name !== LIVE_JOURNAL_FILE && !name.startsWith(LIVE_AGENT_PREFIX)) continue; const full = join(live.dirPath, name); let m = 0; try { m = (await stat(full)).mtimeMs; } catch { continue; // entry vanished — ignore } if (m > newestMtime) newestMtime = m; if (name === LIVE_JOURNAL_FILE) { journalPath = full; journalMtime = m; continue; } // agent-.jsonl (transcript) or agent-.meta.json (queued-slot marker) const stem = name.slice(LIVE_AGENT_PREFIX.length).replace(/\.(meta\.json|jsonl)$/, ''); if (!stem) continue; const slot = agentFiles.get(stem) || { agentId: stem }; if (name.endsWith(LIVE_TRANSCRIPT_SUFFIX)) { slot.transcriptPath = full; slot.transcriptMtime = m; } agentFiles.set(stem, slot); } if (agentFiles.size === 0) return; // nothing to show yet // Skip the expensive parse when nothing changed since the last synthesis. if (this.liveDirMtimes.get(live.dirPath) === newestMtime) return; this.liveDirMtimes.set(live.dirPath, newestMtime); const info = await this.parseLiveDir(live, agentFiles, journalPath, journalMtime, newestMtime); if (!info) return; const existed = this.runs.has(info.runId); this.runs.set(info.runId, info); this.runIdToLiveDir.set(info.runId, live.dirPath); this.emit(existed ? 'run_updated' : 'run_discovered', info); } /** * Build an ACTIVE WorkflowRunInfo from a live transcript dir, ENRICHED so the * floating window / panel show real data mid-run rather than empty slots: * - per-agent tokens + tool-calls + model + last tool, parsed from each * `agent-.jsonl` (mtime-cached — only changed transcripts are re-read); * - per-agent state: 'done' once the journal logs a `result` (→ green badge), * 'progress' once it logs `started` or a transcript exists (→ yellow), else 'start'; * - run name/summary/phases from the live `workflows/scripts/-.js` * (the completion wf_.json, which carries these, only lands at the end); * - run totals = sums of the per-agent stats. */ private async parseLiveDir( live: DiscoveredLiveRun, agentFiles: Map, journalPath: string | null, journalMtime: number, newestMtime: number ): Promise { const { startedOrder, doneIds } = journalPath ? await this.readJournal(journalPath, journalMtime) : { startedOrder: [] as string[], doneIds: new Set() }; const startIndex = new Map(); startedOrder.forEach((id, i) => startIndex.set(id, i)); // Order agents by journal launch order; not-yet-started ones trail (sorted by id). const ids = Array.from(agentFiles.keys()).sort((a, b) => { const ia = startIndex.has(a) ? (startIndex.get(a) as number) : Number.MAX_SAFE_INTEGER; const ib = startIndex.has(b) ? (startIndex.get(b) as number) : Number.MAX_SAFE_INTEGER; if (ia !== ib) return ia - ib; return a < b ? -1 : a > b ? 1 : 0; }); let totalTokens = 0; let totalToolCalls = 0; let defaultModel = ''; const agents: WorkflowAgentInfo[] = []; for (let i = 0; i < ids.length; i++) { const id = ids[i]; const file = agentFiles.get(id) as LiveAgentFile; const stats = file.transcriptPath ? await this.agentTranscriptStats(file.transcriptPath, file.transcriptMtime || 0) : null; const state = doneIds.has(id) ? 'done' : startIndex.has(id) || file.transcriptPath ? 'progress' : 'start'; if (stats) { totalTokens += stats.tokens; totalToolCalls += stats.toolCalls; if (stats.model && !defaultModel) defaultModel = stats.model; } agents.push({ index: i + 1, label: `agent ${i + 1}`, phaseIndex: 1, phaseTitle: '', model: stats?.model || '', state, agentId: id, tokens: stats ? stats.tokens : undefined, toolCalls: stats ? stats.toolCalls : undefined, lastToolName: stats?.lastToolName, promptPreview: stats?.promptPreview, }); } // Script meta is immutable for the life of a run, so derive it at most once. // Re-derive only while we still lack a name (the script can land a poll or two // after the first transcripts, which seed the run before the script exists). let meta = this.liveMetaCache.get(live.runId); if (!meta || !meta.workflowName) { meta = await this.deriveLiveWorkflowMeta(live); this.liveMetaCache.set(live.runId, meta); } return { runId: live.runId, workflowName: meta.workflowName, summary: meta.summary, status: 'running', agentCount: agents.length, totalTokens, totalToolCalls, defaultModel: defaultModel || undefined, phases: meta.phases, agents, sessionUuid: live.sessionUuid, projectHash: live.projectHash, lastActivityAt: newestMtime || 0, }; } /** Parse one agent transcript for token/tool stats; cached on the file's mtime. */ private async agentTranscriptStats(path: string, mtimeMs: number): Promise { // peek (not get) on the hit-check so a cache HIT doesn't churn the LRU — every // active agent is looked up each poll, and get() would delete+reinsert each time. const cached = this.agentStatCache.peek(path); if (cached && cached.mtimeMs === mtimeMs) return cached.stats; let text: string; try { text = await readFile(path, 'utf-8'); } catch { return null; // vanished mid-poll } let tokens = 0; let toolCalls = 0; let model = ''; let lastToolName: string | undefined; let promptPreview: string | undefined; for (const line of text.split('\n')) { if (!line) continue; let entry: { type?: string; message?: unknown }; try { entry = JSON.parse(line) as { type?: string; message?: unknown }; } catch { continue; // tolerate a partially-written trailing line } const msg = entry.message as | { model?: string; usage?: { input_tokens?: number; output_tokens?: number; cache_read_input_tokens?: number; cache_creation_input_tokens?: number; }; content?: Array<{ type?: string; name?: string; text?: string }> | string; } | undefined; if (!promptPreview && entry.type === 'user' && msg) { promptPreview = truncate(extractMessageText(msg.content), PROMPT_PREVIEW_MAX); } if (!msg || typeof msg !== 'object') continue; if (typeof msg.model === 'string' && msg.model) model = msg.model; const u = msg.usage; if (u) { const total = (u.input_tokens || 0) + (u.output_tokens || 0) + (u.cache_read_input_tokens || 0) + (u.cache_creation_input_tokens || 0); // Anthropic usage already reflects cumulative context, so the agent's running // token total ≈ its LATEST usage-bearing message — last non-zero wins. if (total > 0) tokens = total; } if (Array.isArray(msg.content)) { for (const block of msg.content) { if (block && block.type === 'tool_use' && typeof block.name === 'string') { toolCalls++; lastToolName = block.name; } } } } const stats: AgentTranscriptStats = { tokens, toolCalls, model, lastToolName, promptPreview }; this.agentStatCache.set(path, { mtimeMs, stats }); return stats; } /** * Read the run journal: agent ids in `started` order plus the set that already * logged a terminal `result`. (`started` lines appear in launch order.) */ private async readJournal( journalPath: string, mtimeMs: number ): Promise<{ startedOrder: string[]; doneIds: Set }> { const cached = this.journalCache.get(journalPath); if (cached && cached.mtimeMs === mtimeMs) return cached.parsed; const startedOrder: string[] = []; const seenStarted = new Set(); const doneIds = new Set(); let text: string; try { text = await readFile(journalPath, 'utf-8'); } catch { return { startedOrder, doneIds }; // journal not written yet } for (const line of text.split('\n')) { if (!line) continue; try { const ev = JSON.parse(line) as { type?: string; agentId?: string }; if (!ev || typeof ev.agentId !== 'string') continue; if (ev.type === 'started' && !seenStarted.has(ev.agentId)) { seenStarted.add(ev.agentId); startedOrder.push(ev.agentId); } else if (ev.type === 'result') { doneIds.add(ev.agentId); } } catch { // tolerate a partially-written trailing line } } const parsed = { startedOrder, doneIds }; this.journalCache.set(journalPath, { mtimeMs, parsed }); return parsed; } /** * Derive name/summary/phases for a live run from its persisted script file * `…/workflows/scripts/-.js`. The filename yields a reliable name; * `meta.description`/`meta.phases` are best-effort regex extractions (the script is * JS we must not eval), so any parse miss simply leaves those optional fields absent. */ private async deriveLiveWorkflowMeta(live: DiscoveredLiveRun): Promise { const empty: LiveWorkflowMeta = { phases: [] }; const scriptsDir = join(this.projectsDir, live.projectHash, live.sessionUuid, WORKFLOWS_SUBDIR, SCRIPTS_SUBDIR); let names: string[]; try { names = await readdir(scriptsDir); } catch { return empty; } const suffix = `-${live.runId}.js`; const fileName = names.find((n) => n.endsWith(suffix)) || names.find((n) => n.includes(live.runId)); if (!fileName) return empty; const nameFromFile = fileName.endsWith(suffix) ? fileName.slice(0, -suffix.length) : fileName.replace(/\.js$/, ''); let body = ''; try { body = (await readFile(join(scriptsDir, fileName), 'utf-8')).slice(0, SCRIPT_READ_MAX_BYTES); } catch { return { workflowName: nameFromFile || undefined, phases: [] }; } // Scope name/summary/phases parsing to the `meta` literal so later script // declarations (schemas, prose) can't be mistaken for it. const metaBlock = extractMetaBlock(body) || body; return { workflowName: nameFromFile || extractMetaString(metaBlock, 'name') || undefined, summary: extractMetaString(metaBlock, 'description') || undefined, phases: extractMetaPhases(metaBlock), }; } /** * Parse a wf_.json into WorkflowRunInfo, STRIPPING the heavyweight * `script`/`scriptPath`/`result`/`logs` fields (the embedded script alone is * 15–660KB) so they never reach the cache, SSE, or routes. */ private async parseFile(file: DiscoveredRun): Promise { let raw: Record; try { raw = JSON.parse(await readFile(file.filePath, 'utf-8')) as Record; } catch { return null; // mid-write or malformed — next mtime change re-parses } if (!raw || typeof raw !== 'object') return null; const progress = Array.isArray(raw.workflowProgress) ? (raw.workflowProgress as RawProgressEntry[]) : []; const agents: WorkflowAgentInfo[] = progress .filter((e) => e && e.type === 'workflow_agent') .map((e) => this.toAgent(e)); const phases: WorkflowRunPhase[] = Array.isArray(raw.phases) ? (raw.phases as Array>).map((p) => ({ title: typeof p.title === 'string' ? p.title : '', detail: typeof p.detail === 'string' ? p.detail : '', })) : []; let lastProgress = 0; for (const a of agents) { if (typeof a.lastProgressAt === 'number' && a.lastProgressAt > lastProgress) lastProgress = a.lastProgressAt; } const startTime = typeof raw.startTime === 'number' ? raw.startTime : undefined; const lastActivityAt = lastProgress || startTime || 0; return { runId: typeof raw.runId === 'string' ? raw.runId : file.runId, workflowName: typeof raw.workflowName === 'string' ? raw.workflowName : undefined, status: typeof raw.status === 'string' ? raw.status : undefined, summary: typeof raw.summary === 'string' ? raw.summary : undefined, agentCount: typeof raw.agentCount === 'number' ? raw.agentCount : undefined, totalTokens: typeof raw.totalTokens === 'number' ? raw.totalTokens : undefined, totalToolCalls: typeof raw.totalToolCalls === 'number' ? raw.totalToolCalls : undefined, durationMs: typeof raw.durationMs === 'number' ? raw.durationMs : undefined, startTime, timestamp: typeof raw.timestamp === 'string' ? raw.timestamp : undefined, defaultModel: typeof raw.defaultModel === 'string' ? raw.defaultModel : undefined, taskId: typeof raw.taskId === 'string' ? raw.taskId : undefined, error: typeof raw.error === 'string' ? raw.error : undefined, phases, agents, sessionUuid: file.sessionUuid, projectHash: file.projectHash, lastActivityAt, }; } private toAgent(e: RawProgressEntry): WorkflowAgentInfo { return { index: typeof e.index === 'number' ? e.index : 0, label: typeof e.label === 'string' ? e.label : '', phaseIndex: typeof e.phaseIndex === 'number' ? e.phaseIndex : 0, phaseTitle: typeof e.phaseTitle === 'string' ? e.phaseTitle : '', model: typeof e.model === 'string' ? e.model : '', state: typeof e.state === 'string' ? e.state : 'start', queuedAt: e.queuedAt, lastProgressAt: e.lastProgressAt, promptPreview: truncate(e.promptPreview, PROMPT_PREVIEW_MAX), agentId: e.agentId, startedAt: e.startedAt, attempt: e.attempt, tokens: e.tokens, toolCalls: e.toolCalls, lastToolName: e.lastToolName, lastToolSummary: truncate(e.lastToolSummary, RESULT_PREVIEW_MAX), durationMs: e.durationMs, resultPreview: truncate(e.resultPreview, RESULT_PREVIEW_MAX), }; } private ensureDirWatcher(workflowsDir: string): void { if (this.dirWatchers.has(workflowsDir)) return; try { const watcher = chokidarWatch(workflowsDir, { depth: 0, awaitWriteFinish: { stabilityThreshold: 200 }, ignoreInitial: true, persistent: false, }); const handler = () => this.poll(); watcher.on('add', handler); watcher.on('change', handler); watcher.on('unlink', handler); // subagents/workflows/ children are wf_/ DIRS — catch their add/remove too. watcher.on('addDir', handler); watcher.on('unlinkDir', handler); watcher.on('error', () => { // chokidar surfaced an error for this dir — drop the watcher; poll still covers it. this.removeDirWatcher(workflowsDir); }); this.dirWatchers.set(workflowsDir, watcher); } catch { // Watch setup failed — periodic poll still discovers changes. } } private removeDirWatcher(workflowsDir: string): void { const watcher = this.dirWatchers.get(workflowsDir); if (watcher) { watcher.close().catch(() => {}); this.dirWatchers.delete(workflowsDir); } } } /** Process-wide singleton (mirrors subagentWatcher / imageWatcher). */ export const workflowRunWatcher = new WorkflowRunWatcher();