fix(execution): finalize runs before wrapper recovery

Author: PlaneInABottle

apps/sim/background/schedule-execution.ts

@@ -8,7 +8,10 @@ import { createTimeoutAbortController, getTimeoutErrorMessage } from '@/lib/core
 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 { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
+import {
+  executeWorkflowCore,
+  wasExecutionFinalizedByCore,
+} from '@/lib/workflows/executor/execution-core'
 import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager'
 import {
   blockExistsInDeployment,
@@ -249,6 +252,10 @@ async function runWorkflowExecution({
   } catch (error: unknown) {
     logger.error(`[${requestId}] Early failure in scheduled workflow ${payload.workflowId}`, error)
 
+    if (wasExecutionFinalizedByCore(error, executionId)) {
+      throw error
+    }
+
     const executionResult = hasExecutionResult(error) ? error.executionResult : undefined
     const { traceSpans } = executionResult ? buildTraceSpans(executionResult) : { traceSpans: [] }
 

apps/sim/background/webhook-execution.ts

@@ -12,7 +12,10 @@ import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
 import { WebhookAttachmentProcessor } from '@/lib/webhooks/attachment-processor'
 import { fetchAndProcessAirtablePayloads, formatWebhookInput } from '@/lib/webhooks/utils.server'
-import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
+import {
+  executeWorkflowCore,
+  wasExecutionFinalizedByCore,
+} from '@/lib/workflows/executor/execution-core'
 import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager'
 import { loadDeployedWorkflowState } from '@/lib/workflows/persistence/utils'
 import { resolveOAuthAccountId } from '@/app/api/auth/oauth/utils'
@@ -623,6 +626,10 @@ async function executeWebhookJobInternal(
       provider: payload.provider,
     })
 
+    if (wasExecutionFinalizedByCore(error, executionId)) {
+      throw error
+    }
+
     try {
       await loggingSession.safeStart({
         userId: payload.userId,

apps/sim/background/workflow-execution.ts

@@ -5,7 +5,10 @@ import { createTimeoutAbortController, getTimeoutErrorMessage } from '@/lib/core
 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 { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
+import {
+  executeWorkflowCore,
+  wasExecutionFinalizedByCore,
+} from '@/lib/workflows/executor/execution-core'
 import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager'
 import { ExecutionSnapshot } from '@/executor/execution/snapshot'
 import type { ExecutionMetadata } from '@/executor/execution/types'
@@ -178,6 +181,10 @@ export async function executeWorkflowJob(payload: WorkflowExecutionPayload) {
       executionId,
     })
 
+    if (wasExecutionFinalizedByCore(error, executionId)) {
+      throw error
+    }
+
     const executionResult = hasExecutionResult(error) ? error.executionResult : undefined
     const { traceSpans } = executionResult ? buildTraceSpans(executionResult) : { traceSpans: [] }
 

apps/sim/lib/logs/execution/logging-session.test.ts

@@ -0,0 +1,75 @@
+import { describe, expect, it, vi } from 'vitest'
+
+vi.mock('@sim/db', () => ({
+  db: {},
+}))
+
+vi.mock('@sim/db/schema', () => ({
+  workflowExecutionLogs: {},
+}))
+
+vi.mock('@sim/logger', () => ({
+  createLogger: () => ({
+    info: vi.fn(),
+    error: vi.fn(),
+    warn: vi.fn(),
+    debug: vi.fn(),
+  }),
+}))
+
+vi.mock('drizzle-orm', () => ({
+  eq: vi.fn(),
+  sql: vi.fn(),
+}))
+
+vi.mock('@/lib/logs/execution/logger', () => ({
+  executionLogger: {
+    startWorkflowExecution: vi.fn(),
+    completeWorkflowExecution: vi.fn(),
+  },
+}))
+
+vi.mock('@/lib/logs/execution/logging-factory', () => ({
+  calculateCostSummary: vi.fn().mockReturnValue({
+    totalCost: 0,
+    totalInputCost: 0,
+    totalOutputCost: 0,
+    totalTokens: 0,
+    totalPromptTokens: 0,
+    totalCompletionTokens: 0,
+    baseExecutionCharge: 0,
+    modelCost: 0,
+    models: {},
+  }),
+  createEnvironmentObject: vi.fn(),
+  createTriggerObject: vi.fn(),
+  loadDeployedWorkflowStateForLogging: vi.fn(),
+  loadWorkflowStateForExecution: vi.fn(),
+}))
+
+import { LoggingSession } from './logging-session'
+
+describe('LoggingSession completion retries', () => {
+  it('clears failed completion promise so error finalization can retry', async () => {
+    const session = new LoggingSession('workflow-1', 'execution-1', 'api', 'req-1') as any
+
+    const successFinalizeError = new Error('success finalize failed')
+    session.complete = vi.fn().mockRejectedValue(successFinalizeError)
+    session.completeWithCostOnlyLog = vi.fn().mockRejectedValue(successFinalizeError)
+    session.completeWithError = vi.fn().mockResolvedValue(undefined)
+
+    await expect(session.safeComplete({ finalOutput: { ok: true } })).rejects.toThrow(
+      'success finalize failed'
+    )
+
+    await expect(
+      session.safeCompleteWithError({
+        error: { message: 'fallback error finalize' },
+      })
+    ).resolves.toBeUndefined()
+
+    expect(session.complete).toHaveBeenCalledTimes(1)
+    expect(session.completeWithCostOnlyLog).toHaveBeenCalledTimes(1)
+    expect(session.completeWithError).toHaveBeenCalledTimes(1)
+  })
+})

apps/sim/lib/logs/execution/logging-session.ts

@@ -690,7 +690,10 @@ export class LoggingSession {
 
   async safeComplete(params: SessionCompleteParams = {}): Promise<void> {
     if (this.completionPromise) return this.completionPromise
-    this.completionPromise = this._safeCompleteImpl(params)
+    this.completionPromise = this._safeCompleteImpl(params).catch((error) => {
+      this.completionPromise = null
+      throw error
+    })
     return this.completionPromise
   }
 
@@ -715,7 +718,10 @@ export class LoggingSession {
 
   async safeCompleteWithError(params?: SessionErrorCompleteParams): Promise<void> {
     if (this.completionPromise) return this.completionPromise
-    this.completionPromise = this._safeCompleteWithErrorImpl(params)
+    this.completionPromise = this._safeCompleteWithErrorImpl(params).catch((error) => {
+      this.completionPromise = null
+      throw error
+    })
     return this.completionPromise
   }
 
@@ -742,7 +748,10 @@ export class LoggingSession {
 
   async safeCompleteWithCancellation(params?: SessionCancelledParams): Promise<void> {
     if (this.completionPromise) return this.completionPromise
-    this.completionPromise = this._safeCompleteWithCancellationImpl(params)
+    this.completionPromise = this._safeCompleteWithCancellationImpl(params).catch((error) => {
+      this.completionPromise = null
+      throw error
+    })
     return this.completionPromise
   }
 
@@ -768,7 +777,10 @@ export class LoggingSession {
 
   async safeCompleteWithPause(params?: SessionPausedParams): Promise<void> {
     if (this.completionPromise) return this.completionPromise
-    this.completionPromise = this._safeCompleteWithPauseImpl(params)
+    this.completionPromise = this._safeCompleteWithPauseImpl(params).catch((error) => {
+      this.completionPromise = null
+      throw error
+    })
     return this.completionPromise
   }