fix(execution): stop offloading state the engine consumes structurally

Compacting a loop's `items` was a regression: the orchestrator indexes that
collection to derive the current `item`, the resume path rebuilds the scope
verbatim without materializing anything, and the loop resolver asserts no refs
reach it — so an oversized forEach would have traded a failed pause for a
broken resume. `currentIterationOutputs` is excluded for the same reason: the
block executor has already compacted its entries, and they resolve through the
reference path rather than being read raw.

Offloading is now limited to exactly the accumulators the orchestrators
themselves compact when a subflow exits. An oversized `items` collection
therefore still fails the pause; that is the honest outcome until it can be
handled without breaking iteration.

Author: waleedlatif1

apps/sim/executor/execution/snapshot-serializer.test.ts

@@ -299,24 +299,19 @@ describe('compactPauseSnapshotScopes', () => {
     expect(serialize(context).snapshot).toBeTruthy()
   })
 
-  /** A forEach collection is the most common way a loop's state gets large. */
-  it('compacts the forEach collection, not just the iteration outputs', async () => {
+  /**
+   * `items` is consumed structurally — the orchestrator indexes it to derive
+   * `item`, and the loop resolver asserts no refs reach it — so it stays inline
+   * even though that leaves an oversized collection unhandled. Offloading it
+   * would trade a failed pause for a broken resume.
+   */
+  it('leaves the forEach collection inline rather than breaking iteration', async () => {
     const context = loopContext({ items: fatIterations(40, 300_000) })
 
-    expect(() => serialize(context)).toThrow('oversized loop execution state')
     await compactPauseSnapshotScopes(context)
-    expect(serialize(context).snapshot).toBeTruthy()
-  })
-
-  /** A single fat mid-flight block output reaches the limit on its own. */
-  it('compacts in-flight iteration outputs', async () => {
-    const context = loopContext({
-      currentIterationOutputs: new Map([['block-1', { payload: 'x'.repeat(9_000_000) }]]),
-    })
 
-    expect(() => serialize(context)).toThrow('oversized loop execution state')
-    await compactPauseSnapshotScopes(context)
-    expect(serialize(context).snapshot).toBeTruthy()
+    const items = context.loopExecutions?.get('loop-1')?.items as unknown[]
+    expect(JSON.stringify(items)).not.toContain('__simLargeValueRef')
   })
 
   /** The assertion is on the whole record, so per-scope headroom is not enough. */

apps/sim/executor/execution/snapshot-serializer.ts

@@ -1,6 +1,6 @@
 import { recordMaterializedAccessKeys } from '@/lib/execution/payloads/access-keys'
 import { LARGE_VALUE_THRESHOLD_BYTES } from '@/lib/execution/payloads/large-value-ref'
-import { compactExecutionPayload, compactSubflowResults } from '@/lib/execution/payloads/serializer'
+import { compactSubflowResults } from '@/lib/execution/payloads/serializer'
 import type { DAG } from '@/executor/dag/builder'
 import type { EdgeManager } from '@/executor/execution/edge-manager'
 import { ExecutionSnapshot } from '@/executor/execution/snapshot'
@@ -222,10 +222,14 @@ function isSubflowStateOversized(loops?: Map<string, any>, parallels?: Map<strin
  * gone out by then, leaving the approver holding a link to something that was
  * never recorded.
  *
- * Every field that grows without an aggregate bound is covered: a loop's
- * iteration outputs, its in-flight iteration outputs and its `forEach`
- * collection, and the parallel equivalents. Compacting only one of them would
- * leave the same failure reachable by a different route.
+ * Covers exactly the accumulators the orchestrators themselves compact when a
+ * subflow exits — a loop's iteration outputs and a parallel's branch and
+ * accumulated outputs. Deliberately excluded: `items`, which the loop consumes
+ * structurally (`orchestrators/loop.ts` indexes it to derive `item`, and the
+ * loop resolver asserts no refs reach it), and `currentIterationOutputs`, whose
+ * entries the block executor has already compacted and which resolve through
+ * the reference path that materializes refs. Offloading either would trade this
+ * failure for a broken resume.
  *
  * Skipped entirely when the state already serializes small enough, so the
  * common case — a pause per iteration inside a modest loop — pays one bounded
@@ -264,24 +268,10 @@ export async function compactPauseSnapshotScopes(context: ExecutionContext): Pro
     if (scope.allIterationOutputs?.length) {
       scope.allIterationOutputs = await compactList(scope.allIterationOutputs)
     }
-    if (scope.items?.length) {
-      scope.items = await compactList(scope.items)
-    }
-    if (scope.currentIterationOutputs instanceof Map && scope.currentIterationOutputs.size > 0) {
-      for (const [blockId, output] of scope.currentIterationOutputs) {
-        scope.currentIterationOutputs.set(
-          blockId,
-          await compactExecutionPayload(output, { ...buildOptions(), preserveRoot: false })
-        )
-      }
-    }
     recordMaterializedAccessKeys(context, scope)
   }
 
   for (const scope of parallels?.values() ?? []) {
-    if (scope.items?.length) {
-      scope.items = await compactList(scope.items)
-    }
     if (scope.branchOutputs instanceof Map) {
       await compactMapValues(scope.branchOutputs)
     }