fix(async): keep execution finalization state consistent

Retry minimal logging for early failures, only mark core finalization after a log row actually completes, and let paused completions fall back cleanly.

Author: PlaneInABottle

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

@@ -89,4 +89,49 @@ describe('LoggingSession completion retries', () => {
 
     expect(completeWorkflowExecutionMock).toHaveBeenCalledTimes(2)
   })
+
+  it('marks paused completions as completed and deduplicates later attempts', async () => {
+    const session = new LoggingSession('workflow-1', 'execution-1', 'api', 'req-1')
+
+    completeWorkflowExecutionMock.mockResolvedValue({})
+
+    await expect(
+      session.safeCompleteWithPause({
+        endedAt: new Date().toISOString(),
+        totalDurationMs: 10,
+        traceSpans: [],
+        workflowInput: { hello: 'world' },
+      })
+    ).resolves.toBeUndefined()
+
+    expect(session.hasCompleted()).toBe(true)
+
+    await expect(
+      session.safeCompleteWithError({
+        error: { message: 'should be ignored' },
+      })
+    ).resolves.toBeUndefined()
+
+    expect(completeWorkflowExecutionMock).toHaveBeenCalledTimes(1)
+  })
+
+  it('falls back to cost-only logging when paused completion fails', async () => {
+    const session = new LoggingSession('workflow-1', 'execution-2', 'api', 'req-1')
+
+    completeWorkflowExecutionMock
+      .mockRejectedValueOnce(new Error('pause finalize failed'))
+      .mockResolvedValueOnce({})
+
+    await expect(
+      session.safeCompleteWithPause({
+        endedAt: new Date().toISOString(),
+        totalDurationMs: 10,
+        traceSpans: [],
+        workflowInput: { hello: 'world' },
+      })
+    ).resolves.toBeUndefined()
+
+    expect(session.hasCompleted()).toBe(true)
+    expect(completeWorkflowExecutionMock).toHaveBeenCalledTimes(2)
+  })
 })

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

@@ -549,6 +549,11 @@ export class LoggingSession {
   }
 
   async completeWithPause(params: SessionPausedParams = {}): Promise<void> {
+    if (this.completed || this.completing) {
+      return
+    }
+    this.completing = true
+
     try {
       const { endedAt, totalDurationMs, traceSpans, workflowInput } = params
 
@@ -580,6 +585,8 @@ export class LoggingSession {
         status: 'pending',
       })
 
+      this.completed = true
+
       try {
         const { PlatformEvents, createOTelSpansForWorkflowExecution } = await import(
           '@/lib/core/telemetry'
@@ -616,6 +623,7 @@ export class LoggingSession {
         )
       }
     } catch (pauseError) {
+      this.completing = false
       logger.error(`Failed to complete paused logging for execution ${this.executionId}:`, {
         requestId: this.requestId,
         workflowId: this.workflowId,
@@ -699,6 +707,10 @@ export class LoggingSession {
     }
   }
 
+  hasCompleted(): boolean {
+    return this.completed
+  }
+
   private shouldStartNewCompletionAttempt(attempt: CompletionAttempt): boolean {
     return this.completionAttemptFailed && this.completionAttempt !== 'error' && attempt === 'error'
   }

apps/sim/lib/workflows/executor/execution-core.test.ts

@@ -10,6 +10,7 @@ const {
   safeCompleteWithErrorMock,
   safeCompleteWithCancellationMock,
   safeCompleteWithPauseMock,
+  hasCompletedMock,
   updateWorkflowRunCountsMock,
   clearExecutionCancellationMock,
   buildTraceSpansMock,
@@ -25,6 +26,7 @@ const {
   safeCompleteWithErrorMock: vi.fn(),
   safeCompleteWithCancellationMock: vi.fn(),
   safeCompleteWithPauseMock: vi.fn(),
+  hasCompletedMock: vi.fn(),
   updateWorkflowRunCountsMock: vi.fn(),
   clearExecutionCancellationMock: vi.fn(),
   buildTraceSpansMock: vi.fn(),
@@ -102,6 +104,7 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
     safeCompleteWithError: safeCompleteWithErrorMock,
     safeCompleteWithCancellation: safeCompleteWithCancellationMock,
     safeCompleteWithPause: safeCompleteWithPauseMock,
+    hasCompleted: hasCompletedMock,
   }
 
   const createSnapshot = () => ({
@@ -165,11 +168,12 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
     mergeSubblockStateWithValuesMock.mockImplementation((blocks) => blocks)
     serializeWorkflowMock.mockReturnValue({ loops: {}, parallels: {} })
     buildTraceSpansMock.mockReturnValue({ traceSpans: [{ id: 'span-1' }], totalDuration: 123 })
-    safeStartMock.mockResolvedValue(undefined)
+    safeStartMock.mockResolvedValue(true)
     safeCompleteMock.mockResolvedValue(undefined)
     safeCompleteWithErrorMock.mockResolvedValue(undefined)
     safeCompleteWithCancellationMock.mockResolvedValue(undefined)
     safeCompleteWithPauseMock.mockResolvedValue(undefined)
+    hasCompletedMock.mockReturnValue(true)
     updateWorkflowRunCountsMock.mockResolvedValue(undefined)
     clearExecutionCancellationMock.mockResolvedValue(undefined)
   })
@@ -444,4 +448,77 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
 
     expect(safeCompleteWithErrorMock).toHaveBeenCalledTimes(1)
   })
+
+  it('does not mark core finalization when error completion never persists a log row', async () => {
+    const error = new Error('engine failed')
+    executorExecuteMock.mockRejectedValue(error)
+    hasCompletedMock.mockReturnValue(false)
+    const snapshot = {
+      ...createSnapshot(),
+      metadata: {
+        ...createSnapshot().metadata,
+        executionId: 'execution-unfinalized',
+      },
+    }
+
+    await expect(
+      executeWorkflowCore({
+        snapshot: snapshot as any,
+        callbacks: {},
+        loggingSession: loggingSession as any,
+      })
+    ).rejects.toBe(error)
+
+    expect(safeCompleteWithErrorMock).toHaveBeenCalledTimes(1)
+    expect(wasExecutionFinalizedByCore(error, 'execution-unfinalized')).toBe(false)
+  })
+
+  it('starts a minimal log session before error completion when setup fails early', async () => {
+    const envError = new Error('env lookup failed')
+    getPersonalAndWorkspaceEnvMock.mockRejectedValue(envError)
+
+    await expect(
+      executeWorkflowCore({
+        snapshot: createSnapshot() as any,
+        callbacks: {},
+        loggingSession: loggingSession as any,
+      })
+    ).rejects.toBe(envError)
+
+    expect(safeStartMock).toHaveBeenCalledTimes(1)
+    expect(safeStartMock).toHaveBeenCalledWith(
+      expect.objectContaining({
+        userId: 'user-1',
+        workspaceId: 'workspace-1',
+        variables: {},
+      })
+    )
+    expect(safeCompleteWithErrorMock).toHaveBeenCalledTimes(1)
+    expect(wasExecutionFinalizedByCore(envError, 'execution-1')).toBe(true)
+  })
+
+  it('skips core finalization when minimal error logging cannot start', async () => {
+    const envError = new Error('env lookup failed')
+    getPersonalAndWorkspaceEnvMock.mockRejectedValue(envError)
+    safeStartMock.mockResolvedValue(false)
+    const snapshot = {
+      ...createSnapshot(),
+      metadata: {
+        ...createSnapshot().metadata,
+        executionId: 'execution-no-log-start',
+      },
+    }
+
+    await expect(
+      executeWorkflowCore({
+        snapshot: snapshot as any,
+        callbacks: {},
+        loggingSession: loggingSession as any,
+      })
+    ).rejects.toBe(envError)
+
+    expect(safeStartMock).toHaveBeenCalledTimes(1)
+    expect(safeCompleteWithErrorMock).not.toHaveBeenCalled()
+    expect(wasExecutionFinalizedByCore(envError, 'execution-no-log-start')).toBe(false)
+  })
 })

