fix(execution): stop run-buffer failures from stranding resumed executions (#6187)

* fix(resume): stop a failed run-buffer publish from stranding a resumed execution

The run buffer is a replay convenience for stream readers; the durable
execution record is authoritative. A failed terminal event publish was
deciding the outcome of work that had already run: it threw, the resume
was marked failed, and the paused execution was left paused forever.

That path is also not retryable — the workflow had already executed — so
the next resume attempt re-ran its side effects.

Degrade instead. Record the terminal status on the stream meta so readers
are not left polling an 'active' stream, log the failure, and let the
resume settle on its real result. The same treatment applies when the
terminal event is never published at all, which previously synthesized an
error for the same stranding effect.

Removes the now-unreachable TERMINAL_PUBLISH_ERROR constant and its
branch. Pre-execution buffer failures stay fatal and retryable, since no
work has happened yet.

* fix(execution): give per-user Redis budget keys a fixed window

The per-user byte budget refreshed its TTL on every accepted write, so for
any user who never went a full TTL without writing, the key never expired.
The per-execution data it accounted for kept expiring underneath it, so the
counter accrued bytes Redis had already dropped and drifted toward the
ceiling. On reaching it, every subsequent write for that user was rejected
until they stopped writing for a full TTL — and since a rejected write does
not refresh the TTL, it recovered on its own and then refilled.

User keys now get a fixed window: the TTL is set when the key is created and
never extended. Applied to all five Lua scripts that touch the key so the
writers cannot drift apart. Execution-scoped keys keep sliding — they are
refreshed on the same schedule as the data they account for, so their
counter and bytes stay in step.

* fix(execution): heal a user budget key that somehow has no expiry

The no-op branch of the flush script dropped the user-key EXPIRE outright
rather than guarding it like every other site. No current path can create
the key without an expiry, but if one ever did, that branch was the one
place that would never give it one — and a user counter with no expiry is
the unbounded version of the bug this series fixes.

Guard it instead, so the branch heals such a key rather than skipping it,
and so all five scripts read the same way.

* fix(stream): end a replay cleanly when terminal metadata has no terminal event

Terminal metadata is the authoritative end-of-run signal — the happy path
writes it atomically with the terminal event, and a run whose terminal
event could not be buffered records the status on its own. The reader
required both, so the degraded case threw and turned a replay that was
merely incomplete into a broken one.

Metadata without a matching event means the buffer degraded, not that the
run is still going. Log it and close the stream: the reader has already
received every event that was buffered, and the durable record is
unaffected either way.

apps/sim/app/api/workflows/[id]/executions/[executionId]/stream/route.test.ts

@@ -90,7 +90,7 @@ describe('execution stream reconnect route', () => {
     expect(mockReadExecutionEventsState).toHaveBeenNthCalledWith(2, 'exec-1', 3)
   })
 
-  it('errors when terminal metadata has no terminal event to replay', async () => {
+  it('ends the stream cleanly when terminal metadata has no terminal event to replay', async () => {
     mockReadExecutionMetaState
       .mockResolvedValueOnce({
         status: 'found',
@@ -115,9 +115,7 @@ describe('execution stream reconnect route', () => {
     })
 
     expect(response.status).toBe(200)
-    await expect(response.text()).rejects.toThrow(
-      'Execution reached terminal metadata without a terminal event'
-    )
+    await expect(response.text()).resolves.toContain('data: [DONE]')
   })
 
   it('allows replay event id gaps from reserved but unused writer ids', async () => {

apps/sim/app/api/workflows/[id]/executions/[executionId]/stream/route.ts

@@ -142,9 +142,20 @@ export const GET = withRouteHandler(
             if (!closed) controller.close()
           }
 
+          /**
+           * Terminal metadata is the authoritative end-of-run signal. The
+           * happy path writes it atomically with the terminal event, and a run
+           * whose terminal event could not be buffered records the status on
+           * its own — so metadata without a matching event means the buffer
+           * degraded, not that the run is still going. End the stream cleanly:
+           * the reader has already received every event that was buffered, and
+           * failing here would turn a degraded replay into a broken one.
+           */
           const closeAfterTerminalEvent = (events: ExecutionEventEntry[]) => {
             if (!enqueueEvents(events)) {
-              throw new Error('Execution reached terminal metadata without a terminal event')
+              logger.warn('Execution reached terminal metadata without a terminal event', {
+                executionId,
+              })
             }
             closeWithDone()
           }

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

@@ -68,6 +68,10 @@ function isResetScript(script: string): boolean {
   return script.includes('retained_bytes') && script.includes('replayStartEventId')
 }
 
+function countOccurrences(haystack: string, needle: string): number {
+  return haystack.split(needle).length - 1
+}
+
 describe('execution event buffer', () => {
   beforeEach(() => {
     vi.clearAllMocks()
@@ -417,6 +421,37 @@ describe('execution event buffer', () => {
     )
   })
 
+  it('never extends an existing user budget window while flushing events', async () => {
+    let flushScript = ''
+    mockRedis.eval.mockImplementation(async (script: string) => {
+      if (isFlushScript(script)) flushScript = script
+      return [1, false, 0]
+    })
+
+    const writer = createExecutionEventWriter('exec-1', { userId: 'user-1' })
+    await writer.writeTerminal(makeEvent('terminal'), 'complete')
+
+    expect(flushScript).not.toBe('')
+    const userKeyExpires = countOccurrences(flushScript, "redis.call('EXPIRE', KEYS[5]")
+    const userKeyTtlGuards = countOccurrences(flushScript, "redis.call('TTL', KEYS[5]) < 0")
+    expect(userKeyExpires).toBeGreaterThan(0)
+    expect(userKeyTtlGuards).toBe(userKeyExpires)
+  })
+
+  it('keeps sliding the execution budget window, which expires with its own data', async () => {
+    let flushScript = ''
+    mockRedis.eval.mockImplementation(async (script: string) => {
+      if (isFlushScript(script)) flushScript = script
+      return [1, false, 0]
+    })
+
+    const writer = createExecutionEventWriter('exec-1', { userId: 'user-1' })
+    await writer.writeTerminal(makeEvent('terminal'), 'complete')
+
+    expect(countOccurrences(flushScript, "redis.call('TTL', KEYS[4]) < 0")).toBe(0)
+    expect(countOccurrences(flushScript, "redis.call('EXPIRE', KEYS[4]")).toBeGreaterThan(0)
+  })
+
   it('reports pruned replay buffers before reading incomplete events', async () => {
     mockRedis.hgetall.mockResolvedValue({ status: 'active', earliestEventId: '10' })
 

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

@@ -85,7 +85,9 @@ if net_bytes > 0 then
   redis.call('EXPIRE', KEYS[4], budget_ttl_seconds)
   if #KEYS >= 5 then
     redis.call('INCRBY', KEYS[5], net_bytes)
-    redis.call('EXPIRE', KEYS[5], budget_ttl_seconds)
+    if redis.call('TTL', KEYS[5]) < 0 then
+      redis.call('EXPIRE', KEYS[5], budget_ttl_seconds)
+    end
   end
 elseif net_bytes < 0 then
   local release_bytes = -net_bytes
@@ -99,7 +101,7 @@ elseif net_bytes < 0 then
     local user_next = redis.call('DECRBY', KEYS[5], release_bytes)
     if user_next <= 0 then
       redis.call('DEL', KEYS[5])
-    else
+    elseif redis.call('TTL', KEYS[5]) < 0 then
       redis.call('EXPIRE', KEYS[5], budget_ttl_seconds)
     end
   end
@@ -107,7 +109,7 @@ else
   if redis.call('EXISTS', KEYS[4]) == 1 then
     redis.call('EXPIRE', KEYS[4], budget_ttl_seconds)
   end
-  if #KEYS >= 5 and redis.call('EXISTS', KEYS[5]) == 1 then
+  if #KEYS >= 5 and redis.call('EXISTS', KEYS[5]) == 1 and redis.call('TTL', KEYS[5]) < 0 then
     redis.call('EXPIRE', KEYS[5], budget_ttl_seconds)
   end
 end
@@ -148,7 +150,7 @@ if retained_bytes > 0 then
     local user_next = redis.call('DECRBY', KEYS[4], retained_bytes)
     if user_next <= 0 then
       redis.call('DEL', KEYS[4])
-    else
+    elseif redis.call('TTL', KEYS[4]) < 0 then
       redis.call('EXPIRE', KEYS[4], tonumber(ARGV[4]))
     end
   end

apps/sim/lib/execution/redis-budget.server.test.ts

@@ -0,0 +1,76 @@
+/**
+ * @vitest-environment node
+ */
+import { describe, expect, it, vi } from 'vitest'
+import {
+  getExecutionRedisBudgetKeys,
+  reserveExecutionRedisBytes,
+} from '@/lib/execution/redis-budget.server'
+
+function countOccurrences(haystack: string, needle: string): number {
+  return haystack.split(needle).length - 1
+}
+
+async function captureReserveScript(userId?: string): Promise<string> {
+  let script = ''
+  const redis = {
+    eval: vi.fn(async (source: string) => {
+      script = source
+      return [1, 'ok', 0, 0]
+    }),
+  }
+
+  await reserveExecutionRedisBytes(redis as never, {
+    executionId: 'exec-1',
+    userId,
+    category: 'event_buffer',
+    operation: 'write_events',
+    bytes: 128,
+  })
+
+  return script
+}
+
+describe('reserveExecutionRedisBytes', () => {
+  it('scopes the reservation to the execution, and to the user when one is known', () => {
+    expect(
+      getExecutionRedisBudgetKeys({
+        executionId: 'exec-1',
+        category: 'event_buffer',
+        operation: 'write_events',
+        bytes: 1,
+      })
+    ).toEqual(['execution:redis-budget:execution:exec-1'])
+
+    expect(
+      getExecutionRedisBudgetKeys({
+        executionId: 'exec-1',
+        userId: 'user-1',
+        category: 'event_buffer',
+        operation: 'write_events',
+        bytes: 1,
+      })
+    ).toEqual(['execution:redis-budget:execution:exec-1', 'execution:redis-budget:user:user-1'])
+  })
+
+  /**
+   * A user key aggregates across every execution that user runs, so extending
+   * its TTL on each write keeps it alive indefinitely while the per-execution
+   * data it accounts for expires underneath it. The window must be fixed.
+   */
+  it('never extends an existing user budget window', async () => {
+    const script = await captureReserveScript('user-1')
+
+    const userKeyExpires = countOccurrences(script, "redis.call('EXPIRE', KEYS[2]")
+    const userKeyTtlGuards = countOccurrences(script, "redis.call('TTL', KEYS[2]) < 0")
+    expect(userKeyExpires).toBeGreaterThan(0)
+    expect(userKeyTtlGuards).toBe(userKeyExpires)
+  })
+
+  it('keeps sliding the execution budget window, which expires with its own data', async () => {
+    const script = await captureReserveScript('user-1')
+
+    expect(countOccurrences(script, "redis.call('TTL', KEYS[1]) < 0")).toBe(0)
+    expect(countOccurrences(script, "redis.call('EXPIRE', KEYS[1]")).toBeGreaterThan(0)
+  })
+})

apps/sim/lib/execution/redis-budget.server.ts

@@ -12,6 +12,22 @@ const MAX_EXECUTION_REDIS_BYTES = 64 * 1024 * 1024
 const MAX_USER_REDIS_BYTES = 256 * 1024 * 1024
 const REDIS_BUDGET_TTL_SECONDS = 60 * 60
 
+/**
+ * Execution and user budget keys expire differently on purpose.
+ *
+ * An execution key accounts for data that is refreshed on the same schedule as
+ * the key itself, so sliding its TTL on every write keeps the counter and the
+ * bytes it represents in step.
+ *
+ * A user key aggregates across every execution that user runs. Sliding its TTL
+ * on each write keeps it alive indefinitely for any user who stays active,
+ * while the per-execution data it accounts for keeps expiring underneath it —
+ * so the counter accrues bytes Redis has already dropped and eventually pins
+ * the user at their ceiling until they go a full TTL without writing. User
+ * keys therefore get a fixed window: the TTL is set when the key is created
+ * and never extended.
+ */
+
 const RESERVE_REDIS_BYTES_SCRIPT = `
 local bytes = tonumber(ARGV[1])
 local execution_limit = tonumber(ARGV[2])
@@ -32,7 +48,9 @@ redis.call('INCRBY', KEYS[1], bytes)
 redis.call('EXPIRE', KEYS[1], ttl_seconds)
 if #KEYS >= 2 then
   redis.call('INCRBY', KEYS[2], bytes)
-  redis.call('EXPIRE', KEYS[2], ttl_seconds)
+  if redis.call('TTL', KEYS[2]) < 0 then
+    redis.call('EXPIRE', KEYS[2], ttl_seconds)
+  end
 end
 return {1, 'ok', execution_current + bytes, user_current + bytes}
 `

apps/sim/lib/uploads/utils/user-file-base64.server.ts

@@ -59,7 +59,7 @@ if bytes and bytes > 0 then
     local user_next = redis.call('DECRBY', KEYS[4], bytes)
     if user_next <= 0 then
       redis.call('DEL', KEYS[4])
-    else
+    elseif redis.call('TTL', KEYS[4]) < 0 then
       redis.call('EXPIRE', KEYS[4], budget_ttl_seconds)
     end
   end
@@ -129,7 +129,7 @@ if #KEYS >= 4 then
       redis.call('DEL', KEYS[4])
     end
   end
-  if redis.call('EXISTS', KEYS[4]) == 1 then
+  if redis.call('EXISTS', KEYS[4]) == 1 and redis.call('TTL', KEYS[4]) < 0 then
     redis.call('EXPIRE', KEYS[4], budget_ttl_seconds)
   end
 end

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

@@ -14,6 +14,7 @@ import {
   flushExecutionStreamReplayBuffer,
   initializeExecutionStreamMeta,
   resetExecutionStreamBuffer,
+  setExecutionMeta,
   type TerminalExecutionStreamStatus,
 } from '@/lib/execution/event-buffer'
 import {
@@ -71,7 +72,6 @@ const execDb = dbFor('exec')
 
 const logger = createLogger('HumanInTheLoopManager')
 const RUN_BUFFER_UNAVAILABLE_ERROR = 'Run buffer temporarily unavailable'
-const TERMINAL_PUBLISH_ERROR = 'Run buffer terminal event publish failed'
 const RESUMABLE_PAUSED_STATUSES = ['paused', 'partially_resumed'] as const
 const CANCELLABLE_PAUSED_STATUSES = ['paused', 'partially_resumed'] as const
 const AUTOMATIC_RESUME_INTERVENTION_PREFIX = 'Automatic resume requires manual intervention: '
@@ -771,7 +771,7 @@ export class PauseResumeManager {
           preserveForRetry: true,
           retryable: error.retryable,
         })
-      } else if (message === RUN_BUFFER_UNAVAILABLE_ERROR || message === TERMINAL_PUBLISH_ERROR) {
+      } else if (message === RUN_BUFFER_UNAVAILABLE_ERROR) {
         await PauseResumeManager.markResumeAttemptFailed({
           resumeEntryId,
           pausedExecutionId: pausedExecution.id,
@@ -1280,6 +1280,38 @@ export class PauseResumeManager {
     }
 
     let terminalEventPublished = false
+    let terminalPublishDegraded = false
+
+    /**
+     * The run buffer is a replay convenience for stream readers; the durable
+     * execution record is authoritative. A failed terminal publish must not
+     * decide the outcome of work that already ran — the resume is not
+     * retryable at this point, so throwing here would strand the execution as
+     * paused and re-run its side effects on the next attempt. Degrade instead:
+     * record the terminal status on the stream meta so readers are not left
+     * polling an 'active' stream forever, and let the resume settle normally.
+     */
+    const degradeTerminalPublish = async (
+      terminalStatus: TerminalExecutionStreamStatus,
+      error: unknown
+    ) => {
+      terminalPublishDegraded = true
+      logger.warn('Failed to publish resume terminal event', {
+        resumeExecutionId,
+        status: terminalStatus,
+        error: toError(error).message,
+      })
+      const metaPersisted = await setExecutionMeta(resumeExecutionId, {
+        status: terminalStatus,
+      }).catch(() => false)
+      if (!metaPersisted) {
+        logger.warn('Failed to record degraded terminal status on resume stream meta', {
+          resumeExecutionId,
+          status: terminalStatus,
+        })
+      }
+    }
+
     const writeBufferedEvent = async (
       event: ExecutionEvent,
       terminalStatus?: TerminalExecutionStreamStatus
@@ -1287,13 +1319,9 @@ export class PauseResumeManager {
       const isBuffered = !LIVE_ONLY_EXECUTION_EVENT_TYPES.has(event.type)
       if (isBuffered) {
         const entry = terminalStatus
-          ? await eventWriter.writeTerminal(event, terminalStatus).catch((error) => {
-              logger.warn('Failed to publish resume terminal event', {
-                resumeExecutionId,
-                status: terminalStatus,
-                error: toError(error).message,
-              })
-              throw new Error(TERMINAL_PUBLISH_ERROR)
+          ? await eventWriter.writeTerminal(event, terminalStatus).catch(async (error) => {
+              await degradeTerminalPublish(terminalStatus, error)
+              return { eventId: 0, executionId: resumeExecutionId, event }
             })
           : await eventWriter.write(event)
         event.eventId = entry.eventId
@@ -1686,9 +1714,10 @@ export class PauseResumeManager {
           status: finalMetaStatus,
           replayBufferFlushed,
         })
-        if (!executionError) {
-          executionError = new Error(TERMINAL_PUBLISH_ERROR)
-        }
+        await degradeTerminalPublish(
+          finalMetaStatus,
+          new Error('Terminal event was never published')
+        )
       } else {
         await eventWriter.close().catch((error) => {
           logger.warn('Failed to close resume event writer after terminal publish', {
@@ -1707,6 +1736,13 @@ export class PauseResumeManager {
      */
     await loggingSession.waitForPostExecution()
 
+    if (terminalPublishDegraded) {
+      logger.warn('Resume settled with a degraded run buffer', {
+        resumeExecutionId,
+        status: finalMetaStatus,
+      })
+    }
+
     if (executionError || !result) {
       throw executionError ?? new Error('Resume execution did not produce a result')
     }