fix(workflows): preserve completed node state across DAG multi-resume cycles (#1530)

* fix(workflows): preserve completed node state across DAG multi-resume cycles (#1520)

Bug: getCompletedDagNodeOutputs only queried 'node_completed' events, but
resumed runs emit 'node_skipped_prior_success' for already-completed nodes.
On a second resume, previously completed nodes were re-executed because
their skip events were invisible to the query.

Fix:
- Query both 'node_completed' and 'node_skipped_prior_success' event types
  in getCompletedDagNodeOutputs so multi-resume correctly identifies all
  previously completed nodes
- Store original node_output in node_skipped_prior_success event data so
  the output is available for $nodeId.output substitution on next resume
- Improve error messaging when all nodes fail: name the specific failed
  nodes and count skipped downstream nodes instead of the generic
  'no successful nodes' message

Closes #1520

* test: address review feedback — assert node_output, test failure message format

- Assert node_output field in node_skipped_prior_success event test
- Add test for empty-string fallback when node ID has undefined output
- Add dedicated test verifying failure message names failing nodes
- Update stale comment from 'no successful nodes' to new behavior
- Add edge-case test for only node_skipped_prior_success rows (no node_completed)

Author: kagura-agent

packages/core/src/db/workflow-events.test.ts

@@ -208,6 +208,46 @@ describe('workflow-events', () => {
       ]);
     });
 
