improvement(logs): fixed logs for parallel and loop execution flow (#2468)

* fixed logs for parallel and loop execution flow

* Fix array check for collection

* fixed for empty loop and paralle blocks and showing input on dashboard

* extracted utility functions

* fixed the refrencing errors and making sure it propogates to the console

* fix parallel

* fix tests'

---------

Co-authored-by: priyanshu.solanki <priyanshu.solanki@saviynt.com>
Co-authored-by: Siddharth Ganesan <siddharthganesan@gmail.com>
Co-authored-by: Vikhyath Mondreti <vikhyath@simstudio.ai>

apps/sim/executor/orchestrators/loop.ts

@@ -5,14 +5,17 @@ import { buildLoopIndexCondition, DEFAULTS, EDGE } from '@/executor/constants'
 import type { DAG } from '@/executor/dag/builder'
 import type { EdgeManager } from '@/executor/execution/edge-manager'
 import type { LoopScope } from '@/executor/execution/state'
-import type { BlockStateController } from '@/executor/execution/types'
+import type { BlockStateController, ContextExtensions } from '@/executor/execution/types'
 import type { ExecutionContext, NormalizedBlockOutput } from '@/executor/types'
 import type { LoopConfigWithNodes } from '@/executor/types/loop'
 import { replaceValidReferences } from '@/executor/utils/reference-validation'
 import {
+  addSubflowErrorLog,
   buildSentinelEndId,
   buildSentinelStartId,
   extractBaseBlockId,
+  resolveArrayInput,
+  validateMaxCount,
 } from '@/executor/utils/subflow-utils'
 import type { VariableResolver } from '@/executor/variables/resolver'
 import type { SerializedLoop } from '@/serializer/types'
@@ -32,6 +35,7 @@ export interface LoopContinuationResult {
 
 export class LoopOrchestrator {
   private edgeManager: EdgeManager | null = null
+  private contextExtensions: ContextExtensions | null = null
 
   constructor(
     private dag: DAG,
@@ -39,6 +43,10 @@ export class LoopOrchestrator {
     private resolver: VariableResolver
   ) {}
 
+  setContextExtensions(contextExtensions: ContextExtensions): void {
+    this.contextExtensions = contextExtensions
+  }
+
   setEdgeManager(edgeManager: EdgeManager): void {
     this.edgeManager = edgeManager
   }
@@ -48,7 +56,6 @@ export class LoopOrchestrator {
     if (!loopConfig) {
       throw new Error(`Loop config not found: ${loopId}`)
     }
-
     const scope: LoopScope = {
       iteration: 0,
       currentIterationOutputs: new Map(),
@@ -58,15 +65,70 @@ export class LoopOrchestrator {
     const loopType = loopConfig.loopType
 
     switch (loopType) {
-      case 'for':
+      case 'for': {
         scope.loopType = 'for'
-        scope.maxIterations = loopConfig.iterations || DEFAULTS.MAX_LOOP_ITERATIONS
+        const requestedIterations = loopConfig.iterations || DEFAULTS.MAX_LOOP_ITERATIONS
+
+        const iterationError = validateMaxCount(
+          requestedIterations,
+          DEFAULTS.MAX_LOOP_ITERATIONS,
+          'For loop iterations'
+        )
+        if (iterationError) {
+          logger.error(iterationError, { loopId, requestedIterations })
+          this.addLoopErrorLog(ctx, loopId, loopType, iterationError, {
+            iterations: requestedIterations,
+          })
+          scope.maxIterations = 0
+          scope.validationError = iterationError
+          scope.condition = buildLoopIndexCondition(0)
+          ctx.loopExecutions?.set(loopId, scope)
+          throw new Error(iterationError)
+        }
+
+        scope.maxIterations = requestedIterations
         scope.condition = buildLoopIndexCondition(scope.maxIterations)
         break
+      }
 
       case 'forEach': {
         scope.loopType = 'forEach'
-        const items = this.resolveForEachItems(ctx, loopConfig.forEachItems)
+        let items: any[]
+        try {
+          items = this.resolveForEachItems(ctx, loopConfig.forEachItems)
+        } catch (error) {
+          const errorMessage = `ForEach loop resolution failed: ${error instanceof Error ? error.message : String(error)}`
+          logger.error(errorMessage, { loopId, forEachItems: loopConfig.forEachItems })
+          this.addLoopErrorLog(ctx, loopId, loopType, errorMessage, {
+            forEachItems: loopConfig.forEachItems,
+          })
+          scope.items = []
+          scope.maxIterations = 0
+          scope.validationError = errorMessage
+          scope.condition = buildLoopIndexCondition(0)
+          ctx.loopExecutions?.set(loopId, scope)
+          throw new Error(errorMessage)
+        }
+
+        const sizeError = validateMaxCount(
+          items.length,
+          DEFAULTS.MAX_FOREACH_ITEMS,
+          'ForEach loop collection size'
+        )
+        if (sizeError) {
+          logger.error(sizeError, { loopId, collectionSize: items.length })
+          this.addLoopErrorLog(ctx, loopId, loopType, sizeError, {
+            forEachItems: loopConfig.forEachItems,
+            collectionSize: items.length,
+          })
+          scope.items = []
+          scope.maxIterations = 0
+          scope.validationError = sizeError
+          scope.condition = buildLoopIndexCondition(0)
+          ctx.loopExecutions?.set(loopId, scope)
+          throw new Error(sizeError)
+        }
+
         scope.items = items
         scope.maxIterations = items.length
         scope.item = items[0]
@@ -79,15 +141,35 @@ export class LoopOrchestrator {
         scope.condition = loopConfig.whileCondition
         break
 
-      case 'doWhile':
+      case 'doWhile': {
         scope.loopType = 'doWhile'
         if (loopConfig.doWhileCondition) {
           scope.condition = loopConfig.doWhileCondition
         } else {
-          scope.maxIterations = loopConfig.iterations || DEFAULTS.MAX_LOOP_ITERATIONS
+          const requestedIterations = loopConfig.iterations || DEFAULTS.MAX_LOOP_ITERATIONS
+
+          const iterationError = validateMaxCount(
+            requestedIterations,
+            DEFAULTS.MAX_LOOP_ITERATIONS,
+            'Do-While loop iterations'
+          )
+          if (iterationError) {
+            logger.error(iterationError, { loopId, requestedIterations })
+            this.addLoopErrorLog(ctx, loopId, loopType, iterationError, {
+              iterations: requestedIterations,
+            })
+            scope.maxIterations = 0
+            scope.validationError = iterationError
+            scope.condition = buildLoopIndexCondition(0)
+            ctx.loopExecutions?.set(loopId, scope)
+            throw new Error(iterationError)
+          }
+
+          scope.maxIterations = requestedIterations
           scope.condition = buildLoopIndexCondition(scope.maxIterations)
         }
         break
+      }
 
       default:
         throw new Error(`Unknown loop type: ${loopType}`)
@@ -100,6 +182,23 @@ export class LoopOrchestrator {
     return scope
   }
 
+  private addLoopErrorLog(
+    ctx: ExecutionContext,
+    loopId: string,
+    loopType: string,
+    errorMessage: string,
+    inputData?: any
+  ): void {
+    addSubflowErrorLog(
+      ctx,
+      loopId,
+      'loop',
+      errorMessage,
+      { loopType, ...inputData },
+      this.contextExtensions
+    )
+  }
+
   storeLoopNodeOutput(
     ctx: ExecutionContext,
     loopId: string,
@@ -412,54 +511,6 @@ export class LoopOrchestrator {
   }
 
   private resolveForEachItems(ctx: ExecutionContext, items: any): any[] {
-    if (Array.isArray(items)) {
-      return items
-    }
-
-    if (typeof items === 'object' && items !== null) {
-      return Object.entries(items)
-    }
-
-    if (typeof items === 'string') {
-      if (items.startsWith('<') && items.endsWith('>')) {
-        const resolved = this.resolver.resolveSingleReference(ctx, '', items)
-        if (Array.isArray(resolved)) {
-          return resolved
-        }
-        return []
-      }
-
-      try {
-        const normalized = items.replace(/'/g, '"')
-        const parsed = JSON.parse(normalized)
-        if (Array.isArray(parsed)) {
-          return parsed
-        }
-        return []
-      } catch (error) {
-        logger.error('Failed to parse forEach items', { items, error })
-        return []
-      }
-    }
-
-    try {
-      const resolved = this.resolver.resolveInputs(ctx, 'loop_foreach_items', { items }).items
-
-      if (Array.isArray(resolved)) {
-        return resolved
-      }
-
-      logger.warn('ForEach items did not resolve to array', {
-        items,
-        resolved,
-      })
-
-      return []
-    } catch (error: any) {
-      logger.error('Error resolving forEach items, returning empty array:', {
-        error: error.message,
-      })
-      return []
-    }
+    return resolveArrayInput(ctx, items, this.resolver)
   }
 }

apps/sim/executor/orchestrators/parallel.ts

@@ -1,15 +1,19 @@
 import { createLogger } from '@/lib/logs/console/logger'
+import { DEFAULTS } from '@/executor/constants'
 import type { DAG, DAGNode } from '@/executor/dag/builder'
 import type { ParallelScope } from '@/executor/execution/state'
-import type { BlockStateWriter } from '@/executor/execution/types'
+import type { BlockStateWriter, ContextExtensions } from '@/executor/execution/types'
 import type { ExecutionContext, NormalizedBlockOutput } from '@/executor/types'
 import type { ParallelConfigWithNodes } from '@/executor/types/parallel'
 import {
+  addSubflowErrorLog,
   buildBranchNodeId,
   calculateBranchCount,
   extractBaseBlockId,
   extractBranchIndex,
   parseDistributionItems,
+  resolveArrayInput,
+  validateMaxCount,
 } from '@/executor/utils/subflow-utils'
 import type { VariableResolver } from '@/executor/variables/resolver'
 import type { SerializedParallel } from '@/serializer/types'
@@ -32,6 +36,7 @@ export interface ParallelAggregationResult {
 
 export class ParallelOrchestrator {
   private resolver: VariableResolver | null = null
+  private contextExtensions: ContextExtensions | null = null
 
   constructor(
     private dag: DAG,
@@ -42,6 +47,10 @@ export class ParallelOrchestrator {
     this.resolver = resolver
   }
 
+  setContextExtensions(contextExtensions: ContextExtensions): void {
+    this.contextExtensions = contextExtensions
+  }
+
   initializeParallelScope(
     ctx: ExecutionContext,
     parallelId: string,
@@ -49,11 +58,42 @@ export class ParallelOrchestrator {
     terminalNodesCount = 1
   ): ParallelScope {
     const parallelConfig = this.dag.parallelConfigs.get(parallelId)
-    const items = parallelConfig ? this.resolveDistributionItems(ctx, parallelConfig) : undefined
 
-    // If we have more items than pre-built branches, expand the DAG
+    let items: any[] | undefined
+    if (parallelConfig) {
+      try {
+        items = this.resolveDistributionItems(ctx, parallelConfig)
+      } catch (error) {
+        const errorMessage = `Parallel Items did not resolve: ${error instanceof Error ? error.message : String(error)}`
+        logger.error(errorMessage, {
+          parallelId,
+          distribution: parallelConfig.distribution,
+        })
+        this.addParallelErrorLog(ctx, parallelId, errorMessage, {
+          distribution: parallelConfig.distribution,
+        })
+        this.setErrorScope(ctx, parallelId, errorMessage)
+        throw new Error(errorMessage)
+      }
+    }
+
     const actualBranchCount = items && items.length > totalBranches ? items.length : totalBranches
 
+    const branchError = validateMaxCount(
+      actualBranchCount,
+      DEFAULTS.MAX_PARALLEL_BRANCHES,
+      'Parallel branch count'
+    )
+    if (branchError) {
+      logger.error(branchError, { parallelId, actualBranchCount })
+      this.addParallelErrorLog(ctx, parallelId, branchError, {
+        distribution: parallelConfig?.distribution,
+        branchCount: actualBranchCount,
+      })
+      this.setErrorScope(ctx, parallelId, branchError)
+      throw new Error(branchError)
+    }
+
     const scope: ParallelScope = {
       parallelId,
       totalBranches: actualBranchCount,
@@ -108,6 +148,38 @@ export class ParallelOrchestrator {
     return scope
   }
 
+  private addParallelErrorLog(
+    ctx: ExecutionContext,
+    parallelId: string,
+    errorMessage: string,
+    inputData?: any
+  ): void {
+    addSubflowErrorLog(
+      ctx,
+      parallelId,
+      'parallel',
+      errorMessage,
+      inputData || {},
+      this.contextExtensions
+    )
+  }
+
+  private setErrorScope(ctx: ExecutionContext, parallelId: string, errorMessage: string): void {
+    const scope: ParallelScope = {
+      parallelId,
+      totalBranches: 0,
+      branchOutputs: new Map(),
+      completedCount: 0,
+      totalExpectedNodes: 0,
+      items: [],
+      validationError: errorMessage,
+    }
+    if (!ctx.parallelExecutions) {
+      ctx.parallelExecutions = new Map()
+    }
+    ctx.parallelExecutions.set(parallelId, scope)
+  }
+
   /**
    * Dynamically expand the DAG to include additional branch nodes when
    * the resolved item count exceeds the pre-built branch count.
@@ -291,63 +363,19 @@ export class ParallelOrchestrator {
     }
   }
 
-  /**
-   * Resolve distribution items at runtime, handling references like <previousBlock.items>
-   * This mirrors how LoopOrchestrator.resolveForEachItems works.
-   */
   private resolveDistributionItems(ctx: ExecutionContext, config: SerializedParallel): any[] {
-    const rawItems = config.distribution
-
-    if (rawItems === undefined || rawItems === null) {
+    if (config.parallelType === 'count') {
       return []
     }
 
-    // Already an array - return as-is
-    if (Array.isArray(rawItems)) {
-      return rawItems
+    if (
+      config.distribution === undefined ||
+      config.distribution === null ||
+      config.distribution === ''
+    ) {
+      return []
     }
-
-    // Object - convert to entries array (consistent with loop forEach behavior)
-    if (typeof rawItems === 'object') {
-      return Object.entries(rawItems)
-    }
-
-    // String handling
-    if (typeof rawItems === 'string') {
-      // Resolve references at runtime using the variable resolver
-      if (rawItems.startsWith('<') && rawItems.endsWith('>') && this.resolver) {
-        const resolved = this.resolver.resolveSingleReference(ctx, '', rawItems)
-        if (Array.isArray(resolved)) {
-          return resolved
-        }
-        if (typeof resolved === 'object' && resolved !== null) {
-          return Object.entries(resolved)
-        }
-        logger.warn('Distribution reference did not resolve to array or object', {
-          rawItems,
-          resolved,
-        })
-        return []
-      }
-
-      // Try to parse as JSON
-      try {
-        const normalized = rawItems.replace(/'/g, '"')
-        const parsed = JSON.parse(normalized)
-        if (Array.isArray(parsed)) {
-          return parsed
-        }
-        if (typeof parsed === 'object' && parsed !== null) {
-          return Object.entries(parsed)
-        }
-        return []
-      } catch (error) {
-        logger.error('Failed to parse distribution items', { rawItems, error })
-        return []
-      }
-    }
-
-    return []
+    return resolveArrayInput(ctx, config.distribution, this.resolver)
   }
 
   handleParallelBranchCompletion(