fix(execution): offload buffered event values under budget pressure

An execution buffers EVENT_LIMIT events inside a per-execution byte budget, so
a full ring only fits if events average under budget/EVENT_LIMIT. Values were
only offloaded to object storage at the shared 8 MiB cap, far above that, so a
run emitting large block outputs exhausted its budget within a few dozen events
and stayed pinned at its ceiling for the rest of its life.

Applying that ceiling to every run would be worse than the problem: the SSE
stream carries the compacted event and the terminal renders a ref only as a
preview, so ordinary block outputs would stop being readable live, and every
value would cost an object-storage write on the hot path. Engage the tight
ceiling only once a run has actually buffered past half its budget. A short run
keeps full-fidelity output and pays nothing; a runaway one stops accumulating.

Both bounds derive from the existing budget rather than being asserted, and
preserved UserFile base64 is exempt — it is an explicit request for inline
delivery, already bounded by its own cap and the strip-and-recompact fallback.

Also stop a failed resume-path buffer write from failing the run: it was
awaited bare, so the rejection propagated into the executor callback and failed
work that had already completed. The buffer only backs reconnect replay, so
degrade to live-only delivery the way the execute route does.

Author: waleedlatif1

apps/sim/lib/execution/event-buffer.test.ts

@@ -5,6 +5,7 @@ import { redisConfigMockFns, resetRedisConfigMock } from '@sim/testing'
 import { sleep } from '@sim/utils/helpers'
 import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
 import type { ExecutionEventEntry } from '@/lib/execution/event-buffer'
