@elsium-ai/workflows

Multi-step workflow pipelines and DAG execution for ElsiumAI.

Install

npm install @elsium-ai/workflows @elsium-ai/core

What’s Inside

Types

StepStatus

Union type representing the possible states of a step during execution.

type StepStatus = 'pending' | 'running' | 'completed' | 'failed' | 'skipped'

StepConfig<TInput, TOutput>

Configuration object that defines a single step’s behavior, including its handler, optional input validation, retry policy, conditional execution, fallback logic, and timeout.

interface StepConfig<TInput = unknown, TOutput = unknown> {
  name: string
  input?: z.ZodType<TInput>
  handler: (input: TInput, context: StepContext) => Promise<TOutput>
  retry?: RetryConfig
  condition?: (input: TInput, context: StepContext) => boolean
  fallback?: (error: Error, input: TInput) => Promise<TOutput>
  timeoutMs?: number
}

StepContext

Runtime context passed to every step handler and condition function.

interface StepContext {
  workflowName: string
  stepIndex: number
  previousOutputs: Record<string, unknown>
  signal?: AbortSignal
}

StepResult<T>

Outcome returned after a step finishes (or is skipped/fails).

interface StepResult<T = unknown> {
  name: string
  status: StepStatus
  data?: T
  error?: string
  durationMs: number
  retryCount: number
}

RetryConfig

Per-step retry policy with exponential backoff and jitter.

interface RetryConfig {
  maxRetries: number
  baseDelayMs?: number
  maxDelayMs?: number
  shouldRetry?: (error: Error) => boolean
}

Backoff formula: min(baseDelayMs * 2^(attempt-1), maxDelayMs) * random(0.5, 1.0).

WorkflowStatus

Union type representing the possible states of a workflow.

type WorkflowStatus = 'pending' | 'running' | 'completed' | 'failed' | 'paused'

WorkflowConfig

Configuration for a sequential workflow created via defineWorkflow.

interface WorkflowConfig {
  name: string
  steps: StepConfig[]
  onStepComplete?: (result: StepResult) => void | Promise<void>
  onStepError?: (error: Error, stepName: string) => void | Promise<void>
  onComplete?: (result: WorkflowResult) => void | Promise<void>
}

WorkflowResult

Final output returned by workflow.run().

interface WorkflowResult {
  name: string
  status: WorkflowStatus
  steps: StepResult[]
  totalDurationMs: number
  outputs: Record<string, unknown>
}

WorkflowRunOptions

Options passed to workflow.run().

interface WorkflowRunOptions {
  signal?: AbortSignal
}

Steps

step

Factory function that creates a StepConfig by combining a name with the rest of the configuration, providing a concise shorthand for inline step definitions.

function step<TInput, TOutput>(
  name: string,
  config: Omit<StepConfig<TInput, TOutput>, 'name'>,
): StepConfig<TInput, TOutput>

Returns: StepConfig<TInput, TOutput>

import { step } from '@elsium-ai/workflows'
import { z } from 'zod'

const fetchPage = step('fetch-page', {
  input: z.object({ url: z.string().url() }),
  handler: async (input) => {
    const res = await fetch(input.url)
    return res.text()
  },
  retry: { maxRetries: 3, baseDelayMs: 500 },
  timeoutMs: 10_000,
})

executeStep

Runs a single step to completion, handling input validation, condition checks, retries with exponential backoff, timeout enforcement, and fallback execution.

function executeStep<TInput, TOutput>(
  stepConfig: StepConfig<TInput, TOutput>,
  rawInput: unknown,
  context: StepContext,
): Promise<StepResult<TOutput>>

Returns: Promise<StepResult<TOutput>>

The execution order is:

1. Validate rawInput against the Zod schema (if provided). Return 'failed' on validation error.
2. Evaluate the condition guard (if provided). Return 'skipped' when false.
3. Run the handler, retrying on failure up to retry.maxRetries times with backoff.
4. On final failure, invoke fallback (if provided). If the fallback also fails, return 'failed'.

import { step, executeStep } from '@elsium-ai/workflows'
import type { StepContext } from '@elsium-ai/workflows'

const myStep = step('greet', {
  handler: async (input: { name: string }) => `Hello, ${input.name}!`,
})

const context: StepContext = {
  workflowName: 'demo',
  stepIndex: 0,
  previousOutputs: {},
}

const result = await executeStep(myStep, { name: 'World' }, context)
console.log(result.data) // "Hello, World!"

Workflow

Workflow

Interface implemented by all workflow variants (sequential, parallel, and branch).

interface Workflow {
  readonly name: string
  run(input: unknown, options?: WorkflowRunOptions): Promise<WorkflowResult>
}

