opensource/sim
3
0
Fork 0
mirror of https://github.com/simstudioai/sim.git synced 2026-09-21 13:00:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(logs): add durable execution diagnostics foundation (#3564)

Browse Source 
* fix(logs): persist execution diagnostics markers

Store last-started and last-completed block markers with finalization metadata so later read surfaces can explain how a run ended without reconstructing executor state.

* fix(executor): preserve durable diagnostics ordering

Await only the persistence needed to keep diagnostics durable before terminal completion while keeping callback failures from changing execution behavior.

* fix(logs): preserve fallback diagnostics semantics

Keep successful fallback output and accumulated cost intact while tightening progress-write draining and deduplicating trace span counting for diagnostics helpers.

* fix(api): restore async execute route test mock

Add the missing AuthType export to the hybrid auth mock so the async execution route test exercises the 202 queueing path instead of crashing with a 500 in CI.

* fix(executor): align async block error handling

* fix(logs): tighten marker ordering scope

Allow same-millisecond marker writes to replace prior markers and drop the unused diagnostics read helper so this PR stays focused on persistence rather than unread foundation code.

* fix(logs): remove unused finalization type guard

Drop the unused  helper so this PR only ships the persistence-side status types it actually uses.

* fix(executor): await subflow diagnostics callbacks

Ensure empty-subflow and subflow-error lifecycle callbacks participate in progress-write draining before terminal finalization while still swallowing callback failures.

---------

