fix(execution): publish the terminal event when its batch is dropped

A budget-rejected batch carrying the terminal event dropped it with the
rest of the backlog, so a run could end without ever publishing its final
status even though the terminal event alone was small enough to fit.
Retry it on its own once, gated on a budget rejection so a transient Redis
error still leaves the batch queued for retry.

Use the shared sleep helper in the tests instead of a raw setTimeout
promise, which the utils enforcement gate rejects.

Author: waleedlatif1

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

@@ -2,6 +2,7 @@
  * @vitest-environment node
  */
 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 type { ExecutionEvent } from '@/lib/workflows/executor/execution-events'
@@ -490,13 +491,49 @@ describe('execution event buffer', () => {
     // Let writeTerminal's queued body actually enqueue its entry before the
     // in-flight flush resolves — otherwise the scheduled loop finds nothing left
     // to drain and the race under test never forms.
-    await new Promise((resolve) => setTimeout(resolve, 5))
+    await sleep(5)
     releaseFirstFlush?.()
     await terminalWrite
 
     expect(observedTerminalStatuses).toContain('complete')
   })
 
+  /**
+   * The backlog ahead of a terminal event can exceed the budget while the
+   * terminal event itself still fits. Discarding it alongside the backlog would
+   * leave readers without the final status for a run that could have published
+   * one.
+   */
+  it('still publishes the terminal event when the backlog ahead of it is dropped', async () => {
+    mockRedis.incrby.mockResolvedValue(100)
+    const observedTerminalStatuses: string[] = []
+    mockRedis.eval.mockImplementation(async (script: string, ...args: unknown[]) => {
+      if (!isFlushScript(script)) return [1, 'ok', 0, 0]
+      const { terminalStatus, zaddArgs } = parseFlushEvalArgs(args)
+      // Reject anything but a lone entry, standing in for a budget with only
+      // enough headroom left for one small write.
+      if (zaddArgs.length > 2) return [0, 'execution_redis_bytes', 64 * 1024 * 1024]
+      observedTerminalStatuses.push(terminalStatus)
+      for (let i = 0; i < zaddArgs.length; i += 2) {
+        persistedEntries.push(JSON.parse(zaddArgs[i + 1] as string) as ExecutionEventEntry)
+      }
+      return [1, 1, 0]
+    })
+
+    const writer = createExecutionEventWriter('exec-1')
+    for (let i = 0; i < 5; i++) {
+      await writer.write(makeEvent(`block-${i}`)).catch(() => {})
+    }
+
+    await expect(writer.writeTerminal(makeEvent('terminal'), 'complete')).resolves.toMatchObject({
+      executionId: 'exec-1',
+    })
+    expect(observedTerminalStatuses).toContain('complete')
+    expect(
+      persistedEntries.map((entry) => (entry.event.data as { blockId: string }).blockId)
+    ).toContain('terminal')
+  })
+
   /**
    * A terminal publish that threw must not be resurrected. Leaving the status
    * armed would let the next flush stamp the stream terminal for an event that
@@ -571,7 +608,7 @@ describe('execution event buffer', () => {
     const writer = createExecutionEventWriter('exec-1')
     await writer.write(makeEvent('a'))
 
-    await new Promise((resolve) => setTimeout(resolve, 60))
+    await sleep(60)
 
     await expect(writer.flush()).resolves.toBeUndefined()
   })

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

@@ -1052,6 +1052,18 @@ export function createExecutionEventWriter(
       let ok = false
       try {
         ok = await flushPending(false, status)
+        if (!ok && lastResourceLimitError && pendingTerminalStatus) {
+          // The batch carrying the terminal event exceeded the budget and was
+          // dropped. Those bytes are gone either way, and the terminal event is
+          // small enough to plausibly fit on its own — so give it one attempt
+          // 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, status)
+          pending = pending.concat(remaining)
+        }
       } catch (error) {
         discardTerminalEntry(entry)
         throw error