# Workflow Execution URL: https://cogitator.app/docs/workflows/execution > 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. ```typescript 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. ```typescript 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`: ```typescript 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 ```typescript interface WorkflowExecuteOptions { maxConcurrency?: number; maxIterations?: number; checkpoint?: boolean; checkpointStrategy?: 'per-iteration' | 'per-node'; workflowId?: string; skipNodes?: Set; nodeResults?: Record; 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](/docs/workflows/sagas#compensation-in-the-executor)) | `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: ```typescript 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()`: ```typescript interface WorkflowResult { workflowId: string; workflowName: string; state: S; nodeResults: Map; duration: number; checkpointId?: string; error?: Error; } ``` Access individual node outputs: ```typescript 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. ```typescript 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 ```typescript 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: ```typescript 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](/docs/workflows/scheduling#pause-and-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: ```typescript 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 `TimerManager`s 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 ```typescript 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 ```typescript 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 ```typescript interface WorkflowCheckpoint { id: string; workflowId: string; workflowName: string; state: unknown; completedNodes: string[]; nodeResults: Record; 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. ```typescript 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`): ```typescript 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`: ```typescript 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](/docs/workflows/sagas). ## Concurrency Control The `maxConcurrency` option limits how many nodes can run in parallel. Nodes without dependency relationships between them are candidates for parallel execution. ```typescript // 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: ```typescript 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 ```typescript 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: ```typescript const result = await executor.execute(workflow, input, { maxIterations: 50 }); if (result.error?.message.includes('max iterations')) { console.warn('Workflow hit iteration safety limit'); } ```