import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' import { DEFAULTS } 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 { SubflowNodeIdCodec } from '@/executor/utils/subflow-node-id-codec' import type { SerializedWorkflow } from '@/serializer/types' const logger = createLogger('SubflowUtils') /** * Builds the loop sentinel-start node ID for a container. */ export function buildSentinelStartId(loopId: string): string { return SubflowNodeIdCodec.buildLoopSentinelStartId(loopId) } /** * Builds the loop sentinel-end node ID for a container. */ export function buildSentinelEndId(loopId: string): string { return SubflowNodeIdCodec.buildLoopSentinelEndId(loopId) } export function buildParallelSentinelStartId(parallelId: string): string { return SubflowNodeIdCodec.buildParallelSentinelStartId(parallelId) } export function buildParallelSentinelEndId(parallelId: string): string { return SubflowNodeIdCodec.buildParallelSentinelEndId(parallelId) } export function isLoopSentinelNodeId(nodeId: string): boolean { return SubflowNodeIdCodec.isLoopSentinelNodeId(nodeId) } export function isParallelSentinelNodeId(nodeId: string): boolean { return SubflowNodeIdCodec.isParallelSentinelNodeId(nodeId) } export function isSentinelNodeId(nodeId: string): boolean { return isLoopSentinelNodeId(nodeId) || isParallelSentinelNodeId(nodeId) } export function extractLoopIdFromSentinel(sentinelId: string): string | null { return SubflowNodeIdCodec.extractLoopIdFromSentinel(sentinelId) } export function extractParallelIdFromSentinel(sentinelId: string): string | null { return SubflowNodeIdCodec.extractParallelIdFromSentinel(sentinelId) } /** * Build branch node ID with subscript notation * Example: ("blockId", 2) → "blockId₍2₎" */ export function buildBranchNodeId(baseId: string, branchIndex: number): string { return SubflowNodeIdCodec.buildBranchNodeId(baseId, branchIndex) } export function extractBaseBlockId(branchNodeId: string): string { return SubflowNodeIdCodec.extractBaseBlockId(branchNodeId) } export function extractBranchIndex(branchNodeId: string): number | null { return SubflowNodeIdCodec.extractBranchIndex(branchNodeId) } export function isBranchNodeId(nodeId: string): boolean { return SubflowNodeIdCodec.isBranchNodeId(nodeId) } /** * Extracts the outer branch index from a cloned subflow ID. * Cloned IDs follow the pattern `{originalId}__obranch-{index}`. * Returns undefined if the ID is not a clone. */ export function extractOuterBranchIndex(clonedId: string): number | undefined { return SubflowNodeIdCodec.extractOuterBranchIndex(clonedId) } export function extractInnermostOuterBranchIndex(clonedId: string): number | undefined { return SubflowNodeIdCodec.extractInnermostOuterBranchIndex(clonedId) } /** * Strips all clone suffixes (`__obranch-N`) and branch subscripts (`₍N₎`) * from a node ID, returning the original workflow-level block ID. */ export function stripCloneSuffixes(nodeId: string): string { return SubflowNodeIdCodec.stripCloneSuffixes(nodeId) } /** * Builds a stable ID for an output scoped to a global outer parallel branch. */ export function buildOuterBranchScopedId(originalId: string, branchIndex: number): string { return SubflowNodeIdCodec.buildOuterBranchScopedId(originalId, branchIndex) } /** * Builds a cloned subflow ID from an original ID and outer branch index. */ export function buildClonedSubflowId(originalId: string, branchIndex: number): string { return SubflowNodeIdCodec.buildOuterBranchScopedId(originalId, branchIndex) } /** * Strips outer-branch clone suffixes (`__obranch-N`) from an ID, * returning the original workflow-level subflow ID. */ export function stripOuterBranchSuffix(id: string): string { return SubflowNodeIdCodec.stripOuterBranchSuffix(id) } /** * Finds the effective (possibly cloned) container ID for a subflow, * given the current node's ID and an execution map (loopExecutions or parallelExecutions). * * When inside a cloned subflow (e.g., loop-1__obranch-2), the execution scope is * stored under the cloned ID, not the original. This function extracts the `__obranch-N` * suffix from the current node ID, constructs the candidate cloned container ID, and * checks if it exists in the execution map. * * Returns the effective ID (cloned or original) that exists in the map. */ export function findEffectiveContainerId( originalId: string, currentNodeId: string, executionMap: Map, mappedBranchIndex?: number ): string { return SubflowNodeIdCodec.findEffectiveContainerId( originalId, currentNodeId, executionMap, mappedBranchIndex ) } export function normalizeNodeId(nodeId: string): string { return SubflowNodeIdCodec.normalizeNodeId(nodeId) } type SubflowContainerType = 'loop' | 'parallel' function getSubflowNodes( workflow: Pick, type: SubflowContainerType, id: string ): string[] | undefined { return type === 'loop' ? workflow.loops?.[id]?.nodes : workflow.parallels?.[id]?.nodes } export function subflowContainsBlock( workflow: Pick, containerType: SubflowContainerType, containerId: string, baseBlockId: string, visited = new Set() ): boolean { const visitKey = `${containerType}:${containerId}` if (visited.has(visitKey)) return false visited.add(visitKey) const nodes = getSubflowNodes(workflow, containerType, containerId) if (!nodes) return false for (const nodeId of nodes) { if (nodeId === baseBlockId) return true if (workflow.loops?.[nodeId]) { if (subflowContainsBlock(workflow, 'loop', nodeId, baseBlockId, visited)) return true } else if (workflow.parallels?.[nodeId]) { if (subflowContainsBlock(workflow, 'parallel', nodeId, baseBlockId, visited)) return true } } return false } export function isSubflowNestedInside( workflow: Pick, childType: SubflowContainerType, childId: string, ancestorType: SubflowContainerType, ancestorId: string, visited = new Set() ): boolean { const visitKey = `${ancestorType}:${ancestorId}` if (visited.has(visitKey)) return false visited.add(visitKey) const nodes = getSubflowNodes(workflow, ancestorType, ancestorId) if (!nodes) return false for (const nodeId of nodes) { if ( nodeId === childId && (childType === 'loop' ? workflow.loops?.[childId] : workflow.parallels?.[childId]) ) { return true } if (workflow.loops?.[nodeId]) { if (isSubflowNestedInside(workflow, childType, childId, 'loop', nodeId, visited)) { return true } } else if (workflow.parallels?.[nodeId]) { if (isSubflowNestedInside(workflow, childType, childId, 'parallel', nodeId, visited)) { return true } } } return false } /** * Creates and logs an error for a subflow (loop or parallel). */ export async function addSubflowErrorLog( ctx: ExecutionContext, blockId: string, blockType: 'loop' | 'parallel', errorMessage: string, inputData: Record, contextExtensions: ContextExtensions | null ): Promise { const now = new Date().toISOString() const execOrder = getNextExecutionOrder(ctx) const block = ctx.workflow?.blocks?.find((b) => b.id === blockId) const blockName = block?.metadata?.name || (blockType === 'loop' ? 'Loop' : 'Parallel') const blockLog: BlockLog = { blockId, blockName, blockType, startedAt: now, executionOrder: execOrder, endedAt: now, durationMs: 0, success: false, error: errorMessage, input: inputData, output: { error: errorMessage }, ...(blockType === 'loop' ? { loopId: blockId } : { parallelId: blockId }), } ctx.blockLogs.push(blockLog) if (contextExtensions?.onBlockStart) { try { await contextExtensions.onBlockStart(blockId, blockName, blockType, execOrder) } catch (error) { logger.warn('Subflow error start callback failed', { blockId, blockType, error: toError(error).message, }) } } if (contextExtensions?.onBlockComplete) { 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: toError(error).message, }) } } } /** * Emits the BlockLog + onBlockComplete callback for a loop/parallel container that * finished successfully. Without this, successful container runs produce no top-level BlockLog, * which forces the trace-span builder to fall back * to generic counter-based names ("Loop 1", "Parallel 1") instead of the user-configured * block name. */ export async function emitSubflowSuccessEvents( ctx: ExecutionContext, blockId: string, blockType: 'loop' | 'parallel', output: { results: unknown }, contextExtensions: ContextExtensions | null ): Promise { const now = new Date().toISOString() const executionOrder = getNextExecutionOrder(ctx) const block = ctx.workflow?.blocks.find((b) => b.id === blockId) const blockName = block?.metadata?.name ?? blockType const iterationContext = buildContainerIterationContext(ctx, blockId) ctx.blockLogs.push({ blockId, blockName, blockType, startedAt: now, endedAt: now, durationMs: DEFAULTS.EXECUTION_TIME, success: true, output, executionOrder, }) if (contextExtensions?.onBlockComplete) { try { await contextExtensions.onBlockComplete( blockId, blockName, blockType, { output, executionTime: DEFAULTS.EXECUTION_TIME, startedAt: now, executionOrder, endedAt: now, }, iterationContext ) } catch (error) { logger.warn('Subflow success completion callback failed', { blockId, blockType, error: toError(error).message, }) } } }