Co-authored-by: test <test@example.com>
Co-authored-by: Vikhyath Mondreti <vikhyath@simstudio.ai>
This commit is contained in:
PlaneInABottle
2026-03-17 17:24:40 -07:00
committed by GitHub
co-authored by test Vikhyath Mondreti
parent c9f082da1a
commit 67478bbc80
13 changed files with 1219 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/sim/executor/execution/block-executor.ts
+51 -35
View File
@@ -77,7 +77,7 @@ export class BlockExecutor {
if (!isSentinel) {
blockLog = this.createBlockLog(ctx, node.id, block, node)
ctx.blockLogs.push(blockLog)
this.callOnBlockStart(ctx, node, block, blockLog.executionOrder)
await this.callOnBlockStart(ctx, node, block, blockLog.executionOrder)
}
const startTime = performance.now()
@@ -105,7 +105,7 @@ export class BlockExecutor {
}
} catch (error) {
cleanupSelfReference?.()
return this.handleBlockError(
return await this.handleBlockError(
error,
ctx,
node,
@@ -179,7 +179,7 @@ export class BlockExecutor {
const displayOutput = filterOutputForLog(block.metadata?.id || '', normalizedOutput, {
block,
})
this.callOnBlockComplete(
await this.callOnBlockComplete(
ctx,
node,
block,
@@ -195,7 +195,7 @@ export class BlockExecutor {
return normalizedOutput
} catch (error) {
return this.handleBlockError(
return await this.handleBlockError(
error,
ctx,
node,
@@ -226,7 +226,7 @@ export class BlockExecutor {
return this.blockHandlers.find((h) => h.canHandle(block))
}
private handleBlockError(
private async handleBlockError(
error: unknown,
ctx: ExecutionContext,
node: DAGNode,
@@ -236,7 +236,7 @@ export class BlockExecutor {
resolvedInputs: Record<string, any>,
isSentinel: boolean,
phase: 'input_resolution' | 'execution'
): NormalizedBlockOutput {
): Promise<NormalizedBlockOutput> {
const duration = performance.now() - startTime
const errorMessage = normalizeError(error)
const hasResolvedInputs =
@@ -287,7 +287,7 @@ export class BlockExecutor {
? error.childWorkflowInstanceId
: undefined
const displayOutput = filterOutputForLog(block.metadata?.id || '', errorOutput, { block })
this.callOnBlockComplete(
await this.callOnBlockComplete(
ctx,
node,
block,
@@ -439,12 +439,12 @@ export class BlockExecutor {
return redactApiKeys(result)
}
private callOnBlockStart(
private async callOnBlockStart(
ctx: ExecutionContext,
node: DAGNode,
block: SerializedBlock,
executionOrder: number
): void {
): Promise<void> {
const blockId = node.metadata?.originalBlockId ?? node.id
const blockName = block.metadata?.name ?? blockId
const blockType = block.metadata?.id ?? DEFAULTS.BLOCK_TYPE
@@ -452,18 +452,26 @@ export class BlockExecutor {
const iterationContext = getIterationContext(ctx, node?.metadata)
if (this.contextExtensions.onBlockStart) {
this.contextExtensions.onBlockStart(
blockId,
blockName,
blockType,
executionOrder,
iterationContext,
ctx.childWorkflowContext
)
try {
await this.contextExtensions.onBlockStart(
blockId,
blockName,
blockType,
executionOrder,
iterationContext,
ctx.childWorkflowContext
)
} catch (error) {
logger.warn('Block start callback failed', {
blockId,
blockType,
error: error instanceof Error ? error.message : String(error),
})
}
}
}
private callOnBlockComplete(
private async callOnBlockComplete(
ctx: ExecutionContext,
node: DAGNode,
block: SerializedBlock,
@@ -474,7 +482,7 @@ export class BlockExecutor {
executionOrder: number,
endedAt: string,
childWorkflowInstanceId?: string
): void {
): Promise<void> {
const blockId = node.metadata?.originalBlockId ?? node.id
const blockName = block.metadata?.name ?? blockId
const blockType = block.metadata?.id ?? DEFAULTS.BLOCK_TYPE
@@ -482,22 +490,30 @@ export class BlockExecutor {
const iterationContext = getIterationContext(ctx, node?.metadata)
if (this.contextExtensions.onBlockComplete) {
this.contextExtensions.onBlockComplete(
blockId,
blockName,
blockType,
{
input,
output,
executionTime: duration,
startedAt,
executionOrder,
endedAt,
childWorkflowInstanceId,
},
iterationContext,
ctx.childWorkflowContext
)
try {
await this.contextExtensions.onBlockComplete(
blockId,
blockName,
blockType,
{
input,
output,
executionTime: duration,
startedAt,
executionOrder,
endedAt,
childWorkflowInstanceId,
},
iterationContext,
ctx.childWorkflowContext
)
} catch (error) {
logger.warn('Block completion callback failed', {
blockId,
blockType,
error: error instanceof Error ? error.message : String(error),
})
}
}
}
apps/sim/executor/orchestrators/loop.ts
+36 -29
View File
@@ -51,7 +51,7 @@ export class LoopOrchestrator {
private edgeManager: EdgeManager | null = null
) {}
initializeLoopScope(ctx: ExecutionContext, loopId: string): LoopScope {
async initializeLoopScope(ctx: ExecutionContext, loopId: string): Promise<LoopScope> {
const loopConfig = this.dag.loopConfigs.get(loopId) as SerializedLoop | undefined
if (!loopConfig) {
throw new Error(`Loop config not found: ${loopId}`)
@@ -76,7 +76,7 @@ export class LoopOrchestrator {
)
if (iterationError) {
logger.error(iterationError, { loopId, requestedIterations })
this.addLoopErrorLog(ctx, loopId, loopType, iterationError, {
await this.addLoopErrorLog(ctx, loopId, loopType, iterationError, {
iterations: requestedIterations,
})
scope.maxIterations = 0
@@ -99,7 +99,7 @@ export class LoopOrchestrator {
} catch (error) {
const errorMessage = `ForEach loop resolution failed: ${error instanceof Error ? error.message : String(error)}`
logger.error(errorMessage, { loopId, forEachItems: loopConfig.forEachItems })
this.addLoopErrorLog(ctx, loopId, loopType, errorMessage, {
await this.addLoopErrorLog(ctx, loopId, loopType, errorMessage, {
forEachItems: loopConfig.forEachItems,
})
scope.items = []
@@ -117,7 +117,7 @@ export class LoopOrchestrator {
)
if (sizeError) {
logger.error(sizeError, { loopId, collectionSize: items.length })
this.addLoopErrorLog(ctx, loopId, loopType, sizeError, {
await this.addLoopErrorLog(ctx, loopId, loopType, sizeError, {
forEachItems: loopConfig.forEachItems,
collectionSize: items.length,
})
@@ -155,7 +155,7 @@ export class LoopOrchestrator {
)
if (iterationError) {
logger.error(iterationError, { loopId, requestedIterations })
this.addLoopErrorLog(ctx, loopId, loopType, iterationError, {
await this.addLoopErrorLog(ctx, loopId, loopType, iterationError, {
iterations: requestedIterations,
})
scope.maxIterations = 0
@@ -182,14 +182,14 @@ export class LoopOrchestrator {
return scope
}
private addLoopErrorLog(
private async addLoopErrorLog(
ctx: ExecutionContext,
loopId: string,
loopType: string,
errorMessage: string,
inputData?: any
): void {
addSubflowErrorLog(
): Promise<void> {
await addSubflowErrorLog(
ctx,
loopId,
'loop',
@@ -238,7 +238,7 @@ export class LoopOrchestrator {
}
if (isCancelled) {
logger.info('Loop execution cancelled', { loopId, iteration: scope.iteration })
return this.createExitResult(ctx, loopId, scope)
return await this.createExitResult(ctx, loopId, scope)
}
const iterationResults: NormalizedBlockOutput[] = []
@@ -253,7 +253,7 @@ export class LoopOrchestrator {
scope.currentIterationOutputs.clear()
if (!(await this.evaluateCondition(ctx, scope, scope.iteration + 1))) {
return this.createExitResult(ctx, loopId, scope)
return await this.createExitResult(ctx, loopId, scope)
}
scope.iteration++
@@ -269,11 +269,11 @@ export class LoopOrchestrator {
}
}
private createExitResult(
private async createExitResult(
ctx: ExecutionContext,
loopId: string,
scope: LoopScope
): LoopContinuationResult {
): Promise<LoopContinuationResult> {
const results = scope.allIterationOutputs
const output = { results }
this.state.setBlockOutput(loopId, output, DEFAULTS.EXECUTION_TIME)
@@ -282,19 +282,26 @@ export class LoopOrchestrator {
const now = new Date().toISOString()
const iterationContext = buildContainerIterationContext(ctx, loopId)
this.contextExtensions.onBlockComplete(
loopId,
'Loop',
'loop',
{
output,
executionTime: DEFAULTS.EXECUTION_TIME,
startedAt: now,
executionOrder: getNextExecutionOrder(ctx),
endedAt: now,
},
iterationContext
)
try {
await this.contextExtensions.onBlockComplete(
loopId,
'Loop',
'loop',
{
output,
executionTime: DEFAULTS.EXECUTION_TIME,
startedAt: now,
executionOrder: getNextExecutionOrder(ctx),
endedAt: now,
},
iterationContext
)
} catch (error) {
logger.warn('Loop completion callback failed', {
loopId,
error: error instanceof Error ? error.message : String(error),
})
}
}
return {
@@ -597,7 +604,7 @@ export class LoopOrchestrator {
if (!scope.items || scope.items.length === 0) {
logger.info('ForEach loop has empty collection, skipping loop body', { loopId })
this.state.setBlockOutput(loopId, { results: [] }, DEFAULTS.EXECUTION_TIME)
emitEmptySubflowEvents(ctx, loopId, 'loop', this.contextExtensions)
await emitEmptySubflowEvents(ctx, loopId, 'loop', this.contextExtensions)
return false
}
return true
@@ -607,7 +614,7 @@ export class LoopOrchestrator {
if (scope.maxIterations === 0) {
logger.info('For loop has 0 iterations, skipping loop body', { loopId })
this.state.setBlockOutput(loopId, { results: [] }, DEFAULTS.EXECUTION_TIME)
emitEmptySubflowEvents(ctx, loopId, 'loop', this.contextExtensions)
await emitEmptySubflowEvents(ctx, loopId, 'loop', this.contextExtensions)
return false
}
return true
@@ -621,7 +628,7 @@ export class LoopOrchestrator {
if (!scope.condition) {
logger.warn('No condition defined for while loop', { loopId })
this.state.setBlockOutput(loopId, { results: [] }, DEFAULTS.EXECUTION_TIME)
emitEmptySubflowEvents(ctx, loopId, 'loop', this.contextExtensions)
await emitEmptySubflowEvents(ctx, loopId, 'loop', this.contextExtensions)
return false
}
@@ -634,7 +641,7 @@ export class LoopOrchestrator {
if (!result) {
this.state.setBlockOutput(loopId, { results: [] }, DEFAULTS.EXECUTION_TIME)
emitEmptySubflowEvents(ctx, loopId, 'loop', this.contextExtensions)
await emitEmptySubflowEvents(ctx, loopId, 'loop', this.contextExtensions)
}
return result
apps/sim/executor/orchestrators/node.ts
+12 -12
View File
@@ -53,14 +53,14 @@ export class NodeExecutionOrchestrator {
const loopId = node.metadata.loopId
if (loopId && !this.loopOrchestrator.getLoopScope(ctx, loopId)) {
this.loopOrchestrator.initializeLoopScope(ctx, loopId)
await this.loopOrchestrator.initializeLoopScope(ctx, loopId)
}
const parallelId = node.metadata.parallelId
if (parallelId && !this.parallelOrchestrator.getParallelScope(ctx, parallelId)) {
const parallelConfig = this.dag.parallelConfigs.get(parallelId)
const nodesInParallel = parallelConfig?.nodes?.length || 1
this.parallelOrchestrator.initializeParallelScope(ctx, parallelId, nodesInParallel)
await this.parallelOrchestrator.initializeParallelScope(ctx, parallelId, nodesInParallel)
}
if (node.metadata.isSentinel) {
@@ -92,7 +92,7 @@ export class NodeExecutionOrchestrator {
const isParallelSentinel = node.metadata.isParallelSentinel
if (isParallelSentinel) {
return this.handleParallelSentinel(ctx, node, sentinelType, parallelId)
return await this.handleParallelSentinel(ctx, node, sentinelType, parallelId)
}
switch (sentinelType) {
@@ -142,12 +142,12 @@ export class NodeExecutionOrchestrator {
}
}
private handleParallelSentinel(
private async handleParallelSentinel(
ctx: ExecutionContext,
node: DAGNode,
sentinelType: string | undefined,
parallelId: string | undefined
): NormalizedBlockOutput {
): Promise<NormalizedBlockOutput> {
if (!parallelId) {
logger.warn('Parallel sentinel called without parallelId')
return {}
@@ -158,7 +158,7 @@ export class NodeExecutionOrchestrator {
const parallelConfig = this.dag.parallelConfigs.get(parallelId)
if (parallelConfig) {
const nodesInParallel = parallelConfig.nodes?.length || 1
this.parallelOrchestrator.initializeParallelScope(ctx, parallelId, nodesInParallel)
await this.parallelOrchestrator.initializeParallelScope(ctx, parallelId, nodesInParallel)
}
}
@@ -176,7 +176,7 @@ export class NodeExecutionOrchestrator {
}
if (sentinelType === 'end') {
const result = this.parallelOrchestrator.aggregateParallelResults(ctx, parallelId)
const result = await this.parallelOrchestrator.aggregateParallelResults(ctx, parallelId)
return {
results: result.results || [],
sentinelEnd: true,
@@ -210,7 +210,7 @@ export class NodeExecutionOrchestrator {
} else if (isParallelBranch) {
const parallelId = this.findParallelIdForNode(node.id)
if (parallelId) {
this.handleParallelNodeCompletion(ctx, node, output, parallelId)
await this.handleParallelNodeCompletion(ctx, node, output, parallelId)
} else {
this.handleRegularNodeCompletion(ctx, node, output)
}
@@ -229,17 +229,17 @@ export class NodeExecutionOrchestrator {
this.state.setBlockOutput(node.id, output)
}
private handleParallelNodeCompletion(
private async handleParallelNodeCompletion(
ctx: ExecutionContext,
node: DAGNode,
output: NormalizedBlockOutput,
parallelId: string
): void {
): Promise<void> {
const scope = this.parallelOrchestrator.getParallelScope(ctx, parallelId)
if (!scope) {
const parallelConfig = this.dag.parallelConfigs.get(parallelId)
const nodesInParallel = parallelConfig?.nodes?.length || 1
this.parallelOrchestrator.initializeParallelScope(ctx, parallelId, nodesInParallel)
await this.parallelOrchestrator.initializeParallelScope(ctx, parallelId, nodesInParallel)
}
const allComplete = this.parallelOrchestrator.handleParallelBranchCompletion(
ctx,
@@ -248,7 +248,7 @@ export class NodeExecutionOrchestrator {
output
)
if (allComplete) {
this.parallelOrchestrator.aggregateParallelResults(ctx, parallelId)
await this.parallelOrchestrator.aggregateParallelResults(ctx, parallelId)
}
this.state.setBlockOutput(node.id, output)
apps/sim/executor/orchestrators/parallel.test.ts
+142
View File
@@ -0,0 +1,142 @@
/**
* @vitest-environment node
*/
import { beforeEach, describe, expect, it, vi } from 'vitest'
import type { DAG } from '@/executor/dag/builder'
import type { BlockStateWriter, ContextExtensions } from '@/executor/execution/types'
import { ParallelOrchestrator } from '@/executor/orchestrators/parallel'
import type { ExecutionContext } from '@/executor/types'
vi.mock('@sim/logger', () => ({
createLogger: () => ({
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
debug: vi.fn(),
}),
}))
function createDag(): DAG {
return {
nodes: new Map(),
loopConfigs: new Map(),
parallelConfigs: new Map([
[
'parallel-1',
{
id: 'parallel-1',
nodes: ['task-1'],
distribution: [],
parallelType: 'collection',
},
],
]),
}
}
function createState(): BlockStateWriter {
return {
setBlockOutput: vi.fn(),
setBlockState: vi.fn(),
deleteBlockState: vi.fn(),
unmarkExecuted: vi.fn(),
}
}
function createContext(overrides: Partial<ExecutionContext> = {}): ExecutionContext {
return {
workflowId: 'workflow-1',
workspaceId: 'workspace-1',
executionId: 'execution-1',
userId: 'user-1',
blockStates: new Map(),
executedBlocks: new Set(),
blockLogs: [],
metadata: { duration: 0 },
environmentVariables: {},
decisions: {
router: new Map(),
condition: new Map(),
},
completedLoops: new Set(),
activeExecutionPath: new Set(),
workflow: {
version: '1',
blocks: [
{
id: 'parallel-1',
position: { x: 0, y: 0 },
config: { tool: '', params: {} },
inputs: {},
outputs: {},
metadata: { id: 'parallel', name: 'Parallel 1' },
enabled: true,
},
],
connections: [],
loops: {},
parallels: {},
},
...overrides,
}
}
describe('ParallelOrchestrator', () => {
beforeEach(() => {
vi.clearAllMocks()
})
it('awaits empty-subflow lifecycle callbacks before returning the empty scope', async () => {
let releaseStart: (() => void) | undefined
const onBlockStart = vi.fn(
() =>
new Promise<void>((resolve) => {
releaseStart = resolve
})
)
const onBlockComplete = vi.fn()
const contextExtensions: ContextExtensions = {
onBlockStart,
onBlockComplete,
}
const orchestrator = new ParallelOrchestrator(
createDag(),
createState(),
null,
contextExtensions
)
const ctx = createContext()
const initializePromise = orchestrator.initializeParallelScope(ctx, 'parallel-1', 1)
await Promise.resolve()
expect(onBlockStart).toHaveBeenCalledTimes(1)
expect(onBlockComplete).not.toHaveBeenCalled()
releaseStart?.()
const scope = await initializePromise
expect(onBlockComplete).toHaveBeenCalledTimes(1)
expect(scope.isEmpty).toBe(true)
})
it('swallows helper callback failures on empty parallel paths', async () => {
const contextExtensions: ContextExtensions = {
onBlockStart: vi.fn().mockRejectedValue(new Error('start failed')),
onBlockComplete: vi.fn().mockRejectedValue(new Error('complete failed')),
}
const orchestrator = new ParallelOrchestrator(
createDag(),
createState(),
null,
contextExtensions
)
await expect(
orchestrator.initializeParallelScope(createContext(), 'parallel-1', 1)
).resolves.toMatchObject({
parallelId: 'parallel-1',
isEmpty: true,
})
})
})
apps/sim/executor/orchestrators/parallel.ts
+32 -22
View File
@@ -47,11 +47,11 @@ export class ParallelOrchestrator {
private contextExtensions: ContextExtensions | null = null
) {}
initializeParallelScope(
async initializeParallelScope(
ctx: ExecutionContext,
parallelId: string,
terminalNodesCount = 1
): ParallelScope {
): Promise<ParallelScope> {
const parallelConfig = this.dag.parallelConfigs.get(parallelId)
if (!parallelConfig) {
throw new Error(`Parallel config not found: ${parallelId}`)
@@ -69,7 +69,7 @@ export class ParallelOrchestrator {
} catch (error) {
const errorMessage = `Parallel Items did not resolve: ${error instanceof Error ? error.message : String(error)}`
logger.error(errorMessage, { parallelId, distribution: parallelConfig.distribution })
this.addParallelErrorLog(ctx, parallelId, errorMessage, {
await this.addParallelErrorLog(ctx, parallelId, errorMessage, {
distribution: parallelConfig.distribution,
})
this.setErrorScope(ctx, parallelId, errorMessage)
@@ -83,7 +83,7 @@ export class ParallelOrchestrator {
)
if (branchError) {
logger.error(branchError, { parallelId, branchCount })
this.addParallelErrorLog(ctx, parallelId, branchError, {
await this.addParallelErrorLog(ctx, parallelId, branchError, {
distribution: parallelConfig.distribution,
branchCount,
})
@@ -109,7 +109,7 @@ export class ParallelOrchestrator {
this.state.setBlockOutput(parallelId, { results: [] })
emitEmptySubflowEvents(ctx, parallelId, 'parallel', this.contextExtensions)
await emitEmptySubflowEvents(ctx, parallelId, 'parallel', this.contextExtensions)
logger.info('Parallel scope initialized with empty distribution, skipping body', {
parallelId,
@@ -220,13 +220,13 @@ export class ParallelOrchestrator {
return { branchCount: items.length, items }
}
private addParallelErrorLog(
private async addParallelErrorLog(
ctx: ExecutionContext,
parallelId: string,
errorMessage: string,
inputData?: any
): void {
addSubflowErrorLog(
): Promise<void> {
await addSubflowErrorLog(
ctx,
parallelId,
'parallel',
@@ -291,7 +291,10 @@ export class ParallelOrchestrator {
return allComplete
}
aggregateParallelResults(ctx: ExecutionContext, parallelId: string): ParallelAggregationResult {
async aggregateParallelResults(
ctx: ExecutionContext,
parallelId: string
): Promise<ParallelAggregationResult> {
const scope = ctx.parallelExecutions?.get(parallelId)
if (!scope) {
logger.error('Parallel scope not found for aggregation', { parallelId })
@@ -316,19 +319,26 @@ export class ParallelOrchestrator {
const now = new Date().toISOString()
const iterationContext = buildContainerIterationContext(ctx, parallelId)
this.contextExtensions.onBlockComplete(
parallelId,
'Parallel',
'parallel',
{
output,
executionTime: 0,
startedAt: now,
executionOrder: getNextExecutionOrder(ctx),
endedAt: now,
},
iterationContext
)
try {
await this.contextExtensions.onBlockComplete(
parallelId,
'Parallel',
'parallel',
{
output,
executionTime: 0,
startedAt: now,
executionOrder: getNextExecutionOrder(ctx),
endedAt: now,
},
iterationContext
)
} catch (error) {
logger.warn('Parallel completion callback failed', {
parallelId,
error: error instanceof Error ? error.message : String(error),
})
}
}
return {
apps/sim/executor/utils/subflow-utils.ts
+62 -27
View File
@@ -1,9 +1,12 @@
import { createLogger } from '@sim/logger'
import { DEFAULTS, LOOP, PARALLEL, REFERENCE } from '@/executor/constants'
import type { ContextExtensions } from '@/executor/execution/types'
import { type BlockLog, type ExecutionContext, getNextExecutionOrder } from '@/executor/types'
import { buildContainerIterationContext } from '@/executor/utils/iteration-context'
import type { VariableResolver } from '@/executor/variables/resolver'
const logger = createLogger('SubflowUtils')
const BRANCH_PATTERN = new RegExp(`${PARALLEL.BRANCH.PREFIX}\\d+${PARALLEL.BRANCH.SUFFIX}$`)
const BRANCH_INDEX_PATTERN = new RegExp(`${PARALLEL.BRANCH.PREFIX}(\\d+)${PARALLEL.BRANCH.SUFFIX}$`)
const LOOP_SENTINEL_START_PATTERN = new RegExp(
@@ -265,14 +268,14 @@ export function resolveArrayInput(
/**
* Creates and logs an error for a subflow (loop or parallel).
*/
export function addSubflowErrorLog(
export async function addSubflowErrorLog(
ctx: ExecutionContext,
blockId: string,
blockType: 'loop' | 'parallel',
errorMessage: string,
inputData: Record<string, any>,
contextExtensions: ContextExtensions | null
): void {
): Promise<void> {
const now = new Date().toISOString()
const execOrder = getNextExecutionOrder(ctx)
@@ -296,18 +299,34 @@ export function addSubflowErrorLog(
ctx.blockLogs.push(blockLog)
if (contextExtensions?.onBlockStart) {
contextExtensions.onBlockStart(blockId, blockName, blockType, execOrder)
try {
await contextExtensions.onBlockStart(blockId, blockName, blockType, execOrder)
} catch (error) {
logger.warn('Subflow error start callback failed', {
blockId,
blockType,
error: error instanceof Error ? error.message : String(error),
})
}
}
if (contextExtensions?.onBlockComplete) {
contextExtensions.onBlockComplete(blockId, blockName, blockType, {
input: inputData,
output: { error: errorMessage },
executionTime: 0,
startedAt: now,
executionOrder: execOrder,
endedAt: now,
})
try {
await contextExtensions.onBlockComplete(blockId, blockName, blockType, {
input: inputData,
output: { error: errorMessage },
executionTime: 0,
startedAt: now,
executionOrder: execOrder,
endedAt: now,
})
} catch (error) {
logger.warn('Subflow error completion callback failed', {
blockId,
blockType,
error: error instanceof Error ? error.message : String(error),
})
}
}
}
@@ -316,12 +335,12 @@ export function addSubflowErrorLog(
* empty collection or false initial condition. This ensures the container block
* appears in terminal logs, execution snapshots, and edge highlighting.
*/
export function emitEmptySubflowEvents(
export async function emitEmptySubflowEvents(
ctx: ExecutionContext,
blockId: string,
blockType: 'loop' | 'parallel',
contextExtensions: ContextExtensions | null
): void {
): Promise<void> {
const now = new Date().toISOString()
const executionOrder = getNextExecutionOrder(ctx)
const output = { results: [] }
@@ -342,22 +361,38 @@ export function emitEmptySubflowEvents(
})
if (contextExtensions?.onBlockStart) {
contextExtensions.onBlockStart(blockId, blockName, blockType, executionOrder)
try {
await contextExtensions.onBlockStart(blockId, blockName, blockType, executionOrder)
} catch (error) {
logger.warn('Empty subflow start callback failed', {
blockId,
blockType,
error: error instanceof Error ? error.message : String(error),
})
}
}
if (contextExtensions?.onBlockComplete) {
contextExtensions.onBlockComplete(
blockId,
blockName,
blockType,
{
output,
executionTime: DEFAULTS.EXECUTION_TIME,
startedAt: now,
executionOrder,
endedAt: now,
},
iterationContext
)
try {
await contextExtensions.onBlockComplete(
blockId,
blockName,
blockType,
{
output,
executionTime: DEFAULTS.EXECUTION_TIME,
startedAt: now,
executionOrder,
endedAt: now,
},
iterationContext
)
} catch (error) {
logger.warn('Empty subflow completion callback failed', {
blockId,
blockType,
error: error instanceof Error ? error.message : String(error),
})
}
}
}
apps/sim/lib/logs/execution/logger.test.ts
+23 -2
View File
@@ -1,6 +1,6 @@
import { databaseMock, loggerMock } from '@sim/testing'
import { beforeEach, describe, expect, test, vi } from 'vitest'
import { ExecutionLogger } from './logger'
import { ExecutionLogger } from '@/lib/logs/execution/logger'
vi.mock('@sim/db', () => databaseMock)
@@ -112,7 +112,7 @@ describe('ExecutionLogger', () => {
expect(typeof logger.getWorkflowExecution).toBe('function')
})
test('preserves start correlation data when execution completes', () => {
test('preserves correlation and diagnostics when execution completes', () => {
const loggerInstance = new ExecutionLogger() as any
const completedData = loggerInstance.buildCompletedExecutionData({
@@ -140,9 +140,24 @@ describe('ExecutionLogger', () => {
},
},
},
lastStartedBlock: {
blockId: 'block-start',
blockName: 'Start',
blockType: 'agent',
startedAt: '2025-01-01T00:00:00.000Z',
},
lastCompletedBlock: {
blockId: 'block-end',
blockName: 'Finish',
blockType: 'api',
endedAt: '2025-01-01T00:00:05.000Z',
success: true,
},
},
traceSpans: [],
finalOutput: { ok: true },
finalizationPath: 'completed',
completionFailure: 'fallback failure',
executionCost: {
tokens: { input: 0, output: 0, total: 0 },
models: {},
@@ -161,6 +176,12 @@ describe('ExecutionLogger', () => {
})
expect(completedData.correlation).toEqual(completedData.trigger?.data?.correlation)
expect(completedData.finalOutput).toEqual({ ok: true })
expect(completedData.lastStartedBlock?.blockId).toBe('block-start')
expect(completedData.lastCompletedBlock?.blockId).toBe('block-end')
expect(completedData.finalizationPath).toBe('completed')
expect(completedData.completionFailure).toBe('fallback failure')
expect(completedData.hasTraceSpans).toBe(false)
expect(completedData.traceSpanCount).toBe(0)
})
})
apps/sim/lib/logs/execution/logger.ts
+49 -8
View File
@@ -27,6 +27,7 @@ import { snapshotService } from '@/lib/logs/execution/snapshot/service'
import type {
BlockOutputData,
ExecutionEnvironment,
ExecutionFinalizationPath,
ExecutionTrigger,
ExecutionLoggerService as IExecutionLoggerService,
TraceSpan,
@@ -49,11 +50,21 @@ export interface ToolCall {
const logger = createLogger('ExecutionLogger')
function countTraceSpans(traceSpans?: TraceSpan[]): number {
if (!Array.isArray(traceSpans) || traceSpans.length === 0) {
return 0
}
return traceSpans.reduce((count, span) => count + 1 + countTraceSpans(span.children), 0)
}
export class ExecutionLogger implements IExecutionLoggerService {
private buildCompletedExecutionData(params: {
existingExecutionData?: WorkflowExecutionLog['executionData']
traceSpans?: TraceSpan[]
finalOutput: BlockOutputData
finalizationPath?: ExecutionFinalizationPath
completionFailure?: string
executionCost: {
tokens: {
input: number
@@ -64,7 +75,16 @@ export class ExecutionLogger implements IExecutionLoggerService {
}
executionState?: SerializableExecutionState
}): WorkflowExecutionLog['executionData'] {
const { existingExecutionData, traceSpans, finalOutput, executionCost, executionState } = params
const {
existingExecutionData,
traceSpans,
finalOutput,
finalizationPath,
completionFailure,
executionCost,
executionState,
} = params
const traceSpanCount = countTraceSpans(traceSpans)
return {
...(existingExecutionData?.environment
@@ -78,6 +98,17 @@ export class ExecutionLogger implements IExecutionLoggerService {
existingExecutionData?.trigger?.data?.correlation,
}
: {}),
...(existingExecutionData?.error ? { error: existingExecutionData.error } : {}),
...(existingExecutionData?.lastStartedBlock
? { lastStartedBlock: existingExecutionData.lastStartedBlock }
: {}),
...(existingExecutionData?.lastCompletedBlock
? { lastCompletedBlock: existingExecutionData.lastCompletedBlock }
: {}),
...(completionFailure ? { completionFailure } : {}),
...(finalizationPath ? { finalizationPath } : {}),
hasTraceSpans: traceSpanCount > 0,
traceSpanCount,
traceSpans,
finalOutput,
tokens: {
@@ -173,6 +204,8 @@ export class ExecutionLogger implements IExecutionLoggerService {
environment,
trigger,
...(trigger.data?.correlation ? { correlation: trigger.data.correlation } : {}),
hasTraceSpans: false,
traceSpanCount: 0,
},
cost: {
total: BASE_EXECUTION_CHARGE,
@@ -232,6 +265,8 @@ export class ExecutionLogger implements IExecutionLoggerService {
traceSpans?: TraceSpan[]
workflowInput?: any
executionState?: SerializableExecutionState
finalizationPath?: ExecutionFinalizationPath
completionFailure?: string
isResume?: boolean
level?: 'info' | 'error'
status?: 'completed' | 'failed' | 'cancelled' | 'pending'
@@ -245,6 +280,8 @@ export class ExecutionLogger implements IExecutionLoggerService {
traceSpans,
workflowInput,
executionState,
finalizationPath,
completionFailure,
isResume,
level: levelOverride,
status: statusOverride,
@@ -315,6 +352,16 @@ export class ExecutionLogger implements IExecutionLoggerService {
? Math.max(0, Math.round(rawDurationMs))
: 0
const completedExecutionData = this.buildCompletedExecutionData({
existingExecutionData,
traceSpans: redactedTraceSpans,
finalOutput: redactedFinalOutput,
finalizationPath,
completionFailure,
executionCost,
executionState,
})
const [updatedLog] = await db
.update(workflowExecutionLogs)
.set({
@@ -323,13 +370,7 @@ export class ExecutionLogger implements IExecutionLoggerService {
endedAt: new Date(endedAt),
totalDurationMs: totalDuration,
files: executionFiles.length > 0 ? executionFiles : null,
executionData: this.buildCompletedExecutionData({
existingExecutionData,
traceSpans: redactedTraceSpans,
finalOutput: redactedFinalOutput,
executionCost,
executionState,
}),
executionData: completedExecutionData,
cost: executionCost,
})
.where(eq(workflowExecutionLogs.executionId, executionId))
apps/sim/lib/logs/execution/logging-session.test.ts
+267 -4
View File
@@ -1,11 +1,48 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
const dbMocks = vi.hoisted(() => {
const selectLimit = vi.fn()
const selectWhere = vi.fn()
const selectFrom = vi.fn()
const select = vi.fn()
const updateWhere = vi.fn()
const updateSet = vi.fn()
const update = vi.fn()
const execute = vi.fn()
const eq = vi.fn()
const sql = vi.fn((strings: TemplateStringsArray, ...values: unknown[]) => ({ strings, values }))
select.mockReturnValue({ from: selectFrom })
selectFrom.mockReturnValue({ where: selectWhere })
selectWhere.mockReturnValue({ limit: selectLimit })
update.mockReturnValue({ set: updateSet })
updateSet.mockReturnValue({ where: updateWhere })
return {
select,
selectFrom,
selectWhere,
selectLimit,
update,
updateSet,
updateWhere,
execute,
eq,
sql,
}
})
const { completeWorkflowExecutionMock } = vi.hoisted(() => ({
completeWorkflowExecutionMock: vi.fn(),
}))
vi.mock('@sim/db', () => ({
db: {},
db: {
select: dbMocks.select,
update: dbMocks.update,
execute: dbMocks.execute,
},
}))
vi.mock('@sim/db/schema', () => ({
@@ -22,8 +59,8 @@ vi.mock('@sim/logger', () => ({
}))
vi.mock('drizzle-orm', () => ({
eq: vi.fn(),
sql: vi.fn(),
eq: dbMocks.eq,
sql: dbMocks.sql,
}))
vi.mock('@/lib/logs/execution/logger', () => ({
@@ -56,6 +93,9 @@ import { LoggingSession } from './logging-session'
describe('LoggingSession completion retries', () => {
beforeEach(() => {
vi.clearAllMocks()
dbMocks.selectLimit.mockResolvedValue([{ executionData: {} }])
dbMocks.updateWhere.mockResolvedValue(undefined)
dbMocks.execute.mockResolvedValue(undefined)
})
it('keeps completion best-effort when a later error completion retries after full completion and fallback both fail', async () => {
@@ -86,7 +126,6 @@ describe('LoggingSession completion retries', () => {
.mockRejectedValueOnce(new Error('cost only failed'))
await expect(session.safeComplete({ finalOutput: { ok: true } })).resolves.toBeUndefined()
await expect(session.safeComplete({ finalOutput: { ok: true } })).resolves.toBeUndefined()
expect(completeWorkflowExecutionMock).toHaveBeenCalledTimes(2)
@@ -118,6 +157,64 @@ describe('LoggingSession completion retries', () => {
expect(session.hasCompleted()).toBe(true)
})
it('preserves successful final output during fallback completion', async () => {
const session = new LoggingSession('workflow-1', 'execution-5', 'api', 'req-1')
completeWorkflowExecutionMock
.mockRejectedValueOnce(new Error('success finalize failed'))
.mockResolvedValueOnce({})
await expect(
session.safeComplete({ finalOutput: { ok: true, stage: 'done' } })
).resolves.toBeUndefined()
expect(completeWorkflowExecutionMock).toHaveBeenLastCalledWith(
expect.objectContaining({
executionId: 'execution-5',
finalOutput: { ok: true, stage: 'done' },
finalizationPath: 'fallback_completed',
})
)
})
it('preserves accumulated cost during fallback completion', async () => {
const session = new LoggingSession('workflow-1', 'execution-6', 'api', 'req-1') as any
session.accumulatedCost = {
total: 12,
input: 5,
output: 7,
tokens: { input: 11, output: 13, total: 24 },
models: {
'test-model': {
input: 5,
output: 7,
total: 12,
tokens: { input: 11, output: 13, total: 24 },
},
},
}
session.costFlushed = true
completeWorkflowExecutionMock
.mockRejectedValueOnce(new Error('success finalize failed'))
.mockResolvedValueOnce({})
await expect(session.safeComplete({ finalOutput: { ok: true } })).resolves.toBeUndefined()
expect(completeWorkflowExecutionMock).toHaveBeenLastCalledWith(
expect.objectContaining({
executionId: 'execution-6',
costSummary: expect.objectContaining({
totalCost: 12,
totalInputCost: 5,
totalOutputCost: 7,
totalTokens: 24,
}),
})
)
})
it('persists failed error semantics when completeWithError receives non-error trace spans', async () => {
const session = new LoggingSession('workflow-1', 'execution-4', 'api', 'req-1')
const traceSpans = [
@@ -148,6 +245,8 @@ describe('LoggingSession completion retries', () => {
traceSpans,
level: 'error',
status: 'failed',
finalizationPath: 'force_failed',
completionFailure: 'persist me as failed',
})
)
})
@@ -196,4 +295,168 @@ describe('LoggingSession completion retries', () => {
expect(session.hasCompleted()).toBe(true)
expect(completeWorkflowExecutionMock).toHaveBeenCalledTimes(2)
})
it('persists last started block independently from cost accumulation', async () => {
const session = new LoggingSession('workflow-1', 'execution-1', 'api', 'req-1')
await session.onBlockStart('block-1', 'Fetch', 'api', '2025-01-01T00:00:00.000Z')
expect(dbMocks.select).not.toHaveBeenCalled()
expect(dbMocks.execute).toHaveBeenCalledTimes(1)
})
it('enforces started marker monotonicity in the database write path', async () => {
const session = new LoggingSession('workflow-1', 'execution-1', 'api', 'req-1')
await session.onBlockStart('block-1', 'Fetch', 'api', '2025-01-01T00:00:00.000Z')
expect(dbMocks.sql).toHaveBeenCalled()
expect(dbMocks.execute).toHaveBeenCalledTimes(1)
})
it('allows same-millisecond started markers to replace the prior marker', async () => {
const session = new LoggingSession('workflow-1', 'execution-1', 'api', 'req-1')
await session.onBlockStart('block-1', 'Fetch', 'api', '2025-01-01T00:00:00.000Z')
const queryCall = dbMocks.sql.mock.calls.at(-1)
expect(queryCall).toBeDefined()
const [query] = queryCall!
expect(Array.from(query).join(' ')).toContain('<=')
})
it('persists last completed block for zero-cost outputs', async () => {
const session = new LoggingSession('workflow-1', 'execution-1', 'api', 'req-1')
await session.onBlockComplete('block-2', 'Transform', 'function', {
endedAt: '2025-01-01T00:00:01.000Z',
output: { value: true },
})
expect(dbMocks.select).not.toHaveBeenCalled()
expect(dbMocks.execute).toHaveBeenCalledTimes(1)
})
it('allows same-millisecond completed markers to replace the prior marker', async () => {
const session = new LoggingSession('workflow-1', 'execution-1', 'api', 'req-1')
await session.onBlockComplete('block-2', 'Transform', 'function', {
endedAt: '2025-01-01T00:00:01.000Z',
output: { value: true },
})
const queryCall = dbMocks.sql.mock.calls.at(-1)
expect(queryCall).toBeDefined()
const [query] = queryCall!
expect(Array.from(query).join(' ')).toContain('<=')
})
it('drains pending lifecycle writes before terminal completion', async () => {
let releasePersist: (() => void) | undefined
const persistPromise = new Promise<void>((resolve) => {
releasePersist = resolve
})
const session = new LoggingSession('workflow-1', 'execution-1', 'api', 'req-1') as any
session.persistLastStartedBlock = vi.fn(() => persistPromise)
session.complete = vi.fn().mockResolvedValue(undefined)
const startPromise = session.onBlockStart('block-1', 'Fetch', 'api', '2025-01-01T00:00:00.000Z')
const completionPromise = session.safeComplete({ finalOutput: { ok: true } })
await Promise.resolve()
expect(session.complete).not.toHaveBeenCalled()
releasePersist?.()
await startPromise
await completionPromise
expect(session.persistLastStartedBlock).toHaveBeenCalledTimes(1)
expect(session.complete).toHaveBeenCalledTimes(1)
})
it('drains fire-and-forget cost flushes before terminal completion', async () => {
let releaseFlush: (() => void) | undefined
const flushPromise = new Promise<void>((resolve) => {
releaseFlush = resolve
})
const session = new LoggingSession('workflow-1', 'execution-1', 'api', 'req-1') as any
session.flushAccumulatedCost = vi.fn(() => flushPromise)
session.complete = vi.fn().mockResolvedValue(undefined)
await session.onBlockComplete('block-2', 'Transform', 'function', {
endedAt: '2025-01-01T00:00:01.000Z',
output: { value: true },
cost: { total: 1, input: 1, output: 0 },
tokens: { input: 1, output: 0, total: 1 },
model: 'test-model',
})
const completionPromise = session.safeComplete({ finalOutput: { ok: true } })
await Promise.resolve()
expect(session.complete).not.toHaveBeenCalled()
releaseFlush?.()
await completionPromise
expect(session.flushAccumulatedCost).toHaveBeenCalledTimes(1)
expect(session.complete).toHaveBeenCalledTimes(1)
})
it('keeps draining when new progress writes arrive during drain', async () => {
let releaseFirst: (() => void) | undefined
let releaseSecond: (() => void) | undefined
const firstPromise = new Promise<void>((resolve) => {
releaseFirst = resolve
})
const secondPromise = new Promise<void>((resolve) => {
releaseSecond = resolve
})
const session = new LoggingSession('workflow-1', 'execution-1', 'api', 'req-1') as any
void session.trackProgressWrite(firstPromise)
const drainPromise = session.drainPendingProgressWrites()
await Promise.resolve()
void session.trackProgressWrite(secondPromise)
releaseFirst?.()
await Promise.resolve()
let drained = false
void drainPromise.then(() => {
drained = true
})
await Promise.resolve()
expect(drained).toBe(false)
releaseSecond?.()
await drainPromise
expect(session.pendingProgressWrites.size).toBe(0)
})
it('marks pause completion as terminal and prevents duplicate pause finalization', async () => {
const session = new LoggingSession('workflow-1', 'execution-1', 'api', 'req-1') as any
session.completeExecutionWithFinalization = vi.fn().mockResolvedValue(undefined)
await session.completeWithPause({ workflowInput: { ok: true } })
await session.completeWithPause({ workflowInput: { ok: true } })
expect(session.completeExecutionWithFinalization).toHaveBeenCalledTimes(1)
expect(session.completed).toBe(true)
expect(session.completing).toBe(true)
})
})
apps/sim/lib/logs/execution/logging-session.ts
+236 -33
View File
@@ -13,6 +13,9 @@ import {
} from '@/lib/logs/execution/logging-factory'
import type {
ExecutionEnvironment,
ExecutionFinalizationPath,
ExecutionLastCompletedBlock,
ExecutionLastStartedBlock,
ExecutionTrigger,
TraceSpan,
WorkflowState,
@@ -23,6 +26,46 @@ type TriggerData = Record<string, unknown> & {
correlation?: NonNullable<ExecutionTrigger['data']>['correlation']
}
function buildStartedMarkerPersistenceQuery(params: {
executionId: string
marker: ExecutionLastStartedBlock
}) {
const markerJson = JSON.stringify(params.marker)
return sql`UPDATE workflow_execution_logs
SET execution_data = jsonb_set(
COALESCE(execution_data, '{}'::jsonb),
'{lastStartedBlock}',
${markerJson}::jsonb,
true
)
WHERE execution_id = ${params.executionId}
AND COALESCE(
jsonb_extract_path_text(COALESCE(execution_data, '{}'::jsonb), 'lastStartedBlock', 'startedAt'),
''
) <= ${params.marker.startedAt}`
}
function buildCompletedMarkerPersistenceQuery(params: {
executionId: string
marker: ExecutionLastCompletedBlock
}) {
const markerJson = JSON.stringify(params.marker)
return sql`UPDATE workflow_execution_logs
SET execution_data = jsonb_set(
COALESCE(execution_data, '{}'::jsonb),
'{lastCompletedBlock}',
${markerJson}::jsonb,
true
)
WHERE execution_id = ${params.executionId}
AND COALESCE(
jsonb_extract_path_text(COALESCE(execution_data, '{}'::jsonb), 'lastCompletedBlock', 'endedAt'),
''
) <= ${params.marker.endedAt}`
}
const logger = createLogger('LoggingSession')
type CompletionAttempt = 'complete' | 'error' | 'cancelled' | 'paused'
@@ -109,6 +152,7 @@ export class LoggingSession {
tokens: { input: 0, output: 0, total: 0 },
models: {},
}
private pendingProgressWrites = new Set<Promise<void>>()
private costFlushed = false
private postExecutionPromise: Promise<void> | null = null
@@ -124,12 +168,132 @@ export class LoggingSession {
this.requestId = requestId
}
async onBlockStart(
blockId: string,
blockName: string,
blockType: string,
startedAt: string
): Promise<void> {
await this.trackProgressWrite(
this.persistLastStartedBlock({
blockId,
blockName,
blockType,
startedAt,
})
)
}
private async persistLastStartedBlock(marker: ExecutionLastStartedBlock): Promise<void> {
try {
await db.execute(
buildStartedMarkerPersistenceQuery({
executionId: this.executionId,
marker,
})
)
} catch (error) {
logger.error(`Failed to persist last started block for execution ${this.executionId}:`, {
error: error instanceof Error ? error.message : String(error),
})
}
}
private async persistLastCompletedBlock(marker: ExecutionLastCompletedBlock): Promise<void> {
try {
await db.execute(
buildCompletedMarkerPersistenceQuery({
executionId: this.executionId,
marker,
})
)
} catch (error) {
logger.error(`Failed to persist last completed block for execution ${this.executionId}:`, {
error: error instanceof Error ? error.message : String(error),
})
}
}
private async trackProgressWrite(writePromise: Promise<void>): Promise<void> {
this.pendingProgressWrites.add(writePromise)
try {
await writePromise
} finally {
this.pendingProgressWrites.delete(writePromise)
}
}
private async drainPendingProgressWrites(): Promise<void> {
while (this.pendingProgressWrites.size > 0) {
await Promise.allSettled(Array.from(this.pendingProgressWrites))
}
}
private async completeExecutionWithFinalization(params: {
endedAt: string
totalDurationMs: number
costSummary: {
totalCost: number
totalInputCost: number
totalOutputCost: number
totalTokens: number
totalPromptTokens: number
totalCompletionTokens: number
baseExecutionCharge: number
modelCost: number
models: Record<
string,
{
input: number
output: number
total: number
tokens: { input: number; output: number; total: number }
}
>
}
finalOutput: Record<string, unknown>
traceSpans: TraceSpan[]
workflowInput?: unknown
executionState?: SerializableExecutionState
finalizationPath: ExecutionFinalizationPath
completionFailure?: string
level?: 'info' | 'error'
status?: 'completed' | 'failed' | 'cancelled' | 'pending'
}): Promise<void> {
await executionLogger.completeWorkflowExecution({
executionId: this.executionId,
endedAt: params.endedAt,
totalDurationMs: params.totalDurationMs,
costSummary: params.costSummary,
finalOutput: params.finalOutput,
traceSpans: params.traceSpans,
workflowInput: params.workflowInput,
executionState: params.executionState,
finalizationPath: params.finalizationPath,
completionFailure: params.completionFailure,
isResume: this.isResume,
level: params.level,
status: params.status,
})
}
async onBlockComplete(
blockId: string,
blockName: string,
blockType: string,
output: any
): Promise<void> {
await this.trackProgressWrite(
this.persistLastCompletedBlock({
blockId,
blockName,
blockType,
endedAt: output?.endedAt || new Date().toISOString(),
success: !output?.output?.error,
})
)
if (!output?.cost || typeof output.cost.total !== 'number' || output.cost.total <= 0) {
return
}
@@ -165,7 +329,7 @@ export class LoggingSession {
}
}
await this.flushAccumulatedCost()
void this.trackProgressWrite(this.flushAccumulatedCost())
}
private async flushAccumulatedCost(): Promise<void> {
@@ -276,8 +440,7 @@ export class LoggingSession {
const endTime = endedAt || new Date().toISOString()
const duration = totalDurationMs || 0
await executionLogger.completeWorkflowExecution({
executionId: this.executionId,
await this.completeExecutionWithFinalization({
endedAt: endTime,
totalDurationMs: duration,
costSummary,
@@ -285,7 +448,7 @@ export class LoggingSession {
traceSpans: traceSpans || [],
workflowInput,
executionState,
isResume: this.isResume,
finalizationPath: 'completed',
})
this.completed = true
@@ -403,8 +566,7 @@ export class LoggingSession {
const spans = hasProvidedSpans ? traceSpans : [errorSpan]
await executionLogger.completeWorkflowExecution({
executionId: this.executionId,
await this.completeExecutionWithFinalization({
endedAt: endTime.toISOString(),
totalDurationMs: Math.max(1, durationMs),
costSummary,
@@ -412,6 +574,8 @@ export class LoggingSession {
traceSpans: spans,
level: 'error',
status: 'failed',
finalizationPath: 'force_failed',
completionFailure: message,
})
this.completed = true
@@ -490,13 +654,13 @@ export class LoggingSession {
models: {},
}
await executionLogger.completeWorkflowExecution({
executionId: this.executionId,
await this.completeExecutionWithFinalization({
endedAt: endTime.toISOString(),
totalDurationMs: Math.max(1, durationMs),
costSummary,
finalOutput: { cancelled: true },
traceSpans: traceSpans || [],
finalizationPath: 'cancelled',
status: 'cancelled',
})
@@ -577,14 +741,14 @@ export class LoggingSession {
models: {},
}
await executionLogger.completeWorkflowExecution({
executionId: this.executionId,
await this.completeExecutionWithFinalization({
endedAt: endTime.toISOString(),
totalDurationMs: Math.max(1, durationMs),
costSummary,
finalOutput: { paused: true },
traceSpans: traceSpans || [],
workflowInput,
finalizationPath: 'paused',
status: 'pending',
})
@@ -746,7 +910,6 @@ export class LoggingSession {
this.completionAttemptFailed = true
throw error
})
return this.completionPromise
}
@@ -756,6 +919,7 @@ export class LoggingSession {
private async _safeCompleteImpl(params: SessionCompleteParams = {}): Promise<void> {
try {
await this.drainPendingProgressWrites()
await this.complete(params)
} catch (error) {
const errorMsg = error instanceof Error ? error.message : String(error)
@@ -769,6 +933,8 @@ export class LoggingSession {
totalDurationMs: params.totalDurationMs,
errorMessage: `Failed to store trace spans: ${errorMsg}`,
isError: false,
finalizationPath: 'fallback_completed',
finalOutput: params.finalOutput || {},
})
}
}
@@ -779,6 +945,7 @@ export class LoggingSession {
private async _safeCompleteWithErrorImpl(params?: SessionErrorCompleteParams): Promise<void> {
try {
await this.drainPendingProgressWrites()
await this.completeWithError(params)
} catch (error) {
const errorMsg = error instanceof Error ? error.message : String(error)
@@ -793,6 +960,10 @@ export class LoggingSession {
errorMessage:
params?.error?.message || `Execution failed to store trace spans: ${errorMsg}`,
isError: true,
finalizationPath: 'force_failed',
finalOutput: {
error: params?.error?.message || `Execution failed to store trace spans: ${errorMsg}`,
},
status: 'failed',
})
}
@@ -806,6 +977,7 @@ export class LoggingSession {
private async _safeCompleteWithCancellationImpl(params?: SessionCancelledParams): Promise<void> {
try {
await this.drainPendingProgressWrites()
await this.completeWithCancellation(params)
} catch (error) {
const errorMsg = error instanceof Error ? error.message : String(error)
@@ -819,6 +991,8 @@ export class LoggingSession {
totalDurationMs: params?.totalDurationMs,
errorMessage: 'Execution was cancelled',
isError: false,
finalizationPath: 'cancelled',
finalOutput: { cancelled: true },
status: 'cancelled',
})
}
@@ -830,6 +1004,7 @@ export class LoggingSession {
private async _safeCompleteWithPauseImpl(params?: SessionPausedParams): Promise<void> {
try {
await this.drainPendingProgressWrites()
await this.completeWithPause(params)
} catch (error) {
const errorMsg = error instanceof Error ? error.message : String(error)
@@ -843,6 +1018,8 @@ export class LoggingSession {
totalDurationMs: params?.totalDurationMs,
errorMessage: 'Execution paused but failed to store full trace spans',
isError: false,
finalizationPath: 'paused',
finalOutput: { paused: true },
status: 'pending',
})
}
@@ -867,12 +1044,16 @@ export class LoggingSession {
status: 'failed',
executionData: sql`jsonb_set(
jsonb_set(
COALESCE(execution_data, '{}'::jsonb),
ARRAY['error'],
to_jsonb(${message}::text)
jsonb_set(
COALESCE(execution_data, '{}'::jsonb),
ARRAY['error'],
to_jsonb(${message}::text)
),
ARRAY['finalOutput'],
jsonb_build_object('error', ${message}::text)
),
ARRAY['finalOutput'],
jsonb_build_object('error', ${message}::text)
ARRAY['finalizationPath'],
to_jsonb('force_failed'::text)
)`,
})
.where(eq(workflowExecutionLogs.executionId, executionId))
@@ -891,6 +1072,8 @@ export class LoggingSession {
totalDurationMs?: number
errorMessage: string
isError: boolean
finalizationPath: ExecutionFinalizationPath
finalOutput?: Record<string, unknown>
status?: 'completed' | 'failed' | 'cancelled' | 'pending'
}): Promise<void> {
if (this.completed || this.completing) {
@@ -903,28 +1086,48 @@ export class LoggingSession {
)
try {
const costSummary = params.traceSpans?.length
? calculateCostSummary(params.traceSpans)
: {
totalCost: BASE_EXECUTION_CHARGE,
totalInputCost: 0,
totalOutputCost: 0,
totalTokens: 0,
totalPromptTokens: 0,
totalCompletionTokens: 0,
baseExecutionCharge: BASE_EXECUTION_CHARGE,
modelCost: 0,
models: {},
}
const hasAccumulatedCost =
this.costFlushed ||
this.accumulatedCost.total > BASE_EXECUTION_CHARGE ||
this.accumulatedCost.tokens.total > 0 ||
Object.keys(this.accumulatedCost.models).length > 0
await executionLogger.completeWorkflowExecution({
executionId: this.executionId,
const costSummary = hasAccumulatedCost
? {
totalCost: this.accumulatedCost.total,
totalInputCost: this.accumulatedCost.input,
totalOutputCost: this.accumulatedCost.output,
totalTokens: this.accumulatedCost.tokens.total,
totalPromptTokens: this.accumulatedCost.tokens.input,
totalCompletionTokens: this.accumulatedCost.tokens.output,
baseExecutionCharge: BASE_EXECUTION_CHARGE,
modelCost: Math.max(0, this.accumulatedCost.total - BASE_EXECUTION_CHARGE),
models: this.accumulatedCost.models,
}
: params.traceSpans?.length
? calculateCostSummary(params.traceSpans)
: {
totalCost: BASE_EXECUTION_CHARGE,
totalInputCost: 0,
totalOutputCost: 0,
totalTokens: 0,
totalPromptTokens: 0,
totalCompletionTokens: 0,
baseExecutionCharge: BASE_EXECUTION_CHARGE,
modelCost: 0,
models: {},
}
const finalOutput = params.finalOutput || { _fallback: true, error: params.errorMessage }
await this.completeExecutionWithFinalization({
endedAt: params.endedAt || new Date().toISOString(),
totalDurationMs: params.totalDurationMs || 0,
costSummary,
finalOutput: { _fallback: true, error: params.errorMessage },
finalOutput,
traceSpans: [],
isResume: this.isResume,
finalizationPath: params.finalizationPath,
completionFailure: params.errorMessage,
level: params.isError ? 'error' : 'info',
status: params.status,
})
apps/sim/lib/logs/types.ts
+35
View File
@@ -71,6 +71,31 @@ export interface ExecutionStatus {
durationMs?: number
}
export const EXECUTION_FINALIZATION_PATHS = [
'completed',
'fallback_completed',
'force_failed',
'cancelled',
'paused',
] as const
export type ExecutionFinalizationPath = (typeof EXECUTION_FINALIZATION_PATHS)[number]
export interface ExecutionLastStartedBlock {
blockId: string
blockName: string
blockType: string
startedAt: string
}
export interface ExecutionLastCompletedBlock {
blockId: string
blockName: string
blockType: string
endedAt: string
success: boolean
}
export interface WorkflowExecutionSnapshot {
id: string
workflowId: string | null
@@ -105,6 +130,13 @@ export interface WorkflowExecutionLog {
environment?: ExecutionEnvironment
trigger?: ExecutionTrigger
correlation?: AsyncExecutionCorrelation
error?: string
lastStartedBlock?: ExecutionLastStartedBlock
lastCompletedBlock?: ExecutionLastCompletedBlock
hasTraceSpans?: boolean
traceSpanCount?: number
completionFailure?: string
finalizationPath?: ExecutionFinalizationPath
traceSpans?: TraceSpan[]
tokens?: { input?: number; output?: number; total?: number }
models?: Record<
@@ -378,6 +410,9 @@ export interface ExecutionLoggerService {
finalOutput: BlockOutputData
traceSpans?: TraceSpan[]
workflowInput?: any
executionState?: SerializableExecutionState
finalizationPath?: ExecutionFinalizationPath
completionFailure?: string
isResume?: boolean
level?: 'info' | 'error'
status?: 'completed' | 'failed' | 'cancelled' | 'pending'
apps/sim/lib/workflows/executor/execution-core.test.ts
+184 -8
View File
@@ -16,6 +16,8 @@ const {
buildTraceSpansMock,
serializeWorkflowMock,
executorExecuteMock,
onBlockStartPersistenceMock,
executorConstructorMock,
} = vi.hoisted(() => ({
loadWorkflowFromNormalizedTablesMock: vi.fn(),
loadDeployedWorkflowStateMock: vi.fn(),
@@ -32,6 +34,8 @@ const {
buildTraceSpansMock: vi.fn(),
serializeWorkflowMock: vi.fn(),
executorExecuteMock: vi.fn(),
onBlockStartPersistenceMock: vi.fn(),
executorConstructorMock: vi.fn(),
}))
vi.mock('@sim/logger', () => ({
@@ -79,10 +83,13 @@ vi.mock('@/lib/workflows/utils', () => ({
}))
vi.mock('@/executor', () => ({
Executor: vi.fn().mockImplementation(() => ({
execute: executorExecuteMock,
executeFromBlock: executorExecuteMock,
})),
Executor: vi.fn().mockImplementation((args) => {
executorConstructorMock(args)
return {
execute: executorExecuteMock,
executeFromBlock: executorExecuteMock,
}
}),
}))
vi.mock('@/serializer', () => ({
@@ -105,6 +112,8 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
safeCompleteWithCancellation: safeCompleteWithCancellationMock,
safeCompleteWithPause: safeCompleteWithPauseMock,
hasCompleted: hasCompletedMock,
onBlockStart: onBlockStartPersistenceMock,
onBlockComplete: vi.fn(),
setPostExecutionPromise: vi.fn(),
waitForPostExecution: vi.fn().mockResolvedValue(undefined),
}
@@ -176,10 +185,72 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
safeCompleteWithCancellationMock.mockResolvedValue(undefined)
safeCompleteWithPauseMock.mockResolvedValue(undefined)
hasCompletedMock.mockReturnValue(true)
onBlockStartPersistenceMock.mockResolvedValue(undefined)
updateWorkflowRunCountsMock.mockResolvedValue(undefined)
clearExecutionCancellationMock.mockResolvedValue(undefined)
})
it('routes onBlockStart through logging session persistence path', async () => {
executorExecuteMock.mockResolvedValue({
success: true,
status: 'completed',
output: { done: true },
logs: [],
metadata: { duration: 123, startTime: 'start', endTime: 'end' },
})
await executeWorkflowCore({
snapshot: createSnapshot() as any,
callbacks: {
onBlockStart: async (blockId) => {
expect(blockId).toBe('block-1')
},
},
loggingSession: loggingSession as any,
})
const contextExtensions = executorConstructorMock.mock.calls[0]?.[0]?.contextExtensions
await contextExtensions.onBlockStart('block-1', 'Fetch', 'api', 1)
expect(onBlockStartPersistenceMock).toHaveBeenCalledWith(
'block-1',
'Fetch',
'api',
expect.any(String)
)
})
it('does not await user block start callback after persistence completes', async () => {
let releaseCallback: (() => void) | undefined
const callbackPromise = new Promise<void>((resolve) => {
releaseCallback = resolve
})
executorExecuteMock.mockResolvedValue({
success: true,
status: 'completed',
output: { done: true },
logs: [],
metadata: { duration: 123, startTime: 'start', endTime: 'end' },
})
await executeWorkflowCore({
snapshot: createSnapshot() as any,
callbacks: {
onBlockStart: vi.fn(() => callbackPromise),
},
loggingSession: loggingSession as any,
})
const contextExtensions = executorConstructorMock.mock.calls[0]?.[0]?.contextExtensions
await expect(
contextExtensions.onBlockStart('block-1', 'Fetch', 'api', 1)
).resolves.toBeUndefined()
releaseCallback?.()
})
it('awaits terminal completion before updating run counts and returning', async () => {
const callOrder: string[] = []
@@ -222,7 +293,57 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
])
})
it('clears cancellation even when success finalization throws', async () => {
it('awaits wrapped lifecycle persistence before terminal finalization returns', async () => {
let releaseBlockStart: (() => void) | undefined
const blockStartPromise = new Promise<void>((resolve) => {
releaseBlockStart = resolve
})
const callOrder: string[] = []
onBlockStartPersistenceMock.mockImplementation(async () => {
callOrder.push('persist:start')
await blockStartPromise
callOrder.push('persist:end')
})
safeCompleteMock.mockImplementation(async () => {
callOrder.push('safeComplete')
})
executorExecuteMock.mockImplementation(async () => {
const contextExtensions = executorConstructorMock.mock.calls[0]?.[0]?.contextExtensions
const startLifecycle = contextExtensions.onBlockStart('block-1', 'Fetch', 'api', 1)
await Promise.resolve()
callOrder.push('executor:before-release')
releaseBlockStart?.()
await startLifecycle
callOrder.push('executor:after-start')
return {
success: true,
status: 'completed',
output: { done: true },
logs: [],
metadata: { duration: 123, startTime: 'start', endTime: 'end' },
}
})
await executeWorkflowCore({
snapshot: createSnapshot() as any,
callbacks: {},
loggingSession: loggingSession as any,
})
expect(callOrder).toEqual([
'persist:start',
'executor:before-release',
'persist:end',
'executor:after-start',
'safeComplete',
])
})
it('preserves successful execution when success finalization throws', async () => {
executorExecuteMock.mockResolvedValue({
success: true,
status: 'completed',
@@ -244,7 +365,7 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
expect(result.status).toBe('completed')
expect(clearExecutionCancellationMock).toHaveBeenCalledWith('execution-1')
expect(updateWorkflowRunCountsMock).not.toHaveBeenCalled()
expect(updateWorkflowRunCountsMock).toHaveBeenCalledWith('workflow-1')
})
it('routes cancelled executions through safeCompleteWithCancellation', async () => {
@@ -304,6 +425,61 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
expect(updateWorkflowRunCountsMock).not.toHaveBeenCalled()
})
it('swallows wrapped block start callback failures without breaking execution', async () => {
onBlockStartPersistenceMock.mockRejectedValue(new Error('start persistence failed'))
executorExecuteMock.mockImplementation(async () => {
const contextExtensions = executorConstructorMock.mock.calls[0]?.[0]?.contextExtensions
await contextExtensions.onBlockStart('block-1', 'Fetch', 'api', 1)
return {
success: true,
status: 'completed',
output: { done: true },
logs: [],
metadata: { duration: 123, startTime: 'start', endTime: 'end' },
}
})
const result = await executeWorkflowCore({
snapshot: createSnapshot() as any,
callbacks: {},
loggingSession: loggingSession as any,
})
expect(result.status).toBe('completed')
expect(safeCompleteMock).toHaveBeenCalledTimes(1)
})
it('swallows wrapped block complete callback failures without blocking completion', async () => {
executorExecuteMock.mockResolvedValue({
success: true,
status: 'completed',
output: { done: true },
logs: [],
metadata: { duration: 123, startTime: 'start', endTime: 'end' },
})
await executeWorkflowCore({
snapshot: createSnapshot() as any,
callbacks: {
onBlockComplete: vi.fn().mockRejectedValue(new Error('complete callback failed')),
},
loggingSession: loggingSession as any,
})
const contextExtensions = executorConstructorMock.mock.calls[0]?.[0]?.contextExtensions
await expect(
contextExtensions.onBlockComplete('block-1', 'Fetch', 'api', {
output: { ok: true },
executionTime: 1,
startedAt: 'start',
endedAt: 'end',
})
).resolves.toBeUndefined()
})
it('finalizes errors before rethrowing and marks them as core-finalized', async () => {
const error = new Error('engine failed')
const executionResult = {
@@ -445,7 +621,7 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
expect(wasExecutionFinalizedByCore('engine failed', 'execution-a')).toBe(true)
})
it('logs error without rejecting when success finalization rejects', async () => {
it('does not replace a successful outcome when success finalization rejects', async () => {
executorExecuteMock.mockResolvedValue({
success: true,
status: 'completed',
@@ -464,7 +640,7 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
await loggingSession.setPostExecutionPromise.mock.calls[0][0]
expect(result.status).toBe('completed')
expect(result).toMatchObject({ status: 'completed', success: true })
expect(clearExecutionCancellationMock).toHaveBeenCalledWith('execution-1')
expect(safeCompleteWithErrorMock).not.toHaveBeenCalled()
})
apps/sim/lib/workflows/executor/execution-core.ts
+90 -29
View File
@@ -179,33 +179,41 @@ async function finalizeExecutionOutcome(params: {
const endedAt = new Date().toISOString()
try {
if (result.status === 'cancelled') {
await loggingSession.safeCompleteWithCancellation({
endedAt,
totalDurationMs: totalDuration || 0,
traceSpans: traceSpans || [],
})
return
}
try {
if (result.status === 'cancelled') {
await loggingSession.safeCompleteWithCancellation({
endedAt,
totalDurationMs: totalDuration || 0,
traceSpans: traceSpans || [],
})
return
}
if (result.status === 'paused') {
await loggingSession.safeCompleteWithPause({
if (result.status === 'paused') {
await loggingSession.safeCompleteWithPause({
endedAt,
totalDurationMs: totalDuration || 0,
traceSpans: traceSpans || [],
workflowInput,
})
return
}
await loggingSession.safeComplete({
endedAt,
totalDurationMs: totalDuration || 0,
finalOutput: result.output || {},
traceSpans: traceSpans || [],
workflowInput,
executionState: result.executionState,
})
} catch (error) {
logger.warn(`[${requestId}] Post-execution finalization failed`, {
executionId,
status: result.status,
error,
})
return
}
await loggingSession.safeComplete({
endedAt,
totalDurationMs: totalDuration || 0,
finalOutput: result.output || {},
traceSpans: traceSpans || [],
workflowInput,
executionState: result.executionState,
})
} finally {
await clearExecutionCancellationSafely(executionId, requestId)
}
@@ -424,16 +432,69 @@ export async function executeWorkflowCore(
iterationContext?: IterationContext,
childWorkflowContext?: ChildWorkflowContext
) => {
await loggingSession.onBlockComplete(blockId, blockName, blockType, output)
if (onBlockComplete) {
await onBlockComplete(
try {
await loggingSession.onBlockComplete(blockId, blockName, blockType, output)
if (onBlockComplete) {
void onBlockComplete(
blockId,
blockName,
blockType,
output,
iterationContext,
childWorkflowContext
).catch((error) => {
logger.warn(`[${requestId}] Block completion callback failed`, {
executionId,
blockId,
blockType,
error,
})
})
}
} catch (error) {
logger.warn(`[${requestId}] Block completion persistence failed`, {
executionId,
blockId,
blockName,
blockType,
output,
iterationContext,
childWorkflowContext
)
error,
})
}
}
const wrappedOnBlockStart = async (
blockId: string,
blockName: string,
blockType: string,
executionOrder: number,
iterationContext?: IterationContext,
childWorkflowContext?: ChildWorkflowContext
) => {
try {
await loggingSession.onBlockStart(blockId, blockName, blockType, new Date().toISOString())
if (onBlockStart) {
void onBlockStart(
blockId,
blockName,
blockType,
executionOrder,
iterationContext,
childWorkflowContext
).catch((error) => {
logger.warn(`[${requestId}] Block start callback failed`, {
executionId,
blockId,
blockType,
error,
})
})
}
} catch (error) {
logger.warn(`[${requestId}] Block start persistence failed`, {
executionId,
blockId,
blockType,
error,
})
}
}
@@ -445,7 +506,7 @@ export async function executeWorkflowCore(
userId,
isDeployedContext: !metadata.isClientSession,
enforceCredentialAccess: metadata.enforceCredentialAccess ?? false,
onBlockStart,
onBlockStart: wrappedOnBlockStart,
onBlockComplete: wrappedOnBlockComplete,
onStream,
resumeFromSnapshot,