Workflows API

Workflow state types must satisfy WorkflowState (Record<string, unknown>): use a type alias or interface MyState extends WorkflowState.

Building and RunningCopy Anchor Link

import {
  WorkflowBuilder,
  WorkflowExecutor,
  agentNode,
  functionNode,
} from '@cogitator-ai/workflows';
import type { WorkflowState } from '@cogitator-ai/types';

interface ReportState extends WorkflowState {
  topic?: string;
  draft?: string;
  words?: number;
}

const workflow = new WorkflowBuilder<ReportState>('report')
  .initialState({ topic: 'WebAssembly' })
  .addNode(
    'write',
    agentNode<ReportState>(writer, {
      inputMapper: (state) => `Write a short report about ${state.topic}`,
      stateMapper: (result) => ({ draft: result.output }),
    })
  )
  .addNode(
    'count',
    functionNode<ReportState>('count', async (state) => state.draft?.split(/\s+/).length ?? 0, {
      stateMapper: (words) => ({ words: words as number }),
    }),
    { after: ['write'] }
  )
  .build();

const result = await new WorkflowExecutor(cog).execute(workflow);
console.log(result.state.words, result.error);

WorkflowBuilder<S>Copy Anchor Link

NodeFn<S> is (ctx: NodeContext<S>) => Promise<NodeResult<S>> with ctx = { state, input?, nodeId, workflowId, step, reportProgress? } and NodeResult = { state?: Partial<S>; output?; next? }. NodeConfig<S> is { name?, timeout?, retries?, retryDelay?, compensation?: CompensationConfig<S> }; compensation ({ compensate, compensateCondition?, compensateOrder?, compensateTimeout? }) is rolled back by the executor when a later node of the run fails (Sagas). build() throws when there is no single root (independent roots need a root addParallel or .entryPoint()) and when a node after a loop is neither its back nor its exit node. See Builder.

WorkflowExecutorCopy Anchor Link

ExecutorExecuteOptions extends WorkflowExecuteOptions:

These five callbacks are the run observers of WorkflowExecuteOptionsV2, also accepted by the manager's execute(); one that throws is logged and does not break the run.

WorkflowResult<S>: { workflowId, workflowName, state, nodeResults: Map<string, { output, duration }>, duration, checkpointId?, error? }. NodeExecutionError (nodeName) and NodeTimeoutError (nodeName, timeoutMs) are the errors nodes fail with. WorkflowScheduler is the executor's internal DAG scheduler (dependency graph, ready nodes, concurrency), exported for custom executors. See Execution.

NodesCopy Anchor Link

The *WorkflowNode adapters turn the configs built by the helpers below (approvalNode, delayNode, mapNode, subworkflowNode, …) into nodes for addNode. See Nodes.

CheckpointsCopy Anchor Link

CheckpointStore: save(checkpoint), load(id), list(workflowName), delete(id).

Human in the LoopCopy Anchor Link

ApprovalStore: createRequest, getRequest, getPendingRequests(workflowId?), getPendingForAssignee(assignee), submitResponse, getResponse, deleteRequest, onResponse(requestId, callback). HumanNodeResult: { approved, decision, response, state, timedOut?, escalated?, withdrawn? }; a humanWorkflowNode outputs { approved, decision, timedOut, escalated, withdrawn }, where escalated means the escalateTo assignee answered.

Timers and CronCopy Anchor Link

TimerManagerConfig: { pollInterval?, batchSize?, processOverdueOnStart?, cleanupMaxAge?, cleanupInterval?, enableCleanup?, maxRetries?, onError?, onTimerFired?, onTimerMissed?, onDeadLetter?, onClaimLost? }. See Scheduling.

SagasCopy Anchor Link

CompensationHooks: { onStepStart?(nodeId), onStepComplete?(result: CompensationResult) }, called around each step that runs (steps skipped by their condition report neither). The executor runs the same logic for nodes with config.compensation. See Sagas.

Map-Reduce and SubworkflowsCopy Anchor Link

See Patterns.

Run Management and TriggersCopy Anchor Link

schedule() with cron queues one run at the next occurrence; registerCronJob queues a run (with jobOptions, triggerId: 'cron:<jobId>') on every occurrence while the manager is started. pause() needs a checkpointStore and aborts the run; resume() continues it from its last checkpoint with the options it was started with (unless you pass new ones) and resolves once it is running. Scheduled runs get the manager's checkpoint store, tracer, metrics and their timeout ?? defaultTimeout; ScheduleOptions.maxRetries retries a failed scheduled run (counted in metadata.retryAttempt). A run's workflowId defaults to its run id. See Scheduling.

ObservabilityCopy Anchor Link

Pass a tracer and metrics collector to execute() as tracer / metricsCollector. A run that passes none uses the global ones (setGlobalTracer / setGlobalMetrics) when they are enabled — tracer.isSampled() / metrics.isEnabled(); the built-in globals are disabled.

TypesCopy Anchor Link

Re-exported from @cogitator-ai/types: Workflow, WorkflowState, WorkflowNode, WorkflowResult, WorkflowEvent, StreamingWorkflowEvent, WorkflowExecuteOptions, WorkflowCheckpoint, CheckpointStore, NodeContext, NodeResult, NodeConfig, NodeFn, the edge and builder option types, the saga types (RetryConfig, CircuitBreakerConfig, CompensationConfig, DeadLetterEntry, IdempotencyStore, …), the human types (ApprovalRequest, ApprovalResponse, ApprovalStore, ApprovalNotifier, HumanNodeConfig, …), the timer types (TimerEntry, TimerStore), the manager types (WorkflowRun, RunStore, WorkflowManager, ScheduleOptions, …), the trigger types (CronTriggerConfig, WebhookTriggerConfig, EventTriggerConfig, WorkflowTrigger, TriggerManager, …) and the tracing types (TracingConfig, MetricsConfig, WorkflowSpan, …).