feat(core): expose workflow execution state

packages/cli/src/ui/commands/workflowsCommand.test.ts

@@ -28,10 +28,14 @@ function entry(overrides: Partial<WorkflowTask> = {}): 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<string | null, number>(),

packages/cli/src/ui/commands/workflowsCommand.ts

@@ -30,10 +30,14 @@ function snapshotToTask(s: WorkflowSnapshot): WorkflowTask {
     meta: s.meta,
     status: s.status,
     currentPhase: null,
+    currentPhaseVisitId: null,
     phases: s.phases ?? [],
+    phaseVisits: s.phaseVisits ?? [],
+    dispatches: s.dispatches ?? [],
     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 +52,7 @@ function snapshotToTask(s: WorkflowSnapshot): WorkflowTask {
     outputOffset: 0,
     notified: true,
     abortController: new AbortController(),
-  } as WorkflowTask;
+  };
 }
 
 /**

packages/core/src/agents/runtime/workflow-journal.test.ts

@@ -126,6 +126,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();

packages/core/src/agents/runtime/workflow-journal.ts

@@ -168,6 +168,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. */
@@ -183,6 +185,13 @@ export class WorkflowJournal {
 
   /** Append one entry. Rejects only on I/O error (callers `.catch`). */
   append(entry: JournalEntry): Promise<void> {
-    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<void> {
+    return this.pending;
   }
 }

packages/core/src/agents/runtime/workflow-orchestrator.test.ts

@@ -417,6 +417,170 @@ 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<string | undefined> = [];
+    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('emitter subscriber errors do not break the run (defensive try/catch)', async () => {
     const orchestrator = new WorkflowOrchestrator(
       async (prompt) => `mock:${prompt}`,
@@ -813,9 +977,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
@@ -827,6 +995,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 () => {