Running workflows with the WorkflowExecutor, streaming events, checkpointing, and resuming long-running workflows.

WorkflowExecutor

WorkflowExecutor is the main engine that takes a built Workflow and runs it. It resolves the DAG topology, schedules nodes respecting dependencies, manages concurrency, and optionally saves checkpoints.

import { Cogitator } from '@cogitator-ai/core';
import { WorkflowExecutor } from '@cogitator-ai/workflows';

const cogitator = new Cogitator({
  llm: {
    defaultModel: 'openai/gpt-5.5',
    providers: { openai: { apiKey: process.env.OPENAI_API_KEY! } },
  },
});

const executor = new WorkflowExecutor(cogitator);

The executor requires a Cogitator instance because agentNode nodes need it to run agents. You can optionally pass a CheckpointStore as the second argument.

`.execute(workflow, input?, options?)`

Run a workflow to completion.

const result = await executor.execute(workflow, { prompt: 'Explain quantum computing' });

console.log(result.state.result);
console.log(`Completed in ${result.duration}ms`);

Input Merging

The input parameter is merged on top of workflow.initialState:

const workflow = new WorkflowBuilder<{ query: string; limit: number }>('search')
  .initialState({ query: '', limit: 10 })
  .addNode('search', searchNode)
  .build();

// limit stays at 10, query is overridden
const result = await executor.execute(workflow, { query: 'typescript generics' });

Execution Options

interface WorkflowExecuteOptions {
  maxConcurrency?: number;
  maxIterations?: number;
  checkpoint?: boolean;
  checkpointStrategy?: 'per-iteration' | 'per-node';
  workflowId?: string;
  skipNodes?: Set<string>;
  nodeResults?: Record<string, unknown>;
  onNodeStart?: (nodeName: string) => void;
  onNodeComplete?: (nodeName: string, output: unknown, duration: number) => void;
  onNodeError?: (nodeName: string, error: Error) => void;
  onNodeProgress?: (nodeName: string, progress: number) => void;
}

Option	Default	Description
`maxConcurrency`	`4`	Max parallel node executions
`maxIterations`	`100`	Safety limit for loop iterations
`checkpoint`	`false`	Enable checkpoint saving
`checkpointStrategy`	`'per-iteration'`	When to save checkpoints
`workflowId`	generated `wf_…`	Id of the run (nodes see it as `ctx.workflowId`)
`skipNodes`	—	Nodes that already ran: they are passed over, not run again (used by `resume`)
`nodeResults`	—	Outputs of the `skipNodes`, so the nodes after them get their inputs

WorkflowExecutor.execute additionally accepts (ExecutorExecuteOptions):

