From 1c746e0bf1bd3f068b9f91c7b86e301102aa5d3f Mon Sep 17 00:00:00 2001 From: Dawid Aksamski Date: Mon, 24 Aug 2026 09:49:03 +0200 Subject: [PATCH] feat(execution-core): emit node_skipped events --- .../src/components/execution/highlighting.tsx | 2 +- .../components/execution/log-panel.module.css | 12 ++- .../src/components/execution/log-panel.tsx | 9 ++- .../execution/node-markers.module.css | 5 ++ .../src/components/execution/node-markers.tsx | 8 +- .../src/stores/use-execution-store.ts | 6 +- packages/execution-core/README.md | 15 +++- .../graph-runner.replay-determinism.test.ts | 26 +++++++ .../execution-core/src/graph-runner.test.ts | 78 ++++++++++++++++++- packages/execution-core/src/graph-runner.ts | 39 +++++++++- .../workflow-execution/execution-events.ts | 17 ++++ 11 files changed, 205 insertions(+), 12 deletions(-) diff --git a/apps/ai-studio/src/components/execution/highlighting.tsx b/apps/ai-studio/src/components/execution/highlighting.tsx index 79051a9a9..b28db5ec8 100644 --- a/apps/ai-studio/src/components/execution/highlighting.tsx +++ b/apps/ai-studio/src/components/execution/highlighting.tsx @@ -33,7 +33,7 @@ export function ExecutionHighlighting() { const nonIdleNodes = new Set(); for (const [nodeId, state] of Object.entries(nodeStates) as [string, NodeExecutionState][]) { - if (state.status !== 'idle') { + if (state.status !== 'idle' && state.status !== 'skipped') { nonIdleNodes.add(nodeId); } diff --git a/apps/ai-studio/src/components/execution/log-panel.module.css b/apps/ai-studio/src/components/execution/log-panel.module.css index ec687cd99..def1d63dc 100644 --- a/apps/ai-studio/src/components/execution/log-panel.module.css +++ b/apps/ai-studio/src/components/execution/log-panel.module.css @@ -89,7 +89,8 @@ color: var(--ai-studio-status-color--completed); } -.badge--node_started { +.badge--node_started, +.badge--node_skipped { color: var(--ax-txt-secondary, #888); } @@ -108,6 +109,15 @@ opacity: 0.5; } +.reason { + font-size: 0.65rem; + font-style: italic; + opacity: 0.55; + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; +} + .time { margin-left: auto; font-size: 0.65rem; diff --git a/apps/ai-studio/src/components/execution/log-panel.tsx b/apps/ai-studio/src/components/execution/log-panel.tsx index 31bf79880..c357385f7 100644 --- a/apps/ai-studio/src/components/execution/log-panel.tsx +++ b/apps/ai-studio/src/components/execution/log-panel.tsx @@ -2,7 +2,7 @@ import { useSingleSelectedElement } from '@workflowbuilder/sdk'; import clsx from 'clsx'; import { useEffect, useRef, useState } from 'react'; -import type { ExecutionEvent } from '@workflow-builder/types/workflow-execution/execution-events'; +import type { ExecutionEvent, NodeSkipReason } from '@workflow-builder/types/workflow-execution/execution-events'; import styles from './log-panel.module.css'; @@ -10,6 +10,11 @@ import { useRightPanelAnchor } from '../../hooks/use-right-panel-anchor'; import { toggleLog, useExecutionStore } from '../../stores/use-execution-store'; import { extractOutputText } from '../../utils/extract-output-text'; +const SKIP_REASON_LABEL: Record = { + branch_not_taken: 'branch not taken', + upstream_skipped: 'upstream skipped', +}; + const DETAIL_PREVIEW_CHARS = 120; const NODE_ID_PREVIEW_CHARS = 8; const AT_BOTTOM_TOLERANCE_PX = 4; @@ -25,6 +30,7 @@ function EventRow({ event, selectedNodeId }: { event: ExecutionEvent; selectedNo const isNode = typeof nodeId === 'string' && nodeId.length > 0; const isHighlighted = isNode && nodeId === selectedNodeId; const label = event.type.replaceAll('_', ' '); + const skipReason = event.type === 'node_skipped' ? SKIP_REASON_LABEL[event.payload.reason] : undefined; let detail: string | undefined; switch (event.type) { @@ -71,6 +77,7 @@ function EventRow({ event, selectedNodeId }: { event: ExecutionEvent; selectedNo
{label} {isNode && {nodeId.slice(0, NODE_ID_PREVIEW_CHARS)}} + {skipReason && {skipReason}} {formatTime(event.timestamp)} {hasDetail && {isExpanded ? '▲' : '▼'}}
diff --git a/apps/ai-studio/src/components/execution/node-markers.module.css b/apps/ai-studio/src/components/execution/node-markers.module.css index 158f88ef2..a61a3e9d5 100644 --- a/apps/ai-studio/src/components/execution/node-markers.module.css +++ b/apps/ai-studio/src/components/execution/node-markers.module.css @@ -41,6 +41,11 @@ color: var(--ai-studio-status-color--failed); } +.icon--skipped { + color: var(--ax-txt-secondary, #888); + opacity: 0.7; +} + .icon--running { color: var(--ai-studio-status-color--completed); animation: spin 1.6s linear infinite; diff --git a/apps/ai-studio/src/components/execution/node-markers.tsx b/apps/ai-studio/src/components/execution/node-markers.tsx index d5a130ce4..d760aa9b6 100644 --- a/apps/ai-studio/src/components/execution/node-markers.tsx +++ b/apps/ai-studio/src/components/execution/node-markers.tsx @@ -17,7 +17,8 @@ export function ExecutionNodeMarkers({ props }: Props) { if (!nodeState || nodeState.status === 'idle') return null; - const isClickable = nodeState.status === 'completed' || nodeState.status === 'failed'; + const isClickable = + nodeState.status === 'completed' || nodeState.status === 'failed' || nodeState.status === 'skipped'; // The click also selects the node on the canvas, which drives the log highlight. return ( @@ -40,6 +41,11 @@ export function ExecutionNodeMarkers({ props }: Props) { )} + {nodeState.status === 'skipped' && ( + + + + )} ); } diff --git a/apps/ai-studio/src/stores/use-execution-store.ts b/apps/ai-studio/src/stores/use-execution-store.ts index effb405b6..6980eebec 100644 --- a/apps/ai-studio/src/stores/use-execution-store.ts +++ b/apps/ai-studio/src/stores/use-execution-store.ts @@ -7,7 +7,7 @@ import type { ExecutionStatus, } from '@workflow-builder/types/workflow-execution/execution-events'; -type NodeExecutionStatus = 'idle' | 'running' | 'completed' | 'failed'; +type NodeExecutionStatus = 'idle' | 'running' | 'completed' | 'failed' | 'skipped'; export type NodeExecutionState = { status: NodeExecutionStatus; @@ -107,6 +107,10 @@ function applyEventToNodeStates(event: ExecutionEvent, states: Record { expect(records[0]!.activityCallOrder).toEqual(['D', 'B']); }); + it('node_skipped — several skips in one wave keep the same order and reasons', async () => { + // The one event a skipped node ever emits, so its position and payload are all a + // replay has to compare. Emission order comes from `propagate`'s breadth-first walk + // over the dead subtree, seeded from `definition.nodes` order — a switch to a Set + // or an emit-as-you-go inside the wave would re-order these without changing + // anything else the runner records. + const input = makeInput( + [start('D'), trigger('Live'), trigger('C1'), trigger('C2'), trigger('C1prime')], + [ + edge('e1', 'D', 'Live', 'X'), + edge('e2', 'D', 'C1', 'Y'), + edge('e3', 'D', 'C2', 'Z'), + edge('e4', 'C1', 'C1prime'), + ], + ); + + const records = await runNTimes(input, { D: { output: 'd', nextPort: 'X' } }, RUNS); + expectAllRunsIdentical(records); + + expect(records[0]!.events.filter((event) => event.type === 'node_skipped')).toEqual([ + { type: 'node_skipped', nodeId: 'C1', payload: { reason: 'branch_not_taken' } }, + { type: 'node_skipped', nodeId: 'C2', payload: { reason: 'branch_not_taken' } }, + { type: 'node_skipped', nodeId: 'C1prime', payload: { reason: 'upstream_skipped' } }, + ]); + }); + it('node failure — failure path is deterministic too (same error code, same event sequence)', async () => { // The catch branch builds errorPayload from the thrown error. Across // replays the activity returns the same error (cached in history), so diff --git a/packages/execution-core/src/graph-runner.test.ts b/packages/execution-core/src/graph-runner.test.ts index 9d23e6682..a6014aad5 100644 --- a/packages/execution-core/src/graph-runner.test.ts +++ b/packages/execution-core/src/graph-runner.test.ts @@ -92,6 +92,12 @@ function edge(id: string, source: string, target: string, sourceHandle?: string) return { id, sourceNodeId: source, targetNodeId: target, sourceHandle }; } +function skipsFrom(events: EventCall[]): { nodeId: string | undefined; reason: unknown }[] { + return events + .filter((event) => event.type === 'node_skipped') + .map((event) => ({ nodeId: event.nodeId, reason: (event.payload as { reason: string }).reason })); +} + function makeInput(nodes: TestNode[], edges: WorkflowEdgeDefinition[]): WorkflowExecutionInput { const definition: WorkflowDefinition = { workflowId: 'wf-1', nodes, edges }; return { @@ -202,7 +208,7 @@ describe('runGraph — topological scheduling', () => { expect(runner.contexts.C).toEqual({ S: 'out-S', A: 'out-A', Aprime: 'out-Aprime', B: 'out-B' }); }); - it('decision routing — node reachable only via pruned branch is skipped silently', async () => { + it('decision routing — node reachable only via pruned branch is skipped and reported', async () => { // D is a decision picking branch X. C is reachable only via Y → must be skipped. const runner = makeRunner({ D: { output: { matchedBranch: 'X' }, nextPort: 'X' }, @@ -216,12 +222,37 @@ describe('runGraph — topological scheduling', () => { ); expect(runner.callOrder).toEqual(['D', 'B']); - // No node_started for C + // C never ran, but the run says so out loud rather than leaving it indistinguishable + // from a node still pending. D ran and chose another handle → branch_not_taken. expect(events.events.some((event) => event.type === 'node_started' && event.nodeId === 'C')).toBe(false); + expect(skipsFrom(events.events)).toEqual([{ nodeId: 'C', reason: 'branch_not_taken' }]); // Graph still completes successfully expect(events.statuses.at(-1)?.status).toBe('completed'); }); + it('node_skipped lands after the wave that pruned it and before the next wave starts', async () => { + // Ordering is the contract a log panel renders against: the skip must read as a + // consequence of D's completion, not arrive interleaved with B's execution. + const runner = makeRunner({ D: { output: 'd', nextPort: 'X' } }); + const events = makeEvents(); + + await runGraph( + makeInput([start('D'), trigger('B'), trigger('C')], [edge('e1', 'D', 'B', 'X'), edge('e2', 'D', 'C', 'Y')]), + runner.port, + events.port, + ); + + expect(events.events.map((event) => `${event.type}:${event.nodeId ?? '-'}`)).toEqual([ + 'execution_started:-', + 'node_started:D', + 'node_completed:D', + 'node_skipped:C', + 'node_started:B', + 'node_completed:B', + 'execution_completed:-', + ]); + }); + it('decision-pruned fan-in — join executes with only the live predecessor', async () => { // D picks X → B runs, C is skipped. E joins B and C — should run with only B's output. const runner = makeRunner({ @@ -268,6 +299,42 @@ describe('runGraph — topological scheduling', () => { // Only D, B, and E execute expect(runner.callOrder.sort()).toEqual(['B', 'D', 'E']); expect(runner.contexts.E).toEqual({ D: 'd', B: 'out-B' }); + // C is the head of the dead branch (D ran, chose X); Cprime is deeper inside it. + expect(skipsFrom(events.events)).toEqual([ + { nodeId: 'C', reason: 'branch_not_taken' }, + { nodeId: 'Cprime', reason: 'upstream_skipped' }, + ]); + }); + + it("skip reason ignores the order a join's predecessors resolve in", async () => { + // J joins two dead edges of different kinds: C→J (C was itself skipped) resolves + // first, then B→J (B ran and routed to K instead). One live-but-pruned incoming + // edge is enough, so J reports branch_not_taken regardless of which edge lands + // last — a "whichever predecessor resolved last wins" rule would call it + // upstream_skipped here. + const runner = makeRunner({ D: { output: 'd', nextPort: 'X' }, B: { output: 'b', nextPort: 'P' } }); + const events = makeEvents(); + + await runGraph( + makeInput( + [start('D'), trigger('B'), trigger('C'), trigger('J'), trigger('K')], + [ + edge('e1', 'D', 'B', 'X'), + edge('e2', 'D', 'C', 'Y'), + edge('e3', 'C', 'J'), + edge('e4', 'B', 'J', 'Q'), + edge('e5', 'B', 'K', 'P'), + ], + ), + runner.port, + events.port, + ); + + expect(runner.callOrder).toEqual(['D', 'B', 'K']); + expect(skipsFrom(events.events)).toEqual([ + { nodeId: 'C', reason: 'branch_not_taken' }, + { nodeId: 'J', reason: 'branch_not_taken' }, + ]); }); it('failure short-circuits the graph — emits execution_failed and stops', async () => { @@ -714,6 +781,9 @@ describe('runGraph — errorPolicy', () => { expect(runner.callOrder).toEqual(['A', 'B', 'Recovery']); expect(events.events.some((event) => event.type === 'node_started' && event.nodeId === 'Success')).toBe(false); + // B ran (and failed) before routing to the error handle, so the pruned success + // branch is a branch not taken — not an upstream skip. + expect(skipsFrom(events.events)).toEqual([{ nodeId: 'Success', reason: 'branch_not_taken' }]); expect(runner.contexts.Recovery).toEqual({ A: 'out-A', B: { error: { message: 'boom' } } }); expect(events.statuses.at(-1)?.status).toBe('completed'); expect(outcome).toEqual({ status: 'completed' }); @@ -748,6 +818,10 @@ describe('runGraph — errorPolicy', () => { ); expect(runner.callOrder).toEqual(['A', 'B', 'Recovery', 'Done']); + expect(skipsFrom(events.events)).toEqual([ + { nodeId: 'Success', reason: 'branch_not_taken' }, + { nodeId: 'SuccessPrime', reason: 'upstream_skipped' }, + ]); expect(events.statuses.at(-1)?.status).toBe('completed'); }); diff --git a/packages/execution-core/src/graph-runner.ts b/packages/execution-core/src/graph-runner.ts index d09191144..8a33b3b80 100644 --- a/packages/execution-core/src/graph-runner.ts +++ b/packages/execution-core/src/graph-runner.ts @@ -1,3 +1,4 @@ +import type { NodeSkipReason } from '@workflow-builder/types/workflow-execution/execution-events'; import type { BaseNode, NodeErrorPolicy, @@ -38,8 +39,8 @@ export type RunGraphOutcome = { status: 'completed' } | { status: 'failed'; erro // sandbox-safe entry (`./workflow`) and therefore runs inside Temporal's // V8 workflow context, where every call to `new Date()`, `Math.random()`, // or other non-deterministic source poisons history replay. Lifecycle -// signals (execution_started/completed/failed, node_started/completed/failed) -// already flow through EventEmitterPort — operators tail those for run-time +// signals (execution_started/completed/failed, node_started/completed/failed, +// node_skipped) already flow through EventEmitterPort — operators tail those for run-time // observability. Activity executors that need real-time logs (LLM failures, // HTTP retries) hold their own LoggerPort outside the sandbox. export async function runGraph( @@ -67,6 +68,7 @@ export async function runGraph( pendingPredecessors: new Map(inDegree), liveIncoming: new Map(input.definition.nodes.map((node) => [node.id, 0])), status: new Map(input.definition.nodes.map((node) => [node.id, 'pending'])), + prunedFromLiveSource: new Set(), }; let ready: TNode[] = [startNode]; @@ -92,6 +94,7 @@ export async function runGraph( } const newlyReady: TNode[] = []; + const skipped: SkippedNode[] = []; for (const result of results) { if (result.failed) { // 'continue' and 'errorRoute' absorb the error into nodeOutputs so downstream @@ -104,12 +107,22 @@ export async function runGraph( nodeOutputs[result.node.id] = errorOutput; state.status.set(result.node.id, 'completed'); const nextPort = policy === 'errorRoute' ? RESERVED_ERROR_HANDLE : undefined; - propagate(result.node.id, nextPort, true, state, newlyReady); + propagate(result.node.id, nextPort, true, state, newlyReady, skipped); continue; } nodeOutputs[result.node.id] = result.output; state.status.set(result.node.id, 'completed'); - propagate(result.node.id, result.nextPort, true, state, newlyReady); + propagate(result.node.id, result.nextPort, true, state, newlyReady, skipped); + } + + // Emitted once the whole wave has propagated, so a skip reads as a consequence of + // the wave that pruned it rather than arriving mid-wave. Order is a pure function of + // the definition: `results` follows `ready`, which follows `definition.nodes`, and + // `propagate` walks the dead subtree breadth-first from there — nothing wall-clock or + // completion-order dependent, so a replay reproduces it. Emit failures are not + // swallowed: as with every other lifecycle emit, an exhausted `emitEvent` fails the run. + for (const node of skipped) { + await events.emitEvent(input.executionId, 'node_skipped', { reason: node.reason }, node.id); } ready = newlyReady; @@ -152,8 +165,15 @@ type SchedulerState = { pendingPredecessors: Map; liveIncoming: Map; status: Map; + // Nodes with at least one incoming edge pruned while its source was live — an upstream + // node ran and routed elsewhere. Separates the head of a dead branch from the rest of it + // when reporting why a node was skipped. A set, not a flag overwritten by the last edge + // resolved: the reason must not depend on the order a node's predecessors resolve in. + prunedFromLiveSource: Set; }; +type SkippedNode = { id: string; reason: NodeSkipReason }; + // Resolves all outgoing edges from `rootId`. For each successor, decrements its // pending counter, increments live counter if the edge is alive (no decision // pruning, or sourceHandle matches the decision's nextPort). When pending hits @@ -161,12 +181,17 @@ type SchedulerState = { // through its own outgoing edges via the same queue, so unreachable subtrees // don't stall downstream join points and deep dead-branch chains can't blow the // call stack. +// +// Newly ready nodes land in `out`, newly skipped ones in `skippedOut`; both are +// appended in traversal order and neither is emitted from here, so the caller +// keeps control of event ordering. function propagate( rootId: string, rootNextPort: string | undefined, rootSourceLive: boolean, state: SchedulerState, out: TNode[], + skippedOut: SkippedNode[], ): void { const queue: { fromId: string; nextPort: string | undefined; sourceLive: boolean }[] = [ { fromId: rootId, nextPort: rootNextPort, sourceLive: rootSourceLive }, @@ -179,6 +204,8 @@ function propagate( state.pendingPredecessors.set(target.id, (state.pendingPredecessors.get(target.id) ?? 0) - 1); if (edgeLive) { state.liveIncoming.set(target.id, (state.liveIncoming.get(target.id) ?? 0) + 1); + } else if (sourceLive) { + state.prunedFromLiveSource.add(target.id); } if ((state.pendingPredecessors.get(target.id) ?? 0) === 0 && state.status.get(target.id) === 'pending') { @@ -186,6 +213,10 @@ function propagate( out.push(target); } else { state.status.set(target.id, 'skipped'); + skippedOut.push({ + id: target.id, + reason: state.prunedFromLiveSource.has(target.id) ? 'branch_not_taken' : 'upstream_skipped', + }); queue.push({ fromId: target.id, nextPort: undefined, sourceLive: false }); } } diff --git a/packages/types/src/workflow-execution/execution-events.ts b/packages/types/src/workflow-execution/execution-events.ts index e811616f2..851617b62 100644 --- a/packages/types/src/workflow-execution/execution-events.ts +++ b/packages/types/src/workflow-execution/execution-events.ts @@ -4,6 +4,7 @@ export type ExecutionEventType = | 'node_waiting' | 'node_completed' | 'node_failed' + | 'node_skipped' | 'branch_spawned' | 'branches_joined' | 'execution_completed' @@ -30,6 +31,16 @@ export type NodeCompletedPayload = { output: unknown; }; +// Why a node never ran. 'branch_not_taken' — an upstream node completed but +// routed elsewhere (a decision picking another branch, or an 'errorRoute' +// policy pruning the success branch). 'upstream_skipped' — every predecessor +// was itself skipped, so this node sits deeper in an already-dead branch. +export type NodeSkipReason = 'branch_not_taken' | 'upstream_skipped'; + +export type NodeSkippedPayload = { + reason: NodeSkipReason; +}; + export type ExecutionErrorPayload = { error: { message: string; @@ -81,6 +92,11 @@ export type NodeFailedEvent = NodeEvent & { payload: ExecutionErrorPayload; }; +export type NodeSkippedEvent = NodeEvent & { + type: 'node_skipped'; + payload: NodeSkippedPayload; +}; + export type BranchSpawnedEvent = NodeEvent & { type: 'branch_spawned'; payload: BranchSpawnedPayload; @@ -112,6 +128,7 @@ export type ExecutionEvent = | NodeWaitingEvent | NodeCompletedEvent | NodeFailedEvent + | NodeSkippedEvent | BranchSpawnedEvent | BranchesJoinedEvent | ExecutionCompletedEvent