fix(trigger): complete async workflow runs whose failure is already recorded instead of failing them

apps/sim/background/async-preprocessing-correlation.test.ts

@@ -358,46 +358,86 @@ describe('async preprocessing correlation threading', () => {
     )
   })
 
-  it('preserves a core-finalized execution error for task failure semantics', async () => {
-    const rawError = Object.assign(new Error('Function 1 failed with activated-secret-value'), {
-      executionResult: {
-        success: false,
-        output: { error: 'Function failed' },
-        logs: [],
-      },
-    })
-    mockPreprocessExecution.mockResolvedValueOnce({
-      success: true,
-      actorUserId: 'actor-1',
-      workflowRecord: {
-        id: 'workflow-1',
-        userId: 'owner-1',
-        workspaceId: 'workspace-1',
-        variables: {},
-      },
-      billingAttribution,
-      executionTimeout: {},
-    })
-    mockExecuteWorkflowCore.mockRejectedValueOnce(rawError)
-    mockHasExecutionResult.mockImplementation((error) => error === rawError)
-    mockWasExecutionFinalizedByCore.mockReturnValue(true)
+  describe('a workflow failure core already recorded', () => {
+    const recordedError = () =>
+      Object.assign(new Error('Function 1 failed: RuntimeError'), {
+        executionResult: { success: false, output: { error: 'Function failed' }, logs: [] },
+      })
 
-    await expect(
-      executeWorkflowJob({
+    /** Core marks the failure finalized from a post-execution promise it does not await. */
+    function finalizeWhenPostExecutionSettles() {
+      let finalized = false
+      loggingSessionMockFns.mockWaitForPostExecution.mockImplementationOnce(async () => {
+        await Promise.resolve()
+        finalized = true
+      })
+      mockWasExecutionFinalizedByCore.mockImplementation(() => finalized)
+    }
+
+    function runRecordedFailure(executionId: string) {
+      mockPreprocessExecution.mockResolvedValueOnce({
+        success: true,
+        actorUserId: 'actor-1',
+        workflowRecord: {
+          id: 'workflow-1',
+          userId: 'owner-1',
+          workspaceId: 'workspace-1',
+          variables: {},
+        },
+        billingAttribution,
+        executionTimeout: {},
+      })
+      return executeWorkflowJob({
         principal,
         workflowId: 'workflow-1',
         userId: 'actor-1',
         workspaceId: 'workspace-1',
         billingAttribution,
         triggerType: 'api',
-        executionId: 'execution-finalized',
-        requestId: 'request-finalized',
+        executionId,
+        requestId: 'request-recorded',
       })
-    ).rejects.toBe(rawError)
+    }
 
-    expect(loggingSessionMockFns.mockWaitForPostExecution).not.toHaveBeenCalled()
-    expect(mockWasExecutionFinalizedByCore).toHaveBeenCalledWith(rawError, 'execution-finalized')
-    expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled()
+    it('completes the job with the failure once core finalization settles', async () => {
+      const error = recordedError()
+      mockExecuteWorkflowCore.mockRejectedValueOnce(error)
+      mockHasExecutionResult.mockImplementation((value) => value === error)
+      finalizeWhenPostExecutionSettles()
+
+      await expect(runRecordedFailure('execution-recorded')).resolves.toMatchObject({
+        success: false,
+        workflowId: 'workflow-1',
+        executionId: 'execution-recorded',
+        error: 'Function 1 failed: RuntimeError',
+      })
+      expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled()
+    })
+
+    it('still fails the job when the deadline passes while finalization settles', async () => {
+      const error = recordedError()
+      mockExecuteWorkflowCore.mockRejectedValueOnce(error)
+      let timedOut = false
+      mockIsWorkflowTimedOut.mockImplementation(() => timedOut)
+      loggingSessionMockFns.mockWaitForPostExecution.mockImplementationOnce(async () => {
+        await Promise.resolve()
+        timedOut = true
+      })
+      mockWasExecutionFinalizedByCore.mockReturnValue(true)
+
+      await expect(runRecordedFailure('execution-recorded-deadline')).rejects.toBe(error)
+      expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled()
+    })
+
+    it('still fails the job when the run passed its deadline', async () => {
+      const error = recordedError()
+      mockExecuteWorkflowCore.mockRejectedValueOnce(error)
+      mockIsWorkflowTimedOut.mockReturnValue(true)
+      mockWasExecutionFinalizedByCore.mockReturnValue(true)
+
+      await expect(runRecordedFailure('execution-recorded-late')).rejects.toBe(error)
+      expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled()
+    })
   })
 
   it('fails the backing job after a cooperative workflow timeout is finalized', async () => {
@@ -528,6 +568,57 @@ describe('async preprocessing correlation threading', () => {
     )
   })
 
+  describe('a preprocessing rejection', () => {
+    function runRejected(error: Record<string, unknown>) {
+      mockPreprocessExecution.mockResolvedValueOnce({ success: false, error })
+      return executeWorkflowJob({
+        principal,
+        workflowId: 'workflow-1',
+        userId: 'actor-1',
+        workspaceId: 'workspace-1',
+        triggerType: 'api',
+        executionId: 'execution-rejected',
+        requestId: 'request-rejected',
+        billingAttribution,
+      })
+    }
+
+    it.each([
+      { statusCode: 404, message: 'Workflow not found' },
+      { statusCode: 403, message: 'Workflow is not deployed' },
+      { statusCode: 402, message: 'Usage limit exceeded. Please upgrade your plan to continue.' },
+      { statusCode: 429, message: 'Rate limit exceeded. Please try again later.' },
+    ])('completes the job for a non-retryable $statusCode', async ({ statusCode, message }) => {
+      loggingSessionMockFns.mockHasCompleted.mockReturnValueOnce(true)
+      await expect(runRejected({ statusCode, message })).resolves.toMatchObject({
+        success: false,
+        workflowId: 'workflow-1',
+        executionId: 'execution-rejected',
+        error: message,
+      })
+      expect(loggingSessionMockFns.mockSafeCompleteWithError).toHaveBeenCalledWith(
+        expect.objectContaining({ error: expect.objectContaining({ message }) })
+      )
+    })
+
+    it('fails the job when the rejection could not be logged', async () => {
+      loggingSessionMockFns.mockHasCompleted.mockReturnValueOnce(false)
+
+      await expect(
+        runRejected({ statusCode: 403, message: 'Workflow is not deployed' })
+      ).rejects.toThrow('Workflow is not deployed')
+    })
+
+    it.each([
+      { statusCode: 429, message: 'Too many concurrent executions', retryable: true },
+      { statusCode: 402, message: 'Usage is temporarily unavailable', retryable: true },
+      { statusCode: 503, message: 'Usage admission is temporarily unavailable', retryable: true },
+      { statusCode: 500, message: 'Internal error while fetching workflow' },
+    ])('fails the job for $statusCode $message', async (error) => {
+      await expect(runRejected(error)).rejects.toThrow(error.message)
+    })
+  })
+
   it('routes retryable reservation concurrency through bounded infrastructure backoff', async () => {
     mockPreprocessExecution.mockResolvedValueOnce({
       success: false,

apps/sim/background/schedule-execution.ts

@@ -34,11 +34,9 @@ import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
 import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server'
-import {
-  executeWorkflowCore,
-  wasExecutionFinalizedByCore,
-} from '@/lib/workflows/executor/execution-core'
+import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
 import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence'
+import { isRecordedWorkflowFailure } from '@/lib/workflows/executor/recorded-failure'
 import { loadDeployedWorkflowState } from '@/lib/workflows/persistence/utils'
 import { notifyScheduleAutoDisabled } from '@/lib/workflows/schedules/disable-notifications'
 import type { ScheduleDisableReason } from '@/lib/workflows/schedules/disable-reasons'
@@ -730,7 +728,7 @@ async function runWorkflowExecution({
       }
     }
 
-    if (wasExecutionFinalizedByCore(error, executionId)) {
+    if (await isRecordedWorkflowFailure(error, executionId, loggingSession)) {
       throw error
     }
 

apps/sim/background/webhook-execution.ts

@@ -74,9 +74,9 @@ import { readSlackStreamResponseConfig } from '@/lib/webhooks/slack-stream-confi
 import {
   executeWorkflowCore,
   type PreloadedExecutionEnvironment,
-  wasExecutionFinalizedByCore,
 } from '@/lib/workflows/executor/execution-core'
 import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence'
+import { isRecordedWorkflowFailure } from '@/lib/workflows/executor/recorded-failure'
 import {
   loadDeployedWorkflowState,
   loadWorkflowDeploymentVersionState,
@@ -1277,15 +1277,11 @@ async function executeWebhookJobInternal(
       executionId,
     })
 
-    // The finalized flag is set inside a fire-and-forget post-execution promise; await it so the
-    // signal is reliable and the failure is fully persisted before we decide fault vs error.
-    await loggingSession.waitForPostExecution()
-
     // A failure inside workflow execution (block error, provider 4xx, missing required field, etc.)
     // is finalized by core and already recorded in the execution logs. That is a user/workflow error,
     // not a trigger.dev job fault — complete the run normally so we don't fire a false alert. Errors
     // that were not finalized came from the webhook pipeline itself, so we re-throw to fault below.
-    if (wasExecutionFinalizedByCore(error, executionId)) {
+    if (await isRecordedWorkflowFailure(error, executionId, loggingSession)) {
       return {
         success: false,
         workflowId: payload.workflowId,

apps/sim/background/workflow-execution.ts

@@ -1,3 +1,4 @@
+import { trace } from '@opentelemetry/api'
 import {
   parsePrincipal,
   type SerializedPrincipalV1,
@@ -15,6 +16,7 @@ import {
   assertBillingAttributionSnapshot,
   type BillingAttributionSnapshot,
 } from '@/lib/billing/core/attribution'
+import { buildRecordedJobFailureResult } from '@/lib/core/async-jobs/recorded-failure'
 import type { AsyncExecutionCorrelation } from '@/lib/core/async-jobs/types'
 import { logFailureOnce } from '@/lib/core/errors/failure-log'
 import {
@@ -27,15 +29,13 @@ import {
   RESERVATION_TTL_BUFFER_MS,
 } from '@/lib/core/execution-limits'
 import type { RequestAttribution } from '@/lib/core/utils/request-attribution'
-import { preprocessExecution } from '@/lib/execution/preprocessing'
+import { type PreprocessExecutionError, preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
 import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server'
-import {
-  executeWorkflowCore,
-  wasExecutionFinalizedByCore,
-} from '@/lib/workflows/executor/execution-core'
+import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
 import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence'
+import { isRecordedWorkflowFailure } from '@/lib/workflows/executor/recorded-failure'
 import { WORKFLOW_EXECUTION_CONCURRENCY_LIMIT } from '@/background/concurrency-limits'
 import { ExecutionSnapshot } from '@/executor/execution/snapshot'
 import type { ExecutionMetadata } from '@/executor/execution/types'
@@ -104,6 +104,28 @@ type LegacyWorkflowExecutionPayload = Omit<WorkflowExecutionPayload, 'principal'
 
 type WorkflowExecutionJobPayload = WorkflowExecutionPayload | LegacyWorkflowExecutionPayload
 
+/**
+ * Preprocessing rejections that are an outcome of the workflow's or account's
+ * state — deleted or undeployed workflow, suspended account, exhausted usage
+ * or rate quota — rather than a fault. A retryable rejection (concurrency
+ * capacity, an unreadable usage ledger, reservation infrastructure) and every
+ * 5xx still fail the job.
+ */
+const NON_FAULT_PREPROCESSING_STATUS_CODES: ReadonlySet<number> = new Set([402, 403, 404, 429])
+
+function isNonFaultPreprocessingRejection(rejection: PreprocessExecutionError): boolean {
+  return (
+    NON_FAULT_PREPROCESSING_STATUS_CODES.has(rejection.statusCode) && rejection.retryable !== true
+  )
+}
+
+/** Completes the job with a failure already in the execution log, keeping it on the run's span. */
+function completeWithRecordedFailure(workflowId: string, executionId: string, error: unknown) {
+  const failure = toError(error)
+  trace.getActiveSpan()?.recordException(failure)
+  return buildRecordedJobFailureResult({ workflowId, executionId, error: failure.message })
+}
+
 function requireLegacyWorkflowJobString(value: unknown, field: string): string {
   if (typeof value !== 'string' || !value.trim()) {
     throw new Error(`Legacy workflow job ${field} must be a non-empty string`)
@@ -194,6 +216,7 @@ export async function executeWorkflowJob(
         loggingSession.setTrustedExecutionCorrelation(correlation)
       }
       loggingSession.setExecutionDeadlineAt(getExecutionDeadlineAt(timeoutController.signal))
+      let preprocessingRejection: PreprocessExecutionError | undefined
 
       try {
         const preprocessResult = await preprocessExecution({
@@ -217,6 +240,7 @@ export async function executeWorkflowJob(
             workflowId,
             statusCode: preprocessResult.error?.statusCode,
           })
+          preprocessingRejection = preprocessResult.error
 
           throw new Error(preprocessResult.error?.message || 'Preprocessing failed')
         }
@@ -314,8 +338,11 @@ export async function executeWorkflowJob(
 
         if (error instanceof ExecutionTimeoutError) throw error
 
-        if (wasExecutionFinalizedByCore(error, executionId)) {
-          throw error
+        // A failure core already recorded is the workflow's outcome, not a job fault; a run past
+        // its deadline (checked after the await, which can outlast it) still fails the job.
+        if (await isRecordedWorkflowFailure(error, executionId, loggingSession)) {
+          if (timeoutController.isTimedOut()) throw error
+          return completeWithRecordedFailure(workflowId, executionId, error)
         }
 
         const executionResult = hasExecutionResult(error) ? error.executionResult : undefined
@@ -332,6 +359,15 @@ export async function executeWorkflowJob(
           executionState: executionResult?.executionState,
         })
 
+        // Complete only when the rejection is durably logged; an unsaved one must still alert.
+        if (
+          preprocessingRejection &&
+          isNonFaultPreprocessingRejection(preprocessingRejection) &&
+          loggingSession.hasCompleted()
+        ) {
+          return completeWithRecordedFailure(workflowId, executionId, error)
+        }
+
         throw error
       } finally {
         void cleanupExecutionBase64Cache(executionId)

apps/sim/lib/core/async-jobs/backends/database.test.ts

@@ -11,6 +11,7 @@ vi.mock('@/lib/core/execution-limits/metrics', () => ({
 }))
 
 import { DatabaseJobQueue } from '@/lib/core/async-jobs/backends/database'
+import { buildRecordedJobFailureResult } from '@/lib/core/async-jobs/recorded-failure'
 import { AsyncJobEnqueueError, MAX_JOB_DURATION_SECONDS } from '@/lib/core/async-jobs/types'
 
 const EXISTING_JOB = {
@@ -450,3 +451,55 @@ describe('DatabaseJobQueue cancellation', () => {
     })
   })
 })
+
+describe('DatabaseJobQueue a workflow job that completed with a recorded failure', () => {
+  beforeEach(() => {
+    resetDbChainMock()
+  })
+
+  const settledJob = {
+    ...EXISTING_JOB,
+    startedAt: new Date('2026-07-10T00:00:01.000Z'),
+    completedAt: new Date('2026-07-10T00:00:04.000Z'),
+    attempts: 1,
+  }
+
+  it('reads exactly as the row markJobFailed wrote before the runner stopped throwing', async () => {
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      { ...settledJob, status: 'failed', error: 'No start block found', output: null },
+    ])
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      {
+        ...settledJob,
+        status: 'completed',
+        error: null,
+        output: buildRecordedJobFailureResult({
+          workflowId: 'workflow-1',
+          executionId: 'execution-1',
+          error: 'No start block found',
+        }),
+      },
+    ])
+    const queue = new DatabaseJobQueue()
+
+    const failedJob = await queue.getJob('workflow:1')
+    const recordedFailure = await queue.getJob('workflow:1')
+
+    expect(failedJob).toMatchObject({ status: 'failed', error: 'No start block found' })
+    expect(recordedFailure).toStrictEqual(failedJob)
+  })
+
+  it('leaves a completed webhook job that reported an unsuccessful execution completed', async () => {
+    const output = { success: false, workflowId: 'workflow-1', executionId: 'execution-1' }
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      { ...settledJob, type: 'webhook-execution', status: 'completed', output },
+    ])
+    const queue = new DatabaseJobQueue()
+
+    await expect(queue.getJob('workflow:1')).resolves.toMatchObject({
+      status: 'completed',
+      error: undefined,
+      output,
+    })
+  })
+})