Skip to content

Commit f0c083c

Browse files
authored
improvement(executor): make run-from-block refusals actionable (#9636)
* improvement(executor): make run-from-block refusals actionable * fix(executor): name the outermost container and catch parallel children by workflow id * chore(executor): type subflow configs in new run-from-block tests
1 parent 4dd8a14 commit f0c083c

6 files changed

Lines changed: 330 additions & 38 deletions

File tree

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

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -197,7 +197,10 @@ describe('DAGExecutor run-from-block snapshot metadata', () => {
197197
version: '1',
198198
blocks: [
199199
createBlock('start', BlockType.STARTER),
200-
createBlock('producer', BlockType.FUNCTION),
200+
{
201+
...createBlock('producer', BlockType.FUNCTION),
202+
metadata: { id: 'function', name: 'Fetch' },
203+
},
201204
createBlock('consumer', BlockType.FUNCTION),
202205
],
203206
connections: [
@@ -220,7 +223,12 @@ describe('DAGExecutor run-from-block snapshot metadata', () => {
220223
const error = await executor.executeFromBlock('wf', 'consumer', sourceSnapshot).catch((e) => e)
221224

222225
expect(error).toBeInstanceOf(RunFromBlockValidationError)
223-
expect(error.message).toBe('Upstream dependency not executed: producer')
226+
expect(error.message).toBe('Upstream dependency not executed: Fetch')
227+
expect(error.refusal).toEqual({
228+
kind: 'upstream_not_executed',
229+
blockId: 'consumer',
230+
upstreamBlockId: 'producer',
231+
})
224232
})
225233
})
226234

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

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -153,10 +153,15 @@ export class DAGExecutor {
153153
const dag = this.dagBuilder.build(this.workflow, { includeAllBlocks: true })
154154

155155
const executedBlocks = new Set(sourceSnapshot.executedBlocks)
156-
const validation = validateRunFromBlock(startBlockId, dag, executedBlocks)
156+
const blockNames = new Map<string, string>()
157+
for (const block of this.workflow.blocks) {
158+
if (block.metadata?.name) blockNames.set(block.id, block.metadata.name)
159+
}
160+
const validation = validateRunFromBlock(startBlockId, dag, executedBlocks, blockNames)
157161
if (!validation.valid) {
158162
throw new RunFromBlockValidationError(
159-
validation.error ?? `Cannot run from block: ${startBlockId}`
163+
validation.error ?? `Cannot run from block: ${startBlockId}`,
164+
validation.refusal
160165
)
161166
}
162167

‎apps/sim/executor/utils/run-from-block.test.ts‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -333,6 +333,53 @@ describe('validateRunFromBlock', () => {
333333
expect(result.error).toContain(outerParallelId)
334334
})
335335

336+
it('names the outermost container for a block in a nested loop, since the inner loop is refused too', () => {
337+
const dag = createDAG([
338+
createNode('B', [], { isLoopNode: true, subflowId: 'inner-loop', subflowType: 'loop' }),
339+
])
340+
dag.loopConfigs.set('outer-loop', {
341+
id: 'outer-loop',
342+
nodes: ['inner-loop'],
343+
iterations: 2,
344+
loopType: 'for',
345+
})
346+
dag.loopConfigs.set('inner-loop', {
347+
id: 'inner-loop',
348+
nodes: ['B'],
349+
iterations: 2,
350+
loopType: 'for',
351+
})
352+
353+
const result = validateRunFromBlock('B', dag, new Set())
354+
355+
expect(result.valid).toBe(false)
356+
expect(result.refusal).toEqual({
357+
kind: 'inside_subflow',
358+
blockId: 'B',
359+
subflowId: 'outer-loop',
360+
subflowType: 'loop',
361+
})
362+
})
363+
364+
it('refuses a parallel child by its workflow id as inside the parallel, not as missing', () => {
365+
// The DAG holds parallel children only as branch clones, never under the workflow id.
366+
const dag = createDAG([
367+
createNode('P₍0₎', [], { isParallelBranch: true, subflowId: 'par', subflowType: 'parallel' }),
368+
])
369+
dag.parallelConfigs.set('par', { id: 'par', nodes: ['P'], count: 2 })
370+
371+
const result = validateRunFromBlock('P', dag, new Set(), new Map([['par', 'Fan out']]))
372+
373+
expect(result.valid).toBe(false)
374+
expect(result.error).toContain('inside parallel: Fan out')
375+
expect(result.refusal).toEqual({
376+
kind: 'inside_subflow',
377+
blockId: 'P',
378+
subflowId: 'par',
379+
subflowType: 'parallel',
380+
})
381+
})
382+
336383
it('rejects container when sentinel-start upstream dependency was not executed', () => {
337384
const loopId = 'loop-container-1'
338385
const sentinelStartId = `loop-${loopId}-sentinel-start`

‎apps/sim/executor/utils/run-from-block.ts‎

Lines changed: 77 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -29,18 +29,37 @@ export function resolveContainerToSentinelStart(blockId: string, dag: DAG): stri
2929
export interface RunFromBlockValidation {
3030
valid: boolean
3131
error?: string
32+
/** Why the start was refused, by id, for callers that phrase their own guidance. */
33+
refusal?: RunFromBlockRefusal
3234
}
3335

36+
/**
37+
* The structured reason behind a refused run-from-block start. `blockId` is always the
38+
* requested start block.
39+
*/
40+
export type RunFromBlockRefusal =
41+
| { kind: 'block_not_found'; blockId: string }
42+
| {
43+
kind: 'inside_subflow'
44+
blockId: string
45+
subflowId: string
46+
subflowType: 'loop' | 'parallel'
47+
}
48+
| { kind: 'upstream_not_executed'; blockId: string; upstreamBlockId: string }
49+
3450
/**
3551
* A run-from-block start the executor refused before running anything: the block is
3652
* missing, sits inside a loop or parallel, or has an upstream dependency the source
3753
* snapshot never executed. Its own class so the application layer can hand the reason
3854
* to the caller as a validation failure instead of the generic system-error fallback.
3955
*/
4056
export class RunFromBlockValidationError extends Error {
41-
constructor(message: string) {
57+
readonly refusal?: RunFromBlockRefusal
58+
59+
constructor(message: string, refusal?: RunFromBlockRefusal) {
4260
super(message)
4361
this.name = 'RunFromBlockValidationError'
62+
this.refusal = refusal
4463
}
4564
}
4665

@@ -158,40 +177,48 @@ export function computeExecutionSets(dag: DAG, startBlockId: string): ExecutionS
158177
* @param blockId - The block ID to validate
159178
* @param dag - The workflow DAG
160179
* @param executedBlocks - Set of blocks that were executed in the source run
180+
* @param blockNames - Display names by block id, so the message names blocks instead of ids
161181
* @returns Validation result with error message if invalid
162182
*/
163183
export function validateRunFromBlock(
164184
blockId: string,
165185
dag: DAG,
166-
executedBlocks: Set<string>
186+
executedBlocks: Set<string>,
187+
blockNames: ReadonlyMap<string, string> = new Map()
167188
): RunFromBlockValidation {
189+
const label = (id: string) => blockNames.get(id) ?? id
190+
const insideSubflow = (
191+
subflowId: string | undefined,
192+
subflowType: 'loop' | 'parallel'
193+
): RunFromBlockValidation => {
194+
if (!subflowId) {
195+
return { valid: false, error: `Cannot run from block inside ${subflowType}: ${blockId}` }
196+
}
197+
return {
198+
valid: false,
199+
error: `Cannot run from block inside ${subflowType}: ${label(subflowId)}. Run the ${subflowType} or the whole workflow instead.`,
200+
refusal: { kind: 'inside_subflow', blockId, subflowId, subflowType },
201+
}
202+
}
203+
204+
const container = findOutermostContainer(blockId, dag)
205+
if (container) return insideSubflow(container.id, container.type)
206+
168207
const node = dag.nodes.get(blockId)
169208
const isLoopContainer = dag.loopConfigs.has(blockId)
170209
const isParallelContainer = dag.parallelConfigs.has(blockId)
171210
const isContainer = isLoopContainer || isParallelContainer
172211
let validationNode = node
173212

174213
if (!node && !isContainer) {
175-
return { valid: false, error: `Block not found in workflow: ${blockId}` }
214+
return {
215+
valid: false,
216+
error: `Block not found in workflow: ${blockId}`,
217+
refusal: { kind: 'block_not_found', blockId },
218+
}
176219
}
177220

178221
if (isContainer) {
179-
const parentLoopId = findParentLoop(blockId, dag)
180-
if (parentLoopId) {
181-
return {
182-
valid: false,
183-
error: `Cannot run from block inside loop: ${parentLoopId}`,
184-
}
185-
}
186-
187-
const parentParallelId = findParentParallel(blockId, dag)
188-
if (parentParallelId) {
189-
return {
190-
valid: false,
191-
error: `Cannot run from block inside parallel: ${parentParallelId}`,
192-
}
193-
}
194-
195222
const sentinelStartId = resolveContainerToSentinelStart(blockId, dag)
196223
if (!sentinelStartId || !dag.nodes.has(sentinelStartId)) {
197224
return {
@@ -203,19 +230,8 @@ export function validateRunFromBlock(
203230
}
204231

205232
if (node) {
206-
if (node.metadata.isLoopNode) {
207-
return {
208-
valid: false,
209-
error: `Cannot run from block inside loop: ${node.metadata.subflowId}`,
210-
}
211-
}
212-
213-
if (node.metadata.isParallelBranch) {
214-
return {
215-
valid: false,
216-
error: `Cannot run from block inside parallel: ${node.metadata.subflowId}`,
217-
}
218-
}
233+
if (node.metadata.isLoopNode) return insideSubflow(node.metadata.subflowId, 'loop')
234+
if (node.metadata.isParallelBranch) return insideSubflow(node.metadata.subflowId, 'parallel')
219235

220236
if (node.metadata.isSentinel) {
221237
return { valid: false, error: 'Cannot run from sentinel node' }
@@ -236,7 +252,8 @@ export function validateRunFromBlock(
236252
if (!executedBlocks.has(sourceId)) {
237253
return {
238254
valid: false,
239-
error: `Upstream dependency not executed: ${sourceId}`,
255+
error: `Upstream dependency not executed: ${label(sourceId)}`,
256+
refusal: { kind: 'upstream_not_executed', blockId, upstreamBlockId: sourceId },
240257
}
241258
}
242259
}
@@ -245,6 +262,33 @@ export function validateRunFromBlock(
245262
return { valid: true }
246263
}
247264

265+
/**
266+
* The top-level loop or parallel a block sits in, however deeply nested. A nested container
267+
* is refused as a start too, so naming the immediate parent would point at another refusal.
268+
* Parallel children are matched by their workflow id, which the DAG only holds as branch clones.
269+
*/
270+
function findOutermostContainer(
271+
blockId: string,
272+
dag: DAG
273+
): { id: string; type: 'loop' | 'parallel' } | undefined {
274+
let outermost: { id: string; type: 'loop' | 'parallel' } | undefined
275+
const visited = new Set([blockId])
276+
let current = blockId
277+
while (true) {
278+
const loopId = findParentLoop(current, dag)
279+
const parallelId = loopId ? undefined : findParentParallel(current, dag)
280+
const parent = loopId
281+
? { id: loopId, type: 'loop' as const }
282+
: parallelId
283+
? { id: parallelId, type: 'parallel' as const }
284+
: undefined
285+
if (!parent || visited.has(parent.id)) return outermost
286+
visited.add(parent.id)
287+
outermost = parent
288+
current = parent.id
289+
}
290+
}
291+
248292
function findParentLoop(blockId: string, dag: DAG): string | undefined {
249293
for (const [loopId, config] of dag.loopConfigs) {
250294
if (loopId !== blockId && config.nodes?.includes(blockId)) {

0 commit comments

Comments
 (0)