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

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,188 @@ 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)
+  })
+
+  /**
+   * Terminal status is the reader's end-of-run signal: once it lands, a
+   * reconnecting client drains what is in Redis and closes. Stamping it while
+   * lower event ids are still queued strands those events behind a stream the
+   * reader has already finished with.
+   *
+   * Needs a backlog past the single-write cap so chunking leaves a remainder
+   * behind the terminal entry — the only shape where that ordering can invert.
+   */
+  it('does not stamp terminal status while earlier events are still queued', async () => {
+    mockRedis.incrby.mockResolvedValue(100000)
+    const idsAtStamp: number[] = []
+    mockRedis.eval.mockImplementation(async (script: string, ...args: unknown[]) => {
+      if (!isFlushScript(script)) return [1, 'ok', 0, 0]
+      const { terminalStatus, zaddArgs } = parseFlushEvalArgs(args)
+      // Reject any multi-entry batch, forcing the terminal-alone retry path.
+      if (zaddArgs.length > 2) return [0, 'execution_redis_bytes', 64 * 1024 * 1024]
+      for (let i = 0; i < zaddArgs.length; i += 2) {
+        persistedEntries.push(JSON.parse(zaddArgs[i + 1] as string) as ExecutionEventEntry)
+      }
+      if (terminalStatus && idsAtStamp.length === 0) {
+        idsAtStamp.push(...persistedEntries.map((e) => e.eventId))
+      }
+      return [1, 1, 0]
+    })
+
+    // ~3MB per event, so three of them exceed the 8MiB single-write cap and the
+    // chunk boundary leaves a remainder queued behind the terminal entry.
+    const payload = 'x'.repeat(1_500_000)
+    const writer = createExecutionEventWriter('exec-1', {
+      workspaceId: 'ws-1',
+      workflowId: 'wf-1',
+    })
+    for (let i = 0; i < 3; i++) {
+      await writer.write(makeEvent(payload)).catch(() => {})
+    }
+    await writer.writeTerminal(makeEvent('terminal'), 'complete').catch(() => {})
+    await writer.close().catch(() => {})
+
+    const terminalId = Math.max(...persistedEntries.map((e) => e.eventId))
+    const strandedAtStamp = persistedEntries
+      .map((e) => e.eventId)
+      .filter((id) => id < terminalId && !idsAtStamp.includes(id))
+    expect(strandedAtStamp).toEqual([])
+  })
+
+  /**
+   * Pressure has to be measured as events are produced, not once a flush
+   * succeeds. A burst is compacted long before the scheduled flush runs, so
+   * flush-time accounting would let the very batch that exhausts the budget
+   * through at the loose ceiling and drop it instead of offloading it.
+   */
+  it('engages pressure within a burst that has not flushed yet', async () => {
+    mockRedis.incrby.mockResolvedValue(100000)
+    const payload = 'x'.repeat(2 * 1024 * 1024)
+
+    const writer = createExecutionEventWriter('exec-1', {
+      workspaceId: 'ws-1',
+      workflowId: 'wf-1',
+    })
+    // No flush between writes: everything stays pending while the burst builds.
+    for (let i = 0; i < 20; i++) {
+      await writer.write(makeEvent(payload)).catch(() => {})
+    }
+    await writer.flush().catch(() => {})
+
+    // The later events in the burst must have been offloaded, not left inline.
+    const persisted = JSON.stringify(persistedEntries)
+    expect(persisted).toContain(LARGE_VALUE_REF_MARKER)
+  })
+
+  /**
+   * A transient failure while draining the backlog must not cost events, and
+   * must not let the run be marked terminal. Overwriting the queue would drop
+   * entries the budget never rejected, and the drain's final chunk would
+   * otherwise stamp the status before the terminal event is written.
+   */
+  it('retains the backlog and withholds terminal status when the drain fails transiently', async () => {
+    mockRedis.incrby.mockResolvedValue(100000)
+    const stamped: string[] = []
+    let failDrain = true
+    mockRedis.eval.mockImplementation(async (script: string, ...args: unknown[]) => {
+      if (!isFlushScript(script)) return [1, 'ok', 0, 0]
+      const { terminalStatus, zaddArgs } = parseFlushEvalArgs(args)
+      // Reject any multi-entry batch so the terminal-alone retry path is taken.
+      if (zaddArgs.length > 2) return [0, 'execution_redis_bytes', 64 * 1024 * 1024]
+      // The backlog drain hits a transient outage rather than a budget rejection.
+      if (failDrain) {
+        failDrain = false
+        throw new Error('redis unavailable')
+      }
+      for (let i = 0; i < zaddArgs.length; i += 2) {
+        persistedEntries.push(JSON.parse(zaddArgs[i + 1] as string) as ExecutionEventEntry)
+      }
+      if (terminalStatus) stamped.push(terminalStatus)
+      return [1, 1, 0]
+    })
+
+    const payload = 'x'.repeat(1_500_000)
+    const writer = createExecutionEventWriter('exec-1', {
+      workspaceId: 'ws-1',
+      workflowId: 'wf-1',
+    })
+    for (let i = 0; i < 3; i++) {
+      await writer.write(makeEvent(payload)).catch(() => {})
+    }
+    await expect(writer.writeTerminal(makeEvent('terminal'), 'complete')).rejects.toThrow()
+
+    // The transiently-failed backlog is still queued, so it is not lost.
+    expect(stamped).toEqual([])
+    await writer.close().catch(() => {})
+    expect(persistedEntries.length).toBeGreaterThan(0)
+  })
+
+  /**
+   * Offloading under pressure is an optimization. If the value cannot be
+   * persisted durably, the event must still reach the replay buffer inline —
+   * dropping it would leave a reconnecting client permanently missing it.
+   */
+  it('buffers the event inline when a pressure offload cannot be persisted', async () => {
+    mockRedis.incrby.mockResolvedValue(100000)
+    const payload = 'x'.repeat(2 * 1024 * 1024)
+
+    // No workspace/workflow ids, so durable persistence of an offloaded value
+    // fails the way a storage outage would.
+    const writer = createExecutionEventWriter('exec-1')
+    for (let i = 0; i < 20; i++) {
+      await writer.write(makeEvent(payload)).catch(() => {})
+    }
+    await writer.flush().catch(() => {})
+
+    expect(persistedEntries).toHaveLength(20)
+  })
+
   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,27 @@ export function createExecutionEventWriter(
   let maxReservedId = 0
   let flushTimer: ReturnType<typeof setTimeout> | null = null
   let consecutiveFlushFailures = 0
+  /**
+   * Bytes this execution has produced, counted as each event is compacted
+   * rather than once a flush succeeds. A burst can be compacted long before the
+   * scheduled flush runs, so flush-time accounting would let the very batch that
+   * exhausts the budget through at the loose ceiling. Counted gross of
+   * ring-buffer pruning too, so the mark is reached 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
@@ -983,17 +1022,39 @@ export function createExecutionEventWriter(
     }
   }
 
+  /**
+   * Compact an event for the buffer, degrading if pressure offloading fails.
+   *
+   * Offloading under pressure is an optimization: it keeps a heavy run from
+   * exhausting its budget. When durable storage rejects the write, losing the
+   * event from replay entirely is a worse outcome than carrying it inline, so
+   * fall back to the shared cap — exactly what the run would have done before
+   * pressure engaged.
+   */
+  const compactForBuffer = async (event: ExecutionEvent) => {
+    const valueThresholdBytes = getValueThresholdBytes()
+    const options = { ...context, executionId, requireDurablePayloads: true }
+    if (valueThresholdBytes === undefined) return compactEventForBuffer(event, options)
+    try {
+      return await compactEventForBuffer(event, { ...options, valueThresholdBytes })
+    } catch (error) {
+      logger.warn('Pressure offload failed; buffering the event inline instead', {
+        executionId,
+        eventType: event.type,
+        error: toError(error).message,
+      })
+      return compactEventForBuffer(event, options)
+    }
+  }
+
   const writeCore = async (event: ExecutionEvent): Promise<ExecutionEventEntry> => {
     if (nextEventId === 0 || nextEventId > maxReservedId) {
       await reserveIds(1)
     }
     const eventId = nextEventId++
-    const compactEvent = await compactEventForBuffer(event, {
-      ...context,
-      executionId,
-      requireDurablePayloads: true,
-    })
+    const compactEvent = await compactForBuffer(event)
     const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent }
+    bufferedBytes += getJsonSize(entry) ?? 0
     pending.push(entry)
     if (pending.length >= FLUSH_MAX_BATCH) {
       await flushPending()
@@ -1034,12 +1095,9 @@ export function createExecutionEventWriter(
         await reserveIds(1)
       }
       const eventId = nextEventId++
-      const compactEvent = await compactEventForBuffer(event, {
-        ...context,
-        executionId,
-        requireDurablePayloads: true,
-      })
+      const compactEvent = await compactForBuffer(event)
       const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent }
+      bufferedBytes += getJsonSize(entry) ?? 0
       pending.push(entry)
       let ok = false
       try {
@@ -1051,10 +1109,26 @@ export function createExecutionEventWriter(
           // alone rather than losing the run's final status with them. Gated on a
           // budget rejection specifically: a transient Redis error leaves the batch
           // queued for retry, and clearing it here would turn that into data loss.
-          const remaining = pending.filter((pendingEntry) => pendingEntry !== entry)
-          pending = [entry]
-          ok = await flushPending(false)
-          pending = pending.concat(remaining)
+          const terminalStatus = pendingTerminalStatus
+          pending = pending.filter((pendingEntry) => pendingEntry !== entry)
+          if (pending.length > 0) {
+            // Drain what is queued ahead of the terminal event first, with the
+            // status disarmed: `doFlush` stamps it on whichever chunk empties
+            // `pending`, so leaving it armed would mark the run complete before
+            // its terminal event is written. Whatever this cannot persist stays
+            // queued — it must not be overwritten.
+            pendingTerminalStatus = undefined
+            await flushPending(false)
+            pendingTerminalStatus = terminalStatus
+          }
+          if (pending.length === 0) {
+            // Only publish alone once nothing earlier is still queued. Doing so
+            // over a surviving backlog would signal end-of-run to a reader that
+            // has not received those events; failing instead lets the caller
+            // degrade, which records the status without claiming they arrived.
+            pending = [entry]
+            ok = await flushPending(false)
+          }
         }
       } catch (error) {
         discardTerminalEntry(entry)

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

@@ -1323,8 +1323,22 @@ export class PauseResumeManager {
               await degradeTerminalPublish(terminalStatus, error)
               return { eventId: 0, executionId: resumeExecutionId, event }
             })
-          : await eventWriter.write(event)
-        event.eventId = entry.eventId
+          : 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 null
+            })
+        // Leave `eventId` unset when the write failed, matching the execute
+        // route. Assigning 0 here would be persisted as a reconnect cursor and
+        // rewind the client to the start of the run.
+        if (entry) event.eventId = entry.eventId
         terminalEventPublished ||= Boolean(terminalStatus)
       }
       sendEvent?.(event)