loop/parallel results compaction

Author: icecrasher321

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

@@ -3,7 +3,7 @@
  */
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { clearLargeValueCacheForTests } from '@/lib/execution/payloads/cache'
-import { isLargeValueRef } from '@/lib/execution/payloads/large-value-ref'
+import { isLargeArrayManifest } from '@/lib/execution/payloads/large-array-manifest-metadata'
 import { EDGE } from '@/executor/constants'
 import { LoopOrchestrator } from '@/executor/orchestrators/loop'
 import type { ExecutionContext } from '@/executor/types'
@@ -99,6 +99,8 @@ describe('LoopOrchestrator', () => {
     await orchestrator.evaluateLoopContinuation(ctx, 'loop-1')
 
     const output = setBlockOutput.mock.calls[0][1]
-    expect(isLargeValueRef(output.results[0][0].result)).toBe(true)
+    expect(Array.isArray(output.results[0])).toBe(true)
+    expect(isLargeArrayManifest(output.results[0][0].result)).toBe(true)
+    expect(output.results[0][0].result.totalCount).toBe(200_000)
   })
 })

apps/sim/executor/orchestrators/loop.ts

@@ -3,7 +3,7 @@ import { toError } from '@sim/utils/errors'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { isExecutionCancelled, isRedisCancellationEnabled } from '@/lib/execution/cancellation'
 import { executeInIsolatedVM } from '@/lib/execution/isolated-vm'
-import { compactExecutionPayload, compactSubflowResults } from '@/lib/execution/payloads/serializer'
+import { compactSubflowResults } from '@/lib/execution/payloads/serializer'
 import { isLikelyReferenceSegment } from '@/lib/workflows/sanitization/references'
 import { buildLoopIndexCondition, DEFAULTS, EDGE, PARALLEL } from '@/executor/constants'
 import type { DAG } from '@/executor/dag/builder'
@@ -252,13 +252,13 @@ export class LoopOrchestrator {
     }
 
     if (iterationResults.length > 0) {
-      const compactedIterationResults = (await compactExecutionPayload(iterationResults, {
+      const compactedIterationResults = await compactSubflowResults(iterationResults, {
         workspaceId: ctx.workspaceId,
         workflowId: ctx.workflowId,
         executionId: ctx.executionId,
         userId: ctx.userId,
         requireDurable: true,
-      })) as NormalizedBlockOutput[]
+      })
       scope.allIterationOutputs.push(compactedIterationResults)
     }