defineWorkflow

Creates a sequential workflow that executes steps one after another, piping each step’s output as the next step’s input.

function defineWorkflow(config: WorkflowConfig): Workflow

Returns: Workflow

When a step completes, its output is stored in outputs under the step’s name and becomes the input for the next step. If any step fails, the workflow short-circuits and returns with status: 'failed'.

import { defineWorkflow, step } from '@elsium-ai/workflows'

const pipeline = defineWorkflow({
  name: 'etl-pipeline',
  steps: [
    step('extract', {
      handler: async (input: { source: string }) => {
        return await extractData(input.source)
      },
    }),
    step('transform', {
      handler: async (rawData: RawData) => {
        return transformData(rawData)
      },
    }),
    step('load', {
      handler: async (transformed: TransformedData) => {
        await loadData(transformed)
        return { loaded: true }
      },
    }),
  ],
  onStepComplete: (result) => {
    console.log(`Step "${result.name}" finished in ${result.durationMs}ms`)
  },
  onComplete: (result) => {
    console.log(`Workflow "${result.name}" ${result.status}`)
  },
})

const result = await pipeline.run({ source: 'database' })

ParallelWorkflowConfig

Configuration for a parallel workflow created via defineParallelWorkflow.

interface ParallelWorkflowConfig {
  name: string
  steps: StepConfig[]
  onComplete?: (result: WorkflowResult) => void | Promise<void>
}

defineParallelWorkflow

Creates a parallel workflow that executes all steps concurrently using Promise.all, where every step receives the same input.

function defineParallelWorkflow(config: ParallelWorkflowConfig): Workflow

Returns: Workflow

Each step’s output is stored in outputs under its name. The workflow status is 'failed' if any step fails, 'completed' otherwise.

import { defineParallelWorkflow, step } from '@elsium-ai/workflows'

const enrichment = defineParallelWorkflow({
  name: 'enrich-profile',
  steps: [
    step('fetch-social', {
      handler: async (input: { userId: string }) => {
        return await fetchSocialProfile(input.userId)
      },
    }),
    step('fetch-activity', {
      handler: async (input: { userId: string }) => {
        return await fetchActivityLog(input.userId)
      },
    }),
    step('fetch-preferences', {
      handler: async (input: { userId: string }) => {
        return await fetchPreferences(input.userId)
      },
    }),
  ],
})

const result = await enrichment.run({ userId: 'u_123' })
// result.outputs['fetch-social'], result.outputs['fetch-activity'], etc.

BranchConfig

Defines a single branch in a branching workflow, pairing a condition with the workflow to execute when the condition is met.

interface BranchConfig {
  condition: (input: unknown) => boolean
  workflow: Workflow
}

defineBranchWorkflow

Creates a branching workflow that evaluates conditions in order and delegates to the first matching branch’s workflow, with an optional fallback.

function defineBranchWorkflow(
  name: string,
  branches: BranchConfig[],
  fallback?: Workflow,
): Workflow

Returns: Workflow

If no branch matches and no fallback is provided, the workflow returns immediately with status: 'completed', an empty steps array, and empty outputs.

import { defineBranchWorkflow, defineWorkflow, step } from '@elsium-ai/workflows'

const textWorkflow = defineWorkflow({
  name: 'process-text',
  steps: [
    step('analyze-text', {
      handler: async (input: { content: string }) => {
        return await analyzeText(input.content)
      },
    }),
  ],
})

const imageWorkflow = defineWorkflow({
  name: 'process-image',
  steps: [
    step('analyze-image', {
      handler: async (input: { content: string }) => {
        return await analyzeImage(input.content)
      },
    }),
  ],
})

const router = defineBranchWorkflow(
  'content-router',
  [
    { condition: (input: any) => input.type === 'text', workflow: textWorkflow },
    { condition: (input: any) => input.type === 'image', workflow: imageWorkflow },
  ],
)

const result = await router.run({ type: 'text', content: 'Hello world' })

Sequential Workflow

Steps execute in order. Each step receives the previous step’s output as input.

import { step, defineWorkflow } from '@elsium-ai/workflows'

const pipeline = defineWorkflow({
  name: 'data-pipeline',
  steps: [
    step('fetch', {
      handler: async (input: { url: string }) => {
        const res = await fetch(input.url)
        return res.json()
      },
    }),
    step('transform', {
      handler: async (data: unknown) => {
        return processData(data)
      },
    }),
    step('save', {
      handler: async (processed: unknown) => {
        await db.insert(processed)
        return { saved: true }
      },
    }),
  ],
  onStepComplete: (result) => {
    console.log(`Step ${result.name}: ${result.status} (${result.durationMs}ms)`)
  },
  onStepError: (error, stepName) => {
    console.error(`Step ${stepName} failed:`, error.message)
  },
})

