Skip to content

Commit 7f3d93c

Browse files
committed
fix(executor): fail a stop-after run whose routing skips the stop block
A run with stopAfterBlockId only stopped when the stop block completed. When a router, condition, or untaken error path routed the run away from it, the stop never triggered and the run finished every other branch, reporting success as if it had stopped there. A static check before the run cannot see this. - The engine ends the run as soon as every path into the stop block has been deactivated, before any further block starts, and fails it with `Stop block "<name>" (<id>) was not reached: no path this run took leads to it`. - Any run that ends without completing its stop block fails the same way: a stop block missing from the executed graph, or a Response block that ended the run first. - A loop or parallel stop with nothing to run completes at its start sentinel, whose end sentinel never runs, so that exit now counts as reaching it. - The v2 contract and the CLI `--stop-after` help describe the failure. - E2E: a condition fixture checks the stop on the taken branch still stops there, a stop on the skipped branch fails the run before the other branch's slow block finishes, and the CLI exits non-zero.
1 parent 6b3cd82 commit 7f3d93c

9 files changed

Lines changed: 483 additions & 30 deletions

File tree

‎apps/docs/content/docs/cli/reference.mdx‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6704,7 +6704,7 @@ sim workflows run <workflowId> [options]
67046704
| `--mock-payload` | No | Use the selected trigger's server-derived mock payload; runs the current saved workflow state (implies --manual). |
67056705
| `--from-block <blockId>` | No | Run manually from this saved workflow block. |
67066706
| `--source-run <runId>` | No | Prior run whose persisted state supplies upstream outputs (requires --from-block). |
6707-
| `--stop-after <blockId>` | No | Stop the run after this saved block; with --from-block on the same block, re-runs only that block (implies --manual). |
6707+
| `--stop-after <blockId>` | No | Stop the run after this saved block, failing it if the run takes a path that skips the block; with --from-block on the same block, re-runs only that block (implies --manual). |
67086708
| `--follow` | No | Stream the run as it happens; progress on stderr, result on stdout. The stream reports only success and output, so the result omits the run id and timings a non-streaming run returns. |
67096709
| `--include-thinking` | No | Show model reasoning while following (requires --follow). |
67106710
| `--include-tool-calls` | No | Show tool calls while following (requires --follow). |

‎apps/docs/content/docs/cli/workflows.mdx‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -642,7 +642,7 @@ sim workflows run <workflowId> [options]
642642
| `--mock-payload` | No | Use the selected trigger's server-derived mock payload; runs the current saved workflow state (implies --manual). |
643643
| `--from-block <blockId>` | No | Run manually from this saved workflow block. |
644644
| `--source-run <runId>` | No | Prior run whose persisted state supplies upstream outputs (requires --from-block). |
645-
| `--stop-after <blockId>` | No | Stop the run after this saved block; with --from-block on the same block, re-runs only that block (implies --manual). |
645+
| `--stop-after <blockId>` | No | Stop the run after this saved block, failing it if the run takes a path that skips the block; with --from-block on the same block, re-runs only that block (implies --manual). |
646646
| `--follow` | No | Stream the run as it happens; progress on stderr, result on stdout. The stream reports only success and output, so the result omits the run id and timings a non-streaming run returns. |
647647
| `--include-thinking` | No | Show model reasoning while following (requires --follow). |
648648
| `--include-tool-calls` | No | Show tool calls while following (requires --follow). |

‎apps/docs/openapi-v2-workflows.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12617,7 +12617,7 @@
1261712617
]
1261812618
},
1261912619
"stopAfterBlockId": {
12620-
"description": "Saved workflow block after which the run stops; downstream blocks do not execute. Must not be inside a loop or parallel. With a block entry naming the same block, re-runs only that block against the source run.",
12620+
"description": "Saved workflow block after which the run stops; downstream blocks do not execute. Must not be inside a loop or parallel. If a router, condition, or untaken error path routes the run away from the block, the run fails as soon as that is decided, without running the other branches. With a block entry naming the same block, re-runs only that block against the source run.",
1262112621
"type": "string",
1262212622
"minLength": 1
1262312623
}

‎apps/sim/executor/execution/edge-manager.ts‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -122,6 +122,17 @@ export class EdgeManager {
122122
return node.incomingEdges.size === 0 || this.countActiveIncomingEdges(node) === 0
123123
}
124124

