Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion .agents/skills/ship/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
3 changes: 2 additions & 1 deletion .claude/commands/ship.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
3 changes: 2 additions & 1 deletion .cursor/commands/ship.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
92 changes: 91 additions & 1 deletion apps/sim/lib/workflows/executor/execute-workflow.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ const {
loggingSessionConstructorMock,
projectDiagnosticErrorMock,
safeStartMock,
waitForPostExecutionMock,
setTrustedExecutionCorrelationMock,
} = vi.hoisted(() => ({
captureServerEventMock: vi.fn(),
Expand All @@ -21,6 +22,7 @@ const {
loggingSessionConstructorMock: vi.fn(),
projectDiagnosticErrorMock: vi.fn(),
safeStartMock: vi.fn(),
waitForPostExecutionMock: vi.fn(),
setTrustedExecutionCorrelationMock: vi.fn(),
}))

Expand All @@ -32,6 +34,7 @@ vi.mock('@/lib/logs/execution/logging-session', () => ({
LoggingSession: class {
projectDiagnosticError = projectDiagnosticErrorMock
safeStart = safeStartMock
waitForPostExecution = waitForPostExecutionMock
setTrustedExecutionCorrelation = setTrustedExecutionCorrelationMock

constructor(...args: unknown[]) {
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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',
Expand Down
7 changes: 7 additions & 0 deletions apps/sim/lib/workflows/executor/execute-workflow.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -109,6 +110,7 @@ export async function executeWorkflow(
if (streamConfig?.trustedExecutionCorrelation) {
loggingSession.setTrustedExecutionCorrelation(streamConfig.trustedExecutionCorrelation)
}
let postExecutionOwnershipTransferred = false

try {
const metadata: ExecutionMetadata = {
Expand Down Expand Up @@ -207,6 +209,7 @@ export async function executeWorkflow(
await handlePostExecutionPauseState({ result, workflowId, executionId, loggingSession })

if (streamConfig?.skipLoggingComplete) {
postExecutionOwnershipTransferred = true
return {
...result,
_streamingMetadata: {
Expand Down Expand Up @@ -237,5 +240,9 @@ export async function executeWorkflow(
)

throw error
} finally {
if (!postExecutionOwnershipTransferred) {
await loggingSession.waitForPostExecution()
}
}
}
Loading