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

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,15 +101,15 @@ 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
 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,20 +1280,48 @@ 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
     ) => {
       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')
     }