Option	Description
`signal`	Cancels the run between steps; passed to nodes as `ctx.signal` (agent/tool nodes and timers honour it)
`tracer`	`WorkflowTracer` — one workflow span plus a child span per node execution; defaults to the global tracer when it is enabled
`metricsCollector`	`WorkflowMetricsCollector` — workflow/node counts, latencies and retries; defaults to the global collector when it is enabled
`defaultRetry`	Retry policy for nodes without their own `config.retries` (honours `isRetryable`)
`defaultCircuitBreaker`	Per-node circuit breaker shared across runs of the executor
`deadLetterQueue`	Receives an entry for every node that finally failed
`idempotencyStore`	Reuses results of nodes that already completed for the same `workflowId` and step
`approvalStore` / `approvalNotifier` / `timerStore`	Run-level defaults for human and timer nodes
`depth`	Subworkflow nesting depth, exposed to nodes as `ctx.depth`
`onApprovalRequired(request)`	A human node opened an approval request — each chain step and escalation included
`onTimerScheduled(entry)`	A timer node stored a persisted timer in the `timerStore` (`persist: true` configs only)
`onDeadLetter(entry)`	A node that finally failed was written to the `deadLetterQueue` (`entry.id` is the queue's id)
`onCompensationStart(nodeId)` / `onCompensationComplete(nodeId)`	A node's `config.compensation` starts / finishes rolling back after a later node failed (Sagas)

resume() takes the same ExecutorExecuteOptions, and stream() takes them without the four node callbacks (onNodeStart, onNodeComplete, onNodeError, onNodeProgress — it turns those into events). The five run observers in the table above are guarded: one that throws is logged and does not break the run.

Lifecycle Hooks

Track execution progress with callbacks:

const result = await executor.execute(
  workflow,
  { topic: 'AI agents' },
  {
    onNodeStart: (node) => console.log(`Starting: ${node}`),
    onNodeComplete: (node, output, duration) => {
      console.log(`Completed: ${node} in ${duration}ms`);
    },
    onNodeError: (node, error) => {
      console.error(`Failed: ${node}`, error.message);
    },
    onNodeProgress: (node, progress) => {
      console.log(`${node}: ${progress}%`);
    },
  }
);

WorkflowResult

The return value of .execute():

interface WorkflowResult<S> {
  workflowId: string;
  workflowName: string;
  state: S;
  nodeResults: Map<string, { output: unknown; duration: number }>;
  duration: number;
  checkpointId?: string;
  error?: Error;
}

Access individual node outputs:

const result = await executor.execute(workflow, input);

const searchOutput = result.nodeResults.get('search');
console.log(searchOutput?.output);
console.log(searchOutput?.duration);

if (result.error) {
  console.error('Workflow failed:', result.error.message);
}

Streaming Events

Use .stream() to get an async iterable of events as the workflow executes. This is ideal for real-time UIs.

for await (const event of executor.stream(workflow, { topic: 'GraphQL' })) {
  switch (event.type) {
    case 'workflow_started':
      console.log(`Workflow ${event.workflowName} started`);
      break;

    case 'node_started':
      console.log(`Node ${event.nodeName} started`);
      break;

    case 'node_progress':
      console.log(`Node ${event.nodeName}: ${event.progress}%`);
      break;

    case 'node_completed':
      console.log(`Node ${event.nodeName} done in ${event.duration}ms`);
      break;

    case 'node_error':
      console.error(`Node ${event.nodeName} failed:`, event.error);
      break;

    case 'workflow_completed':
      console.log(`Done in ${event.duration}ms`);
      console.log(event.result.state);
      break;
  }
}

Event Types

Event	Fields	When
`workflow_started`	`workflowId`, `workflowName`, `timestamp`	Workflow begins
`node_started`	`nodeName`, `timestamp`	Node execution starts
`node_progress`	`nodeName`, `progress`, `timestamp`	Node reports progress
`node_completed`	`nodeName`, `output`, `duration`	Node finishes
`node_error`	`nodeName`, `error`, `timestamp`	Node throws
`workflow_completed`	`workflowId`, `result`, `duration`	Workflow finishes

Checkpointing

For long-running workflows, enable checkpointing to persist state after each iteration. If the process crashes, you can resume from the last checkpoint.

Checkpoint Stores

import { InMemoryCheckpointStore, FileCheckpointStore } from '@cogitator-ai/workflows';

const memoryStore = new InMemoryCheckpointStore();

const fileStore = new FileCheckpointStore('/tmp/checkpoints');

const executor = new WorkflowExecutor(cogitator, fileStore);

For several processes, or a process that may be replaced, keep checkpoints in Redis or Postgres. The stores take the client you already have — an @cogitator-ai/redis client or ioredis, a pg Pool — and add no dependency:

import { RedisCheckpointStore, PostgresCheckpointStore } from '@cogitator-ai/workflows';

const redisStore = new RedisCheckpointStore({ client: redis, keyPrefix: 'myapp:checkpoints' });

const pgStore = new PostgresCheckpointStore({ client: pool }); // creates cogitator_workflow_checkpoints on first use

A checkpoint saved by one process resumes in any other that reads the same store. A WorkflowManager with a checkpointStore checkpoints every run it starts — with execute(), from its schedule or on resume() — so replay(workflow, runId, node) can run that node and everything after it again, while the nodes before it keep their results, and a paused run can resume from its last checkpoint.

Runs, approvals and timers in Redis or Postgres

The other workflow state has durable stores of the same kind, so a deployment with several processes — a web server and workers, or a worker that gets replaced — shares it:

import {
  PostgresCheckpointStore,
  PostgresRunStore,
  PostgresTimerStore,
  RedisApprovalStore,
  createTimerManager,
  createWorkflowManager,
} from '@cogitator-ai/workflows';

const manager = createWorkflowManager({
  cogitator,
  runStore: new PostgresRunStore({ client: pool }), // cogitator_workflow_runs
  checkpointStore: new PostgresCheckpointStore({ client: pool }),
});

const approvalStore = new RedisApprovalStore({ client: redis, pollInterval: 1000 });
const timerStore = new PostgresTimerStore({ client: pool, claimTtl: 60_000 });
const timers = createTimerManager(timerStore);

Runs — listRuns, getStats and cleanup work across processes; Postgres filters, sorts and counts in SQL, Redis keeps sorted-set indexes so nothing scans the keyspace.
Approvals — a human node in one process resumes when another answers: onResponse fires at once for answers given through the same store and otherwise checks every pollInterval ms.
Timers — several TimerManagers can poll one store: getOverdue() claims the timers it returns for claimTtl ms (Postgres with FOR UPDATE SKIP LOCKED), so each fires once. The manager renews a claim every claimTtl / 3 while its handler runs, so a handler slower than the lease is not picked up by another worker (onClaimLost reports a claim that ran out anyway), and releases a timer it has no handler for so a manager that has one runs it right away. A timer whose worker died becomes available again when the claim runs out; one whose handler failed stays claimed until then, which spaces out the retries.

Tables are created on first use and each store accepts its own table / tablePrefix or keyPrefix. In Redis Cluster put a hash tag in the prefix ({myapp}:approvals) so a store's keys share a slot.

Enabling Checkpoints

const result = await executor.execute(workflow, input, {
  checkpoint: true,
  checkpointStrategy: 'per-node',
});

console.log('Checkpoint:', result.checkpointId);

Strategies:

'per-iteration' -- Save after each batch of parallel nodes completes
'per-node' -- Save immediately after every single node completes (more granular, slightly slower)

Resuming from Checkpoint

const result = await executor.resume(workflow, checkpointId, {
  maxConcurrency: 4,
});

The executor loads the checkpoint, restores state and completed node set, then continues execution from where it left off.

Checkpoint Data

interface WorkflowCheckpoint {
  id: string;
  workflowId: string;
  workflowName: string;
  state: unknown;
  completedNodes: string[];
  nodeResults: Record<string, unknown>;
  timestamp: number;
}

Error Handling

If a node throws, the executor catches the error, calls onNodeError and stores the node's original error in result.error — execute() itself does not throw. Nodes of the same batch that are already running finish, the ones not yet started are skipped, and result.state is the state after the last completed step.

const result = await executor.execute(workflow, input);

if (result.error) {
  console.error(`Failed at node: ${result.error.message}`);
  console.log('Completed nodes:', [...result.nodeResults.keys()]);
  console.log('Partial state:', result.state);
}

Per-node timeouts and retries

Every node can carry a NodeConfig (from a factory or the config option of addNode):

builder.addNode('call-api', callApiFn, {
  after: ['prepare'],
  config: { timeout: 10_000, retries: 3, retryDelay: 1_000 },
});

timeout applies to each attempt (a timed-out attempt fails with NodeTimeoutError and aborts the attempt's ctx.signal, so agent and tool nodes stop their in-flight calls), retries adds extra attempts and retryDelay waits between them. Nodes without their own retries use the defaultRetry execute option, which also honours its isRetryable:

import { InMemoryDLQ, InMemoryIdempotencyStore } from '@cogitator-ai/workflows';

const result = await executor.execute(workflow, input, {
  defaultRetry: { maxRetries: 3, backoff: 'exponential', initialDelay: 500 },
  defaultCircuitBreaker: { threshold: 5, resetTimeout: 30_000 },
  deadLetterQueue: new InMemoryDLQ(),
  idempotencyStore: new InMemoryIdempotencyStore(),
});

A node that still fails after its retries is written to the deadLetterQueue. The circuit breaker is kept per node name and shared by every run of the same executor that passes the same config object; while it is open the node fails at once with CircuitBreakerOpenError. With an idempotencyStore, a node that already completed for the same workflowId and step returns its stored result instead of running again.

NodeConfig also takes compensation, a rollback the executor runs when a later node fails. For more sophisticated error handling (retry policies, circuit breakers, compensation), see Sagas & Compensation.

Concurrency Control

The maxConcurrency option limits how many nodes can run in parallel. Nodes without dependency relationships between them are candidates for parallel execution.

// run up to 2 nodes at a time
await executor.execute(workflow, input, { maxConcurrency: 2 });

// fully sequential
await executor.execute(workflow, input, { maxConcurrency: 1 });

// aggressive parallelism
await executor.execute(workflow, input, { maxConcurrency: 16 });

The internal WorkflowScheduler uses the DAG topology to determine which nodes can safely run in parallel at each iteration.

Cancellation

Pass an AbortSignal to stop a run:

const controller = new AbortController();
setTimeout(() => controller.abort(), 30_000);

const result = await executor.execute(workflow, input, { signal: controller.signal });

if (controller.signal.aborted) {
  console.log('Cancelled:', result.error?.message);
}

The executor checks the signal between steps and passes it to nodes as ctx.signal: agent nodes cancel their LLM request, tool nodes hand it to the tool, timer nodes stop waiting, and subworkflows cancel their child runs.

Tracing and Metrics

import { createTracer, createMetricsCollector } from '@cogitator-ai/workflows';

const tracer = createTracer({
  serviceName: 'billing-workflows',
  exporter: 'otlp', // 'console' (default) | 'otlp' | 'jaeger' | 'zipkin' | 'noop'
  exporterEndpoint: 'http://localhost:4318/v1/traces',
});
const metrics = createMetricsCollector({ prefix: 'billing' });

await executor.execute(workflow, input, { tracer, metricsCollector: metrics });

console.log(metrics.toPrometheusFormat());
await tracer.shutdown();

The tracer records one workflow span and a child span per node execution (with node.retries); the collector counts runs and node executions with their latencies and retries. exporter: 'noop' keeps the tracer in place while sending spans nowhere, and enabled: false does the same. A WorkflowManager takes a tracer and metrics collector for all its runs, and execute(..., { tracing, metrics: { enabled: false } }) overrides them per run. A run that gets neither uses the global ones from setGlobalTracer / setGlobalMetrics when they are enabled (the defaults are disabled, so nothing is recorded until you set enabled ones).

Iteration Safety

The maxIterations limit prevents infinite loops. If the workflow hits this limit, the result will contain an error:

const result = await executor.execute(workflow, input, { maxIterations: 50 });

if (result.error?.message.includes('max iterations')) {
  console.warn('Workflow hit iteration safety limit');
}

Workflow Execution

WorkflowExecutor

`.execute(workflow, input?, options?)`

Input Merging

Execution Options

Lifecycle Hooks

WorkflowResult

Streaming Events

Event Types

Checkpointing

Checkpoint Stores

Runs, approvals and timers in Redis or Postgres

Enabling Checkpoints

Resuming from Checkpoint

Checkpoint Data

Error Handling

Per-node timeouts and retries

Concurrency Control

Cancellation

Tracing and Metrics

Iteration Safety

On this page

Workflow Execution

WorkflowExecutorCopy Anchor Link

.execute(workflow, input?, options?)Copy Anchor Link

Input MergingCopy Anchor Link

Execution OptionsCopy Anchor Link

Lifecycle HooksCopy Anchor Link

WorkflowResultCopy Anchor Link

Streaming EventsCopy Anchor Link

Event TypesCopy Anchor Link

CheckpointingCopy Anchor Link

Checkpoint StoresCopy Anchor Link

Runs, approvals and timers in Redis or PostgresCopy Anchor Link

Enabling CheckpointsCopy Anchor Link

Resuming from CheckpointCopy Anchor Link

Checkpoint DataCopy Anchor Link

Error HandlingCopy Anchor Link

Per-node timeouts and retriesCopy Anchor Link

Concurrency ControlCopy Anchor Link

CancellationCopy Anchor Link

Tracing and MetricsCopy Anchor Link

Iteration SafetyCopy Anchor Link

On this page

WorkflowExecutor

`.execute(workflow, input?, options?)`

Input Merging

Execution Options

Lifecycle Hooks

WorkflowResult

Streaming Events

Event Types

Checkpointing

Checkpoint Stores

Runs, approvals and timers in Redis or Postgres

Enabling Checkpoints

Resuming from Checkpoint

Checkpoint Data

Error Handling

Per-node timeouts and retries

Concurrency Control

Cancellation

Tracing and Metrics

Iteration Safety