@@ -20,6 +20,7 @@ import {
2020} from '@/executor/handlers'
2121import { LoopManager } from '@/executor/loops/loops'
2222import { ParallelManager } from '@/executor/parallels/parallels'
23+ import { ParallelRoutingUtils } from '@/executor/parallels/utils'
2324import { PathTracker } from '@/executor/path/path'
2425import { InputResolver } from '@/executor/resolver/resolver'
2526import type {
@@ -31,6 +32,7 @@ import type {
3132 StreamingExecution ,
3233} from '@/executor/types'
3334import { streamingResponseFormatProcessor } from '@/executor/utils'
35+ import { VirtualBlockUtils } from '@/executor/utils/virtual-blocks'
3436import type { SerializedBlock , SerializedWorkflow } from '@/serializer/types'
3537import { useExecutionStore } from '@/stores/execution/store'
3638import { useConsoleStore } from '@/stores/panel/console/store'
@@ -1111,7 +1113,7 @@ export class Executor {
11111113 if ( parallelState ) {
11121114 let allVirtualInstancesExecuted = true
11131115 for ( let i = 0 ; i < parallelState . parallelCount ; i ++ ) {
1114- const virtualBlockId = ` ${ block . id } _parallel_ ${ insideParallel } _iteration_ ${ i } `
1116+ const virtualBlockId = VirtualBlockUtils . generateParallelId ( block . id , insideParallel , i )
11151117 if ( ! executedBlocks . has ( virtualBlockId ) ) {
11161118 allVirtualInstancesExecuted = false
11171119 break
@@ -1205,7 +1207,9 @@ export class Executor {
12051207 }
12061208
12071209 /**
1208- * Check if a specific parallel iteration is complete (all blocks executed).
1210+ * Check if a specific parallel iteration is complete (all blocks that should execute have executed).
1211+ * This method now considers conditional execution paths - only blocks in the active execution
1212+ * path are expected to execute.
12091213 *
12101214 * @param parallelId - ID of the parallel block
12111215 * @param iteration - Iteration index to check
@@ -1223,15 +1227,124 @@ export class Executor {
12231227 return true
12241228 }
12251229
1226- for ( const nodeId of parallel . nodes ) {
1227- const virtualBlockId = `${ nodeId } ${ parallelId } ${ iteration }
1230+ const expectedBlocks = this . getExpectedBlocksForIteration (
1231+ parallelId ,
1232+ iteration ,
1233+ parallel ,
1234+ context
1235+ )
1236+
1237+ // Check if all expected blocks have been executed
1238+ for ( const nodeId of expectedBlocks ) {
1239+ const virtualBlockId = VirtualBlockUtils . generateParallelId ( nodeId , parallelId , iteration )
12281240 if ( ! context . executedBlocks . has ( virtualBlockId ) ) {
12291241 return false
12301242 }
12311243 }
12321244 return true
12331245 }
12341246
1247+ /**
1248+ * Get the blocks that are expected to execute in a parallel iteration based on
1249+ * the active execution path. This handles conditional logic where some blocks
1250+ * may not execute due to condition or router blocks.
1251+ *
1252+ * @param parallelId - ID of the parallel block
1253+ * @param iteration - Iteration index
1254+ * @param parallel - Parallel configuration
1255+ * @param context - Execution context
1256+ * @returns Array of node IDs that should execute in this iteration
1257+ */
1258+ private getExpectedBlocksForIteration (
1259+ parallelId : string ,
1260+ iteration : number ,
1261+ parallel : any ,
1262+ context : ExecutionContext
1263+ ) : string [ ] {
1264+ if ( ! parallel || ! parallel . nodes ) {
1265+ return [ ]
1266+ }
1267+
1268+ const expectedBlocks : string [ ] = [ ]
1269+
1270+ for ( const nodeId of parallel . nodes ) {
1271+ const block = this . actualWorkflow . blocks . find ( ( b ) => b . id === nodeId )
1272+
1273+ // If block doesn't exist in workflow, fall back to original behavior (assume it should execute)
1274+ // This maintains compatibility with tests and edge cases
1275+ if ( ! block ) {
1276+ expectedBlocks . push ( nodeId )
1277+ continue
1278+ }
1279+
1280+ if ( ! block . enabled ) {
1281+ continue
1282+ }
1283+
1284+ const virtualBlockId = VirtualBlockUtils . generateParallelId ( nodeId , parallelId , iteration )
1285+
1286+ // Skip blocks that have already been executed
1287+ if ( context . executedBlocks . has ( virtualBlockId ) ) {
1288+ expectedBlocks . push ( nodeId )
1289+ continue
1290+ }
1291+
1292+ // Check if this block should execute based on the active execution path
1293+ // We need to check if the original block is reachable based on current routing decisions
1294+ try {
1295+ const shouldExecute = this . shouldBlockExecuteInParallelIteration (
1296+ nodeId ,
1297+ parallelId ,
1298+ iteration ,
1299+ context
1300+ )
1301+
1302+ if ( shouldExecute ) {
1303+ expectedBlocks . push ( nodeId )
1304+ }
1305+ } catch ( error ) {
1306+ // If path checking fails, default to including the block to maintain existing behavior
1307+ logger . warn (
1308+ `Path check failed for block ${ nodeId } ${ parallelId } ${ iteration } ,
1309+ error
1310+ )
1311+ expectedBlocks . push ( nodeId )
1312+ }
1313+ }
1314+
1315+ return expectedBlocks
1316+ }
1317+
1318+ /**
1319+ * Determines if a block should execute in a specific parallel iteration
1320+ * based on conditional routing and active execution paths.
1321+ *
1322+ * Blocks are excluded from execution if they are completely unconnected (no incoming connections).
1323+ * Starting blocks (with external connections only) and conditionally routed blocks execute as expected.
1324+ *
1325+ * @param nodeId - ID of the block to check
1326+ * @param parallelId - ID of the parallel block
1327+ * @param iteration - Current iteration index
1328+ * @param context - Execution context
1329+ * @returns Whether the block should execute
1330+ */
1331+ private shouldBlockExecuteInParallelIteration (
1332+ nodeId : string ,
1333+ parallelId : string ,
1334+ iteration : number ,
1335+ context : ExecutionContext
1336+ ) : boolean {
1337+ const parallel = this . actualWorkflow . parallels ?. [ parallelId ]
1338+ if ( ! parallel ) return false
1339+
1340+ return ParallelRoutingUtils . shouldBlockExecuteInParallelIteration (
1341+ nodeId ,
1342+ parallel ,
1343+ iteration ,
1344+ context
1345+ )
1346+ }
1347+
12351348 /**
12361349 * Check if there are more parallel iterations to process.
12371350 * This ensures the execution loop continues when iterations are being processed sequentially.
@@ -1269,6 +1382,8 @@ export class Executor {
12691382
12701383 /**
12711384 * Process a single parallel iteration with topological ordering of dependencies.
1385+ * Now includes conditional execution logic - only processes blocks that should execute
1386+ * based on the active execution path (handles conditions, routers, etc.).
12721387 *
12731388 * @param parallelId - ID of the parallel block
12741389 * @param iteration - Current iteration index
@@ -1293,9 +1408,9 @@ export class Executor {
12931408 }
12941409 > ( )
12951410
1296- // Build dependency graph for this iteration
1411+ // Build dependency graph for this iteration - only include blocks that should execute
12971412 for ( const nodeId of parallel . nodes ) {
1298- const virtualBlockId = ` ${ nodeId } _parallel_ ${ parallelId } _iteration_ ${ iteration } `
1413+ const virtualBlockId = VirtualBlockUtils . generateParallelId ( nodeId , parallelId , iteration )
12991414 const isExecuted = context . executedBlocks . has ( virtualBlockId )
13001415
13011416 if ( isExecuted ) {
@@ -1305,6 +1420,26 @@ export class Executor {
13051420 const block = this . actualWorkflow . blocks . find ( ( b ) => b . id === nodeId )
13061421 if ( ! block || ! block . enabled ) continue
13071422
1423+ // Check if this block should execute in this iteration based on conditional paths
1424+ try {
1425+ const shouldExecute = this . shouldBlockExecuteInParallelIteration (
1426+ nodeId ,
1427+ parallelId ,
1428+ iteration ,
1429+ context
1430+ )
1431+
1432+ if ( ! shouldExecute ) {
1433+ continue
1434+ }
1435+ } catch ( error ) {
1436+ // If path checking fails, default to processing the block to maintain existing behavior
1437+ logger . warn (
1438+ `Path check failed for block ${ nodeId } ${ parallelId } ${ iteration } ,
1439+ error
1440+ )
1441+ }
1442+
13081443 // Find dependencies within this iteration
13091444 const incomingConnections = this . actualWorkflow . connections . filter (
13101445 ( conn ) => conn . target === nodeId
@@ -1314,7 +1449,11 @@ export class Executor {
13141449 for ( const conn of incomingConnections ) {
13151450 // Check if the source is within the same parallel
13161451 if ( parallel . nodes . includes ( conn . source ) ) {
1317- const sourceDependencyId = `${ conn . source } ${ parallelId } ${ iteration }
1452+ const sourceDependencyId = VirtualBlockUtils . generateParallelId (
1453+ conn . source ,
1454+ parallelId ,
1455+ iteration
1456+ )
13181457 dependencies . push ( sourceDependencyId )
13191458 } else {
13201459 // External dependency - check if it's met
@@ -1422,7 +1561,11 @@ export class Executor {
14221561 sourceBlock &&
14231562 this . actualWorkflow . parallels ?. [ insideParallel ] ?. nodes . includes ( conn . source )
14241563 ) {
1425- sourceId = `${ conn . source } ${ insideParallel } ${ iterationIndex }
1564+ sourceId = VirtualBlockUtils . generateParallelId (
1565+ conn . source ,
1566+ insideParallel ,
1567+ iterationIndex
1568+ )
14261569 }
14271570 }
14281571
@@ -1769,6 +1912,21 @@ export class Executor {
17691912 )
17701913 }
17711914
1915+ // Store result for loops (IDENTICAL to parallel logic)
1916+ const containingLoopId = this . resolver . getContainingLoopId ( block . id )
1917+ if ( containingLoopId && ! parallelInfo ) {
1918+ // Only store for loops if not already in a parallel (avoid double storage)
1919+ const currentIteration = context . loopIterations . get ( containingLoopId )
1920+ if ( currentIteration !== undefined ) {
1921+ this . loopManager . storeIterationResult (
1922+ context ,
1923+ containingLoopId ,
1924+ currentIteration - 1 , // Convert to 0-based index
1925+ output
1926+ )
1927+ }
1928+ }
1929+
17721930 // Update the execution log
17731931 blockLog . success = true
17741932 blockLog . output = output
@@ -1886,6 +2044,21 @@ export class Executor {
18862044 )
18872045 }
18882046
2047+ // Store result for loops (IDENTICAL to parallel logic)
2048+ const containingLoopId = this . resolver . getContainingLoopId ( block . id )
2049+ if ( containingLoopId && ! parallelInfo ) {
2050+ // Only store for loops if not already in a parallel (avoid double storage)
2051+ const currentIteration = context . loopIterations . get ( containingLoopId )
2052+ if ( currentIteration !== undefined ) {
2053+ this . loopManager . storeIterationResult (
2054+ context ,
2055+ containingLoopId ,
2056+ currentIteration - 1 , // Convert to 0-based index
2057+ output
2058+ )
2059+ }
2060+ }
2061+
18892062 // Update the execution log
18902063 blockLog . success = true
18912064 blockLog . output = output
0 commit comments