const result = await pipeline.run({ url: 'https://api.example.com/data' })

Parallel Workflow

All steps receive the same initial input and execute in parallel.

import { step, defineParallelWorkflow } from '@elsium-ai/workflows'

const fanOut = defineParallelWorkflow({
  name: 'multi-search',
  steps: [
    step('search-web', {
      handler: async (query: { text: string }) => searchWeb(query.text),
    }),
    step('search-db', {
      handler: async (query: { text: string }) => searchDB(query.text),
    }),
    step('search-cache', {
      handler: async (query: { text: string }) => searchCache(query.text),
    }),
  ],
  onComplete: (result) => {
    console.log(`All searches completed in ${result.totalDurationMs}ms`)
  },
})

const result = await fanOut.run({ text: 'ElsiumAI documentation' })
// result.outputs contains each step's output keyed by step name

Branch Workflow

Routes input to the first branch whose condition returns true. An optional fallback workflow handles unmatched inputs.

import { defineWorkflow, defineBranchWorkflow } from '@elsium-ai/workflows'

const textPipeline = defineWorkflow({ name: 'text', steps: [/* ... */] })
const imagePipeline = defineWorkflow({ name: 'image', steps: [/* ... */] })
const defaultPipeline = defineWorkflow({ name: 'default', steps: [/* ... */] })

const router = defineBranchWorkflow('content-router', [
  {
    condition: (input: unknown) => (input as any).type === 'text',
    workflow: textPipeline,
  },
  {
    condition: (input: unknown) => (input as any).type === 'image',
    workflow: imagePipeline,
  },
], defaultPipeline)

const result = await router.run({ type: 'text', data: '...' })

Step Configuration Details

Retry

Automatic retry with exponential backoff.

step('unreliable-api', {
  handler: async (input) => callExternalAPI(input),
  retry: {
    maxRetries: 3,
    baseDelayMs: 500,
    maxDelayMs: 5000,
    shouldRetry: (error) => error.message.includes('timeout'),
  },
})

Conditional Execution

Skip a step based on input or prior context.

step('optional-enrichment', {
  condition: (input, ctx) => input.needsEnrichment === true,
  handler: async (input) => enrichData(input),
})

Fallback

Provide a fallback value when a step fails.

step('fetch-with-fallback', {
  handler: async (input) => fetchFromPrimary(input),
  fallback: async (error, input) => {
    return { data: null, source: 'fallback' }
  },
})

Input Validation

Validate step input with a Zod schema.

import { z } from 'zod'

step('validated-step', {
  input: z.object({ url: z.string().url(), limit: z.number().positive() }),
  handler: async (input) => {
    // input is typed as { url: string, limit: number }
    return fetch(input.url)
  },
})

DAG Workflow

Steps declare their dependencies via dependsOn. The workflow computes a topological order and executes independent steps concurrently in waves. A step receives the output of its first declared dependency as input (falling back to the workflow input when it has none). Execution stops after the first wave that contains a failed step. Unknown dependencies and cycles throw at run time.

import { step, defineDagWorkflow } from '@elsium-ai/workflows'

const build = defineDagWorkflow({
  name: 'build-pipeline',
  steps: [
    step('compile', { handler: async (src: { files: string[] }) => compile(src) }),
    { ...step('test', { handler: async (artifact) => runTests(artifact) }), dependsOn: ['compile'] },
    { ...step('lint', { handler: async (artifact) => runLint(artifact) }), dependsOn: ['compile'] },
    { ...step('package', { handler: async (artifact) => bundle(artifact) }), dependsOn: ['test', 'lint'] },
  ],
  onStepComplete: (result) => console.log(`${result.name}: ${result.status}`),
})

const result = await build.run({ files: ['a.ts', 'b.ts'] })
// 'compile' runs first; 'test' and 'lint' run in parallel; 'package' runs last.

DagWorkflowConfig

interface DagWorkflowConfig {
  name: string
  steps: DagStepConfig[]
  onStepComplete?: (result: StepResult) => void | Promise<void>
  onStepError?: (error: Error, stepName: string) => void | Promise<void>
  onComplete?: (result: WorkflowResult) => void | Promise<void>
}

DagStepConfig

Extends StepConfig with a dependency list.

interface DagStepConfig<TInput, TOutput> extends StepConfig<TInput, TOutput> {
  dependsOn?: string[]  // names of steps that must complete first
}

Resumable Workflows