apps/sim/lib/workflows/executor/execution-core.ts

@@ -234,7 +234,7 @@ async function finalizeExecutionError(params: {
       traceSpans,
     })
 
-    return true
+    return loggingSession.hasCompleted()
   } catch (postExecError) {
     logger.error(`[${requestId}] Post-execution error logging failed`, {
       error: postExecError,
@@ -270,13 +270,14 @@ export async function executeWorkflowCore(
   }
 
   let processedInput = input || {}
+  let deploymentVersionId: string | undefined
+  let loggingStarted = false
 
   try {
     let blocks
     let edges: Edge[]
     let loops
     let parallels
-    let deploymentVersionId: string | undefined
 
     // Use workflowStateOverride if provided (for diff workflows)
     if (metadata.workflowStateOverride) {
@@ -333,7 +334,7 @@ export async function executeWorkflowCore(
     // Use already-decrypted values for execution (no redundant decryption)
     const decryptedEnvVars: Record<string, string> = { ...personalDecrypted, ...workspaceDecrypted }
 
-    await loggingSession.safeStart({
+    loggingStarted = await loggingSession.safeStart({
       userId,
       workspaceId: providedWorkspaceId,
       variables,
@@ -520,12 +521,25 @@ export async function executeWorkflowCore(
   } catch (error: unknown) {
     logger.error(`[${requestId}] Execution failed:`, error)
 
-    const finalized = await finalizeExecutionError({
-      error,
-      loggingSession,
-      executionId,
-      requestId,
-    })
+    if (!loggingStarted) {
+      loggingStarted = await loggingSession.safeStart({
+        userId,
+        workspaceId: providedWorkspaceId,
+        variables: {},
+        triggerData: metadata.correlation ? { correlation: metadata.correlation } : undefined,
+        skipLogCreation,
+        deploymentVersionId,
+      })
+    }
+
+    const finalized = loggingStarted
+      ? await finalizeExecutionError({
+          error,
+          loggingSession,
+          executionId,
+          requestId,
+        })
+      : false
 
     if (finalized) {
       markExecutionFinalizedByCore(error, executionId)