fix(trigger): keep alerting on deadline-passed and unlogged async workflow failures

Re-check the deadline after awaiting core's finalization, since that wait can
outlast it, so a timed-out run still fails its job. Complete a non-fault
preprocessing rejection only when the logging session confirms the failure
row was written (hasCompleted, the same signal core uses to mark a failure
finalized); otherwise throw as before so an unsaved rejection still alerts.

Author: waleedlatif1

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

@@ -414,6 +414,20 @@ describe('async preprocessing correlation threading', () => {
       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)
+    })
+
     it('still fails the job when the run passed its deadline', async () => {
       const error = recordedError()
       mockExecuteWorkflowCore.mockRejectedValueOnce(error)
@@ -573,6 +587,7 @@ describe('async preprocessing correlation threading', () => {
       { 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',
@@ -584,6 +599,14 @@ describe('async preprocessing correlation threading', () => {
       )
     })
 
+    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 },

apps/sim/background/workflow-execution.ts

@@ -332,10 +332,10 @@ export async function executeWorkflowJob(
         if (error instanceof ExecutionTimeoutError) throw error
 
         // A failure core already recorded is the workflow's outcome, not a job fault; a run past
-        // its deadline still fails the job so timeouts keep alerting.
+        // its deadline (checked after the await, which can outlast it) still fails the job.
         if (
-          !timeoutController.isTimedOut() &&
-          (await isRecordedWorkflowFailure(error, executionId, loggingSession))
+          (await isRecordedWorkflowFailure(error, executionId, loggingSession)) &&
+          !timeoutController.isTimedOut()
         ) {
           trace.getActiveSpan()?.recordException(toError(error))
           return buildRecordedJobFailureResult({
@@ -359,7 +359,12 @@ export async function executeWorkflowJob(
           executionState: executionResult?.executionState,
         })
 
-        if (preprocessingRejection && isNonFaultPreprocessingRejection(preprocessingRejection)) {
+        // Complete only when the rejection is durably logged; an unsaved one must still alert.
+        if (
+          preprocessingRejection &&
+          isNonFaultPreprocessingRejection(preprocessingRejection) &&
+          loggingSession.hasCompleted()
+        ) {
           trace.getActiveSpan()?.recordException(toError(error))
           return buildRecordedJobFailureResult({
             workflowId,