fix(execution): keep preserved base64 inline and stop a failed buffer write failing a resume

The buffered-event threshold was applied even when a caller explicitly asked to
preserve UserFile base64 inline, so anything past that threshold came back as a
ref the caller cannot read — defeating the request. Preserve the shared cap for
that path; the existing strip-and-recompact fallback still bounds the event.

The resume path awaited a non-terminal buffer write bare, so a failed write
propagated into the executor callback and failed work that had already run.
The buffer only backs reconnect replay, so degrade to live-only delivery the
same way the execute route does.

Author: waleedlatif1

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

@@ -4,6 +4,7 @@
 import { redisConfigMockFns, resetRedisConfigMock } from '@sim/testing'
 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(() => {
@@ -390,6 +391,38 @@ describe('execution event buffer', () => {
     expect(persistedBytes).toBeLessThan(256 * 1024)
   })
 
+  /**
+   * `preserveUserFileBase64` is an explicit request for inline delivery. The
+   * tighter buffered-event threshold must not quietly convert that base64 into
+   * a ref the caller cannot read back.
+   */
+  it('keeps preserved UserFile base64 inline above the buffered-event threshold', async () => {
+    mockRedis.incrby.mockResolvedValue(100)
+    const base64 = Buffer.from('y'.repeat(400 * 1024)).toString('base64')
+
+    const writer = createExecutionEventWriter('exec-1', {
+      workspaceId: 'ws-1',
+      workflowId: 'wf-1',
+      preserveUserFileBase64: true,
+    })
+    await writer.writeTerminal(
+      {
+        type: 'execution:completed',
+        timestamp: new Date().toISOString(),
+        executionId: 'exec-1',
+        workflowId: 'wf-1',
+        data: {
+          output: { file: { type: 'file', name: 'a.txt', size: 1, key: 'k', base64 } },
+        },
+      } as unknown as ExecutionEvent,
+      'complete'
+    )
+
+    const persisted = JSON.stringify(persistedEntries[0])
+    expect(persisted).toContain(base64)
+    expect(persisted).not.toContain(LARGE_VALUE_REF_MARKER)
+  })
+
   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

@@ -312,6 +312,11 @@ async function compactEventForBuffer(
   let compactedData = await compactExecutionPayload(event.data, {
     ...baseOptions,
     preserveUserFileBase64: context.preserveUserFileBase64,
+    // Preserving base64 is an explicit request for inline delivery, so the
+    // tighter buffer threshold must not turn it straight back into a ref the
+    // caller cannot read. Oversize is still bounded — the strip-and-recompact
+    // fallback below runs once the whole event exceeds the shared cap.
+    ...(context.preserveUserFileBase64 ? { thresholdBytes: LARGE_VALUE_THRESHOLD_BYTES } : {}),
   })
   let eventData = trimFinalBlockLogsForEventData(compactedData)
   let eventDataSize = getJsonSize(eventData)

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

@@ -1323,7 +1323,17 @@ 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. A failed buffer write must not fail work
+              // that already ran, which a bare await here would do.
+              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)
       }