Function executeWorkflowCore

apps/sim/lib/workflows/executor/execution-core.ts:313–868

(
  options: ExecuteWorkflowCoreOptions
)

}

export async function executeWorkflowCore(
  options: ExecuteWorkflowCoreOptions
): Promise<ExecutionResult> {
  const {
    snapshot,
    callbacks,
    loggingSession,
    skipLogCreation,
    abortSignal,
    includeFileBase64,
    base64MaxBytes,
    stopAfterBlockId,
    runFromBlock,
  } = options
  const { metadata, workflow, input, workflowVariables, selectedOutputs } = snapshot
  const { requestId, workflowId, userId, triggerType, executionId, triggerBlockId, useDraftState } =
    metadata
  const { onBlockStart, onBlockComplete, onStream, onChildWorkflowInstanceReady } = callbacks

  const providedWorkspaceId = metadata.workspaceId
  if (!providedWorkspaceId) {
    throw new Error(`Execution metadata missing workspaceId for workflow ${workflowId}`)
  }

  let processedInput = input || {}
  let deploymentVersionId: string | undefined
  let loggingStarted = false
  const pendingLifecycleCallbacks = new Set<Promise<void>>()

  const trackLifecycleCallback = (promise: Promise<void>) => {
    pendingLifecycleCallbacks.add(promise)
    void promise
      .finally(() => {
        pendingLifecycleCallbacks.delete(promise)
      })
      .catch(() => {})
  }

  const waitForLifecycleCallbacks = async () => {
    while (pendingLifecycleCallbacks.size > 0) {
      await Promise.allSettled([...pendingLifecycleCallbacks])
    }
  }

  try {
    const personalEnvUserId =
      metadata.isClientSession && metadata.sessionUserId
        ? metadata.sessionUserId
        : metadata.workflowUserId

    if (!personalEnvUserId) {
      throw new Error('Missing workflowUserId in execution metadata')
    }

    /**
     * Resolves the workflow state from the override, the draft tables, or the
     * deployed snapshot. The async load (draft/deployed) has no data dependency
     * on the environment load, so the two are awaited concurrently below.