125+
/**
126+
* Whether a node that has not been queued can still run: it has received an activated edge, or
127+
* an incoming edge is still undecided. False means every path into it was deactivated (a router
128+
* or condition chose another route, or an error path was not taken), so nothing will queue it.
129+
*/
130+
canNodeStillRun(nodeId: string): boolean {
131+
if (this.nodesWithActivatedEdge.has(nodeId)) return true
132+
const node = this.dag.nodes.get(nodeId)
133+
return node !== undefined && this.countActiveIncomingEdges(node) > 0
134+
}
135+
125136
restoreIncomingEdge(targetNodeId: string, sourceNodeId: string): void {
126137
const targetNode = this.dag.nodes.get(targetNodeId)
127138
if (!targetNode) {

‎apps/sim/executor/execution/engine.test.ts‎

Lines changed: 270 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -21,13 +21,17 @@ vi.mock('@/lib/execution/cancellation', () => ({
2121
},
2222
}))
2323

24-
import { EDGE } from '@/executor/constants'
25-
import type { DAG, DAGNode } from '@/executor/dag/builder'
26-
import type { EdgeManager } from '@/executor/execution/edge-manager'
24+
import { BlockType, EDGE } from '@/executor/constants'
25+
import { type DAG, DAGBuilder, type DAGNode } from '@/executor/dag/builder'
26+
import { EdgeManager } from '@/executor/execution/edge-manager'
2727
import type { NodeExecutionOrchestrator } from '@/executor/orchestrators/node'
2828
import type { ExecutionContext, ExecutionResult } from '@/executor/types'
2929
import { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry'
30-
import type { SerializedBlock } from '@/serializer/types'
30+
import {
31+
buildLoopSentinelEndId,
32+
buildLoopSentinelStartId,
33+
} from '@/executor/utils/subflow-node-id-codec'
34+
import type { SerializedBlock, SerializedWorkflow } from '@/serializer/types'
3135
import { ExecutionEngine } from './engine'
3236

3337
const executionEngineLoggerCallIndex = loggerMock.createLogger.mock.calls.findIndex(
@@ -1088,4 +1092,266 @@ describe('ExecutionEngine', () => {
10881092
expect(result.output).toEqual({ data: { response: true }, status: 200, headers: {} })
10891093
})
10901094
})
1095+
/**
1096+
* A stop-after run is a promise that the run ends with the stop block. These run the real DAG
1097+
* builder and edge manager, so a router or condition deciding a path is the same decision the
1098+
* engine makes in production.
1099+
*/
1100+
describe('Stop-after block', () => {
1101+
function block(id: string, type = BlockType.FUNCTION): SerializedBlock {
1102+
return { ...createMockBlock(id), metadata: { id: type, name: id } }
1103+
}
1104+
1105+
function buildRun(
1106+
workflow: SerializedWorkflow,
1107+
stopAfterBlockId: string,
1108+
outputs: Record<string, ExecutionResult['output']> = {}
1109+
) {
1110+
const dag = new DAGBuilder().build(workflow, { triggerBlockId: 'start' })
1111+
const executed: string[] = []
1112+
const nodeOrchestrator = {
1113+
executeNode: vi.fn(async (_ctx: ExecutionContext, nodeId: string) => {
1114+
executed.push(nodeId)
1115+
return { nodeId, output: outputs[nodeId] ?? {}, isFinalOutput: false }
1116+
}),
1117+
handleNodeCompletion: vi.fn(),
1118+
} as unknown as NodeExecutionOrchestrator
1119+
const engine = new ExecutionEngine(
1120+
createMockContext({ stopAfterBlockId }),
1121+
dag,
1122+
new EdgeManager(dag),
1123+
nodeOrchestrator
1124+
)
1125+
return { engine, executed }
1126+
}
1127+
1128+
/** start → condition; `if` → taken → takenTail, `else` → stop → after. */
1129+
const conditionWorkflow: SerializedWorkflow = {
1130+
version: '1',
1131+
blocks: [
1132+
block('start', BlockType.STARTER),
1133+
block('condition', BlockType.CONDITION),
1134+
block('taken'),
1135+
block('takenTail'),
1136+
block('stop'),
1137+
block('after'),
1138+
],
1139+
connections: [
1140+
{ source: 'start', target: 'condition' },
1141+
{ source: 'condition', target: 'taken', sourceHandle: 'condition-if' },
1142+
{ source: 'condition', target: 'stop', sourceHandle: 'condition-else' },
1143+
{ source: 'taken', target: 'takenTail' },
1144+
{ source: 'stop', target: 'after' },
1145+
],
1146+
loops: {},
1147+
parallels: {},
1148+
}
1149+
1150+
it('fails as soon as a condition routes away from the stop block', async () => {
1151+
const { engine, executed } = buildRun(conditionWorkflow, 'stop', {
1152+
condition: { selectedOption: 'if' },
1153+
})
1154+
1155+
await expect(engine.run('start')).rejects.toThrow('Stop block "stop" (stop) was not reached')
1156+
expect(executed).toEqual(['start', 'condition'])
1157+
})
1158+
1159+
it('fails as soon as a router routes away from the stop block', async () => {
1160+
const { engine, executed } = buildRun(
1161+
{
1162+
version: '1',
1163+
blocks: [
1164+
block('start', BlockType.STARTER),
1165+
block('router', BlockType.ROUTER_V2),
1166+
block('taken'),
1167+
block('takenTail'),
1168+
block('stop'),
1169+
],
1170+
connections: [
1171+
{ source: 'start', target: 'router' },
1172+
{ source: 'router', target: 'taken', sourceHandle: 'router-route-a' },
1173+
{ source: 'router', target: 'stop', sourceHandle: 'router-route-b' },
1174+
{ source: 'taken', target: 'takenTail' },
1175+
],
1176+
loops: {},
1177+
parallels: {},
1178+
},
1179+
'stop',
1180+
{ router: { selectedRoute: 'route-a' } }
1181+
)
1182+
1183+
await expect(engine.run('start')).rejects.toThrow('Stop block "stop" (stop) was not reached')
1184+
expect(executed).toEqual(['start', 'router'])
1185+
})
1186+
1187+
it('fails when the stop block sits on an error path the run never takes', async () => {
1188+
const { engine, executed } = buildRun(
1189+
{
1190+
version: '1',
1191+
blocks: [block('start', BlockType.STARTER), block('work'), block('next'), block('stop')],
1192+
connections: [
1193+
{ source: 'start', target: 'work' },
1194+
{ source: 'work', target: 'next', sourceHandle: 'source' },
1195+
{ source: 'work', target: 'stop', sourceHandle: 'error' },
1196+
],
1197+
loops: {},
1198+
parallels: {},
1199+
},
1200+
'stop'
1201+
)
1202+
1203+
await expect(engine.run('start')).rejects.toThrow('Stop block "stop" (stop) was not reached')
1204+
expect(executed).toEqual(['start', 'work'])
1205+
})
1206+
1207+
it('stops after the stop block when the condition routes to it', async () => {
1208+
const { engine, executed } = buildRun(conditionWorkflow, 'stop', {
1209+
condition: { selectedOption: 'else' },
1210+
})
1211+
1212+
const result = await engine.run('start')
1213+
1214+
expect(result.success).toBe(true)
1215+
expect(executed).toEqual(['start', 'condition', 'stop'])
1216+
})
1217+
1218+
it('runs a join reached through one taken and one skipped branch', async () => {
1219+
const { engine, executed } = buildRun(
1220+
{
1221+
version: '1',
1222+
blocks: [
1223+
block('start', BlockType.STARTER),
1224+
block('condition', BlockType.CONDITION),
1225+
block('taken'),
1226+
block('skipped'),
1227+
block('stop'),
1228+
block('after'),
1229+
],
1230+
connections: [
1231+
{ source: 'start', target: 'condition' },
1232+
{ source: 'condition', target: 'taken', sourceHandle: 'condition-if' },
1233+
{ source: 'condition', target: 'skipped', sourceHandle: 'condition-else' },
1234+
{ source: 'taken', target: 'stop' },
1235+
{ source: 'skipped', target: 'stop' },
1236+
{ source: 'stop', target: 'after' },
1237+
],
1238+
loops: {},
1239+
parallels: {},
1240+
},
1241+
'stop',
1242+
{ condition: { selectedOption: 'if' } }
1243+
)
1244+
1245+
const result = await engine.run('start')
1246+
1247+
expect(result.success).toBe(true)
1248+
expect(executed).toEqual(['start', 'condition', 'taken', 'stop'])
1249+
})
1250+
1251+
it('runs a loop stop block queued by a dead end inside the loop', async () => {
1252+
const sentinelStart = buildLoopSentinelStartId('loop')
1253+
const sentinelEnd = buildLoopSentinelEndId('loop')
1254+
const { engine, executed } = buildRun(
1255+
{
1256+
version: '1',
1257+
blocks: [
1258+
block('start', BlockType.STARTER),
1259+
block('loop', BlockType.LOOP),
1260+
block('condition', BlockType.CONDITION),
1261+
block('inner'),
1262+
block('after'),
1263+
],
1264+
connections: [
1265+
{ source: 'start', target: 'loop' },
1266+
{ source: 'loop', target: 'condition', sourceHandle: 'loop-start-source' },
1267+
{ source: 'condition', target: 'inner', sourceHandle: 'condition-if' },
1268+
{ source: 'loop', target: 'after', sourceHandle: 'loop-end-source' },
1269+
],
1270+
loops: { loop: { id: 'loop', nodes: ['condition', 'inner'], iterations: 1 } },
1271+
parallels: {},
1272+
},
1273+
sentinelEnd,
1274+
{ condition: { selectedOption: 'else' } }
1275+
)
1276+
1277+
const result = await engine.run('start')
1278+
1279+
expect(result.success).toBe(true)
1280+
expect(executed).toEqual(['start', sentinelStart, 'condition', sentinelEnd])
1281+
})
1282+
1283+
it('stops after a loop stop block whose loop has nothing to run', async () => {
1284+
const sentinelStart = buildLoopSentinelStartId('loop')
1285+
const { engine, executed } = buildRun(
1286+
{
1287+
version: '1',
1288+
blocks: [
1289+
block('start', BlockType.STARTER),
1290+
block('loop', BlockType.LOOP),
1291+
block('inner'),
1292+
block('after'),
1293+
],
1294+
connections: [
1295+
{ source: 'start', target: 'loop' },
1296+
{ source: 'loop', target: 'inner', sourceHandle: 'loop-start-source' },
1297+
{ source: 'loop', target: 'after', sourceHandle: 'loop-end-source' },
1298+
],
1299+
loops: { loop: { id: 'loop', nodes: ['inner'], iterations: 0 } },
1300+
parallels: {},
1301+
},
1302+
buildLoopSentinelEndId('loop'),
1303+
{
1304+
[sentinelStart]: { sentinelStart: true, shouldExit: true, selectedRoute: EDGE.LOOP_EXIT },
1305+
}
1306+
)
1307+
1308+
const result = await engine.run('start')
1309+
1310+
expect(result.success).toBe(true)
1311+
expect(executed).toEqual(['start', sentinelStart])
1312+
})
1313+
1314+
it('succeeds when the stop block is a Response block, and fails when one ends the run first', async () => {
1315+
const workflow: SerializedWorkflow = {
1316+
version: '1',
1317+
blocks: [
1318+
block('start', BlockType.STARTER),
1319+
block('respond', BlockType.RESPONSE),
1320+
block('stop'),
1321+
],
1322+
connections: [
1323+
{ source: 'start', target: 'respond' },
1324+
{ source: 'respond', target: 'stop' },
1325+
],
1326+
loops: {},
1327+
parallels: {},
1328+
}
1329+
1330+
const atResponse = buildRun(workflow, 'respond')
1331+
expect((await atResponse.engine.run('start')).success).toBe(true)
1332+
1333+
const pastResponse = buildRun(workflow, 'stop')
1334+
await expect(pastResponse.engine.run('start')).rejects.toThrow(
1335+
'Stop block "stop" (stop) was not reached'
1336+
)
1337+
expect(pastResponse.executed).toEqual(['start', 'respond'])
1338+
})
1339+
1340+
it('stops after the entry block when the stop block is the entry', async () => {
1341+
const { engine, executed } = buildRun(conditionWorkflow, 'start')
1342+
1343+
const result = await engine.run('start')
1344+
1345+
expect(result.success).toBe(true)
1346+
expect(executed).toEqual(['start'])
1347+
})
1348+
1349+
it('fails when the stop block is not in the graph the run executes', async () => {
1350+
const { engine } = buildRun(conditionWorkflow, 'missing', {
1351+
condition: { selectedOption: 'if' },
1352+
})
1353+
1354+
await expect(engine.run('start')).rejects.toThrow('Stop block missing was not reached')
1355+
})
1356+
})
10911357
})

0 commit comments

Comments
 (0)