fix(workflows): await execution log finalization (#6309)

Author: TheodoreSpeaks

.agents/skills/ship/SKILL.md

@@ -69,7 +69,8 @@ When the user runs `/ship`:
   for s in check:boundaries check:api-validation:strict check:desktop-bridge check:desktop-ipc \
            check:utils check:zustand-v5 \
            check:react-query check:client-boundary check:bare-icons check:icon-paths \
-           check:realtime-prune check:tool-registry-boundary tool-metadata:check \
+           check:realtime-prune check:tool-registry-boundary check:tool-request-boundary \
+           tool-metadata:check \
            integration-catalog:check skills:check agent-stream-docs:check; do
     ( bun run "$s" >"/tmp/ship-audit-${s//:/-}.log" 2>&1; echo "$? $s" >>/tmp/ship-audit-results ) &
   done

.claude/commands/ship.md

@@ -68,7 +68,8 @@ When the user runs `/ship`:
   for s in check:boundaries check:api-validation:strict check:desktop-bridge check:desktop-ipc \
            check:utils check:zustand-v5 \
            check:react-query check:client-boundary check:bare-icons check:icon-paths \
-           check:realtime-prune check:tool-registry-boundary tool-metadata:check \
+           check:realtime-prune check:tool-registry-boundary check:tool-request-boundary \
+           tool-metadata:check \
            integration-catalog:check skills:check agent-stream-docs:check; do
     ( bun run "$s" >"/tmp/ship-audit-${s//:/-}.log" 2>&1; echo "$? $s" >>/tmp/ship-audit-results ) &
   done

.cursor/commands/ship.md

@@ -63,7 +63,8 @@ When the user runs `/ship`:
   for s in check:boundaries check:api-validation:strict check:desktop-bridge check:desktop-ipc \
            check:utils check:zustand-v5 \
            check:react-query check:client-boundary check:bare-icons check:icon-paths \
-           check:realtime-prune check:tool-registry-boundary tool-metadata:check \
+           check:realtime-prune check:tool-registry-boundary check:tool-request-boundary \
+           tool-metadata:check \
            integration-catalog:check skills:check agent-stream-docs:check; do
     ( bun run "$s" >"/tmp/ship-audit-${s//:/-}.log" 2>&1; echo "$? $s" >>/tmp/ship-audit-results ) &
   done

apps/sim/lib/workflows/executor/execute-workflow.test.ts

@@ -13,6 +13,7 @@ const {
   loggingSessionConstructorMock,
   projectDiagnosticErrorMock,
   safeStartMock,
+  waitForPostExecutionMock,
   setTrustedExecutionCorrelationMock,
 } = vi.hoisted(() => ({
   captureServerEventMock: vi.fn(),
@@ -21,6 +22,7 @@ const {
   loggingSessionConstructorMock: vi.fn(),
   projectDiagnosticErrorMock: vi.fn(),
   safeStartMock: vi.fn(),
+  waitForPostExecutionMock: vi.fn(),
   setTrustedExecutionCorrelationMock: vi.fn(),
 }))
 
@@ -32,6 +34,7 @@ vi.mock('@/lib/logs/execution/logging-session', () => ({
   LoggingSession: class {
     projectDiagnosticError = projectDiagnosticErrorMock
     safeStart = safeStartMock
+    waitForPostExecution = waitForPostExecutionMock
     setTrustedExecutionCorrelation = setTrustedExecutionCorrelationMock
 
     constructor(...args: unknown[]) {
@@ -89,10 +92,11 @@ const workflow = {
   variables: {},
 }
 
-describe('executeWorkflow billing attribution', () => {
+describe('executeWorkflow', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     safeStartMock.mockResolvedValue(true)
+    waitForPostExecutionMock.mockResolvedValue(undefined)
     projectDiagnosticErrorMock.mockImplementation(
       (error: unknown, details: Record<string, unknown> = {}) => ({
         ...details,
@@ -208,6 +212,92 @@ describe('executeWorkflow billing attribution', () => {
     )
   })
 
+  it('waits for post-execution persistence before resolving', async () => {
+    let resolvePostExecution!: () => void
+    waitForPostExecutionMock.mockReturnValueOnce(
+      new Promise<void>((resolve) => {
+        resolvePostExecution = resolve
+      })
+    )
+
+    let executionSettled = false
+    const executionPromise = executeWorkflow(
+      workflow,
+      'request-1',
+      { prompt: 'hello' },
+      'actor-1',
+      {
+        enabled: true,
+        billingAttribution,
+      }
+    ).then((result) => {
+      executionSettled = true
+      return result
+    })
+
+    await vi.waitFor(() => expect(waitForPostExecutionMock).toHaveBeenCalledOnce())
+    expect(executionSettled).toBe(false)
+
+    resolvePostExecution()
+    await executionPromise
+
+    expect(executionSettled).toBe(true)
+  })
+
+  it('waits for post-execution persistence before rejecting', async () => {
+    const executionError = new Error('Request body size limit exceeded (10MB)')
+    executeWorkflowCoreMock.mockRejectedValueOnce(executionError)
+
+    let resolvePostExecution!: () => void
+    waitForPostExecutionMock.mockReturnValueOnce(
+      new Promise<void>((resolve) => {
+        resolvePostExecution = resolve
+      })
+    )
+
+    let executionSettled = false
+    const executionPromise = executeWorkflow(workflow, 'request-1', undefined, 'actor-1', {
+      enabled: true,
+      billingAttribution,
+    }).catch((error: unknown) => {
+      executionSettled = true
+      throw error
+    })
+
+    await vi.waitFor(() => expect(waitForPostExecutionMock).toHaveBeenCalledOnce())
+    expect(executionSettled).toBe(false)
+
+    resolvePostExecution()
+    await expect(executionPromise).rejects.toBe(executionError)
+    expect(executionSettled).toBe(true)
+  })
+
+  it('transfers post-execution ownership with successful streaming metadata', async () => {
+    const result = await executeWorkflow(workflow, 'request-1', undefined, 'actor-1', {
+      enabled: true,
+      skipLoggingComplete: true,
+      billingAttribution,
+    })
+
+    expect(waitForPostExecutionMock).not.toHaveBeenCalled()
+    expect(result._streamingMetadata?.loggingSession).toBeDefined()
+  })
+
+  it('retains post-execution ownership when streaming execution rejects', async () => {
+    const executionError = new Error('Streaming execution failed')
+    executeWorkflowCoreMock.mockRejectedValueOnce(executionError)
+
+    await expect(
+      executeWorkflow(workflow, 'request-1', undefined, 'actor-1', {
+        enabled: true,
+        skipLoggingComplete: true,
+        billingAttribution,
+      })
+    ).rejects.toBe(executionError)
+
+    expect(waitForPostExecutionMock).toHaveBeenCalledOnce()
+  })
+
   it('persists server-issued workflow-group correlation in execution metadata', async () => {
     const correlation = {
       executionId: 'execution-1',

apps/sim/lib/workflows/executor/execute-workflow.ts

@@ -36,6 +36,7 @@ export interface ExecuteWorkflowOptions {
     executionOrder: number
   ) => Promise<void>
   onBlockComplete?: (blockId: string, output: unknown) => Promise<void>
+  /** Transfers post-execution logging ownership to the streaming caller after execution succeeds. */
   skipLoggingComplete?: boolean
   includeFileBase64?: boolean
   base64MaxBytes?: number
@@ -109,6 +110,7 @@ export async function executeWorkflow(
   if (streamConfig?.trustedExecutionCorrelation) {
     loggingSession.setTrustedExecutionCorrelation(streamConfig.trustedExecutionCorrelation)
   }
+  let postExecutionOwnershipTransferred = false
 
   try {
     const metadata: ExecutionMetadata = {
@@ -207,6 +209,7 @@ export async function executeWorkflow(
     await handlePostExecutionPauseState({ result, workflowId, executionId, loggingSession })
 
     if (streamConfig?.skipLoggingComplete) {
+      postExecutionOwnershipTransferred = true
       return {
         ...result,
         _streamingMetadata: {
@@ -237,5 +240,9 @@ export async function executeWorkflow(
     )
 
     throw error
+  } finally {
+    if (!postExecutionOwnershipTransferred) {
+      await loggingSession.waitForPostExecution()
+    }
   }
 }