A resumable workflow saves a WorkflowCheckpoint before each step. If a step fails, the checkpoint is persisted with status: 'failed' and the run returns. Calling resume(workflowId) reloads the checkpoint and continues from the step that failed; resuming an already-completed workflow returns its stored result. Provide your own CheckpointStore implementation for durable persistence — the package ships only the in-memory adapter.

import { step, defineResumableWorkflow, createInMemoryCheckpointStore } from '@elsium-ai/workflows'

const store = createInMemoryCheckpointStore()

const order = defineResumableWorkflow({
  name: 'process-order',
  checkpointStore: store,
  steps: [
    step('reserve', { handler: async (o: { id: string }) => reserveStock(o) }),
    step('charge', { handler: async (o) => chargeCard(o) }),
    step('ship', { handler: async (o) => createShipment(o) }),
  ],
})

const first = await order.run({ id: 'ord-42' }, { workflowId: 'ord-42' })

if (first.status === 'failed') {
  // ...fix the underlying issue, then continue from the failed step
  const final = await order.resume('ord-42')
}

const checkpoint = await order.getCheckpoint('ord-42')
const all = await order.listCheckpoints()

ResumableWorkflowConfig

Extends WorkflowConfig with a checkpoint store.

interface ResumableWorkflowConfig extends WorkflowConfig {
  checkpointStore: CheckpointStore
}

ResumableWorkflowRunOptions

Extends WorkflowRunOptions. Pass a workflowId to use a stable identifier (so the run can be resumed by that id); one is generated when omitted.

interface ResumableWorkflowRunOptions extends WorkflowRunOptions {
  workflowId?: string  // defaults to a generated id
}

ResumableWorkflow

interface ResumableWorkflow {
  readonly name: string
  run(input: unknown, options?: ResumableWorkflowRunOptions): Promise<WorkflowResult>
  resume(workflowId: string, options?: WorkflowRunOptions): Promise<WorkflowResult>
  getCheckpoint(workflowId: string): Promise<WorkflowCheckpoint | null>
  listCheckpoints(): Promise<WorkflowCheckpoint[]>
}

CheckpointStore

The port a backend must implement to persist checkpoints.

interface CheckpointStore {
  save(checkpoint: WorkflowCheckpoint): Promise<void>
  load(workflowId: string): Promise<WorkflowCheckpoint | null>
  delete(workflowId: string): Promise<void>
  list(workflowName?: string): Promise<WorkflowCheckpoint[]>
}

WorkflowCheckpoint

interface WorkflowCheckpoint {
  workflowId: string
  workflowName: string
  status: 'running' | 'completed' | 'failed' | 'paused'
  input: unknown
  currentStepIndex: number
  stepResults: StepResult[]
  outputs: Record<string, unknown>
  createdAt: number
  updatedAt: number
}

Idempotent Steps

Step-level deduplication for side-effectful workflows (POST to an external API, DB writes, sending email). When a workflow is resumed after a crash, an idempotent step that already ran is served from the store instead of re-executing. The package ships only the in-memory adapter; implement IdempotentCheckpointStore against your backend for durability.

A step opts in by setting idempotent: true (see IdempotentStepConfig). executeIdempotentStep resolves the key; if the step is not idempotent it falls through to a normal executeStep. Otherwise it checks the store for an existing record under (workflowId, stepName, idempotencyKey) and returns it as a StepResult (with durationMs: 0) on a hit, or runs the step and records the outcome on a miss.

import {
  step,
  executeIdempotentStep,
  createInMemoryIdempotentCheckpointStore,
} from '@elsium-ai/workflows'

const store = createInMemoryIdempotentCheckpointStore()

const charge: IdempotentStepConfig<{ orderId: string }, { receiptId: string }> = {
  ...step('charge', { handler: async (o) => chargeCard(o) }),
  idempotent: true,
  idempotencyKey: (o) => o.orderId,  // optional; defaults to a hash of the input
}

const result = await executeIdempotentStep({
  workflowId: 'ord-42',
  step: charge,
  input: { orderId: 'ord-42' },
  context: { workflowName: 'process-order', stepIndex: 1, previousOutputs: {} },
  store,
})
// A second call with the same input returns the recorded result without re-charging.

IdempotentStepConfig

Extends StepConfig with idempotency opt-in.

interface IdempotentStepConfig<TInput, TOutput> extends StepConfig<TInput, TOutput> {
  readonly idempotent?: boolean
  readonly idempotencyKey?: (input: TInput) => string  // defaults to defaultIdempotencyKey
}

Part of ElsiumAI

This package is the workflow orchestration layer of the ElsiumAI framework. See the Getting Started guide or the Fundamentals for comprehensive coverage of every feature.

License

MIT