fix(execution): keep an event in the buffer when a pressure offload fails

Durable compaction runs before an event is queued, so a storage or metadata
failure dropped it from replay entirely — a reconnecting client would never see
it, even though the live path carried on. Offloading under pressure is only an
optimization that keeps a heavy run from exhausting its budget, so when the
value cannot be persisted, fall back to buffering it inline: exactly what the
run would have done before pressure engaged.

Author: waleedlatif1

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

@@ -785,6 +785,26 @@ describe('execution event buffer', () => {
     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

@@ -1022,17 +1022,37 @@ 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,
-      valueThresholdBytes: getValueThresholdBytes(),
-    })
+    const compactEvent = await compactForBuffer(event)
     const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent }
     bufferedBytes += getJsonSize(entry) ?? 0
     pending.push(entry)
@@ -1075,12 +1095,7 @@ export function createExecutionEventWriter(
         await reserveIds(1)
       }
       const eventId = nextEventId++
-      const compactEvent = await compactEventForBuffer(event, {
-        ...context,
-        executionId,
-        requireDurablePayloads: true,
-        valueThresholdBytes: getValueThresholdBytes(),
-      })
+      const compactEvent = await compactForBuffer(event)
       const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent }
       bufferedBytes += getJsonSize(entry) ?? 0
       pending.push(entry)