fix(execution): do not lose the backlog or publish terminal status early

Draining the backlog ahead of the terminal event left the terminal status armed,
so whichever chunk emptied the queue stamped the run complete before its
terminal event was written — the inverse of the ordering the drain was added to
guarantee. Disarm the status for the drain and restore it afterwards.

The drain's result was also discarded: a transient Redis failure requeues its
batch, and the unconditional reassignment that followed dropped those events
even though the budget never rejected them. Keep whatever could not be
persisted, and publish the terminal event alone only once nothing earlier is
still queued — failing otherwise lets the caller degrade, which records the
status without claiming the missing events arrived.

Leave eventId unset on a failed resume-path write. Assigning 0 was persisted by
clients as a reconnect cursor and rewound them to the start of the run.

Author: waleedlatif1

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

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

@@ -1094,17 +1094,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)
-          // Drain what is queued ahead of the terminal event first. Terminal
-          // status is the reader's end-of-run signal: stamping it while lower
-          // event ids are still queued strands them behind a stream the reader
-          // has already drained and closed.
-          if (remaining.length > 0) {
-            pending = 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)
           }
-          pending = [entry]
-          ok = await flushPending(false)
         }
       } catch (error) {
         discardTerminalEntry(entry)

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

@@ -1333,9 +1333,12 @@ export class PauseResumeManager {
                 eventType: event.type,
                 error: toError(error).message,
               })
-              return { eventId: 0, executionId: resumeExecutionId, event }
+              return null
             })
-        event.eventId = entry.eventId
+        // 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)