+    test('returns outputs from node_skipped_prior_success events (multi-resume)', async () => {
+      mockQuery.mockResolvedValueOnce(
+        createQueryResult([
+          { step_name: 'node-a', data: { node_output: 'output A' } },
+          { step_name: 'node-b', data: { reason: 'prior_success', node_output: 'output B' } },
+        ])
+      );
+
+      const result = await getCompletedDagNodeOutputs('run-resume');
+
+      expect(result.size).toBe(2);
+      expect(result.get('node-a')).toBe('output A');
+      expect(result.get('node-b')).toBe('output B');
+      expect(mockQuery).toHaveBeenCalledWith(
+        expect.stringContaining('node_skipped_prior_success'),
+        ['run-resume']
+      );
+    });
+
+    test('returns outputs when only node_skipped_prior_success rows exist (no node_completed)', async () => {
+      mockQuery.mockResolvedValueOnce(
+        createQueryResult([
+          {
+            step_name: 'node-x',
+            data: { reason: 'prior_success', node_output: 'skipped output X' },
+          },
+          {
+            step_name: 'node-y',
+            data: { reason: 'prior_success', node_output: 'skipped output Y' },
+          },
+        ])
+      );
+
+      const result = await getCompletedDagNodeOutputs('run-all-skipped');
+
+      expect(result.size).toBe(2);
+      expect(result.get('node-x')).toBe('skipped output X');
+      expect(result.get('node-y')).toBe('skipped output Y');
+    });
+
     test('parses JSON string data (SQLite path)', async () => {
       mockQuery.mockResolvedValueOnce(
         createQueryResult([

packages/core/src/db/workflow-events.ts

@@ -127,7 +127,7 @@ export async function getCompletedDagNodeOutputs(
     data: string | Record<string, unknown>;
   }>(
     `SELECT step_name, data FROM remote_agent_workflow_events
-     WHERE workflow_run_id = $1 AND event_type = 'node_completed'
+     WHERE workflow_run_id = $1 AND event_type IN ('node_completed', 'node_skipped_prior_success')
      ORDER BY created_at ASC`,
     [workflowRunId]
   );

packages/workflows/src/dag-executor.test.ts

@@ -1292,10 +1292,10 @@ describe('executeDagWorkflow -- bash nodes', () => {
     );
 
     // The workflow should complete (it handles failures) but the node failed
-    // The mock platform should have received a failure message about no successful nodes
+    // The mock platform should have received a failure message about the failed node
     const sendMessage = platform.sendMessage as ReturnType<typeof mock>;
     const messages = sendMessage.mock.calls.map((call: unknown[]) => call[1] as string);
-    const failMsg = messages.find((m: string) => m.includes('no successful nodes'));
+    const failMsg = messages.find((m: string) => m.includes('failed') && m.includes('fail'));
     expect(failMsg).toBeDefined();
   });
 
@@ -3239,6 +3239,54 @@ describe('executeDagWorkflow -- resume with priorCompletedNodes', () => {
         (call[0] as { step_name: string }).step_name === 'step1'
     );
     expect(skippedEvent).toBeDefined();
+    expect(skippedEvent[0].data.node_output).toBe('prior output');
+  });
+
+  it('emits node_skipped_prior_success with empty output when node ID not in map', async () => {
+    const store = createMockStore();
+    const mockDeps = createMockDeps(store);
+    const platform = createMockPlatform();
+    const workflowRun = makeWorkflowRun('resume-empty-output');
+
+    // priorCompletedNodes has step1 but with undefined value to test the ?? '' fallback
+    const priorCompletedNodes = new Map<string, string>([
+      ['step1', undefined as unknown as string],
+    ]);
+
+    await executeDagWorkflow(
+      mockDeps,
+      platform,
+      'conv-resume',
+      testDir,
+      {
+        name: 'two-step',
+        nodes: [
+          { id: 'step1', command: 'step1' },
+          { id: 'step2', command: 'step2', depends_on: ['step1'] },
+        ],
+      },
+      workflowRun,
+      'claude',
+      undefined,
+      join(testDir, 'artifacts'),
+      join(testDir, 'logs'),
+      'main',
+      'docs/',
+      minimalConfig,
+      undefined,
+      undefined,
+      priorCompletedNodes
+    );
+
+    const eventCalls = (store.createWorkflowEvent as ReturnType<typeof mock>).mock.calls;
+    const skippedEvent = eventCalls.find(
+      (call: unknown[]) =>
+        (call[0] as { event_type: string }).event_type === 'node_skipped_prior_success' &&
+        (call[0] as { step_name: string }).step_name === 'step1'
+    );
+    expect(skippedEvent).toBeDefined();
+    // The ?? '' fallback kicks in when the map value is undefined
+    expect(skippedEvent[0].data.node_output).toBe('');
   });
 
   it('runs all nodes when priorCompletedNodes is empty', async () => {
@@ -4988,7 +5036,7 @@ describe('executeDagWorkflow -- terminal node output selection', () => {
 
     expect(store.failWorkflowRun).toHaveBeenCalled();
     // The "unknown provider" detail surfaces on the node_failed event; the
-    // workflow-level fail message is a generic "no successful nodes" summary.
+    // workflow-level fail message names the failing node(s).
     const eventCalls = (store.createWorkflowEvent as ReturnType<typeof mock>).mock.calls;
     const nodeFailedEvents = eventCalls.filter(
       (call: unknown[]) => (call[0] as Record<string, unknown>).event_type === 'node_failed'
@@ -5001,6 +5049,44 @@ describe('executeDagWorkflow -- terminal node output selection', () => {
     expect(nodeFailedData.error).toContain("unknown provider 'claud'");
   });
 
+  it('failure message names the failing node instead of generic summary', async () => {
+    const store = createMockStore();
+    const mockDeps = createMockDeps(store);
+    const platform = createMockPlatform();
+    const workflowRun = makeWorkflowRun();
+
+    await executeDagWorkflow(
+      mockDeps,
+      platform,
+      'conv-dag',
+      testDir,
+      {
+        name: 'fail-msg-test',
+        nodes: [
+          {
+            id: 'fail-node',
+            command: 'my-cmd',
+            provider: 'nonexistent',
+          },
+        ],
+      },
+      workflowRun,
+      'claude',
+      undefined,
+      join(testDir, 'artifacts'),
+      join(testDir, 'logs'),
+      'main',
+      'docs/',
+      minimalConfig
+    );
+
+    expect(store.failWorkflowRun).toHaveBeenCalled();
+    const failCall = (store.failWorkflowRun as ReturnType<typeof mock>).mock.calls[0];
+    const failMsg = failCall[1] as string;
+    expect(failMsg).toContain('fail-node failed');
+    expect(failMsg).not.toContain('no successful nodes');
+  });
+
   it('excludes intermediate nodes with dependents from terminal set (fan-in DAG)', async () => {
     let callCount = 0;
     mockSendQueryDag.mockImplementation(async function* () {
@@ -6545,7 +6631,7 @@ describe('executeDagWorkflow -- script nodes', () => {
 
     const sendMessage = platform.sendMessage as ReturnType<typeof mock>;
     const messages = sendMessage.mock.calls.map((call: unknown[]) => call[1] as string);
-    const failMsg = messages.find((m: string) => m.includes('no successful nodes'));
+    const failMsg = messages.find((m: string) => m.includes('failed') && m.includes('fail-script'));
     expect(failMsg).toBeDefined();
   });
 
@@ -6640,7 +6726,7 @@ describe('executeDagWorkflow -- script nodes', () => {
     const sendMessage = platform.sendMessage as ReturnType<typeof mock>;
     const messages = sendMessage.mock.calls.map((call: unknown[]) => call[1] as string);
     // Workflow fails because the only node failed (timeout)
-    const failMsg = messages.find((m: string) => m.includes('no successful nodes'));
+    const failMsg = messages.find((m: string) => m.includes('failed') && m.includes('slow-script'));
     expect(failMsg).toBeDefined();
   }, 10000);
 

packages/workflows/src/dag-executor.ts

@@ -2647,7 +2647,10 @@ export async function executeDagWorkflow(
                 workflow_run_id: workflowRun.id,
                 event_type: 'node_skipped_prior_success',
                 step_name: node.id,
-                data: { reason: 'prior_success' },
+                data: {
+                  reason: 'prior_success',
+                  node_output: priorCompletedNodes.get(node.id) ?? '',
+                },
               })
               .catch((err: Error) => {
                 getLog().error(
@@ -3146,9 +3149,16 @@ export async function executeDagWorkflow(
 
   if (!anyCompleted) {
     if (await skipIfStatusChanged('dag.skip_fail_status_changed')) return;
+    const failedNodes: string[] = [];
+    for (const [nodeId, o] of nodeOutputs) {
+      if (o.state === 'failed') failedNodes.push(nodeId);
+    }
     const failMsg =
-      `DAG workflow '${workflow.name}' completed with no successful nodes. ` +
-      'Check node conditions, trigger rules, and upstream failures.';
+      failedNodes.length > 0
+        ? `DAG workflow '${workflow.name}' failed: node${failedNodes.length > 1 ? 's' : ''} ${failedNodes.join(', ')} failed. ` +
+          `${nodeCounts.skipped} downstream node${nodeCounts.skipped !== 1 ? 's were' : ' was'} skipped.`
+        : `DAG workflow '${workflow.name}' completed with no successful nodes. ` +
+          'Check node conditions, trigger rules, and upstream failures.';
     // Note: nodeCounts not stored for failed runs — failWorkflowRun only stores { error }.
     // Frontend guards with isValidNodeCounts so missing node_counts is safe.
     await deps.store.failWorkflowRun(workflowRun.id, failMsg).catch((dbErr: Error) => {