address more comments

Author: icecrasher321

apps/sim/executor/orchestrators/parallel.test.ts

@@ -8,6 +8,14 @@ import { ParallelOrchestrator } from '@/executor/orchestrators/parallel'
 import type { ExecutionContext } from '@/executor/types'
 import { buildBranchNodeId } from '@/executor/utils/subflow-utils'
 
+const { mockCompactSubflowResults } = vi.hoisted(() => ({
+  mockCompactSubflowResults: vi.fn(async (results: unknown) => results),
+}))
+
+vi.mock('@/lib/execution/payloads/serializer', () => ({
+  compactSubflowResults: mockCompactSubflowResults,
+}))
+
 function createDag(): DAG {
   return {
     nodes: new Map(),
@@ -76,6 +84,7 @@ function createContext(overrides: Partial<ExecutionContext> = {}): ExecutionCont
 describe('ParallelOrchestrator', () => {
   beforeEach(() => {
     vi.clearAllMocks()
+    mockCompactSubflowResults.mockImplementation(async (results: unknown) => results)
   })
 
   it('awaits empty-subflow lifecycle callbacks before returning the empty scope', async () => {
@@ -244,4 +253,57 @@ describe('ParallelOrchestrator', () => {
     expect(state.unmarkExecuted).toHaveBeenCalledWith(incomingBranchId)
     expect(state.unmarkExecuted).not.toHaveBeenCalledWith(previousBranchId)
   })
+
+  it('compacts accumulated outputs before scheduling later batches', async () => {
+    const dag = createDag()
+    const templateBranchId = buildBranchNodeId('task-1', 0)
+    dag.nodes.set(templateBranchId, {
+      id: templateBranchId,
+      block: {
+        id: 'task-1',
+        position: { x: 0, y: 0 },
+        config: { tool: '', params: {} },
+        inputs: {},
+        outputs: {},
+        metadata: { id: 'function', name: 'Task 1' },
+        enabled: true,
+      },
+      incomingEdges: new Set(),
+      outgoingEdges: new Set(),
+      metadata: { parallelId: 'parallel-1', isParallelBranch: true, branchIndex: 0 },
+    })
+    const orchestrator = new ParallelOrchestrator(dag, createState(), null, {})
+    const previousOutputs = [{ output: 'previous' }]
+    const incomingOutputs = [{ output: 'incoming' }]
+    const compactedPrevious = [{ output: 'compacted-previous' }]
+    const compactedIncoming = [{ output: 'compacted-incoming' }]
+    mockCompactSubflowResults.mockResolvedValueOnce([compactedPrevious, compactedIncoming])
+    const scope = {
+      parallelId: 'parallel-1',
+      totalBranches: 3,
+      batchSize: 1,
+      currentBatchStart: 0,
+      currentBatchSize: 2,
+      accumulatedOutputs: new Map([[0, previousOutputs]]),
+      branchOutputs: new Map([[1, incomingOutputs]]),
+    }
+    const ctx = createContext({
+      parallelExecutions: new Map([['parallel-1', scope]]),
+    })
+
+    const result = await orchestrator.aggregateParallelResults(ctx, 'parallel-1')
+
+    expect(result).toMatchObject({ allBranchesComplete: false, completedBranches: 2 })
+    expect(mockCompactSubflowResults).toHaveBeenCalledWith(
+      [previousOutputs, incomingOutputs],
+      expect.objectContaining({
+        workspaceId: 'workspace-1',
+        workflowId: 'workflow-1',
+        executionId: 'execution-1',
+        requireDurable: true,
+      })
+    )
+    expect(scope.accumulatedOutputs.get(0)).toBe(compactedPrevious)
+    expect(scope.accumulatedOutputs.get(1)).toBe(compactedIncoming)
+  })
 })

apps/sim/executor/orchestrators/parallel.ts

@@ -333,6 +333,30 @@ export class ParallelOrchestrator {
     const nextBatchStart =
       (scope.currentBatchStart ?? 0) + (scope.currentBatchSize ?? scope.totalBranches)
     if (nextBatchStart < scope.totalBranches) {
+      /**
+       * Compact accumulated outputs before scheduling the next batch. Each
+       * block output is already individually compacted by `block-executor`, but
+       * many below-threshold branch results can still exceed the aggregate
+       * threshold over time. Re-running the existing subflow compactor over the
+       * accumulated entries forces aggregate-size spills while existing
+       * LargeValueRefs stay stable.
+       */
+      if (accumulatedOutputs.size > 0) {
+        const accumulatedBranchIndexes = Array.from(accumulatedOutputs.keys()).sort((a, b) => a - b)
+        const accumulatedResults = accumulatedBranchIndexes.map(
+          (idx) => accumulatedOutputs.get(idx) ?? []
+        )
+        const compactedAccumulated = await compactSubflowResults(accumulatedResults, {
+          workspaceId: ctx.workspaceId,
+          workflowId: ctx.workflowId,
+          executionId: ctx.executionId,
+          userId: ctx.userId,
+          requireDurable: true,
+        })
+        accumulatedBranchIndexes.forEach((branchIdx, position) => {
+          accumulatedOutputs.set(branchIdx, compactedAccumulated[position])
+        })
+      }
       await this.scheduleNextBatch(ctx, scope, nextBatchStart)
       return {
         allBranchesComplete: false,

apps/sim/lib/execution/isolated-vm-worker.cjs

@@ -27,6 +27,21 @@ const SANDBOX_BUNDLE_FILES = {
 const bundleSourceCache = new Map()
 const activeIsolates = new Map()
 
+/**
+ * Sends an IPC request and reports only actual delivery failures.
+ * Node queues messages under backpressure, so the boolean return value is not
+ * a failure signal.
+ */
+function sendIpcRequest(message, onError) {
+  try {
+    process.send(message, (err) => {
+      if (err) onError(err)
+    })
+  } catch (error) {
+    onError(error instanceof Error ? error : new Error(String(error)))
+  }
+}
+
 function getBundleSource(bundleName) {
   const cached = bundleSourceCache.get(bundleName)
   if (cached) return cached
@@ -233,13 +248,19 @@ async function executeCode(request, executionId) {
           }
         }, FETCH_TIMEOUT_MS)
         pendingFetches.set(fetchId, { resolve, timeout })
-        if (process.send && process.connected) {
-          process.send({ type: 'fetch', fetchId, requestId, url, optionsJson })
-        } else {
+        if (!process.send || !process.connected) {
           clearTimeout(timeout)
           pendingFetches.delete(fetchId)
           resolve(JSON.stringify({ error: 'Parent process disconnected' }))
+          return
         }
+        sendIpcRequest({ type: 'fetch', fetchId, requestId, url, optionsJson }, (err) => {
+          const pending = pendingFetches.get(fetchId)
+          if (!pending) return
+          clearTimeout(pending.timeout)
+          pendingFetches.delete(fetchId)
+          pending.resolve(JSON.stringify({ error: `Fetch IPC send failed: ${err.message}` }))
+        })
       })
     })
     await jail.set('__fetchRef', fetchCallback)
@@ -254,13 +275,19 @@ async function executeCode(request, executionId) {
           }
         }, BROKER_TIMEOUT_MS)
         pendingBrokerCalls.set(brokerId, { resolve, timeout, executionId })
-        if (process.send && process.connected) {
-          process.send({ type: 'broker', brokerId, executionId, brokerName, argsJson })
-        } else {
+        if (!process.send || !process.connected) {
           clearTimeout(timeout)
           pendingBrokerCalls.delete(brokerId)
           resolve(JSON.stringify({ error: 'Parent process disconnected' }))
+          return
         }
+        sendIpcRequest({ type: 'broker', brokerId, executionId, brokerName, argsJson }, (err) => {
+          const pending = pendingBrokerCalls.get(brokerId)
+          if (!pending) return
+          clearTimeout(pending.timeout)
+          pendingBrokerCalls.delete(brokerId)
+          pending.resolve(JSON.stringify({ error: `Broker IPC send failed: ${err.message}` }))
+        })
       })
     })
     await jail.set('__brokerRef', brokerCallback)
@@ -732,13 +759,19 @@ async function executeTask(request, executionId) {
           }
         }, BROKER_TIMEOUT_MS)
         pendingBrokerCalls.set(brokerId, { resolve, timeout, executionId })
-        if (process.send && process.connected) {
-          process.send({ type: 'broker', brokerId, executionId, brokerName, argsJson })
-        } else {
+        if (!process.send || !process.connected) {
           clearTimeout(timeout)
           pendingBrokerCalls.delete(brokerId)
           resolve(JSON.stringify({ error: 'Parent process disconnected' }))
+          return
         }
+        sendIpcRequest({ type: 'broker', brokerId, executionId, brokerName, argsJson }, (err) => {
+          const pending = pendingBrokerCalls.get(brokerId)
+          if (!pending) return
+          clearTimeout(pending.timeout)
+          pendingBrokerCalls.delete(brokerId)
+          pending.resolve(JSON.stringify({ error: `Broker IPC send failed: ${err.message}` }))
+        })
       })
     })
     releaseables.push(brokerRef)