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.

Author: waleedlatif1

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')
     }