fix(trigger): complete runs whose workflow was deleted or undeployed before they started

Not-found, archived, undeployed, unattached and suspended rejections come
from a successful read and never wait on an execution log (not-found and
undeployed never write one), so they complete without the hasCompleted gate.
Quota rejections (402/429) still complete only once their log is saved.

Author: waleedlatif1

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

@@ -683,28 +683,51 @@ describe('async preprocessing correlation threading', () => {
 
     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({
+      { statusCode: 403, message: 'Workflow is not deployed', code: 'WORKFLOW_NOT_DEPLOYED' },
+      {
+        statusCode: 403,
+        message:
+          'This workflow is not attached to a workspace. Personal workflows are deprecated and cannot execute.',
+      },
+      { statusCode: 403, message: 'Account suspended', code: 'ACCOUNT_SUSPENDED' },
+    ])('completes the job for a $statusCode $message that never started a log', async (error) => {
+      loggingSessionMockFns.mockHasCompleted.mockReturnValueOnce(false)
+
+      await expect(runRejected(error)).resolves.toMatchObject({
         success: false,
         workflowId: 'workflow-1',
         executionId: 'execution-rejected',
-        error: message,
+        error: error.message,
       })
-      expect(loggingSessionMockFns.mockSafeCompleteWithError).toHaveBeenCalledWith(
-        expect.objectContaining({ error: expect.objectContaining({ message }) })
-      )
     })
 
-    it('fails the job when the rejection could not be logged', async () => {
+    it.each([
+      { 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 logged $statusCode quota rejection',
+      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.each([
+      { statusCode: 402, message: 'Usage limit exceeded. Please upgrade your plan to continue.' },
+      { statusCode: 429, message: 'Rate limit exceeded. Please try again later.' },
+    ])('fails the job when a $statusCode quota rejection could not be logged', async (error) => {
       loggingSessionMockFns.mockHasCompleted.mockReturnValueOnce(false)
 
-      await expect(
-        runRejected({ statusCode: 403, message: 'Workflow is not deployed' })
-      ).rejects.toThrow('Workflow is not deployed')
+      await expect(runRejected(error)).rejects.toThrow(error.message)
     })
 
     it.each([

apps/sim/background/workflow-execution.ts

@@ -109,18 +109,34 @@ 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
+ * Rejections read from the workflow's own state — deleted, archived, undeployed,
+ * not attached to a workspace, or a suspended account. Preprocessing returns
+ * these only after a successful read (a failed read is a 5xx), and not-found
+ * and undeployed rejections never write an execution log, so the job completes
+ * without waiting for one.
+ */
+const WORKFLOW_STATE_REJECTION_STATUS_CODES: ReadonlySet<number> = new Set([403, 404])
+
+/**
+ * Exhausted usage or rate quota. These rejections write an execution log, so
+ * the job completes only once that log is saved; an unsaved one (an outage)
+ * still fails the job and alerts.
+ */
+const QUOTA_REJECTION_STATUS_CODES: ReadonlySet<number> = new Set([402, 429])
+
+/**
+ * Whether a preprocessing rejection is an outcome of the workflow's or
+ * account's state 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
-  )
+function isNonFaultPreprocessingRejection(
+  rejection: PreprocessExecutionError,
+  rejectionLogged: boolean
+): boolean {
+  if (rejection.retryable === true) return false
+  if (WORKFLOW_STATE_REJECTION_STATUS_CODES.has(rejection.statusCode)) return true
+  return QUOTA_REJECTION_STATUS_CODES.has(rejection.statusCode) && rejectionLogged
 }
 
 /** Completes the job with a failure already in the execution log, keeping it on the run's span. */
@@ -378,11 +394,9 @@ export async function executeWorkflowJob(
         const scopeDeletedOnCompletion = loggingSession.getDeletedScope()
         if (scopeDeletedOnCompletion) return completeWithDeletedScope(scopeDeletedOnCompletion)
 
-        // Complete only when the rejection is durably logged; an unsaved one must still alert.
         if (
           preprocessingRejection &&
-          isNonFaultPreprocessingRejection(preprocessingRejection) &&
-          loggingSession.hasCompleted()
+          isNonFaultPreprocessingRejection(preprocessingRejection, loggingSession.hasCompleted())
         ) {
           return completeWithRecordedFailure(workflowId, executionId, error)
         }