feat(loop): added loop block & parallel block logic (#411)

* feat(subflows): implement parallel and loop nodes

* fixed linter errors

* added loop block logic

* added parallel executioon

* added input resolution for collections for parallel & loop blocks

* refactored tests, cleaned up some unnecessary logic

* prevent self connections now that we have loop blocks

* refactored path tracker

* lint

* acknowledged PR comments, added tests

* added tests for loops

---------

Co-authored-by: Emir Karabeg <emirkarabeg@berkeley.edu>

Author: waleedlatif1

.prettierignore

@@ -335,8 +335,8 @@ export async function executeWorkflowForChat(
   }
 
   // Use deployed state for execution
-  const state = (workflowResult[0].deployedState || workflowResult[0].state) as WorkflowState
-  const { blocks, edges, loops } = state
+  const state = workflowResult[0].deployedState || workflowResult[0].state
+  const { blocks, edges, loops, parallels } = state as WorkflowState
 
   // Prepare for execution, similar to use-workflow-execution.ts
   const mergedStates = mergeSubblockState(blocks)
@@ -386,7 +386,12 @@ export async function executeWorkflowForChat(
   }
 
   // Create serialized workflow
-  const serializedWorkflow = new Serializer().serializeWorkflow(mergedStates, edges, loops)
+  const serializedWorkflow = new Serializer().serializeWorkflow(
+    mergedStates,
+    edges,
+    loops,
+    parallels
+  )
 
   // Decrypt environment variables
   const decryptedEnvVars: Record<string, string> = {}

apps/sim/app/api/chat/utils.ts

@@ -150,7 +150,7 @@ export async function GET(req: NextRequest) {
         }
 
         const state = workflowRecord.state as WorkflowState
-        const { blocks, edges, loops } = state
+        const { blocks, edges, loops, parallels } = state
 
         const mergedStates = mergeSubblockState(blocks)
 
@@ -224,7 +224,12 @@ export async function GET(req: NextRequest) {
           }
         }
 
-        const serializedWorkflow = new Serializer().serializeWorkflow(mergedStates, edges, loops)
+        const serializedWorkflow = new Serializer().serializeWorkflow(
+          mergedStates,
+          edges,
+          loops,
+          parallels
+        )
 
         const input = {
           workflowId: schedule.workflowId,

apps/sim/app/api/schedules/execute/route.ts

@@ -97,7 +97,7 @@ async function executeWorkflow(workflow: any, requestId: string, input?: any) {
     }
 
     const state = workflowState as WorkflowState
-    const { blocks, edges, loops } = state
+    const { blocks, edges, loops, parallels } = state
 
     // Use the same execution flow as in scheduled executions
     const mergedStates = mergeSubblockState(blocks)
@@ -227,7 +227,12 @@ async function executeWorkflow(workflow: any, requestId: string, input?: any) {
 
     // Serialize and execute the workflow
     logger.debug(`[${requestId}] Serializing workflow: ${workflowId}`)
-    const serializedWorkflow = new Serializer().serializeWorkflow(mergedStates, edges, loops)
+    const serializedWorkflow = new Serializer().serializeWorkflow(
+      mergedStates,
+      edges,
+      loops,
+      parallels
+    )
 
     const executor = new Executor(
       serializedWorkflow,

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -21,7 +21,8 @@ const MarketplaceDataSchema = z
 const WorkflowStateSchema = z.object({
   blocks: z.record(z.any()),
   edges: z.array(z.any()),
-  loops: z.record(z.any()),
+  loops: z.record(z.any()).default({}),
+  parallels: z.record(z.any()).default({}),
   lastSaved: z.number().optional(),
   isDeployed: z.boolean().optional(),
   deployedAt: z

apps/sim/app/api/workflows/sync/route.ts

@@ -8,7 +8,12 @@
 }
 
 .workflow-container .react-flow__node {
-  z-index: 20 !important;
+  z-index: 21 !important;
+}
+
+.workflow-container .react-flow__node-loopNode,
+.workflow-container .react-flow__node-parallelNode {
+  z-index: -1 !important;
 }
 
 .workflow-container .react-flow__handle {

apps/sim/app/globals.css

@@ -26,6 +26,7 @@ interface DeployedWorkflowModalProps {
     blocks: Record<string, any>
     edges: Array<any>
     loops: Record<string, any>
+    parallels: Record<string, any>
   }
 }
 
@@ -43,9 +44,11 @@ export function DeployedWorkflowModal({
     blocks: activeWorkflowId ? mergeSubblockState(state.blocks, activeWorkflowId) : state.blocks,
     edges: state.edges,
     loops: state.loops,
+    parallels: state.parallels,
   }))
 
   const handleRevert = () => {
+    // Revert to the deployed state
     revertToDeployedState(deployedWorkflowState)
     setShowRevertDialog(false)
     onClose()

apps/sim/app/w/[id]/components/control-bar/components/deployment-controls/components/deployed-workflow-modal.tsx

@@ -0,0 +1,57 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { useWorkflowStore } from '@/stores/workflows/workflow/store'
+
+// Mock the store
+vi.mock('@/stores/workflows/workflow/store', () => ({
+  useWorkflowStore: vi.fn(),
+}))
+
+describe('LoopBadges Store Integration', () => {
+  const mockUpdateLoopType = vi.fn()
+  const mockUpdateLoopCount = vi.fn()
+  const mockUpdateLoopCollection = vi.fn()
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+
+    ;(useWorkflowStore as any).mockImplementation((selector: any) => {
+      const state = {
+        updateLoopType: mockUpdateLoopType,
+        updateLoopCount: mockUpdateLoopCount,
+        updateLoopCollection: mockUpdateLoopCollection,
+      }
+      return selector(state)
+    })
+  })
+
+  it('should call updateLoopType when changing loop type', () => {
+    // When we update loop type in the UI, it should call the store method
+    const nodeId = 'loop1'
+    const newType = 'forEach'
+
+    // Simulate the handler being called
+    mockUpdateLoopType(nodeId, newType)
+
+    expect(mockUpdateLoopType).toHaveBeenCalledWith(nodeId, newType)
+  })
+
+  it('should call updateLoopCount when changing loop count', () => {
+    const nodeId = 'loop1'
+    const newCount = 15
+
+    // Simulate the handler being called
+    mockUpdateLoopCount(nodeId, newCount)
+
+    expect(mockUpdateLoopCount).toHaveBeenCalledWith(nodeId, newCount)
+  })
+
+  it('should call updateLoopCollection when changing collection', () => {
+    const nodeId = 'loop1'
+    const newCollection = '["item1", "item2", "item3"]'
+
+    // Simulate the handler being called
+    mockUpdateLoopCollection(nodeId, newCollection)
+
+    expect(mockUpdateLoopCollection).toHaveBeenCalledWith(nodeId, newCollection)
+  })
+})