+import { LARGE_VALUE_REF_MARKER } from '@/lib/execution/payloads/large-value-ref'
 import type { ExecutionEvent } from '@/lib/workflows/executor/execution-events'
 
 const { mockRedis, persistedEntries } = vi.hoisted(() => {
@@ -622,6 +623,54 @@ describe('execution event buffer', () => {
     await expect(writer.flush()).resolves.toBeUndefined()
   })
 
+  /**
+   * A short run must keep full-fidelity output: the SSE stream carries the
+   * compacted event, and the terminal renders a ref only as a preview, so
+   * offloading ordinary block outputs would make them unreadable live.
+   */
+  it('keeps values inline while the execution is below the offload pressure mark', async () => {
+    mockRedis.incrby.mockResolvedValue(100)
+    const payload = 'x'.repeat(512 * 1024)
+
+    const writer = createExecutionEventWriter('exec-1', {
+      workspaceId: 'ws-1',
+      workflowId: 'wf-1',
+    })
+    await writer.write(makeEvent(payload))
+    await writer.flush()
+
+    const persisted = JSON.stringify(persistedEntries[0])
+    expect(persisted).toContain(payload)
+    expect(persisted).not.toContain(LARGE_VALUE_REF_MARKER)
+  })
+
+  /**
+   * Once a run has buffered its way into the danger zone the tight ceiling
+   * engages, so it stops accumulating against its budget instead of pinning
+   * itself at the ceiling for the rest of its life.
+   */
+  it('offloads values once the execution crosses the offload pressure mark', async () => {
+    mockRedis.incrby.mockResolvedValue(100000)
+    const payload = 'x'.repeat(2 * 1024 * 1024)
+
+    const writer = createExecutionEventWriter('exec-1', {
+      workspaceId: 'ws-1',
+      workflowId: 'wf-1',
+    })
+    // Push past half the per-execution budget so the next write is under pressure.
+    for (let i = 0; i < 17; i++) {
+      await writer.write(makeEvent(payload))
+      await writer.flush()
+    }
+    persistedEntries.length = 0
+    await writer.write(makeEvent(payload))
+    await writer.flush()
+
+    const persisted = JSON.stringify(persistedEntries[0])
+    expect(persisted).toContain(LARGE_VALUE_REF_MARKER)
+    expect(persisted).not.toContain(payload)
+  })
+
   it('preserves requested UserFile base64 when buffering terminal events', async () => {
     mockRedis.incrby.mockResolvedValue(100)
     const base64 = Buffer.from('hello').toString('base64')

apps/sim/lib/execution/event-buffer.ts

@@ -27,6 +27,21 @@ const FLUSH_INTERVAL_MS = 15
 const FLUSH_MAX_RETRY_INTERVAL_MS = 1000
 const FLUSH_MAX_BATCH = 200
 const MAX_PENDING_EVENTS = 1000
+/**
+ * Bytes a single execution may buffer before its events start offloading
+ * aggressively, and the per-value threshold applied once it does.
+ *
+ * The buffer holds `EVENT_LIMIT` events inside the per-execution byte budget,
+ * so a full ring only fits if events average under budget/EVENT_LIMIT. Applying
+ * that ceiling to every run would offload ordinary block outputs into refs the
+ * terminal cannot display — the SSE stream carries the compacted event, and a
+ * ref renders only as a preview. Instead the tight ceiling engages only once a
+ * run has actually buffered its way into the danger zone, so a short run keeps
+ * full-fidelity output and a runaway one stops accumulating.
+ */
+const EXECUTION_EVENT_OFFLOAD_PRESSURE_BYTES = getExecutionRedisBudgetLimits().maxExecutionBytes / 2
+const EXECUTION_EVENT_PRESSURE_VALUE_BYTES =
+  getExecutionRedisBudgetLimits().maxExecutionBytes / EVENT_LIMIT
 const ACTIVE_META_ATTEMPTS = 3
 const FINALIZE_FLUSH_ATTEMPTS = 2
 const FLUSH_EVENTS_SCRIPT = `
@@ -282,6 +297,8 @@ export interface ExecutionEventWriter {
 export interface ExecutionEventWriterContext extends LargeValueStoreContext {
   requireDurablePayloads?: boolean
   preserveUserFileBase64?: boolean
+  /** Offload ceiling for individual values; defaults to the shared large-value cap. */
+  valueThresholdBytes?: number
 }
 
 async function compactEventForBuffer(
@@ -297,6 +314,7 @@ async function compactEventForBuffer(
     executionId: context.executionId ?? event.executionId,
     requireDurable: context.requireDurablePayloads,
     preserveRoot: true,
+    thresholdBytes: context.valueThresholdBytes,
   }
 
   let compactedData = await compactExecutionPayload(event.data, {
@@ -746,6 +764,24 @@ export function createExecutionEventWriter(
   let maxReservedId = 0
   let flushTimer: ReturnType<typeof setTimeout> | null = null
   let consecutiveFlushFailures = 0
+  /**
+   * Bytes this execution has successfully buffered. Counted gross rather than
+   * net of ring-buffer pruning, so it reaches the pressure mark early — erring
+   * toward offloading sooner is the safe direction.
+   */
+  let bufferedBytes = 0
+
+  /**
+   * Preserved base64 is an explicit request for inline delivery and is already
+   * bounded by its own cap and the strip-and-recompact fallback, so pressure
+   * never rewrites it into a ref the caller cannot read.
+   */
+  const getValueThresholdBytes = () => {
+    if (context.preserveUserFileBase64) return undefined
+    return bufferedBytes >= EXECUTION_EVENT_OFFLOAD_PRESSURE_BYTES
+      ? EXECUTION_EVENT_PRESSURE_VALUE_BYTES
+      : undefined
+  }
 
   const getFlushDelayMs = () => {
     if (consecutiveFlushFailures === 0) return FLUSH_INTERVAL_MS
@@ -911,6 +947,7 @@ export function createExecutionEventWriter(
       }
       consecutiveFlushFailures = 0
       lastResourceLimitError = null
+      bufferedBytes += batchBytes
       if (chunkTerminalStatus) pendingTerminalStatus = undefined
       return true
     } catch (error) {
@@ -992,6 +1029,7 @@ export function createExecutionEventWriter(
       ...context,
       executionId,
       requireDurablePayloads: true,
+      valueThresholdBytes: getValueThresholdBytes(),
     })
     const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent }
     pending.push(entry)
@@ -1038,6 +1076,7 @@ export function createExecutionEventWriter(
         ...context,
         executionId,
         requireDurablePayloads: true,
+        valueThresholdBytes: getValueThresholdBytes(),
       })
       const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent }
       pending.push(entry)

apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts

@@ -1323,7 +1323,18 @@ export class PauseResumeManager {
               await degradeTerminalPublish(terminalStatus, error)
               return { eventId: 0, executionId: resumeExecutionId, event }
             })
-          : await eventWriter.write(event)
+          : await eventWriter.write(event).catch((error) => {
+              // The buffer only backs reconnect replay; the live stream is the
+              // primary delivery path. Awaiting this bare let a failed write
+              // propagate into the executor callback and fail work that had
+              // already run, so degrade the same way the execute route does.
+              logger.warn('Resume event buffer write failed; delivering live only', {
+                resumeExecutionId,
+                eventType: event.type,
+                error: toError(error).message,
+              })
+              return { eventId: 0, executionId: resumeExecutionId, event }
+            })
         event.eventId = entry.eventId
         terminalEventPublished ||= Boolean(terminalStatus)
       }