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

@@ -11,12 +11,14 @@ const {
   handlePostExecutionPauseStateMock,
   loggingSessionConstructorMock,
   safeStartMock,
+  waitForPostExecutionMock,
 } = vi.hoisted(() => ({
   captureServerEventMock: vi.fn(),
   executeWorkflowCoreMock: vi.fn(),
   handlePostExecutionPauseStateMock: vi.fn(),
   loggingSessionConstructorMock: vi.fn(),
   safeStartMock: vi.fn(),
+  waitForPostExecutionMock: vi.fn(),
 }))
 
 vi.mock('@sim/utils/id', () => ({
@@ -26,6 +28,7 @@ vi.mock('@sim/utils/id', () => ({
 vi.mock('@/lib/logs/execution/logging-session', () => ({
   LoggingSession: class {
     safeStart = safeStartMock
+    waitForPostExecution = waitForPostExecutionMock
 
     constructor(...args: unknown[]) {
       loggingSessionConstructorMock(...args)
@@ -75,10 +78,11 @@ const workflow = {
   variables: {},
 }
 
-describe('executeWorkflow billing attribution', () => {
+describe('executeWorkflow', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     safeStartMock.mockResolvedValue(true)
+    waitForPostExecutionMock.mockResolvedValue(undefined)
     handlePostExecutionPauseStateMock.mockResolvedValue(undefined)
     executeWorkflowCoreMock.mockImplementation(
       async (params: {
@@ -186,4 +190,89 @@ describe('executeWorkflow billing attribution', () => {
       expect.objectContaining({ trustedInitialResolvedSecretTraceProvenance: provenance })
     )
   })
+  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()
+  })
 })

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
@@ -104,6 +105,7 @@ export async function executeWorkflow(
   const executionId = providedExecutionId || generateId()
   const triggerType = streamConfig?.workflowTriggerType || 'api'
   const loggingSession = new LoggingSession(workflowId, executionId, triggerType, requestId)
+  let postExecutionOwnershipTransferred = false
 
   try {
     const metadata: ExecutionMetadata = {
@@ -201,6 +203,7 @@ export async function executeWorkflow(
     await handlePostExecutionPauseState({ result, workflowId, executionId, loggingSession })
 
     if (streamConfig?.skipLoggingComplete) {
+      postExecutionOwnershipTransferred = true
       return {
         ...result,
         _streamingMetadata: {
@@ -227,5 +230,9 @@ export async function executeWorkflow(
     )
 
     throw error
+  } finally {
+    if (!postExecutionOwnershipTransferred) {
+      await loggingSession.waitForPostExecution()
+    }
   }
 }