From 00ea813e702dcdc74d18323283d2b604e11692b9 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 24 Sep 2026 19:23:52 -0700 Subject: [PATCH] fix(executor): stop run-from-block reusing the source run's branch state --- apps/sim/executor/execution/edge-manager.ts | 3 +- .../execution/executor.run-from-block.test.ts | 215 ++++++++++++++++++ apps/sim/executor/execution/executor.ts | 9 + 3 files changed, 226 insertions(+), 1 deletion(-) create mode 100644 apps/sim/executor/execution/executor.run-from-block.test.ts diff --git a/apps/sim/executor/execution/edge-manager.ts b/apps/sim/executor/execution/edge-manager.ts index d0911492db6..c190beafc6d 100644 --- a/apps/sim/executor/execution/edge-manager.ts +++ b/apps/sim/executor/execution/edge-manager.ts @@ -176,7 +176,8 @@ export class EdgeManager { } /** - * Clear deactivated edges for a set of nodes (used when restoring loop state for next iteration). + * Clear deactivated edges for a set of nodes that is about to re-execute (a loop's next + * iteration, a parallel batch, or a run-from-block dirty set). * * Only clears edges whose SOURCE is in the provided set. Edges pointing INTO a node in the set * whose source lives outside (e.g. an external branch whose path was cascade-deactivated) must diff --git a/apps/sim/executor/execution/executor.run-from-block.test.ts b/apps/sim/executor/execution/executor.run-from-block.test.ts new file mode 100644 index 00000000000..0b92b9f6470 --- /dev/null +++ b/apps/sim/executor/execution/executor.run-from-block.test.ts @@ -0,0 +1,215 @@ +/** + * @vitest-environment node + */ +import type { SessionPrincipal } from '@sim/auth/principal' +import { createSerializedBlock, createSerializedWorkflow } from '@sim/testing' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { executed, gateChoice } = vi.hoisted(() => ({ + executed: [] as string[], + gateChoice: { value: 'if' as 'if' | 'else' }, +})) + +vi.mock('@/executor/handlers/registry', () => ({ + createBlockHandlers: () => [ + { + canHandle: () => true, + execute: async ( + _ctx: unknown, + block: { id: string; metadata?: { id?: string; name?: string } } + ) => { + executed.push(block.id) + if (block.metadata?.id === 'condition') { + return { selectedOption: `${block.metadata.name}-${gateChoice.value}` } + } + return { ok: true } + }, + }, + ], +})) + +import { BlockType } from '@/executor/constants' +import { DAGExecutor } from '@/executor/execution/executor' +import { stripCloneSuffixes } from '@/executor/utils/subflow-utils' +import type { SerializedWorkflow } from '@/serializer/types' + +const PRINCIPAL: SessionPrincipal = { kind: 'session', userId: 'user-1', sessionId: 'session-1' } + +/** Block ids double as their type for the special blocks; every other block is a function. */ +const BLOCK_TYPES: Record = { + start: BlockType.STARTER, + gate: BlockType.CONDITION, + loop: BlockType.LOOP, + fanOut: BlockType.PARALLEL, +} + +function workflow( + connections: SerializedWorkflow['connections'], + loops: SerializedWorkflow['loops'] = {}, + parallels: SerializedWorkflow['parallels'] = {} +): SerializedWorkflow { + const ids = new Set(connections.flatMap((c) => [c.source, c.target])) + const blocks = [...ids].map((id) => + createSerializedBlock({ id, name: id, type: BLOCK_TYPES[id] ?? BlockType.FUNCTION }) + ) + return { ...createSerializedWorkflow(blocks, connections), loops, parallels } +} + +function createExecutor(wf: SerializedWorkflow, executionId: string): DAGExecutor { + return new DAGExecutor({ + workflow: wf, + contextExtensions: { workspaceId: 'ws', executionId, principal: PRINCIPAL }, + }) +} + +/** A full run with one gate decision, then a run-from-block seeded from that run's snapshot. */ +async function rerunFromBlock( + wf: SerializedWorkflow, + startBlockId: string, + first: 'if' | 'else', + second: 'if' | 'else' +): Promise { + gateChoice.value = first + const run1 = await createExecutor(wf, 'exec-1').execute('wf') + expect(run1.success).toBe(true) + expect(run1.executionState).toBeDefined() + + executed.length = 0 + gateChoice.value = second + const run2 = await createExecutor(wf, 'exec-2').executeFromBlock( + 'wf', + startBlockId, + run1.executionState! + ) + expect(run2.success).toBe(true) + return [...executed] +} + +const exclusive = workflow([ + { source: 'start', target: 'prep' }, + { source: 'prep', target: 'gate' }, + { source: 'gate', target: 'readDoc', sourceHandle: 'condition-gate-if' }, + { source: 'gate', target: 'noDoc', sourceHandle: 'condition-gate-else' }, +]) + +const diamond = workflow([ + { source: 'start', target: 'prep' }, + { source: 'prep', target: 'gate' }, + { source: 'gate', target: 'readDoc', sourceHandle: 'condition-gate-if' }, + { source: 'gate', target: 'docPlan', sourceHandle: 'condition-gate-else' }, + { source: 'readDoc', target: 'docPlan' }, +]) + +const siblings = workflow([ + { source: 'start', target: 'a' }, + { source: 'start', target: 'b' }, + { source: 'a', target: 'join' }, + { source: 'b', target: 'join' }, +]) + +const looped = workflow( + [ + { source: 'start', target: 'prep' }, + { source: 'prep', target: 'loop' }, + { source: 'loop', target: 'gate', sourceHandle: 'loop-start-source' }, + { source: 'gate', target: 'readDoc', sourceHandle: 'condition-gate-if' }, + { source: 'gate', target: 'noDoc', sourceHandle: 'condition-gate-else' }, + ], + { loop: { id: 'loop', nodes: ['gate', 'readDoc', 'noDoc'], iterations: 1, loopType: 'for' } } +) + +describe('DAGExecutor run-from-block edge state', () => { + beforeEach(() => { + executed.length = 0 + }) + + it.each([ + { + title: 'does not run an unselected branch the source execution had activated', + wf: exclusive, + start: 'prep', + first: 'if', + second: 'else', + expected: ['prep', 'gate', 'noDoc'], + }, + { + title: 'runs the branch the source execution had deactivated when it is selected', + wf: exclusive, + start: 'prep', + first: 'else', + second: 'if', + expected: ['prep', 'gate', 'readDoc'], + }, + { + title: 'waits for the live input of a join the source execution had released early', + wf: diamond, + start: 'prep', + first: 'else', + second: 'if', + expected: ['prep', 'gate', 'readDoc', 'docPlan'], + }, + { + title: 'still runs the join directly when its other input is deselected', + wf: diamond, + start: 'prep', + first: 'if', + second: 'else', + expected: ['prep', 'gate', 'docPlan'], + }, + { + title: 'does not wait on a cached sibling input outside the re-run region', + wf: siblings, + start: 'a', + first: 'if', + second: 'if', + expected: ['a', 'join'], + }, + { + title: 'does not run a stale branch inside a loop', + wf: looped, + start: 'prep', + first: 'if', + second: 'else', + expected: ['prep', 'gate', 'noDoc'], + }, + ] as const)('$title', async ({ wf, start, first, second, expected }) => { + expect(await rerunFromBlock(wf, start, first, second)).toEqual(expected) + }) + + it('runs a join once after every re-run input completes', async () => { + const fanIn = workflow([ + { source: 'start', target: 'prep' }, + { source: 'prep', target: 'a' }, + { source: 'prep', target: 'b' }, + { source: 'a', target: 'join' }, + { source: 'b', target: 'join' }, + ]) + const order = await rerunFromBlock(fanIn, 'prep', 'if', 'if') + expect(order.filter((id) => id === 'join')).toHaveLength(1) + expect(order.indexOf('join')).toBeGreaterThan(Math.max(order.indexOf('a'), order.indexOf('b'))) + }) + + it('does not run a stale branch in any parallel branch copy', async () => { + const parallel = workflow( + [ + { source: 'start', target: 'prep' }, + { source: 'prep', target: 'fanOut' }, + { source: 'fanOut', target: 'gate', sourceHandle: 'parallel-start-source' }, + { source: 'gate', target: 'readDoc', sourceHandle: 'condition-gate-if' }, + { source: 'gate', target: 'noDoc', sourceHandle: 'condition-gate-else' }, + ], + {}, + { + fanOut: { + id: 'fanOut', + nodes: ['gate', 'readDoc', 'noDoc'], + count: 3, + parallelType: 'count', + }, + } + ) + const ids = (await rerunFromBlock(parallel, 'prep', 'if', 'else')).map(stripCloneSuffixes) + expect(ids).not.toContain('readDoc') + expect(ids.filter((id) => id === 'noDoc')).toHaveLength(3) + }) +}) diff --git a/apps/sim/executor/execution/executor.ts b/apps/sim/executor/execution/executor.ts index 6c5e5fdb37b..4329b883762 100644 --- a/apps/sim/executor/execution/executor.ts +++ b/apps/sim/executor/execution/executor.ts @@ -383,6 +383,15 @@ export class DAGExecutor { snapshotState?.deactivatedEdges, snapshotState?.nodesWithActivatedEdge ) + /** + * Run-from-block re-executes its dirty set from scratch, so the source execution's edge state + * for those nodes must not carry over: a stale activation runs an unselected branch, and a + * stale deactivation releases a join before its live input completes. + */ + const dirtySet = context.runFromBlockContext?.dirtySet + if (dirtySet) { + edgeManager.clearDeactivatedEdgesForNodes(dirtySet) + } const nodeOrchestrator = new NodeExecutionOrchestrator( dag, state,