diff --git a/.agents/skills/ship/SKILL.md b/.agents/skills/ship/SKILL.md index d1bfd8b4a0e..8f144a04aa6 100644 --- a/.agents/skills/ship/SKILL.md +++ b/.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 diff --git a/.claude/commands/ship.md b/.claude/commands/ship.md index 326abe9fcb3..6b673b18f8e 100644 --- a/.claude/commands/ship.md +++ b/.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 diff --git a/.cursor/commands/ship.md b/.cursor/commands/ship.md index 58ceae2ccae..77ec67d04a8 100644 --- a/.cursor/commands/ship.md +++ b/.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 diff --git a/apps/sim/lib/workflows/executor/execute-workflow.test.ts b/apps/sim/lib/workflows/executor/execute-workflow.test.ts index 324fcd6b5a1..447bde9c047 100644 --- a/apps/sim/lib/workflows/executor/execute-workflow.test.ts +++ b/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 = {}) => ({ ...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((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((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', diff --git a/apps/sim/lib/workflows/executor/execute-workflow.ts b/apps/sim/lib/workflows/executor/execute-workflow.ts index 6810fb48a4a..d054976c871 100644 --- a/apps/sim/lib/workflows/executor/execute-workflow.ts +++ b/apps/sim/lib/workflows/executor/execute-workflow.ts @@ -36,6 +36,7 @@ export interface ExecuteWorkflowOptions { executionOrder: number ) => Promise onBlockComplete?: (blockId: string, output: unknown) => Promise + /** 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() + } } }