diff --git a/packages/cli/src/ui/commands/workflowsCommand.test.ts b/packages/cli/src/ui/commands/workflowsCommand.test.ts index bb22bfeeab6..62d6e2730fa 100644 --- a/packages/cli/src/ui/commands/workflowsCommand.test.ts +++ b/packages/cli/src/ui/commands/workflowsCommand.test.ts @@ -8,7 +8,7 @@ import { vi, describe, it, expect, beforeEach, afterEach } from 'vitest'; import { promises as fs } from 'node:fs'; import os from 'node:os'; import path from 'node:path'; -import { workflowsCommand } from './workflowsCommand.js'; +import { workflowsCommand, snapshotToTask } from './workflowsCommand.js'; import { type CommandContext } from './types.js'; import { createMockCommandContext } from '../../test-utils/mockCommandContext.js'; import type { WorkflowTask, WorkflowSnapshot } from '@qwen-code/qwen-code-core'; @@ -28,10 +28,14 @@ function entry(overrides: Partial = {}): WorkflowTask { isBackgrounded: true, abortController: new AbortController(), currentPhase: null, + currentPhaseVisitId: null, phases: [], + phaseVisits: [], + dispatches: [], agentsDispatched: 0, agentsCompleted: 0, recentLogs: [], + events: [], tokensSpent: 0, tokenBudgetTotal: null, perPhaseTokens: new Map(), @@ -670,3 +674,28 @@ describe('workflowsCommand', () => { }); }); }); + +describe('snapshotToTask', () => { + it('preserves persisted lineage fields across the restart boundary', () => { + const task = snapshotToTask({ + runId: 'wf_lineage', + sourceRunId: 'wf_origin', + startMode: 'retry', + meta: null, + status: 'completed', + script: '', + phases: [], + agentsDispatched: 0, + agentsCompleted: 0, + tokensSpent: 0, + tokenBudgetTotal: null, + perPhaseTokens: [], + recentLogs: [], + startTime: 1_700_000_000_000, + endTime: 1_700_000_005_000, + }); + + expect(task.sourceRunId).toBe('wf_origin'); + expect(task.startMode).toBe('retry'); + }); +}); diff --git a/packages/cli/src/ui/commands/workflowsCommand.ts b/packages/cli/src/ui/commands/workflowsCommand.ts index 335dbf08d4c..bcb3182370c 100644 --- a/packages/cli/src/ui/commands/workflowsCommand.ts +++ b/packages/cli/src/ui/commands/workflowsCommand.ts @@ -21,7 +21,7 @@ import { formatDuration, formatTokenCount } from '../utils/formatters.js'; * `outputFile`, etc.) are filled with inert values — a snapshot is always * terminal, so the controls that read those fields are never reached. */ -function snapshotToTask(s: WorkflowSnapshot): WorkflowTask { +export function snapshotToTask(s: WorkflowSnapshot): WorkflowTask { return { id: s.runId, kind: 'workflow', @@ -30,10 +30,16 @@ function snapshotToTask(s: WorkflowSnapshot): WorkflowTask { meta: s.meta, status: s.status, currentPhase: null, + currentPhaseVisitId: null, phases: s.phases ?? [], + phaseVisits: s.phaseVisits ?? [], + dispatches: s.dispatches ?? [], + sourceRunId: s.sourceRunId, + startMode: s.startMode, agentsDispatched: s.agentsDispatched ?? 0, agentsCompleted: s.agentsCompleted ?? 0, recentLogs: s.recentLogs ?? [], + events: s.events ?? [], tokensSpent: s.tokensSpent ?? 0, tokenBudgetTotal: s.tokenBudgetTotal ?? null, perPhaseTokens: new Map(s.perPhaseTokens ?? []), @@ -48,7 +54,7 @@ function snapshotToTask(s: WorkflowSnapshot): WorkflowTask { outputOffset: 0, notified: true, abortController: new AbortController(), - } as WorkflowTask; + }; } /** diff --git a/packages/core/src/agents/runtime/workflow-journal.test.ts b/packages/core/src/agents/runtime/workflow-journal.test.ts index 75b746be168..581786beb61 100644 --- a/packages/core/src/agents/runtime/workflow-journal.test.ts +++ b/packages/core/src/agents/runtime/workflow-journal.test.ts @@ -154,6 +154,23 @@ describe('WorkflowJournal', () => { expect(replay.started.get('k1')).toHaveLength(1); }); + it('drain waits for fire-and-forget appends', async () => { + const j = new WorkflowJournal(path.join(dir, 'sub', 'journal.jsonl')); + void j.append({ type: 'started', key: 'k1', agentId: '1' }); + void j.append({ + type: 'result', + key: 'k1', + agentId: '1', + result: 'done', + }); + + await j.drain(); + + const replay = await j.load(); + expect(replay.started.get('k1')).toHaveLength(1); + expect(replay.results.get('k1')?.result).toBe('done'); + }); + it('load on a missing file returns empty maps', async () => { const j = new WorkflowJournal(path.join(dir, 'nope.jsonl')); const replay = await j.load(); diff --git a/packages/core/src/agents/runtime/workflow-journal.ts b/packages/core/src/agents/runtime/workflow-journal.ts index 714edfb258a..0da1db61d11 100644 --- a/packages/core/src/agents/runtime/workflow-journal.ts +++ b/packages/core/src/agents/runtime/workflow-journal.ts @@ -179,6 +179,8 @@ export function buildReplay(entries: JournalEntry[]): JournalReplay { * failure must not fail the dispatch). */ export class WorkflowJournal { + private pending = Promise.resolve(); + constructor(readonly path: string) {} /** Load + parse all entries into replay maps. Empty maps if no file. */ @@ -194,6 +196,13 @@ export class WorkflowJournal { /** Append one entry. Rejects only on I/O error (callers `.catch`). */ append(entry: JournalEntry): Promise { - return writeLine(this.path, entry); + const operation = this.pending.then(() => writeLine(this.path, entry)); + this.pending = operation.catch(() => undefined); + return operation; + } + + /** Wait until every append issued so far has settled. */ + drain(): Promise { + return this.pending; } } diff --git a/packages/core/src/agents/runtime/workflow-orchestrator.test.ts b/packages/core/src/agents/runtime/workflow-orchestrator.test.ts index 96f008118bb..fd2d0c83c6c 100644 --- a/packages/core/src/agents/runtime/workflow-orchestrator.test.ts +++ b/packages/core/src/agents/runtime/workflow-orchestrator.test.ts @@ -451,6 +451,238 @@ describe('WorkflowOrchestrator', () => { }); }); + it('records dependency tails across sequential, parallel, and pipeline dispatches', async () => { + const orchestrator = new WorkflowOrchestrator( + async (prompt) => `mock:${prompt}`, + ); + const queued: Array<{ + id: string; + label?: string; + dependsOn: string[]; + }> = []; + + await orchestrator.run({ + script: ` + phase('Inspect'); + await agent('inspect', { label: 'inspect' }); + phase('Review'); + await parallel([ + () => agent('correctness', { label: 'correctness' }), + () => agent('architecture', { label: 'architecture' }), + ]); + phase('Fix'); + await pipeline( + ['a', 'b'], + (_prev, item) => agent('verify ' + item, { label: 'verify-' + item }), + (_prev, item) => agent('fix ' + item, { label: 'fix-' + item }), + ); + `, + args: undefined, + emitter: { + dispatchQueued: (event) => queued.push(event), + }, + }); + + const ids = new Map(queued.map((event) => [event.label, event.id])); + const dependencies = (label: string) => + queued + .find((event) => event.label === label)! + .dependsOn.map((id) => queued.find((event) => event.id === id)!.label); + + expect(dependencies('inspect')).toEqual([]); + expect(dependencies('correctness')).toEqual(['inspect']); + expect(dependencies('architecture')).toEqual(['inspect']); + expect(dependencies('verify-a')).toEqual(['correctness', 'architecture']); + expect(dependencies('verify-b')).toEqual(['correctness', 'architecture']); + expect(dependencies('fix-a')).toEqual(['verify-a']); + expect(dependencies('fix-b')).toEqual(['verify-b']); + expect(new Set(ids.values()).size).toBe(7); + }); + + it.each(['parallel', 'pipeline'] as const)( + 'preserves newer parent dependencies when an un-awaited %s settles', + async (kind) => { + const queued: Array<{ + id: string; + label?: string; + dependsOn: string[]; + }> = []; + const orchestrator = new WorkflowOrchestrator( + async (prompt) => `${prompt}-done`, + ); + const fanout = + kind === 'parallel' + ? `parallel([() => agent('fanout', { label: 'fanout' })])` + : `pipeline([0], () => agent('fanout', { label: 'fanout' }))`; + + await orchestrator.run({ + script: ` + const pending = ${fanout}; + await agent('parent', { label: 'parent' }); + await pending; + await agent('joined', { label: 'joined' }); + `, + args: undefined, + scheduler: new WorkflowDispatchScheduler(2), + emitter: { + dispatchQueued: (event) => queued.push(event), + }, + }); + + const labelsById = new Map( + queued.map((event) => [event.id, event.label]), + ); + const joined = queued.find((event) => event.label === 'joined'); + expect(joined?.dependsOn.map((id) => labelsById.get(id)).sort()).toEqual([ + 'fanout', + 'parent', + ]); + }, + ); + + it('emits queued, started, and settled lifecycle events for one dispatch', async () => { + const orchestrator = new WorkflowOrchestrator(async () => 'done'); + const events: string[] = []; + + await orchestrator.run({ + script: `await agent('inspect', { label: 'scope' });`, + args: undefined, + emitter: { + dispatchQueued: ({ id, label }) => events.push(`queued:${id}:${label}`), + dispatchStarted: (id) => events.push(`started:${id}`), + dispatchSettled: (id, error) => + events.push(`settled:${id}:${error ?? 'ok'}`), + }, + }); + + expect(events).toHaveLength(3); + const dispatchId = events[0]!.split(':')[1]; + expect(events).toEqual([ + `queued:${dispatchId}:scope`, + `started:${dispatchId}`, + `settled:${dispatchId}:ok`, + ]); + }); + + it('passes the recorded dispatch id into the production dispatch boundary', async () => { + const receivedIds: Array = []; + const orchestrator = new WorkflowOrchestrator( + async (_prompt, _opts, dispatchId) => { + receivedIds.push(dispatchId); + return 'done'; + }, + ); + + await orchestrator.run({ + script: `await agent('inspect', { label: 'scope' });`, + args: undefined, + }); + + expect(receivedIds).toEqual(['dispatch-1']); + }); + + it('preserves the dependency tail across empty parallel helpers', async () => { + const orchestrator = new WorkflowOrchestrator(async () => 'done'); + const queued: Array<{ + id: string; + label?: string; + dependsOn: string[]; + }> = []; + + await orchestrator.run({ + script: ` + await agent('before', { label: 'before' }); + await parallel([]); + await agent('after parallel', { label: 'after-parallel' }); + await pipeline([], () => agent('unused')); + await agent('after pipeline', { label: 'after-pipeline' }); + `, + args: undefined, + emitter: { + dispatchQueued: (event) => queued.push(event), + }, + }); + + const labelById = new Map( + queued.map((event) => [event.id, event.label ?? event.id]), + ); + const dependsOn = (label: string) => + queued + .find((event) => event.label === label)! + .dependsOn.map((id) => labelById.get(id)); + + expect(dependsOn('after-parallel')).toEqual(['before']); + expect(dependsOn('after-pipeline')).toEqual(['after-parallel']); + }); + + it('does not re-inject inherited tails from a fan-out branch that never dispatches', async () => { + const orchestrator = new WorkflowOrchestrator( + async (prompt) => `${prompt}`, + ); + const queued: Array<{ + id: string; + label?: string; + dependsOn: string[]; + }> = []; + + await orchestrator.run({ + script: ` + await agent('a', { label: 'a' }); + const pending = parallel([ + () => agent('b', { label: 'b' }), + () => 42, + ]); + await agent('m', { label: 'm' }); + await pending; + await agent('z', { label: 'z' }); + `, + args: undefined, + scheduler: new WorkflowDispatchScheduler(2), + emitter: { + dispatchQueued: (event) => queued.push(event), + }, + }); + + const labelsById = new Map(queued.map((event) => [event.id, event.label])); + const dependsOn = (label: string) => + queued + .find((event) => event.label === label)! + .dependsOn.map((id) => labelsById.get(id)) + .sort(); + + expect(dependsOn('b')).toEqual(['a']); + expect(dependsOn('m')).toEqual(['a']); + // The no-dispatch branch must not re-inject the ancestor 'a' edge. + expect(dependsOn('z')).toEqual(['b', 'm']); + }); + + it('keeps inherited tails when no fan-out branch issues a dispatch', async () => { + const orchestrator = new WorkflowOrchestrator(async () => 'done'); + const queued: Array<{ + id: string; + label?: string; + dependsOn: string[]; + }> = []; + + await orchestrator.run({ + script: ` + await agent('before', { label: 'before' }); + await parallel([() => 1, () => 2]); + await agent('after', { label: 'after' }); + `, + args: undefined, + emitter: { + dispatchQueued: (event) => queued.push(event), + }, + }); + + const labelById = new Map( + queued.map((event) => [event.id, event.label ?? event.id]), + ); + const after = queued.find((event) => event.label === 'after'); + expect(after?.dependsOn.map((id) => labelById.get(id))).toEqual(['before']); + }); + it('emitter subscriber errors do not break the run (defensive try/catch)', async () => { const orchestrator = new WorkflowOrchestrator( async (prompt) => `mock:${prompt}`, @@ -472,6 +704,15 @@ describe('WorkflowOrchestrator', () => { logAppended: () => { throw new Error('log-subscriber-boom'); }, + dispatchQueued: () => { + throw new Error('queued-subscriber-boom'); + }, + dispatchStarted: () => { + throw new Error('started-subscriber-boom'); + }, + dispatchSettled: () => { + throw new Error('settled-subscriber-boom'); + }, }; const outcome = await orchestrator.run({ script: ` @@ -847,9 +1088,13 @@ describe('WorkflowOrchestrator', () => { const orchestrator = new WorkflowOrchestrator(() => Promise.reject(new Error('nested-boom')), ); + const appendedLogs: string[] = []; const outcome = await orchestrator.run({ script: `return 'parent:' + (await workflow('child'));`, args: undefined, + emitter: { + logAppended: (line) => appendedLogs.push(line), + }, resolveSavedWorkflow: async () => ({ // The fire-and-forget dispatch fails but the nested script // still completes — the only trace of the failure is the @@ -861,6 +1106,9 @@ describe('WorkflowOrchestrator', () => { expect(outcome.logs).toContain( 'dispatch failed (result not consumed): nested-boom', ); + expect(appendedLogs).toEqual([ + 'dispatch failed (result not consumed): nested-boom', + ]); }); it('keeps a nested agent result behind the shared pause gate', async () => { @@ -1834,24 +2082,31 @@ describe('createProductionDispatch', () => { signal?.addEventListener('abort', () => resolve(), { once: true }); }); }; - const installed: AgentEventEmitter[] = []; + const installed: Array<{ + emitter: AgentEventEmitter; + dispatchId?: string; + }> = []; const cleaned: AgentEventEmitter[] = []; const dispatch = createProductionDispatch( fakeConfig(), undefined, undefined, - (emitter) => { - installed.push(emitter); + (emitter, dispatchId) => { + installed.push({ emitter, dispatchId }); return () => cleaned.push(emitter); }, ); - await expect(dispatch('hello', { label: 'h1', stallMs: 5 })).resolves.toBe( - 'headless-said:hello', - ); + await expect( + dispatch('hello', { label: 'h1', stallMs: 5 }, 'dispatch-1'), + ).resolves.toBe('headless-said:hello'); expect(installed).toHaveLength(2); - expect(installed[0]).not.toBe(installed[1]); - expect(cleaned).toEqual(installed); + expect(installed[0]?.emitter).not.toBe(installed[1]?.emitter); + expect(installed.map(({ dispatchId }) => dispatchId)).toEqual([ + 'dispatch-1', + 'dispatch-1', + ]); + expect(cleaned).toEqual(installed.map(({ emitter }) => emitter)); }); it('bubbles a production subagent approval through the run registry and resumes after ProceedOnce', async () => { diff --git a/packages/core/src/agents/runtime/workflow-orchestrator.ts b/packages/core/src/agents/runtime/workflow-orchestrator.ts index 15ae25e1479..d9cf519e078 100644 --- a/packages/core/src/agents/runtime/workflow-orchestrator.ts +++ b/packages/core/src/agents/runtime/workflow-orchestrator.ts @@ -5,6 +5,7 @@ */ import { randomBytes } from 'node:crypto'; +import { AsyncLocalStorage } from 'node:async_hooks'; import * as os from 'node:os'; import type { Config } from '../../config/config.js'; import { @@ -404,6 +405,7 @@ export interface WorkflowRunOutcome { export type WorkflowAgentDispatch = ( prompt: string, opts: WorkflowAgentOpts, + dispatchId?: string, ) => Promise; function generateRunId(): string { @@ -465,9 +467,12 @@ export function createProductionDispatch( * just without budget recording. */ onTokens?: (outputTokens: number, opts: WorkflowAgentOpts) => void, - bridgeApprovalEvents?: (emitter: AgentEventEmitter) => () => void, + bridgeApprovalEvents?: ( + emitter: AgentEventEmitter, + dispatchId?: string, + ) => () => void, ): WorkflowAgentDispatch { - return async (prompt, opts) => { + return async (prompt, opts, dispatchId) => { // An empty or non-string prompt seeds no `user` record, so the // transcript would carry no evidence of what the agent was asked — // and a stall retry on top would open the file with an orphaned @@ -499,7 +504,10 @@ export function createProductionDispatch( return runStallResilient( async (attemptSignal, emitter) => { attempt += 1; - const cleanupApprovalBridge = bridgeApprovalEvents?.(emitter); + const cleanupApprovalBridge = bridgeApprovalEvents?.( + emitter, + dispatchId, + ); const cleanupTranscript = attachDispatchTranscript( config, workflowAgentId, @@ -1646,8 +1654,33 @@ export class WorkflowOrchestrator { // cap regardless of launch path (increment-then-check: calls 1..max pass, // the (max+1)th throws), and scheduler.run enforces the dispatch window. let agentCount = 0; + let dispatchTraceCount = 0; const emitter = req.emitter; const budget = req.budget; + const dependencyContext = new AsyncLocalStorage<{ tails: string[] }>(); + const issueDispatchTrace = ( + prompt: string, + opts: WorkflowAgentOpts, + cached = false, + ): string => { + const id = `dispatch-${(dispatchTraceCount += 1)}`; + const store = dependencyContext.getStore(); + const dependsOn = Array.from(new Set(store?.tails ?? [])); + if (store) store.tails = [id]; + try { + emitter?.dispatchQueued?.({ + id, + ...(typeof opts.label === 'string' ? { label: opts.label } : {}), + prompt, + dependsOn, + queuedAt: Date.now(), + ...(cached ? { cached: true } : {}), + }); + } catch (e) { + debugLogger.warn('emitter.dispatchQueued threw:', e); + } + return id; + }; // P6: resume journal state. `prefixHash` chains across sequential // agent() calls; `hadMiss` enforces the "first miss invalidates the @@ -1700,6 +1733,7 @@ export class WorkflowOrchestrator { } const label = typeof opts.label === 'string' ? opts.label : undefined; + const dispatchId = issueDispatchTrace(prompt, opts, true); try { emitter?.agentDispatched?.(label); } catch (e) { @@ -1710,6 +1744,11 @@ export class WorkflowOrchestrator { } catch (e) { debugLogger.warn('emitter.agentCompleted threw:', e); } + try { + emitter?.dispatchSettled?.(dispatchId, undefined, Date.now()); + } catch (e) { + debugLogger.warn('emitter.dispatchSettled threw:', e); + } // Resolve even if the gate aborts: rejecting an already-cached // result at teardown would surface an unobserved rejection for // fire-and-forget calls on a correctly-cancelled run. @@ -1781,6 +1820,7 @@ export class WorkflowOrchestrator { // settles (success or thrown) — defensive try/catch on both so a // subscriber error never propagates into the script. const label = typeof opts.label === 'string' ? opts.label : undefined; + const dispatchId = issueDispatchTrace(prompt, opts); try { emitter?.agentDispatched?.(label); } catch (e) { @@ -1801,10 +1841,20 @@ export class WorkflowOrchestrator { } catch (e) { debugLogger.warn('emitter.agentCompleted threw:', e); } + try { + emitter?.dispatchSettled?.(dispatchId, message, Date.now()); + } catch (e) { + debugLogger.warn('emitter.dispatchSettled threw:', e); + } }; return scheduler .run(async () => { try { + try { + emitter?.dispatchStarted?.(dispatchId, Date.now()); + } catch (e) { + debugLogger.warn('emitter.dispatchStarted threw:', e); + } // P5 R1 (Critical #2): re-check the gate at slot-acquire time so // queued thunks see budget updates from already-completed in- // flight dispatches. Without this, the entry gate above is @@ -1821,7 +1871,7 @@ export class WorkflowOrchestrator { budget.spent(), ); } - const result = await this.dispatch(prompt, opts); + const result = await this.dispatch(prompt, opts, dispatchId); emitCompletion(); // P6: append the live result to the journal so a later resume // serves it from cache. Only JSON-serializable results are @@ -1904,8 +1954,8 @@ export class WorkflowOrchestrator { ); }; - const parallelImpl = makeParallelImpl(signal); - const pipelineImpl = makePipelineImpl(signal); + const parallelImpl = makeParallelImpl(signal, dependencyContext); + const pipelineImpl = makePipelineImpl(signal, dependencyContext); // P-nested: build the host-side `workflow(nameOrRef, args)` impl. Only // wired at the top level (when a resolver is provided). The nested @@ -1947,13 +1997,9 @@ export class WorkflowOrchestrator { // so the parent can try/catch it like any other async failure. return await nestedSandbox.run(resolved.script); } finally { - // Nested logs (script log() lines AND the unconsumed- - // rejection mirror) reach no production surface on their - // own — getLogs() is only ever read on the top-level - // sandbox and the production emitter's logAppended is a - // deliberate no-op. Merge them into the parent run's logs - // at nested settlement (after the nested flush ran) so a - // failed nested dispatch leaves a visible trace. + // The shared emitter already publishes nested logs live. Merge + // them into the parent buffer without re-emitting so the final + // outcome retains the same lines exactly once. for (const line of nestedSandbox.getLogs()) { parentSandboxRef.current?.appendLog(line); } @@ -1975,7 +2021,9 @@ export class WorkflowOrchestrator { }); parentSandboxRef.current = sandbox; try { - const result = await sandbox.run(req.script); + const result = await dependencyContext.run({ tails: [] }, () => + sandbox.run(req.script), + ); return { runId, result, @@ -2091,7 +2139,8 @@ async function settleToNullArray( * array never reaches the script directly. */ function makeParallelImpl( - signal?: AbortSignal, + signal: AbortSignal | undefined, + dependencyContext: AsyncLocalStorage<{ tails: string[] }>, ): (thunks: Array<() => Promise>) => Promise { return (thunks) => { if (!Array.isArray(thunks)) { @@ -2111,7 +2160,28 @@ function makeParallelImpl( ); } } - return settleToNullArray(thunks, signal); + const parent = dependencyContext.getStore(); + const inheritedTails = parent?.tails ?? []; + const branches = thunks.map((thunk) => { + const store = { tails: [...inheritedTails] }; + return { + store, + thunk: () => dependencyContext.run(store, thunk), + }; + }); + return settleToNullArray( + branches.map(({ thunk }) => thunk), + signal, + ).then((result) => { + if (parent && branches.length > 0) { + parent.tails = mergeFanoutTails( + parent.tails, + inheritedTails, + branches.flatMap(({ store }) => store.tails), + ); + } + return result; + }); }; } @@ -2128,7 +2198,8 @@ function makeParallelImpl( * per-element vm-realm revival. */ function makePipelineImpl( - signal?: AbortSignal, + signal: AbortSignal | undefined, + dependencyContext: AsyncLocalStorage<{ tails: string[] }>, ): ( items: unknown[], ...stages: Array< @@ -2153,13 +2224,57 @@ function makePipelineImpl( ); } } - const chains = items.map( - (item, idx) => () => runPipelineChain(item, idx, stages), - ); - return settleToNullArray(chains, signal, 'pipeline'); + const parent = dependencyContext.getStore(); + const inheritedTails = parent?.tails ?? []; + const branches = items.map((item, idx) => { + const store = { tails: [...inheritedTails] }; + return { + store, + thunk: () => + dependencyContext.run(store, () => + runPipelineChain(item, idx, stages), + ), + }; + }); + return settleToNullArray( + branches.map(({ thunk }) => thunk), + signal, + 'pipeline', + ).then((result) => { + if (parent && branches.length > 0) { + parent.tails = mergeFanoutTails( + parent.tails, + inheritedTails, + branches.flatMap(({ store }) => store.tails), + ); + } + return result; + }); }; } +function mergeFanoutTails( + currentParentTails: readonly string[], + inheritedTails: readonly string[], + branchTails: readonly string[], +): string[] { + const inherited = new Set(inheritedTails); + // A dispatching branch always ends on its fresh dispatch id; a branch that + // never dispatched still carries its inherited seed. Merging that seed back + // would re-inject ancestor ids as redundant transitive dependsOn edges — + // unless no branch dispatched at all, where the inherited tails must pass + // through unchanged (same contract as an empty fan-out). + const newBranchTails = branchTails.filter((tail) => !inherited.has(tail)); + const merged = + newBranchTails.length > 0 + ? [ + ...newBranchTails, + ...currentParentTails.filter((tail) => !inherited.has(tail)), + ] + : currentParentTails; + return Array.from(new Set(merged)); +} + /** * Run one item through every stage in order. `null` is the universal drop * sentinel: a stage that returns `null` (or throws — surfaced as a rejection diff --git a/packages/core/src/agents/runtime/workflow-runner.test.ts b/packages/core/src/agents/runtime/workflow-runner.test.ts index a306f8ca33b..9302bef754f 100644 --- a/packages/core/src/agents/runtime/workflow-runner.test.ts +++ b/packages/core/src/agents/runtime/workflow-runner.test.ts @@ -10,17 +10,22 @@ import type { Config } from '../../config/config.js'; import { isTerminalWorkflowStatus, WorkflowRunRegistry, + type WorkflowTask, } from '../workflow-run-registry.js'; import { AgentEventEmitter } from './agent-events.js'; import { WorkflowRunner } from './workflow-runner.js'; const { createProductionDispatchMock, + journalWrites, logWorkflowRunMock, + writeLineMock, writeWorkflowSnapshotMock, } = vi.hoisted(() => ({ createProductionDispatchMock: vi.fn(), + journalWrites: [] as Array<() => void>, logWorkflowRunMock: vi.fn(), + writeLineMock: vi.fn(), writeWorkflowSnapshotMock: vi.fn().mockResolvedValue(undefined), })); @@ -32,6 +37,12 @@ vi.mock('../workflow-snapshot.js', () => ({ writeWorkflowSnapshot: writeWorkflowSnapshotMock, })); +vi.mock('../../utils/jsonl-utils.js', async (importOriginal) => { + const actual = + await importOriginal(); + return { ...actual, writeLine: writeLineMock }; +}); + vi.mock('./workflow-orchestrator.js', async (importOriginal) => { const actual = await importOriginal(); @@ -78,8 +89,12 @@ function observeSettlement(registry: WorkflowRunRegistry): { describe('WorkflowRunner', () => { beforeEach(() => { createProductionDispatchMock.mockReset(); + journalWrites.length = 0; logWorkflowRunMock.mockClear(); + writeLineMock.mockReset(); + writeLineMock.mockResolvedValue(undefined); writeWorkflowSnapshotMock.mockClear(); + writeWorkflowSnapshotMock.mockResolvedValue(undefined); }); it('passes the registry approval bridge only to production dispatch', async () => { @@ -103,16 +118,18 @@ describe('WorkflowRunner', () => { const bridgeApprovalEvents = createProductionDispatchMock.mock .calls[0]?.[3] as - | ((emitter: AgentEventEmitter) => () => void) + | ((emitter: AgentEventEmitter, dispatchId?: string) => () => void) | undefined; expect(bridgeApprovalEvents).toEqual(expect.any(Function)); const emitter = new AgentEventEmitter(); const cleanup = vi.fn(); productionBridge.mockReturnValue(cleanup); - expect(bridgeApprovalEvents?.(emitter)).toBe(cleanup); + expect(bridgeApprovalEvents?.(emitter, 'dispatch-1')).toBe(cleanup); expect(productionBridge).toHaveBeenCalledWith( productionHandle.runId, emitter, + 'dispatch-1', + production.registry.get(productionHandle.runId), ); const injected = configWithRegistry(); @@ -131,6 +148,90 @@ describe('WorkflowRunner', () => { expect(injectedBridge).not.toHaveBeenCalled(); }); + it('retains the original args needed to retry a failed run from its journal', async () => { + const { config, registry } = configWithRegistry(); + const args = { target: 'web-shell', checks: ['correctness'] }; + const handle = await WorkflowRunner.start({ + config, + signal: new AbortController().signal, + script: 'return args.target', + args, + runInBackground: true, + dispatch: async () => 'unused', + }); + + await handle.completion; + + expect(registry.get(handle.runId)?.args).toEqual(args); + }); + + it('records sandbox logs in the replay event ledger', async () => { + const { config, registry } = configWithRegistry(); + const handle = await WorkflowRunner.start({ + config, + signal: new AbortController().signal, + script: 'log("repository loaded"); return "done";', + args: undefined, + runInBackground: true, + dispatch: async () => 'unused', + }); + + await handle.completion; + + expect(registry.get(handle.runId)?.events).toEqual([ + expect.objectContaining({ + type: 'log', + message: 'repository loaded', + }), + expect.objectContaining({ type: 'workflow-completed' }), + ]); + }); + + it('keeps sandbox and registry phase projections equal for normalization-colliding titles', async () => { + const { config, registry } = configWithRegistry(); + const handle = await WorkflowRunner.start({ + config, + signal: new AbortController().signal, + script: + 'phase("\\u001b[1mBuild\\u001b[0m");' + + 'phase("Build");' + + 'await agent("x", { phase: "\\u001b[1mBuild\\u001b[0m" });' + + 'return 1;', + args: undefined, + runInBackground: true, + dispatch: async () => 'unused', + }); + + const settlement = await handle.completion; + + expect(settlement.ok).toBe(true); + const outcomePhases = settlement.ok ? settlement.outcome.phases : []; + expect(registry.get(handle.runId)?.phases).toEqual(['Build']); + expect(outcomePhases).toEqual(registry.get(handle.runId)?.phases); + }); + + it('records a journal retry as sourced from the same run', async () => { + const { config, registry } = configWithRegistry(); + const runId = 'wf_1234abcd'; + const handle = await WorkflowRunner.start({ + config, + signal: new AbortController().signal, + script: 'return "retried"', + args: undefined, + resumeFromRunId: runId, + runInBackground: true, + dispatch: async () => 'unused', + }); + + await handle.completion; + + expect(registry.get(runId)).toMatchObject({ + runId, + sourceRunId: runId, + startMode: 'retry', + }); + }); + it('keeps one registry-owned handle through exactly-once completion', async () => { const { config, registry } = configWithRegistry(); const observed = observeSettlement(registry); @@ -218,6 +319,39 @@ describe('WorkflowRunner', () => { expect(logWorkflowRunMock).toHaveBeenCalledTimes(2); }); + it('records caller-aborted dispatches as cancelled', async () => { + const { config, registry } = configWithRegistry(); + const caller = new AbortController(); + let rejectDispatch: ((error: Error) => void) | undefined; + const handle = await WorkflowRunner.start({ + config, + signal: caller.signal, + script: 'return await agent("work")', + args: undefined, + dispatch: () => + new Promise((_resolve, reject) => { + rejectDispatch = reject; + }), + }); + await vi.waitFor(() => expect(rejectDispatch).toBeDefined()); + + caller.abort(); + rejectDispatch?.(new Error('Request was aborted')); + await handle.completion; + + expect(registry.get(handle.runId)?.dispatches).toEqual([ + expect.objectContaining({ status: 'cancelled' }), + ]); + expect(registry.get(handle.runId)?.dispatches[0]).not.toHaveProperty( + 'error', + ); + expect(registry.get(handle.runId)?.events).not.toEqual( + expect.arrayContaining([ + expect.objectContaining({ type: 'dispatch-failed' }), + ]), + ); + }); + it('keeps background runs alive after the caller turn ends', async () => { const { config, registry } = configWithRegistry(); const observed = observeSettlement(registry); @@ -247,6 +381,92 @@ describe('WorkflowRunner', () => { expect(observed.abortCount()).toBe(1); }); + it('persists terminal runs without live fire-and-forget dispatches', async () => { + const { config, registry } = configWithRegistry(); + let snapshotDispatchStatuses: string[] | undefined; + writeWorkflowSnapshotMock.mockImplementation( + (_config, snapshotEntry: WorkflowTask) => { + snapshotDispatchStatuses = snapshotEntry.dispatches.map( + (dispatch) => dispatch.status, + ); + return Promise.resolve(); + }, + ); + const handle = await WorkflowRunner.start({ + config, + signal: new AbortController().signal, + script: 'agent("fire and forget"); return "done"', + args: undefined, + runInBackground: true, + dispatch: () => new Promise(() => undefined), + }); + + await expect(handle.completion).resolves.toMatchObject({ ok: true }); + + expect(registry.get(handle.runId)).toMatchObject({ status: 'completed' }); + expect(snapshotDispatchStatuses).toEqual(['cancelled']); + expect(registry.get(handle.runId)?.dispatches).toEqual([ + expect.objectContaining({ status: 'cancelled' }), + ]); + }); + + it('freezes snapshot and telemetry before late dispatches drain', async () => { + const { config, registry } = configWithRegistry(); + Object.assign(config, { + storage: { + getWorkflowRunJournalPath: () => 'probe-journal.jsonl', + }, + }); + writeLineMock.mockImplementation( + () => + new Promise((resolve) => { + journalWrites.push(resolve); + }), + ); + let snapshotAgentsCompleted: number | undefined; + writeWorkflowSnapshotMock.mockImplementation((_config, entry) => { + snapshotAgentsCompleted = entry.agentsCompleted; + return Promise.resolve(); + }); + let finishDispatch: ((result: string) => void) | undefined; + + const handle = await WorkflowRunner.start({ + config, + signal: new AbortController().signal, + script: 'agent("fire and forget"); return "done"', + args: undefined, + runInBackground: true, + dispatch: () => + new Promise((resolve) => { + finishDispatch = resolve; + }), + }); + + await vi.waitFor(() => { + expect(registry.get(handle.runId)?.status).toBe('completed'); + expect(journalWrites).toHaveLength(1); + }); + finishDispatch?.('late result'); + await vi.waitFor(() => + expect(registry.get(handle.runId)?.agentsCompleted).toBe(1), + ); + let settled = false; + void handle.completion.then(() => { + settled = true; + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(settled).toBe(false); + journalWrites[0]?.(); + await expect(handle.completion).resolves.toMatchObject({ ok: true }); + + const telemetry = logWorkflowRunMock.mock.calls[0]?.[1] as + | { agents_completed: number } + | undefined; + expect(telemetry?.agents_completed).toBe(0); + expect(snapshotAgentsCompleted).toBe(0); + for (const resolve of journalWrites) resolve(); + }); + it('holds an in-flight agent result until a paused run resumes', async () => { const { config, registry } = configWithRegistry(); let resolveDispatch: ((value: string) => void) | undefined; @@ -587,6 +807,56 @@ describe('WorkflowRunner', () => { expect(registry.get(runId)?.result).toBe('original'); }); + it('ignores late dispatch callbacks from a prior retry entry', async () => { + const { config, registry } = configWithRegistry(); + const runId = 'wf_1234abcd'; + let rejectOriginal: ((error: Error) => void) | undefined; + const original = await WorkflowRunner.start({ + config, + signal: new AbortController().signal, + script: 'agent("original"); throw new Error("original failed")', + args: undefined, + resumeFromRunId: runId, + runInBackground: true, + dispatch: () => + new Promise((_resolve, reject) => { + rejectOriginal = reject; + }), + }); + await expect(original.completion).resolves.toMatchObject({ ok: false }); + + let resolveRetry: ((value: string) => void) | undefined; + const retry = await WorkflowRunner.start({ + config, + signal: new AbortController().signal, + script: 'return await agent("retry")', + args: undefined, + resumeFromRunId: runId, + runInBackground: true, + dispatch: () => + new Promise((resolve) => { + resolveRetry = resolve; + }), + }); + await vi.waitFor(() => expect(resolveRetry).toBeDefined()); + + rejectOriginal?.(new Error('aborted by old controller')); + resolveRetry?.('done'); + await expect(retry.completion).resolves.toMatchObject({ ok: true }); + + expect(registry.get(runId)?.dispatches).toEqual([ + expect.objectContaining({ status: 'completed' }), + ]); + expect(registry.get(runId)?.events).not.toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: 'dispatch-failed', + error: 'aborted by old controller', + }), + ]), + ); + }); + it('classifies a background failure after caller abort as failed', async () => { const { config, registry } = configWithRegistry(); const caller = new AbortController(); diff --git a/packages/core/src/agents/runtime/workflow-runner.ts b/packages/core/src/agents/runtime/workflow-runner.ts index cc51e80fb4c..46f7f9df41d 100644 --- a/packages/core/src/agents/runtime/workflow-runner.ts +++ b/packages/core/src/agents/runtime/workflow-runner.ts @@ -35,6 +35,7 @@ import { resolveSavedWorkflowScript } from './workflow-saved.js'; export interface WorkflowRunnerOptions { config: Config; signal: AbortSignal; + toolUseId?: string; script?: string; scriptPath?: string; args: unknown; @@ -104,10 +105,14 @@ export class WorkflowRunner { throw new Error('Background workflow start was cancelled.'); } const callerWasAbortedBeforeStart = options.signal.aborted; + const registry = config.getWorkflowRunRegistry?.(); + let entry: WorkflowTask | undefined; + const isCurrentEntry = (): boolean => + registry === undefined || + (entry !== undefined && registry.get(runId) === entry); const controller = runInBackground ? createAbortController() : createChildAbortController(options.signal); - const registry = config.getWorkflowRunRegistry?.(); const dispatch = options.dispatch ?? createProductionDispatch( @@ -115,14 +120,22 @@ export class WorkflowRunner { controller.signal, (outputTokens) => budget.recordSpent(outputTokens), registry - ? (emitter) => registry.bridgeApprovalEvents(runId, emitter) + ? (emitter, dispatchId) => + isCurrentEntry() + ? registry.bridgeApprovalEvents( + runId, + emitter, + dispatchId, + entry, + ) + : () => undefined : undefined, ); const orchestrator = new WorkflowOrchestrator(dispatch); - let entry: WorkflowTask | undefined; try { entry = registry?.register({ runId, + toolUseId: options.toolUseId, meta: null, status: 'running', startTime: Date.now(), @@ -131,6 +144,13 @@ export class WorkflowRunner { tokenBudgetTotal: budget.total, script, scriptPath, + args: options.args, + ...(options.resumeFromRunId + ? { + sourceRunId: options.resumeFromRunId, + startMode: 'retry' as const, + } + : {}), isBackgrounded: runInBackground, }); } catch (error) { @@ -138,7 +158,7 @@ export class WorkflowRunner { throw error; } const emitUpdate = (): void => { - if (!entry || !options.onUpdate) return; + if (!entry || !options.onUpdate || !isCurrentEntry()) return; try { options.onUpdate(entry); } catch { @@ -147,22 +167,50 @@ export class WorkflowRunner { }; const emitter: WorkflowOrchestratorEmitter = { phaseStarted: (title) => { + if (!isCurrentEntry()) return; registry?.onPhaseStarted(runId, title); emitUpdate(); }, agentDispatched: () => { + if (!isCurrentEntry()) return; registry?.onAgentDispatched(runId); emitUpdate(); }, agentCompleted: () => { + if (!isCurrentEntry()) return; // No emitUpdate: budgetUpdated fires right after and renders both // updates together (avoids 2x TUI redraws per agent). registry?.onAgentCompleted(runId); }, - // Deliberate no-op: logs are snapshotted at terminal via - // setRecentLogs; per-line emit would cause up to 10k TUI redraws. - logAppended: () => {}, + dispatchQueued: (event) => { + if (!isCurrentEntry()) return; + registry?.onDispatchQueued(runId, event); + emitUpdate(); + }, + dispatchStarted: (dispatchId, startedAt) => { + if (!isCurrentEntry()) return; + registry?.onDispatchStarted(runId, dispatchId, startedAt); + emitUpdate(); + }, + dispatchSettled: (dispatchId, error, endedAt) => { + if (!isCurrentEntry()) return; + registry?.onDispatchSettled( + runId, + dispatchId, + error, + endedAt, + !runInBackground && options.signal.aborted, + ); + emitUpdate(); + }, + // The registry records this without firing a status update, avoiding a + // TUI redraw per line while retaining the real replay timestamp. + logAppended: (line) => { + if (!isCurrentEntry()) return; + registry?.onLogAppended(runId, line); + }, budgetUpdated: (spent, total) => { + if (!isCurrentEntry()) return; registry?.onBudgetUpdated(runId, spent, total); emitUpdate(); }, @@ -171,7 +219,10 @@ export class WorkflowRunner { const scheduler = new WorkflowDispatchScheduler( resolveConcurrencyLimit(), controller.signal, - ({ state }) => registry?.onDispatchStateChange(runId, state), + ({ state }) => { + if (!isCurrentEntry()) return; + registry?.onDispatchStateChange(runId, state); + }, ); const handle: WorkflowRunHandle = new WorkflowRunHandle( @@ -253,6 +304,7 @@ export class WorkflowRunner { duration_ms: (entry.endTime ?? entry.startTime) - entry.startTime, }); await writeWorkflowSnapshot(config, entry); + await journal?.drain(); try { logWorkflowRun(config, telemetryEvent); } catch { diff --git a/packages/core/src/agents/runtime/workflow-sandbox.test.ts b/packages/core/src/agents/runtime/workflow-sandbox.test.ts index b7d8357624e..2124881adfd 100644 --- a/packages/core/src/agents/runtime/workflow-sandbox.test.ts +++ b/packages/core/src/agents/runtime/workflow-sandbox.test.ts @@ -695,14 +695,17 @@ describe('createWorkflowSandbox security', () => { // SEC-I2: log() must cap at MAX_LOG_LINES and add a truncation marker. it('log() caps at MAX_LOG_LINES with a truncation marker', async () => { + const emitted: string[] = []; const sandbox = createWorkflowSandbox({ args: undefined, dispatch: async () => 'ignored', + emitter: { logAppended: (line) => emitted.push(line) }, }); await sandbox.run(`for (let i = 0; i < 10100; i++) log(i); return 0;`); const logs = sandbox.getLogs(); expect(logs.length).toBe(10_001); // 10_000 entries + 1 truncation marker expect(logs[10_000]).toMatch(/truncated/); + expect(emitted.at(-1)).toBe(logs[10_000]); }); // FIX-C5 (SEC-2-I1): same cap pattern for phases array — protects host diff --git a/packages/core/src/agents/runtime/workflow-sandbox.ts b/packages/core/src/agents/runtime/workflow-sandbox.ts index 30ed487f66d..47890cea9bc 100644 --- a/packages/core/src/agents/runtime/workflow-sandbox.ts +++ b/packages/core/src/agents/runtime/workflow-sandbox.ts @@ -363,6 +363,7 @@ function isRegexContext(source: string, i: number): boolean { import * as vm from 'node:vm'; import { createDebugLogger } from '../../utils/debugLogger.js'; +import { stripAnsiAndControl } from '../../utils/textUtils.js'; import type { WorkflowDispatchScheduler } from './workflow-dispatch-scheduler.js'; // Shared with workflow-orchestrator (avoids a duplicate createDebugLogger @@ -470,6 +471,19 @@ export interface WorkflowOrchestratorEmitter { agentDispatched?(label?: string): void; /** `dispatch(...)` settled (success or thrown). `error` set on rejection. */ agentCompleted?(label?: string, error?: string): void; + /** A dispatch was issued and joined to the runtime dependency graph. */ + dispatchQueued?(event: { + id: string; + label?: string; + prompt: string; + dependsOn: string[]; + queuedAt: number; + cached?: boolean; + }): void; + /** A queued dispatch acquired a scheduler slot. */ + dispatchStarted?(id: string, startedAt: number): void; + /** A dispatch reached a terminal state. */ + dispatchSettled?(id: string, error?: string, endedAt?: number): void; /** * P5: cumulative `spent` re-snapshot after each successful agent * completion. `total` is `null` when no per-run cap is set @@ -684,13 +698,7 @@ export interface WorkflowSandbox { getPhases(): string[]; /** Log lines emitted by the script in order. */ getLogs(): string[]; - /** - * Append a log line produced by a nested workflow run. Nested logs - * reach no production surface on their own (the nested sandbox's - * buffer is never read by the orchestrator), so the orchestrator - * merges them into the parent run's logs at nested settlement — - * including the nested unconsumed-rejection mirror lines. - */ + /** Merge a nested workflow log into the parent buffer without re-emitting. */ appendLog(line: string): void; /** * The script's `export const meta = {...}` declaration, validated and @@ -751,26 +759,36 @@ export function createWorkflowSandbox(opts: SandboxOptions): WorkflowSandbox { const phases: string[] = []; const logs: string[] = []; - const safeLog = (msg: unknown): void => { + const emitLog = (line: string): void => { + try { + opts.emitter?.logAppended?.(line); + } catch (e) { + debugLogger.warn('emitter.logAppended threw:', e); + } + }; + + const safeLog = (msg: unknown, notify = true): void => { if (logs.length < MAX_LOG_LINES) { const line = String(msg); logs.push(line); // P4b: emit to host-side subscriber (registry). Defensive try/catch // because a subscriber error must not interrupt script execution // — the script body has no business knowing about UI plumbing. - try { - opts.emitter?.logAppended?.(line); - } catch (e) { - debugLogger.warn('emitter.logAppended threw:', e); - } + if (notify) emitLog(line); } else if (logs.length === MAX_LOG_LINES) { - logs.push(`[workflow log truncated at ${MAX_LOG_LINES} lines]`); + const line = `[workflow log truncated at ${MAX_LOG_LINES} lines]`; + logs.push(line); + if (notify) emitLog(line); } }; const safePhase = (title: string): void => { if (phases.length < MAX_PHASE_ENTRIES) { - const t = String(title); + // Normalize before collapse/push/emit so the sandbox list and the + // registry mirror (same rule at its boundary) compare the same value: + // titles colliding only after ANSI/control stripping or the 200-char + // cap previously diverged the two phase surfaces of the same run. + const t = stripAnsiAndControl(String(title)).slice(0, 200) || 'phase'; // R7 (wenshao): collapse consecutive identical titles so the // sandbox is the single source of truth for the phase list. // Without this, `outcome.phases` (terminal `returnDisplay` JSON) @@ -1876,7 +1894,7 @@ export function createWorkflowSandbox(opts: SandboxOptions): WorkflowSandbox { }, getPhases: () => [...phases], getLogs: () => [...logs], - appendLog: (line: string) => safeLog(line), + appendLog: (line: string) => safeLog(line, false), getMeta: () => extractedMeta, }; } diff --git a/packages/core/src/agents/workflow-run-registry.test.ts b/packages/core/src/agents/workflow-run-registry.test.ts index cdddaf139a8..54fdc2f9817 100644 --- a/packages/core/src/agents/workflow-run-registry.test.ts +++ b/packages/core/src/agents/workflow-run-registry.test.ts @@ -64,6 +64,69 @@ function approvalEvent( } describe('WorkflowRunRegistry', () => { + it('records rerun lineage and notifies status observers', () => { + const r = new WorkflowRunRegistry(); + const onStatusChange = vi.fn(); + r.setStatusChangeCallback(onStatusChange); + r.register(reg('wf_rerun')); + onStatusChange.mockClear(); + + expect(r.setLineage('wf_rerun', 'wf_source', 'rerun')).toBe(true); + expect(r.get('wf_rerun')).toMatchObject({ + sourceRunId: 'wf_source', + startMode: 'rerun', + }); + expect(onStatusChange).toHaveBeenCalledWith( + expect.objectContaining({ runId: 'wf_rerun' }), + ); + expect(r.setLineage('wf_missing', 'wf_source', 'rerun')).toBe(false); + }); + + it('binds a pending approval to the dispatch that owns its event channel', () => { + const r = new WorkflowRunRegistry(); + r.register(reg('wf_dispatch_approval')); + r.setApprovalChangeCallback(() => {}); + r.onDispatchQueued('wf_dispatch_approval', { + id: 'dispatch-1', + prompt: 'Review the change', + label: 'Correctness', + dependsOn: [], + queuedAt: 1_700_000_000_010, + }); + const emitter = new AgentEventEmitter(); + r.bridgeApprovalEvents('wf_dispatch_approval', emitter, 'dispatch-1'); + + emitter.emit( + AgentEventType.TOOL_WAITING_APPROVAL, + approvalEvent({ subagentId: 'correctness-agent-1' }), + ); + + expect(r.get('wf_dispatch_approval')?.dispatches[0]).toMatchObject({ + id: 'dispatch-1', + subagentId: 'correctness-agent-1', + }); + }); + + it('ignores approval events from a replaced run entry', () => { + const r = new WorkflowRunRegistry(); + r.setApprovalChangeCallback(() => {}); + const original = r.register(reg('wf_replaced_approval')); + const emitter = new AgentEventEmitter(); + const cleanup = r.bridgeApprovalEvents( + original.runId, + emitter, + undefined, + original, + ); + r.complete(original.runId, 'done', 1_700_000_000_200); + const replacement = r.register(reg(original.runId)); + + emitter.emit(AgentEventType.TOOL_WAITING_APPROVAL, approvalEvent()); + cleanup(); + + expect(replacement.pendingApprovals).toEqual([]); + }); + it('parks a workflow-agent approval and resolves it exactly once', async () => { const r = new WorkflowRunRegistry(); r.register(reg('wf_approval')); @@ -102,6 +165,14 @@ describe('WorkflowRunRegistry', () => { expect(approval).not.toHaveProperty('args'); expect(approval).not.toHaveProperty('respond'); expect(onApprovalChange).toHaveBeenCalledTimes(1); + expect(r.get('wf_approval')?.events).toEqual([ + { + id: 'event-1', + type: 'approval-requested', + at: 1_700_000_000_100, + name: 'Shell', + }, + ]); await expect( r.resolvePendingApproval( @@ -122,6 +193,14 @@ describe('WorkflowRunRegistry', () => { ToolConfirmationOutcome.ProceedOnce, undefined, ); + expect(r.get('wf_approval')?.events[1]).toMatchObject({ + id: 'event-2', + type: 'approval-settled', + name: 'Shell', + }); + expect(r.get('wf_approval')?.events[1]).not.toHaveProperty('approvalId'); + expect(r.get('wf_approval')?.events[1]).not.toHaveProperty('callId'); + expect(r.get('wf_approval')?.events[1]).not.toHaveProperty('description'); cleanup(); }); @@ -549,6 +628,9 @@ describe('WorkflowRunRegistry', () => { expect(respond).toHaveBeenCalledWith(ToolConfirmationOutcome.Cancel); }); expect(r.get('wf_sync_failed_channel')?.pendingApprovals).toEqual([]); + expect( + r.get('wf_sync_failed_channel')?.events.map((event) => event.type), + ).toEqual(['approval-requested', 'approval-settled']); }); it('fails the run and drains siblings when resolving respond throws', async () => { @@ -660,6 +742,11 @@ describe('WorkflowRunRegistry', () => { expect(r.get('wf_tool_result')?.pendingApprovals).toEqual([]); expect(respond).not.toHaveBeenCalled(); + expect(r.get('wf_tool_result')?.events.at(-1)).toMatchObject({ + type: 'approval-settled', + at: 1_700_000_000_200, + name: 'Shell', + }); }); it('aborts the host request signal when an attempt cleans up', async () => { @@ -788,6 +875,250 @@ describe('WorkflowRunRegistry', () => { expect(e.agentsCompleted).toBe(1); }); + it('records phase visits and dispatch lifecycle without inferring dependencies', () => { + const r = new WorkflowRunRegistry(); + const entry = r.register(reg('wf_graph')); + + r.onPhaseStarted(entry.runId, 'Inspect', 1_100); + r.onDispatchQueued(entry.runId, { + id: 'dispatch-1', + label: 'Scope mapper', + prompt: 'Inspect the repository', + dependsOn: [], + queuedAt: 1_110, + }); + r.onDispatchStarted(entry.runId, 'dispatch-1', 1_120); + r.onDispatchSettled(entry.runId, 'dispatch-1', undefined, 1_180); + r.onPhaseStarted(entry.runId, 'Review', 1_200); + r.onDispatchQueued(entry.runId, { + id: 'dispatch-2', + label: 'Correctness', + prompt: 'Review correctness', + dependsOn: ['dispatch-1'], + queuedAt: 1_210, + }); + + expect(entry.phaseVisits).toEqual([ + { + id: 'phase-1', + index: 0, + title: 'Inspect', + startedAt: 1_100, + endedAt: 1_200, + }, + { + id: 'phase-2', + index: 1, + title: 'Review', + startedAt: 1_200, + }, + ]); + expect(entry.dispatches).toEqual([ + expect.objectContaining({ + id: 'dispatch-1', + phaseVisitId: 'phase-1', + status: 'completed', + startedAt: 1_120, + endedAt: 1_180, + dependsOn: [], + }), + expect.objectContaining({ + id: 'dispatch-2', + phaseVisitId: 'phase-2', + status: 'queued', + dependsOn: ['dispatch-1'], + }), + ]); + }); + + it('records the runtime sequence used by workflow replay', () => { + const r = new WorkflowRunRegistry(); + const onStatusChange = vi.fn(); + r.setStatusChangeCallback(onStatusChange); + const entry = r.register(reg('wf_events')); + onStatusChange.mockClear(); + + r.onPhaseStarted(entry.runId, 'Inspect', 1_100); + r.onLogAppended( + entry.runId, + '\u001b[31mrepository\u0000 loaded\u001b[0m', + 1_105, + ); + expect(onStatusChange).toHaveBeenCalledTimes(1); + r.onDispatchQueued(entry.runId, { + id: 'dispatch-1', + label: 'Correctness', + prompt: 'Review correctness', + dependsOn: [], + queuedAt: 1_110, + }); + r.onDispatchStarted(entry.runId, 'dispatch-1', 1_120); + r.onDispatchSettled(entry.runId, 'dispatch-1', undefined, 1_180); + r.complete(entry.runId, 'done', 1_200); + + expect(entry.recentLogs).toEqual(['repository loaded']); + expect(entry.events).toEqual([ + { + id: 'event-1', + type: 'phase-started', + at: 1_100, + phaseVisitId: 'phase-1', + title: 'Inspect', + }, + { + id: 'event-2', + type: 'log', + at: 1_105, + message: 'repository loaded', + }, + { + id: 'event-3', + type: 'dispatch-queued', + at: 1_110, + dispatchId: 'dispatch-1', + }, + { + id: 'event-4', + type: 'dispatch-started', + at: 1_120, + dispatchId: 'dispatch-1', + }, + { + id: 'event-5', + type: 'dispatch-completed', + at: 1_180, + dispatchId: 'dispatch-1', + }, + { + id: 'event-6', + type: 'phase-completed', + at: 1_200, + phaseVisitId: 'phase-1', + }, + { + id: 'event-7', + type: 'workflow-completed', + at: 1_200, + }, + ]); + }); + + it('records empty dispatch errors as failures', () => { + const r = new WorkflowRunRegistry(); + const entry = r.register(reg('wf_empty_dispatch_error')); + r.onDispatchQueued(entry.runId, { + id: 'dispatch-1', + label: 'Empty failure', + prompt: 'Fail without a message', + dependsOn: [], + queuedAt: 1_100, + }); + r.onDispatchStarted(entry.runId, 'dispatch-1', 1_120); + + r.onDispatchSettled(entry.runId, 'dispatch-1', '', 1_180); + + expect(entry.dispatches[0]).toMatchObject({ + status: 'failed', + error: '', + }); + expect(entry.events.at(-1)).toMatchObject({ + type: 'dispatch-failed', + error: 'Dispatch failed.', + }); + }); + + it('cancels unfinished dispatches before the workflow terminal event', () => { + const r = new WorkflowRunRegistry(); + const entry = r.register(reg('wf_late_dispatch')); + r.onDispatchQueued(entry.runId, { + id: 'dispatch-1', + prompt: 'Fire and forget', + dependsOn: [], + queuedAt: 1_100, + }); + r.onDispatchStarted(entry.runId, 'dispatch-1', 1_200); + r.complete(entry.runId, 'done', 1_300); + + r.onDispatchSettled(entry.runId, 'dispatch-1', undefined, 1_400); + + expect(entry.dispatches[0]).toMatchObject({ + status: 'cancelled', + endedAt: 1_300, + }); + expect(entry.events.at(-1)).toMatchObject({ + type: 'workflow-completed', + at: 1_300, + }); + expect(entry.events.at(-2)).toMatchObject({ + type: 'dispatch-cancelled', + at: 1_300, + }); + }); + + it('cancels unfinished dispatches before a failed workflow is persisted', () => { + const r = new WorkflowRunRegistry(); + const entry = r.register(reg('wf_failed_dispatch')); + r.onDispatchQueued(entry.runId, { + id: 'dispatch-1', + prompt: 'Fire and forget', + dependsOn: [], + queuedAt: 1_100, + }); + r.onDispatchStarted(entry.runId, 'dispatch-1', 1_200); + + r.fail(entry.runId, 'workflow failed', 1_300); + + expect(entry.dispatches[0]).toMatchObject({ + status: 'cancelled', + endedAt: 1_300, + }); + expect(entry.events.at(-2)).toMatchObject({ + type: 'dispatch-cancelled', + at: 1_300, + }); + expect(entry.events.at(-1)).toMatchObject({ + type: 'workflow-failed', + at: 1_300, + }); + }); + + it('fail() sanitizes and caps entry.error like the sibling persisted strings', () => { + const r = new WorkflowRunRegistry(); + const entry = r.register(reg('wf_failed_error')); + + r.fail(entry.runId, `\u001b[2J\u001b[H\u0000${'x'.repeat(5_000)}`, 2_000); + + expect(entry.error).toHaveLength(4_096); + expect(entry.error).not.toContain('\u001b'); + expect(entry.error).not.toContain('\u0000'); + expect(entry.events.at(-1)).toMatchObject({ + type: 'workflow-failed', + error: entry.error, + }); + }); + + it('marks live dispatches cancelled when the workflow is stopped', () => { + const r = new WorkflowRunRegistry(); + const entry = r.register(reg('wf_graph_cancel')); + r.onPhaseStarted(entry.runId, 'Fix', 1_100); + r.onDispatchQueued(entry.runId, { + id: 'dispatch-1', + label: 'Fix boundary', + prompt: 'Fix it', + dependsOn: [], + queuedAt: 1_110, + }); + r.onDispatchStarted(entry.runId, 'dispatch-1', 1_120); + + r.cancel(entry.runId, 1_200); + + expect(entry.dispatches[0]).toMatchObject({ + status: 'cancelled', + endedAt: 1_200, + }); + expect(entry.phaseVisits[0]).toMatchObject({ endedAt: 1_200 }); + }); + it.each(['running', 'pausing', 'paused'] as const)( 'treats %s workflows as active until a terminal transition', (status) => { @@ -958,6 +1289,126 @@ describe('WorkflowRunRegistry', () => { expect(e.recentLogs[99]).toBe('line 249'); }); + it('setRecentLogs sanitizes the mirrored sandbox tail', () => { + const r = new WorkflowRunRegistry(); + r.register(reg('wf_sanitized_logs')); + + r.setRecentLogs('wf_sanitized_logs', [ + '\u001b[2J\u001b[Hchecks\u0000 passed', + ]); + + expect(r.get('wf_sanitized_logs')?.recentLogs).toEqual(['checks passed']); + }); + + it('setRecentLogs truncates the mirrored tail to the persisted line cap', () => { + const r = new WorkflowRunRegistry(); + r.register(reg('wf_long_mirrored_logs')); + + r.setRecentLogs('wf_long_mirrored_logs', ['y'.repeat(5_000)]); + + expect(r.get('wf_long_mirrored_logs')?.recentLogs?.[0]).toHaveLength(4_096); + }); + + it('setRecentLogs resyncs the log event window to the settlement tail', () => { + const r = new WorkflowRunRegistry(); + const entry = r.register(reg('wf_settlement_resync')); + r.onPhaseStarted(entry.runId, 'Build', 1_000); + // Live mirror: the shared emitter publishes lines as they are emitted — + // including lines the sandbox buffer tail later reorders or never + // receives (nested merges via appendLog notify nobody, and the overflow + // sentinel can be pushed without emission). + r.onLogAppended(entry.runId, 'parent-1', 1_001); + r.onLogAppended(entry.runId, 'nested-1', 1_002); + r.onLogAppended(entry.runId, 'nested-2', 1_003); + + r.setRecentLogs(entry.runId, [ + 'parent-1', + '[workflow log truncated at 10000 lines]', + ]); + + // Both persisted log projections must keep agreeing after settlement. + expect(entry.events.filter((event) => event.type === 'log')).toEqual( + entry.recentLogs.map((message) => + expect.objectContaining({ type: 'log', message }), + ), + ); + expect(entry.recentLogs).toEqual([ + 'parent-1', + '[workflow log truncated at 10000 lines]', + ]); + // Non-log events survive the resync. + expect(entry.events[0]).toMatchObject({ type: 'phase-started' }); + }); + + it('onLogAppended keeps recentLogs and log events at the last 100 lines', () => { + const r = new WorkflowRunRegistry(); + const entry = r.register(reg('wf_log_eviction')); + r.onPhaseStarted(entry.runId, 'Build', 1_000); + + for (let i = 1; i <= 150; i++) { + r.onLogAppended(entry.runId, `line-${i}`, 1_000 + i); + } + + expect(entry.recentLogs).toHaveLength(100); + expect(entry.recentLogs[0]).toBe('line-51'); + expect(entry.recentLogs[99]).toBe('line-150'); + const logEvents = entry.events.filter((event) => event.type === 'log'); + expect(logEvents).toHaveLength(100); + expect(logEvents[0]).toMatchObject({ type: 'log', message: 'line-51' }); + expect(logEvents[99]).toMatchObject({ type: 'log', message: 'line-150' }); + // Non-log events survive the log eviction window. + expect(entry.events[0]).toMatchObject({ type: 'phase-started' }); + }); + + it('onLogAppended truncates long lines like the sibling persisted strings', () => { + const r = new WorkflowRunRegistry(); + const entry = r.register(reg('wf_long_log_line')); + + r.onLogAppended(entry.runId, 'x'.repeat(5_000), 1_000); + + expect(entry.recentLogs[0]).toHaveLength(4_096); + expect(entry.events.at(-1)).toMatchObject({ + type: 'log', + message: 'x'.repeat(4_096), + }); + }); + + it('onLogAppended after a cancel transition still reaches both projections', () => { + const r = new WorkflowRunRegistry(); + const entry = r.register(reg('wf_late_cancel_log')); + r.onLogAppended(entry.runId, 'early line', 1_000); + + r.cancel(entry.runId, 1_100); + r.onLogAppended(entry.runId, 'late line', 1_200); + + expect(entry.recentLogs).toEqual(['early line', 'late line']); + expect(entry.events.filter((event) => event.type === 'log')).toEqual([ + expect.objectContaining({ message: 'early line' }), + expect.objectContaining({ message: 'late line' }), + ]); + }); + + it('normalizes phase titles at the registry boundary', () => { + const r = new WorkflowRunRegistry(); + const entry = r.register(reg('wf_phase_titles')); + + r.onPhaseStarted(entry.runId, '\u001b[31mRed\u001b[0m phase', 1_000); + r.onPhaseStarted(entry.runId, 'x'.repeat(500), 1_100); + r.onPhaseStarted(entry.runId, '\u001b[2J', 1_200); + + const titles = ['Red phase', 'x'.repeat(200), 'phase']; + expect(entry.phases).toEqual(titles); + expect(entry.currentPhase).toBe('phase'); + expect(entry.phaseVisits.map((visit) => visit.title)).toEqual(titles); + expect( + entry.events.filter((event) => event.type === 'phase-started'), + ).toEqual([ + expect.objectContaining({ title: 'Red phase' }), + expect.objectContaining({ title: 'x'.repeat(200) }), + expect.objectContaining({ title: 'phase' }), + ]); + }); + it('complete settles the entry and ignores subsequent transitions', () => { const r = new WorkflowRunRegistry(); r.register(reg('wf_1')); diff --git a/packages/core/src/agents/workflow-run-registry.ts b/packages/core/src/agents/workflow-run-registry.ts index fff995c9531..45455b74883 100644 --- a/packages/core/src/agents/workflow-run-registry.ts +++ b/packages/core/src/agents/workflow-run-registry.ts @@ -61,6 +61,7 @@ export type WorkflowTerminalStatus = Extract< WorkflowStatus, 'completed' | 'failed' | 'cancelled' >; +export type WorkflowRunStartMode = 'retry' | 'rerun'; export function isActiveWorkflowStatus( status: WorkflowStatus, @@ -92,6 +93,93 @@ export interface WorkflowApproval { at: number; } +export type WorkflowDispatchTraceStatus = + | 'queued' + | 'running' + | 'completed' + | 'failed' + | 'cancelled' + | 'cached'; + +export interface WorkflowPhaseVisit { + id: string; + index: number; + title: string; + startedAt: number; + endedAt?: number; +} + +export interface WorkflowDispatchTrace { + id: string; + phaseVisitId: string | null; + label: string; + prompt: string; + subagentId?: string; + status: WorkflowDispatchTraceStatus; + dependsOn: string[]; + queuedAt: number; + startedAt?: number; + endedAt?: number; + error?: string; +} + +export interface WorkflowDispatchQueued { + id: string; + label?: string; + prompt: string; + dependsOn: string[]; + queuedAt: number; + cached?: boolean; +} + +interface WorkflowEventBase { + at: number; +} + +type WorkflowEventPayload = + | (WorkflowEventBase & { + type: 'phase-started'; + phaseVisitId: string; + title: string; + }) + | (WorkflowEventBase & { + type: 'phase-completed'; + phaseVisitId: string; + }) + | (WorkflowEventBase & { + type: + | 'dispatch-queued' + | 'dispatch-started' + | 'dispatch-completed' + | 'dispatch-cancelled' + | 'dispatch-cached'; + dispatchId: string; + }) + | (WorkflowEventBase & { + type: 'dispatch-failed'; + dispatchId: string; + error: string; + }) + | (WorkflowEventBase & { + type: 'log'; + message: string; + }) + | (WorkflowEventBase & { + type: 'approval-requested' | 'approval-settled'; + name: string; + dispatchId?: string; + }) + | (WorkflowEventBase & { + type: 'workflow-completed' | 'workflow-cancelled'; + }) + | (WorkflowEventBase & { + type: 'workflow-failed'; + error: string; + }); + +/** Ordered, JSON-safe facts captured while a workflow runs. */ +export type WorkflowEvent = WorkflowEventPayload & { id: string }; + /** * Workflow kind of `TaskState`. Tracks one orchestrator run — the * top-level `Workflow` tool call, not its internal subagent dispatches @@ -104,6 +192,12 @@ export interface WorkflowTask extends TaskBase { kind: 'workflow'; /** Run identifier (e.g. `wf_<8hex>`); aliased to `TaskBase.id`. */ runId: string; + /** Tool call in the parent session that launched this workflow. */ + toolUseId?: string; + /** Run whose result or journal led to this attempt. */ + sourceRunId?: string; + /** Whether this attempt reused the journal or started from scratch. */ + startMode?: WorkflowRunStartMode; /** * Parsed `export const meta = {...}` from the workflow script, or * `null` if the script had no meta declaration. The pill / dialog @@ -121,12 +215,20 @@ export interface WorkflowTask extends TaskBase { * `MAX_PHASE_ENTRIES` (10_000) by the sandbox. */ phases: string[]; + /** Chronological phase entries; unlike `phases`, each revisit has a stable id. */ + phaseVisits: WorkflowPhaseVisit[]; + /** Current phase visit used to associate newly-issued dispatches. */ + currentPhaseVisitId: string | null; + /** Dispatch-level execution graph for live UI consumers. */ + dispatches: WorkflowDispatchTrace[]; /** Cumulative `agent()` dispatches issued by this run. */ agentsDispatched: number; /** Cumulative `agent()` dispatches that have resolved (success or thrown). */ agentsCompleted: number; /** Most recent log lines from the sandbox's `getLogs()`. Capped at 100 for the UI. */ recentLogs: string[]; + /** Ordered runtime facts used to replay this run after it settles. */ + events: WorkflowEvent[]; /** * P5: cumulative output tokens spent by this run's `agent()` dispatches. * Mirrored from `budget.spent()` after each successful completion via @@ -160,6 +262,8 @@ export interface WorkflowTask extends TaskBase { * don't supply it. */ script: string; + /** Original structured arguments, retained so a failed run can resume the same journal prefix. */ + args?: unknown; /** * P7b: the path the script was loaded from, when the run was launched * from a saved workflow (`Workflow({scriptPath})` or a `/workflow-name` @@ -188,9 +292,13 @@ export type WorkflowTaskRegistration = Omit< TaskRegistration, | 'currentPhase' | 'phases' + | 'phaseVisits' + | 'currentPhaseVisitId' + | 'dispatches' | 'agentsDispatched' | 'agentsCompleted' | 'recentLogs' + | 'events' | 'tokensSpent' | 'tokenBudgetTotal' | 'perPhaseTokens' @@ -309,6 +417,10 @@ export class WorkflowRunRegistry { this.statusChangeCallback = cb; } + clearStatusChangeCallback(cb: WorkflowRunStatusChangeCallback): void { + if (this.statusChangeCallback === cb) this.statusChangeCallback = undefined; + } + setNotificationCallback( cb: WorkflowRunNotificationCallback | undefined, ): void { @@ -408,9 +520,13 @@ export class WorkflowRunRegistry { entry.todoWorkChainId ??= todoWorkChainContext.getStore(); entry.currentPhase = null; entry.phases = []; + entry.phaseVisits = []; + entry.currentPhaseVisitId = null; + entry.dispatches = []; entry.agentsDispatched = 0; entry.agentsCompleted = 0; entry.recentLogs = []; + entry.events = []; entry.tokensSpent = 0; // Preserve a caller-supplied cap; default to "no cap" otherwise. // Note: the registration's optional `tokenBudgetTotal` shape is the @@ -482,16 +598,30 @@ export class WorkflowRunRegistry { if (this.handles.get(runId) === handle) this.handles.delete(runId); } - bridgeApprovalEvents(runId: string, emitter: AgentEventEmitter): () => void { + bridgeApprovalEvents( + runId: string, + emitter: AgentEventEmitter, + dispatchId?: string, + expectedEntry?: WorkflowTask, + ): () => void { const ownedApprovalIds = new Set(); const seenSources = new Set(); + const isCurrentEntry = () => + expectedEntry === undefined || this.entries.get(runId) === expectedEntry; const onWaiting = (event: AgentApprovalRequestEvent) => { + if (!isCurrentEntry()) return; + if (dispatchId) { + const dispatch = this.entries + .get(runId) + ?.dispatches.find(({ id }) => id === dispatchId); + if (dispatch) dispatch.subagentId = event.subagentId; + } const sourceKey = JSON.stringify([event.subagentId, event.callId]); // Re-emission of an already-settled call: respond is idempotent via // the runtime's responded set, so silently dropping it is safe. if (seenSources.has(sourceKey)) return; seenSources.add(sourceKey); - const parked = this.parkPendingApproval(runId, event); + const parked = this.parkPendingApproval(runId, event, dispatchId); if (parked === 'duplicate') return; if (parked === 'rejected') { this.rejectResponder(event.respond); @@ -500,13 +630,20 @@ export class WorkflowRunRegistry { ownedApprovalIds.add(parked); }; const onResult = (event: AgentToolResultEvent) => { - this.clearPendingApproval(runId, event.subagentId, event.callId); + if (!isCurrentEntry()) return; + this.clearPendingApproval( + runId, + event.subagentId, + event.callId, + event.timestamp, + ); }; emitter.on(AgentEventType.TOOL_WAITING_APPROVAL, onWaiting); emitter.on(AgentEventType.TOOL_RESULT, onResult); return () => { emitter.off(AgentEventType.TOOL_WAITING_APPROVAL, onWaiting); emitter.off(AgentEventType.TOOL_RESULT, onResult); + if (!isCurrentEntry()) return; this.rejectPendingApprovals(runId, (approval) => ownedApprovalIds.has(approval.approvalId), ); @@ -526,6 +663,7 @@ export class WorkflowRunRegistry { ); if (!approval) return false; const runtime = this.approvalRuntimes.get(approvalId); + this.appendApprovalEvent(entry, approval, 'approval-settled', Date.now()); entry.pendingApprovals = entry.pendingApprovals.filter( (candidate) => candidate !== approval, ); @@ -566,6 +704,7 @@ export class WorkflowRunRegistry { runId: string, subagentId: string, callId: string, + at = Date.now(), ): boolean { const entry = this.entries.get(runId); const approval = entry?.pendingApprovals.find( @@ -573,6 +712,7 @@ export class WorkflowRunRegistry { candidate.subagentId === subagentId && candidate.callId === callId, ); if (!entry || !approval) return false; + this.appendApprovalEvent(entry, approval, 'approval-settled', at); entry.pendingApprovals = entry.pendingApprovals.filter( (candidate) => candidate !== approval, ); @@ -586,6 +726,7 @@ export class WorkflowRunRegistry { private parkPendingApproval( runId: string, event: AgentApprovalRequestEvent, + dispatchId?: string, ): string | 'duplicate' | 'rejected' { const entry = this.entries.get(runId); if ( @@ -647,6 +788,12 @@ export class WorkflowRunRegistry { requestController, }); entry.pendingApprovals = [...entry.pendingApprovals, approval]; + this.appendEvent(entry, { + type: 'approval-requested', + at: approval.at, + name: approval.name, + ...(dispatchId ? { dispatchId } : {}), + }); this.emitApprovalChange(entry); if ( approvalRequestCallback && @@ -670,6 +817,12 @@ export class WorkflowRunRegistry { }); } catch (error) { debugLogger.error('Workflow approval channel failed:', error); + this.appendApprovalEvent( + entry, + approval, + 'approval-settled', + Date.now(), + ); entry.pendingApprovals = entry.pendingApprovals.filter( (candidate) => candidate.approvalId !== approvalId, ); @@ -687,18 +840,158 @@ export class WorkflowRunRegistry { * a phase identical to the most recent entry is treated as the same * phase and not re-appended. `currentPhase` is set unconditionally. * - * @param runId the run to update - * @param title the phase title from the sandbox `phase()` call + * @param runId the run to update + * @param rawTitle the phase title from the sandbox `phase()` call */ - onPhaseStarted(runId: string, title: string): void { + onPhaseStarted(runId: string, rawTitle: string, at = Date.now()): void { const entry = this.entries.get(runId); if (!entry || !isActiveWorkflowStatus(entry.status)) return; + // Script-derived titles reach persisted snapshots and TUI rendering: + // normalize at this registry boundary like every sibling string. + const title = stripAnsiAndControl(rawTitle).slice(0, 200) || 'phase'; entry.currentPhase = title; const last = entry.phases[entry.phases.length - 1]; - if (last !== title) entry.phases.push(title); + if (last !== title) { + entry.phases.push(title); + const priorVisit = entry.phaseVisits[entry.phaseVisits.length - 1]; + if (priorVisit && priorVisit.endedAt === undefined) { + this.closeCurrentPhase(entry, at); + } + const index = entry.phaseVisits.length; + const visit: WorkflowPhaseVisit = { + id: `phase-${index + 1}`, + index, + title, + startedAt: at, + }; + entry.phaseVisits.push(visit); + entry.currentPhaseVisitId = visit.id; + this.appendEvent(entry, { + type: 'phase-started', + at, + phaseVisitId: visit.id, + title, + }); + } + this.emitStatusChange(entry); + } + + onDispatchQueued(runId: string, event: WorkflowDispatchQueued): void { + const entry = this.entries.get(runId); + if (!entry || !isActiveWorkflowStatus(entry.status)) return; + if (entry.dispatches.some((dispatch) => dispatch.id === event.id)) return; + const fallbackLabel = `Agent ${entry.dispatches.length + 1}`; + entry.dispatches.push({ + id: event.id, + phaseVisitId: entry.currentPhaseVisitId, + label: + stripAnsiAndControl(event.label ?? '').slice(0, 200) || fallbackLabel, + prompt: stripAnsiAndControl(event.prompt).slice(0, 4_096), + status: event.cached ? 'cached' : 'queued', + dependsOn: Array.from(new Set(event.dependsOn)).filter((id) => + entry.dispatches.some((dispatch) => dispatch.id === id), + ), + queuedAt: event.queuedAt, + ...(event.cached ? { endedAt: event.queuedAt } : {}), + }); + this.appendEvent(entry, { + type: 'dispatch-queued', + at: event.queuedAt, + dispatchId: event.id, + }); + if (event.cached) { + this.appendEvent(entry, { + type: 'dispatch-cached', + at: event.queuedAt, + dispatchId: event.id, + }); + } + this.emitStatusChange(entry); + } + + onDispatchStarted(runId: string, dispatchId: string, at = Date.now()): void { + const entry = this.entries.get(runId); + const dispatch = entry?.dispatches.find(({ id }) => id === dispatchId); + if (!entry || !dispatch || dispatch.status !== 'queued') return; + dispatch.status = 'running'; + dispatch.startedAt = at; + this.appendEvent(entry, { + type: 'dispatch-started', + at, + dispatchId, + }); + this.emitStatusChange(entry); + } + + onDispatchSettled( + runId: string, + dispatchId: string, + error?: string, + at = Date.now(), + cancelRequested = false, + ): void { + const entry = this.entries.get(runId); + const dispatch = entry?.dispatches.find(({ id }) => id === dispatchId); + if (!entry || !dispatch || dispatch.endedAt !== undefined) return; + const shouldRecordEvent = isActiveWorkflowStatus(entry.status); + dispatch.status = + entry.status === 'cancelled' || cancelRequested + ? 'cancelled' + : error !== undefined + ? 'failed' + : dispatch.status === 'cached' + ? 'cached' + : 'completed'; + dispatch.endedAt = at; + if (error !== undefined && dispatch.status !== 'cancelled') + dispatch.error = stripAnsiAndControl(error).slice(0, 4_096); + if (!shouldRecordEvent) { + this.emitStatusChange(entry); + return; + } + if (dispatch.status === 'failed') { + this.appendEvent(entry, { + type: 'dispatch-failed', + at, + dispatchId, + error: dispatch.error || 'Dispatch failed.', + }); + } else { + this.appendEvent(entry, { + type: + dispatch.status === 'cached' + ? 'dispatch-cached' + : dispatch.status === 'cancelled' + ? 'dispatch-cancelled' + : 'dispatch-completed', + at, + dispatchId, + }); + } this.emitStatusChange(entry); } + /** Record one sandbox log line without forcing a TUI redraw per line. */ + onLogAppended(runId: string, line: string, at = Date.now()): void { + const entry = this.entries.get(runId); + // Mirrors setRecentLogs's 'cancelled' allowance: a dialog cancel flips + // the status before the sandbox's run-end flush fires its last mirror + // lines, and the two persisted log projections must keep agreeing. + if ( + !entry || + (!isActiveWorkflowStatus(entry.status) && entry.status !== 'cancelled') + ) + return; + const message = stripAnsiAndControl(line).slice(0, 4_096); + if (entry.recentLogs.length === 100) { + entry.recentLogs.shift(); + const firstLog = entry.events.findIndex((event) => event.type === 'log'); + if (firstLog >= 0) entry.events.splice(firstLog, 1); + } + entry.recentLogs.push(message); + this.appendEvent(entry, { type: 'log', at, message }); + } + /** Cumulative dispatch counter — incremented before each `agent()` call resolves. */ onAgentDispatched(runId: string): void { const entry = this.entries.get(runId); @@ -787,17 +1080,31 @@ export class WorkflowRunRegistry { if (!isActiveWorkflowStatus(entry.status) && entry.status !== 'cancelled') return; const tail = logs.length > 100 ? logs.slice(-100) : Array.from(logs); - entry.recentLogs = tail; + entry.recentLogs = tail.map((line) => + stripAnsiAndControl(line).slice(0, 4_096), + ); + // The sandbox buffer tail is the run's final log account: nested merges + // reach it via appendLog without re-emitting, and the overflow sentinel + // can be pushed without emission, so the live-mirrored 'log' window can + // disagree with it in membership AND order. Rebuild the window from the + // same tail so the two persisted log projections keep agreeing. + entry.events = entry.events.filter((event) => event.type !== 'log'); + for (const message of entry.recentLogs) { + this.appendEvent(entry, { type: 'log', at: Date.now(), message }); + } this.emitStatusChange(entry); } complete(runId: string, result: unknown, endTime: number): void { const entry = this.entries.get(runId); if (!entry || !isActiveWorkflowStatus(entry.status)) return; - this.rejectPendingApprovals(runId); + this.rejectPendingApprovals(runId, undefined, endTime); entry.status = 'completed'; entry.endTime = endTime; + this.closeCurrentPhase(entry, endTime); + this.cancelLiveDispatches(entry, endTime); entry.result = result; + this.appendEvent(entry, { type: 'workflow-completed', at: endTime }); entry.notified = true; this.emitStatusChange(entry); this.emitNotification(entry); @@ -808,10 +1115,20 @@ export class WorkflowRunRegistry { fail(runId: string, message: string, endTime: number): void { const entry = this.entries.get(runId); if (!entry || !isActiveWorkflowStatus(entry.status)) return; - this.rejectPendingApprovals(runId); + this.rejectPendingApprovals(runId, undefined, endTime); entry.status = 'failed'; entry.endTime = endTime; - entry.error = message; + this.closeCurrentPhase(entry, endTime); + this.cancelLiveDispatches(entry, endTime); + // Script-derived failure text rides into the snapshot, the /workflows + // render, and the completion-notification XML: normalize it once at + // this boundary and persist the same string in both projections. + entry.error = stripAnsiAndControl(message).slice(0, 4_096); + this.appendEvent(entry, { + type: 'workflow-failed', + at: endTime, + error: entry.error, + }); entry.notified = true; this.emitStatusChange(entry); this.emitNotification(entry); @@ -827,9 +1144,12 @@ export class WorkflowRunRegistry { cancel(runId: string, endTime: number): void { const entry = this.entries.get(runId); if (!entry || !isActiveWorkflowStatus(entry.status)) return; - this.rejectPendingApprovals(runId); + this.rejectPendingApprovals(runId, undefined, endTime); entry.status = 'cancelled'; entry.endTime = endTime; + this.closeCurrentPhase(entry, endTime); + this.cancelLiveDispatches(entry, endTime); + this.appendEvent(entry, { type: 'workflow-cancelled', at: endTime }); entry.notified = true; try { (this.handles.get(runId) ?? entry.abortController).abort(); @@ -844,6 +1164,19 @@ export class WorkflowRunRegistry { return this.entries.get(runId); } + setLineage( + runId: string, + sourceRunId: string, + startMode: WorkflowRunStartMode, + ): boolean { + const entry = this.entries.get(runId); + if (!entry) return false; + entry.sourceRunId = sourceRunId; + entry.startMode = startMode; + this.emitStatusChange(entry); + return true; + } + /** All entries (active + terminal, no filter). Iteration order = registration order. */ list(): WorkflowTask[] { return Array.from(this.entries.values()); @@ -922,9 +1255,12 @@ export class WorkflowRunRegistry { let lastCancelled: WorkflowTask | undefined; for (const entry of Array.from(this.entries.values())) { if (!isActiveWorkflowStatus(entry.status)) continue; - this.rejectPendingApprovals(entry.runId); + this.rejectPendingApprovals(entry.runId, undefined, endTime); entry.status = 'cancelled'; entry.endTime = endTime; + this.closeCurrentPhase(entry, endTime); + this.cancelLiveDispatches(entry, endTime); + this.appendEvent(entry, { type: 'workflow-cancelled', at: endTime }); entry.notified = true; try { (this.handles.get(entry.runId) ?? entry.abortController).abort(); @@ -940,6 +1276,62 @@ export class WorkflowRunRegistry { this.evictTerminal(); } + private closeCurrentPhase(entry: WorkflowTask, endTime: number): void { + const current = entry.phaseVisits[entry.phaseVisits.length - 1]; + if (current && current.endedAt === undefined) { + current.endedAt = endTime; + this.appendEvent(entry, { + type: 'phase-completed', + at: endTime, + phaseVisitId: current.id, + }); + } + } + + private cancelLiveDispatches(entry: WorkflowTask, endTime: number): void { + for (const dispatch of entry.dispatches) { + if (dispatch.status !== 'queued' && dispatch.status !== 'running') { + continue; + } + dispatch.status = 'cancelled'; + dispatch.endedAt = endTime; + this.appendEvent(entry, { + type: 'dispatch-cancelled', + at: endTime, + dispatchId: dispatch.id, + }); + } + } + + private appendEvent( + entry: WorkflowTask, + payload: WorkflowEventPayload, + ): void { + const lastId = entry.events.at(-1)?.id; + const nextId = lastId ? Number(lastId.slice('event-'.length)) + 1 : 1; + entry.events.push({ + id: `event-${nextId}`, + ...payload, + }); + } + + private appendApprovalEvent( + entry: WorkflowTask, + approval: WorkflowApproval, + type: 'approval-requested' | 'approval-settled', + at: number, + ): void { + const dispatchId = entry.dispatches.find( + (dispatch) => dispatch.subagentId === approval.subagentId, + )?.id; + this.appendEvent(entry, { + type, + at, + name: approval.name, + ...(dispatchId ? { dispatchId } : {}), + }); + } + /** * Sweep terminal entries when they exceed `MAX_RETAINED_TERMINAL_WORKFLOWS`. * Active entries are always retained. Oldest terminal entries @@ -972,6 +1364,7 @@ export class WorkflowRunRegistry { private rejectPendingApprovals( runId: string, predicate: (approval: WorkflowApproval) => boolean = () => true, + at = Date.now(), ): void { const entry = this.entries.get(runId); if (!entry) return; @@ -980,6 +1373,9 @@ export class WorkflowRunRegistry { const rejectedIds = new Set( rejected.map((approval) => approval.approvalId), ); + for (const approval of rejected) { + this.appendApprovalEvent(entry, approval, 'approval-settled', at); + } entry.pendingApprovals = entry.pendingApprovals.filter( (approval) => !rejectedIds.has(approval.approvalId), ); diff --git a/packages/core/src/agents/workflow-snapshot.test.ts b/packages/core/src/agents/workflow-snapshot.test.ts index 0f1228f49d5..501774bfb6e 100644 --- a/packages/core/src/agents/workflow-snapshot.test.ts +++ b/packages/core/src/agents/workflow-snapshot.test.ts @@ -14,6 +14,7 @@ import { toSnapshot, writeWorkflowSnapshot, listWorkflowSnapshots, + deleteWorkflowSnapshot, MAX_RETAINED_SNAPSHOTS, } from './workflow-snapshot.js'; import type { WorkflowTask } from './workflow-run-registry.js'; @@ -38,9 +39,25 @@ function task(overrides: Partial = {}): WorkflowTask { abortController: new AbortController(), currentPhase: null, phases: ['Plan', 'Build'], + phaseVisits: [], + currentPhaseVisitId: null, + dispatches: [], agentsDispatched: 3, agentsCompleted: 3, recentLogs: ['log1'], + events: [ + { + id: 'event-1', + type: 'log', + at: 1_700_000_004_000, + message: 'log1', + }, + { + id: 'event-2', + type: 'workflow-completed', + at: 1_700_000_005_000, + }, + ], tokensSpent: 450, tokenBudgetTotal: 1000, perPhaseTokens: new Map([ @@ -65,7 +82,13 @@ describe('toSnapshot', () => { ); it('flattens perPhaseTokens Map into [phaseOrNull, tokens] pairs', () => { - const s = toSnapshot(task()); + const s = toSnapshot( + task({ + description: 'Review and fix', + sourceRunId: 'wf_source', + startMode: 'rerun', + }), + ); expect(s.perPhaseTokens).toEqual([ ['Plan', 200], [null, 50], @@ -73,6 +96,11 @@ describe('toSnapshot', () => { expect(s.runId).toBe('wf_a'); expect(s.script).toBe('return 1;'); expect(s.result).toEqual({ answer: 42 }); + expect(s).toMatchObject({ + description: 'Review and fix', + sourceRunId: 'wf_source', + startMode: 'rerun', + }); }); it('replaces a non-JSON-serializable result with a placeholder string', () => { @@ -85,7 +113,9 @@ describe('toSnapshot', () => { const t = task(); const s = toSnapshot(t); t.phases.push('Mutated'); + t.events[0]!.at = 0; expect(s.phases).toEqual(['Plan', 'Build']); + expect(s.events?.[0]?.at).toBe(1_700_000_004_000); }); it('never projects live pending approval data', () => { @@ -119,6 +149,7 @@ describe('toSnapshot', () => { expect(serialized).not.toContain('PRIVATE_DESCRIPTION_SENTINEL'); expect(serialized).not.toContain('PRIVATE_DIFF_SENTINEL'); expect(toSnapshot(live)).not.toHaveProperty('pendingApprovals'); + expect(toSnapshot(live).events).toEqual(live.events); }); }); @@ -142,6 +173,38 @@ describe('writeWorkflowSnapshot + listWorkflowSnapshots', () => { ['Plan', 200], [null, 50], ]); + expect(list[0].events).toEqual([ + { + id: 'event-1', + type: 'log', + at: 1_700_000_004_000, + message: 'log1', + }, + { + id: 'event-2', + type: 'workflow-completed', + at: 1_700_000_005_000, + }, + ]); + }); + + it('loads a legacy snapshot without an event ledger', async () => { + const config = fakeConfig(projectDir); + await writeWorkflowSnapshot(config, task({ runId: 'wf_legacy' })); + const snapshotPath = config.storage.getWorkflowRunSnapshotPath('wf_legacy'); + const parsed = JSON.parse( + await fs.readFile(snapshotPath, 'utf8'), + ) as Record; + delete parsed['events']; + delete parsed['phaseVisits']; + delete parsed['dispatches']; + delete parsed['description']; + await fs.writeFile(snapshotPath, JSON.stringify(parsed), 'utf8'); + + const list = await listWorkflowSnapshots(config); + + expect(list).toHaveLength(1); + expect(list[0].events).toBeUndefined(); }); it('freezes the snapshot projection before the first fs await', async () => { @@ -198,6 +261,92 @@ describe('writeWorkflowSnapshot + listWorkflowSnapshots', () => { expect(list.map((s) => s.runId)).toEqual(['wf_good']); }); + it('skips parseable files that do not match the snapshot contract', async () => { + const config = fakeConfig(projectDir); + await writeWorkflowSnapshot(config, task({ runId: 'wf_good' })); + const dir = config.storage.getWorkflowRunsDir(); + await fs.writeFile( + path.join(dir, 'wf_invalid.json'), + JSON.stringify({ runId: 'wf_invalid', status: 'completed' }), + 'utf8', + ); + + const list = await listWorkflowSnapshots(config); + + expect(list.map((s) => s.runId)).toEqual(['wf_good']); + }); + + it('deletes one saved run and its resume journal', async () => { + const config = fakeConfig(projectDir); + const runId = 'wf_abcd'; + await writeWorkflowSnapshot(config, task({ runId })); + const journalPath = config.storage.getWorkflowRunJournalPath(runId); + await fs.mkdir(path.dirname(journalPath), { recursive: true }); + await fs.writeFile(journalPath, '{}\n', 'utf8'); + + await expect(deleteWorkflowSnapshot(config, runId)).resolves.toBe(true); + + await expect( + fs.access(config.storage.getWorkflowRunSnapshotPath(runId)), + ).rejects.toThrow(); + await expect(fs.access(path.dirname(journalPath))).rejects.toThrow(); + await expect(listWorkflowSnapshots(config)).resolves.toEqual([]); + }); + + it('keeps the snapshot and reports failure when journal deletion fails', async () => { + const config = fakeConfig(projectDir); + const runId = 'wf_dead'; + await writeWorkflowSnapshot(config, task({ runId })); + const journalPath = config.storage.getWorkflowRunJournalPath(runId); + await fs.mkdir(path.dirname(journalPath), { recursive: true }); + await fs.writeFile(journalPath, '{}\n', 'utf8'); + const rmSpy = vi + .spyOn(fs, 'rm') + .mockRejectedValueOnce( + Object.assign(new Error('busy'), { code: 'EBUSY' }), + ); + + await expect(deleteWorkflowSnapshot(config, runId)).resolves.toBe(false); + expect(rmSpy).toHaveBeenCalledTimes(1); + + rmSpy.mockRestore(); + await expect( + fs.access(config.storage.getWorkflowRunSnapshotPath(runId)), + ).resolves.toBeUndefined(); + await expect(fs.access(path.dirname(journalPath))).resolves.toBeUndefined(); + }); + + it('rejects traversal-shaped run ids without touching project files', async () => { + const config = fakeConfig(projectDir); + // Extensionless on purpose: for input '../CANARY' an unguarded recursive + // rm targets /CANARY exactly, so bypassing the guard makes + // the read-back below fail instead of only the boolean assertion. + const canary = path.join(projectDir, 'CANARY'); + await fs.writeFile(canary, 'keep', 'utf8'); + + await expect(deleteWorkflowSnapshot(config, '../CANARY')).resolves.toBe( + false, + ); + await expect(deleteWorkflowSnapshot(config, 'wf_bad/path')).resolves.toBe( + false, + ); + + await expect(fs.readFile(canary, 'utf8')).resolves.toBe('keep'); + }); + + it('rejects malformed run ids without deleting another snapshot', async () => { + const config = fakeConfig(projectDir); + const runId = 'wf_abcd'; + await writeWorkflowSnapshot(config, task({ runId })); + const snapshotPath = config.storage.getWorkflowRunSnapshotPath(runId); + + await expect(deleteWorkflowSnapshot(config, `${runId}.json`)).resolves.toBe( + false, + ); + + await expect(fs.access(snapshotPath)).resolves.toBeUndefined(); + }); + it('prunes the oldest beyond MAX_RETAINED_SNAPSHOTS, journal dirs too', async () => { const config = fakeConfig(projectDir); const dir = config.storage.getWorkflowRunsDir(); diff --git a/packages/core/src/agents/workflow-snapshot.ts b/packages/core/src/agents/workflow-snapshot.ts index 743c9ab29b5..5f68800e9ac 100644 --- a/packages/core/src/agents/workflow-snapshot.ts +++ b/packages/core/src/agents/workflow-snapshot.ts @@ -19,6 +19,10 @@ import { createDebugLogger } from '../utils/debugLogger.js'; import type { WorkflowMeta } from './runtime/workflow-sandbox.js'; import { isTerminalWorkflowStatus, + type WorkflowDispatchTrace, + type WorkflowEvent, + type WorkflowPhaseVisit, + type WorkflowRunStartMode, type WorkflowTask, type WorkflowTerminalStatus, } from './workflow-run-registry.js'; @@ -31,11 +35,21 @@ export const MAX_RETAINED_SNAPSHOTS = 30; /** JSON-serializable projection of a terminal workflow run. */ export interface WorkflowSnapshot { runId: string; + /** Human-readable fallback when a workflow has no exported meta block. */ + description?: string; + /** Prior run used by retry or rerun. Absent on legacy snapshots. */ + sourceRunId?: string; + /** How this run was started from sourceRunId. */ + startMode?: WorkflowRunStartMode; meta: WorkflowMeta | null; status: WorkflowTerminalStatus; script: string; scriptPath?: string; phases: string[]; + /** Absent on snapshots written before workflow graph tracing existed. */ + phaseVisits?: WorkflowPhaseVisit[]; + /** Absent on snapshots written before workflow graph tracing existed. */ + dispatches?: WorkflowDispatchTrace[]; agentsDispatched: number; agentsCompleted: number; tokensSpent: number; @@ -43,6 +57,8 @@ export interface WorkflowSnapshot { /** `perPhaseTokens` flattened to `[phaseOrNull, tokens]` pairs. */ perPhaseTokens: Array<[string | null, number]>; recentLogs: string[]; + /** Absent on snapshots written before runtime event tracing existed. */ + events?: WorkflowEvent[]; startTime: number; endTime?: number; result?: unknown; @@ -56,17 +72,26 @@ export function toSnapshot(task: WorkflowTask): WorkflowSnapshot { } return { runId: task.runId, + description: task.description, + sourceRunId: task.sourceRunId, + startMode: task.startMode, meta: task.meta, status: task.status, script: task.script ?? '', scriptPath: task.scriptPath, phases: [...task.phases], + phaseVisits: task.phaseVisits.map((visit) => ({ ...visit })), + dispatches: task.dispatches.map((dispatch) => ({ + ...dispatch, + dependsOn: [...dispatch.dependsOn], + })), agentsDispatched: task.agentsDispatched, agentsCompleted: task.agentsCompleted, tokensSpent: task.tokensSpent, tokenBudgetTotal: task.tokenBudgetTotal, perPhaseTokens: Array.from(task.perPhaseTokens.entries()), recentLogs: [...task.recentLogs], + events: task.events.map((event) => ({ ...event })), startTime: task.startTime, endTime: task.endTime, result: safeResult(task.result), @@ -136,7 +161,12 @@ export async function listWorkflowSnapshots( for (const file of files) { try { const raw = await fs.readFile(`${dir}/${file}`, 'utf8'); - snapshots.push(JSON.parse(raw) as WorkflowSnapshot); + const parsed: unknown = JSON.parse(raw); + if (!isWorkflowSnapshot(parsed)) { + debugLogger.warn(`skipping invalid workflow snapshot ${file}`); + continue; + } + snapshots.push(parsed); } catch (e) { debugLogger.warn(`skipping unparseable snapshot ${file}: ${e}`); } @@ -145,6 +175,232 @@ export async function listWorkflowSnapshots( return snapshots; } +/** + * Delete one persisted run summary and its resume journal. The run id must be + * a well-formed workflow run id because both targets live below the project + * runs dir. + * Returns true when the safe target is absent after this call. + */ +export async function deleteWorkflowSnapshot( + config: Config, + runId: string, +): Promise { + const storage = config.storage; + if (!storage || !/^wf_[0-9a-f]+$/.test(runId)) return false; + try { + await fs.rm(`${storage.getWorkflowRunsDir()}/${runId}`, { + recursive: true, + force: true, + }); + } catch (error) { + debugLogger.warn(`delete workflow journal failed for ${runId}: ${error}`); + return false; + } + try { + await fs.unlink(storage.getWorkflowRunSnapshotPath(runId)); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') { + debugLogger.warn(`deleteWorkflowSnapshot failed for ${runId}: ${error}`); + return false; + } + } + return true; +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function isFiniteNumber(value: unknown): value is number { + return typeof value === 'number' && Number.isFinite(value); +} + +function isOptionalString(value: unknown): value is string | undefined { + return value === undefined || typeof value === 'string'; +} + +function isStringArray(value: unknown): value is string[] { + return ( + Array.isArray(value) && value.every((item) => typeof item === 'string') + ); +} + +function isWorkflowMeta(value: unknown): value is WorkflowMeta | null { + if (value === null) return true; + if (!isRecord(value)) return false; + if ( + typeof value['name'] !== 'string' || + typeof value['description'] !== 'string' || + !isOptionalString(value['whenToUse']) + ) { + return false; + } + const phases = value['phases']; + return ( + phases === undefined || + (Array.isArray(phases) && + phases.every( + (phase) => + isRecord(phase) && + typeof phase['title'] === 'string' && + isOptionalString(phase['detail']) && + isOptionalString(phase['model']), + )) + ); +} + +function isWorkflowPhaseVisit(value: unknown): value is WorkflowPhaseVisit { + return ( + isRecord(value) && + typeof value['id'] === 'string' && + isFiniteNumber(value['index']) && + typeof value['title'] === 'string' && + isFiniteNumber(value['startedAt']) && + (value['endedAt'] === undefined || isFiniteNumber(value['endedAt'])) + ); +} + +function isWorkflowDispatch(value: unknown): value is WorkflowDispatchTrace { + if (!isRecord(value)) return false; + const status = value['status']; + return ( + typeof value['id'] === 'string' && + (value['phaseVisitId'] === null || + typeof value['phaseVisitId'] === 'string') && + typeof value['label'] === 'string' && + typeof value['prompt'] === 'string' && + isOptionalString(value['subagentId']) && + (status === 'queued' || + status === 'running' || + status === 'completed' || + status === 'failed' || + status === 'cancelled' || + status === 'cached') && + isStringArray(value['dependsOn']) && + isFiniteNumber(value['queuedAt']) && + (value['startedAt'] === undefined || isFiniteNumber(value['startedAt'])) && + (value['endedAt'] === undefined || isFiniteNumber(value['endedAt'])) && + isOptionalString(value['error']) + ); +} + +function hasOnlyKeys( + value: Record, + keys: readonly string[], +): boolean { + const allowed = new Set(keys); + return Object.keys(value).every((key) => allowed.has(key)); +} + +function isWorkflowEvent(value: unknown): value is WorkflowEvent { + if ( + !isRecord(value) || + typeof value['id'] !== 'string' || + !isFiniteNumber(value['at']) || + typeof value['type'] !== 'string' + ) { + return false; + } + const base = ['id', 'type', 'at']; + switch (value['type']) { + case 'phase-started': + return ( + hasOnlyKeys(value, [...base, 'phaseVisitId', 'title']) && + typeof value['phaseVisitId'] === 'string' && + typeof value['title'] === 'string' + ); + case 'phase-completed': + return ( + hasOnlyKeys(value, [...base, 'phaseVisitId']) && + typeof value['phaseVisitId'] === 'string' + ); + case 'dispatch-queued': + case 'dispatch-started': + case 'dispatch-completed': + case 'dispatch-cancelled': + case 'dispatch-cached': + return ( + hasOnlyKeys(value, [...base, 'dispatchId']) && + typeof value['dispatchId'] === 'string' + ); + case 'dispatch-failed': + return ( + hasOnlyKeys(value, [...base, 'dispatchId', 'error']) && + typeof value['dispatchId'] === 'string' && + typeof value['error'] === 'string' + ); + case 'log': + return ( + hasOnlyKeys(value, [...base, 'message']) && + typeof value['message'] === 'string' + ); + case 'approval-requested': + case 'approval-settled': + return ( + hasOnlyKeys(value, [...base, 'name', 'dispatchId']) && + typeof value['name'] === 'string' && + isOptionalString(value['dispatchId']) + ); + case 'workflow-completed': + case 'workflow-cancelled': + return hasOnlyKeys(value, base); + case 'workflow-failed': + return ( + hasOnlyKeys(value, [...base, 'error']) && + typeof value['error'] === 'string' + ); + default: + return false; + } +} + +function isWorkflowSnapshot(value: unknown): value is WorkflowSnapshot { + if (!isRecord(value)) return false; + const status = value['status']; + const phaseVisits = value['phaseVisits']; + const dispatches = value['dispatches']; + const events = value['events']; + const perPhaseTokens = value['perPhaseTokens']; + return ( + typeof value['runId'] === 'string' && + value['runId'].length > 0 && + isOptionalString(value['description']) && + isOptionalString(value['sourceRunId']) && + (value['startMode'] === undefined || + value['startMode'] === 'retry' || + value['startMode'] === 'rerun') && + isWorkflowMeta(value['meta']) && + (status === 'completed' || status === 'failed' || status === 'cancelled') && + typeof value['script'] === 'string' && + isOptionalString(value['scriptPath']) && + isStringArray(value['phases']) && + (phaseVisits === undefined || + (Array.isArray(phaseVisits) && + phaseVisits.every(isWorkflowPhaseVisit))) && + (dispatches === undefined || + (Array.isArray(dispatches) && dispatches.every(isWorkflowDispatch))) && + (events === undefined || + (Array.isArray(events) && events.every(isWorkflowEvent))) && + isFiniteNumber(value['agentsDispatched']) && + isFiniteNumber(value['agentsCompleted']) && + isFiniteNumber(value['tokensSpent']) && + (value['tokenBudgetTotal'] === null || + isFiniteNumber(value['tokenBudgetTotal'])) && + Array.isArray(perPhaseTokens) && + perPhaseTokens.every( + (entry) => + Array.isArray(entry) && + entry.length === 2 && + (entry[0] === null || typeof entry[0] === 'string') && + isFiniteNumber(entry[1]), + ) && + isStringArray(value['recentLogs']) && + isFiniteNumber(value['startTime']) && + (value['endTime'] === undefined || isFiniteNumber(value['endTime'])) && + isOptionalString(value['error']) + ); +} + /** Remove the oldest snapshots beyond the retention cap. */ async function pruneSnapshots(dir: string): Promise { let files: string[]; diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 767207fa5c2..ee525f89e22 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -185,6 +185,7 @@ export { FORK_SUBAGENT_TYPE } from './tools/agent/fork-subagent.js'; export type { WorkflowTool, WorkflowParams, + WorkflowToolResult, } from './tools/workflow/workflow.js'; export type { TodoWriteTool, diff --git a/packages/core/src/services/chatRecordingService.ts b/packages/core/src/services/chatRecordingService.ts index c6b3443edda..7620b6ff7c7 100644 --- a/packages/core/src/services/chatRecordingService.ts +++ b/packages/core/src/services/chatRecordingService.ts @@ -420,7 +420,7 @@ export interface NotificationRecordPayload { backgroundTask?: { taskId: string; status: string; - kind: 'agent' | 'monitor' | 'shell'; + kind: 'agent' | 'monitor' | 'shell' | 'workflow'; toolUseId?: string; /** Structured fields for i18n rendering (persisted for page refresh). */ description?: string; diff --git a/packages/core/src/tools/workflow/workflow.test.ts b/packages/core/src/tools/workflow/workflow.test.ts index 1257af61742..a2c80389b2a 100644 --- a/packages/core/src/tools/workflow/workflow.test.ts +++ b/packages/core/src/tools/workflow/workflow.test.ts @@ -311,18 +311,25 @@ describe('WorkflowTool', () => { }), }); const updateOutput = vi.fn(); - const execution = tool - .build({ - script: `phase('slow'); return await agent('work');`, - run_in_background: true, - }) - .execute(new AbortController().signal, updateOutput); + const invocation = tool.build({ + script: `phase('slow'); return await agent('work');`, + run_in_background: true, + }); + ( + invocation as unknown as { setCallId: (callId: string) => void } + ).setCallId('workflow-tool-call'); + const execution = invocation.execute( + new AbortController().signal, + updateOutput, + ); await vi.waitFor(() => expect(resolveDispatch).toBeDefined()); const result = await execution; const entry = registry.list()[0]!; expect(entry.status).toBe('running'); expect(entry.isBackgrounded).toBe(true); + expect(entry.toolUseId).toBe('workflow-tool-call'); + expect(result.workflowRunId).toBe(entry.runId); expect(result.llmContent).toEqual([ { text: `Workflow started in background.\nRun ID: ${entry.runId}\nStatus: running`, diff --git a/packages/core/src/tools/workflow/workflow.ts b/packages/core/src/tools/workflow/workflow.ts index 63b58f54dbe..13fddc21e80 100644 --- a/packages/core/src/tools/workflow/workflow.ts +++ b/packages/core/src/tools/workflow/workflow.ts @@ -84,6 +84,11 @@ export interface WorkflowToolOptions { dispatch?: WorkflowAgentDispatch; } +export interface WorkflowToolResult extends ToolResult { + /** Exact run started by a successfully admitted background invocation. */ + workflowRunId?: string; +} + const WORKFLOW_PARAM_SCHEMA = { type: 'object', properties: { @@ -193,8 +198,10 @@ const WORKFLOW_PARAM_SCHEMA = { class WorkflowToolInvocation extends BaseToolInvocation< WorkflowParams, - ToolResult + WorkflowToolResult > { + private callId?: string; + constructor( private readonly config: Config, private readonly toolOptions: WorkflowToolOptions, @@ -203,6 +210,10 @@ class WorkflowToolInvocation extends BaseToolInvocation< super(params); } + setCallId(callId: string): void { + this.callId = callId; + } + getDescription(): string { if (this.params.scriptPath && this.params.script === undefined) { return `Run saved workflow (${path.basename(this.params.scriptPath)})`; @@ -222,7 +233,7 @@ class WorkflowToolInvocation extends BaseToolInvocation< signal: AbortSignal, updateOutput?: (output: ToolResultDisplay) => void, _shellExecutionConfig?: ShellExecutionConfig, - ): Promise { + ): Promise { const runInBackground = this.params.run_in_background === true; if (runInBackground && signal.aborted) { return backgroundStartCancelledResult(); @@ -232,6 +243,7 @@ class WorkflowToolInvocation extends BaseToolInvocation< handle = await WorkflowRunner.start({ config: this.config, signal, + toolUseId: this.callId, script: this.params.script, scriptPath: this.params.scriptPath, args: this.params.args, @@ -257,6 +269,7 @@ class WorkflowToolInvocation extends BaseToolInvocation< handle.budget.total, ); return { + workflowRunId: handle.runId, llmContent: [ { text: `Workflow started in background.\nRun ID: ${handle.runId}\nStatus: ${status}`, @@ -369,7 +382,7 @@ class WorkflowToolInvocation extends BaseToolInvocation< } } -function backgroundStartCancelledResult(): ToolResult { +function backgroundStartCancelledResult(): WorkflowToolResult { return { llmContent: 'Workflow was cancelled before it could start.', returnDisplay: 'Workflow cancelled.', @@ -605,7 +618,7 @@ These shapes are a starting point, not a menu; compose the harness the task actu export class WorkflowTool extends BaseDeclarativeTool< WorkflowParams, - ToolResult + WorkflowToolResult > { constructor( private readonly config: Config, @@ -664,7 +677,7 @@ export class WorkflowTool extends BaseDeclarativeTool< protected createInvocation( params: WorkflowParams, - ): ToolInvocation { + ): ToolInvocation { return new WorkflowToolInvocation(this.config, this.toolOptions, params); } }