Node Types

OverviewCopy Anchor Link

Every node in a workflow is a WorkflowNode<S> with a name and an async fn that receives a NodeContext<S> and returns a NodeResult<S>. Instead of writing these by hand, use the built-in node factories.

import { agentNode, toolNode, functionNode, customNode } from '@cogitator-ai/workflows';

Node ContextCopy Anchor Link

Every node function receives a NodeContext<S>:

interface NodeContext<S> {
  state: S;
  nodeId: string;
  workflowId: string;
  step: number;
  input?: unknown;
  reportProgress?: (progress: number) => void;
}

• state -- Current workflow state snapshot
• input -- Output from upstream dependency nodes (single value if one dep, array if multiple)
• step -- The executor iteration the node runs in
• reportProgress -- Report 0-100 progress (clamped) for onNodeProgress and node_progress stream events; the executor always sets it, the type marks it optional

At run time the executor passes an ExtendedNodeContext<S> (exported from @cogitator-ai/workflows) that adds cogitator, signal (the run's AbortSignal, or a per-attempt signal when the node has a timeout), depth (subworkflow nesting) and the run-level approvalStore, approvalNotifier, timerStore, onApprovalRequired and onTimerScheduled. Cast to it in a custom node when you need them.

Node ResultCopy Anchor Link

Every node function returns a NodeResult<S>:

interface NodeResult<S> {
  state?: Partial<S>;
  output?: unknown;
  next?: string | string[];
}

• state -- Partial state updates merged into the workflow state
• output -- Value passed to downstream nodes via ctx.input
• next -- Override automatic edge routing with explicit next node(s)

agentNodeCopy Anchor Link

Runs a Cogitator Agent as a workflow node. The executor injects a cogitator instance into the context automatically.

import { Agent } from '@cogitator-ai/core';
import { agentNode } from '@cogitator-ai/workflows';

type ResearchState = { topic: string; findings?: string };

const researcher = new Agent({
  name: 'researcher',
  model: 'openai/gpt-5.5',
  instructions: 'You are a research assistant.',
  temperature: 0.3,
});

const node = agentNode<ResearchState>(researcher, {
  inputMapper: (state) => `Research the topic: ${state.topic}`,
  stateMapper: (result) => ({ findings: result.output }),
  runOptions: { timeout: 60_000 },
});

AgentNodeOptionsCopy Anchor Link

BehaviorCopy Anchor Link

1. inputMapper converts the current state into a prompt string
2. The agent runs via cogitator.run(agent, { input, signal, ...runOptions }) — the run's abort signal (or the node's timeout) cancels the LLM call
3. stateMapper extracts state updates from the RunResult
4. The agent's raw output is passed downstream via output

A failed run is rethrown as an error naming the agent, workflow and node, with the original error as cause.

builder.addNode(
  'research',
  agentNode(researcher, {
    inputMapper: (state) => state.query,
    stateMapper: (result) => ({ answer: result.output }),
  })
);

toolNodeCopy Anchor Link

Runs a single Tool directly without an agent wrapper.

import { tool } from '@cogitator-ai/core';
import { toolNode } from '@cogitator-ai/workflows';
import { z } from 'zod';

const fetchUrl = tool({
  name: 'fetch_url',
  description: 'Fetch content from a URL',
  parameters: z.object({ url: z.string() }),
  execute: async ({ url }) => {
    const res = await fetch(url);
    return res.text();
  },
});

const node = toolNode(fetchUrl, {
  argsMapper: (state) => ({ url: state.targetUrl }),
  stateMapper: (result) => ({ htmlContent: result as string }),
});

ToolNodeOptionsCopy Anchor Link

BehaviorCopy Anchor Link

1. argsMapper builds typed arguments from state
2. The tool's execute function runs with those arguments and a ToolContext whose agentId is workflow:<workflowId>:<nodeId> and runId is the workflow id
3. stateMapper extracts state updates from the result
4. The raw tool result is passed downstream

functionNodeCopy Anchor Link

Runs an arbitrary async function. The simplest node type for data transformation, validation, or any custom logic.

import { functionNode } from '@cogitator-ai/workflows';

type TextState = { rawData?: string; cleanData?: string };

const node = functionNode<TextState>(
  'normalize',
  async (state) => {
    const data = state.rawData ?? '';
    return data.trim().toLowerCase().replace(/\s+/g, ' ');
  },
  {
    stateMapper: (output) => ({ cleanData: output as string }),
  }
);

SignatureCopy Anchor Link

function functionNode<S, O>(
  name: string,
  fn: (state: S, input?: unknown) => Promise<O>,
  options?: { stateMapper?: (output: unknown) => Partial<S> }
): WorkflowNode<S>;

The function receives the full state and optional input from upstream nodes. Return any value -- it becomes the node's output.

ExamplesCopy Anchor Link

Validation:

type SignupState = { email?: string };

const validate = functionNode<SignupState>('validate', async (state) => {
  if (!state.email?.includes('@')) {
    throw new Error('Invalid email');
  }
  return { valid: true };
});

Data transformation:

type Item = { name: string; score: number };
type ItemState = { items: Item[]; filteredItems?: Item[] };

const transform = functionNode<ItemState>(
  'transform',
  async (state) => {
    return state.items.filter((item) => item.score > 0.5);
  },
  {
    stateMapper: (output) => ({ filteredItems: output as Item[] }),
  }
);

customNodeCopy Anchor Link

Full control over the node context and result. Use when you need access to workflowId, step, progress reporting, or want to explicitly set next routing.

import { customNode } from '@cogitator-ai/workflows';

const node = customNode<StepState>('complex-step', async (ctx) => {
  ctx.reportProgress?.(0);

  const data = await fetchExternalAPI(ctx.state.apiEndpoint);
  ctx.reportProgress?.(50);

  const processed = await processData(data);
  ctx.reportProgress?.(100);

  return {
    state: { processedData: processed, lastStep: ctx.nodeId },
    output: processed,
    next: processed.needsReview ? 'review' : 'finalize',
  };
});

SignatureCopy Anchor Link

function customNode<S>(
  name: string,
  fn: (ctx: NodeContext<S>) => Promise<NodeResult<S>>
): WorkflowNode<S>;

The next field in the result lets you override the DAG routing dynamically at runtime: the executor runs the named nodes next instead of following the node's edges.

To reach the runtime extras (Cogitator, abort signal, stores), cast the context:

import { customNode, type ExtendedNodeContext } from '@cogitator-ai/workflows';

const summarize = customNode<StepState>('summarize', async (ctx) => {
  const { cogitator, signal } = ctx as ExtendedNodeContext<StepState>;
  const result = await cogitator.run(summarizer, { input: ctx.state.apiEndpoint, signal });
  return { output: result.output };
});

Human-in-the-Loop NodesCopy Anchor Link

approvalNode, choiceNode, inputNode, ratingNode, chainNode, managementChain and humanNode build approval configurations (NamedHumanNodeConfig, a HumanNodeConfig with its name set); humanWorkflowNode turns them into workflow nodes that pause until a response arrives in the ApprovalStore. The store comes from the node options or from the approvalStore execute option; a human node with neither throws.

import {
  WorkflowBuilder,
  approvalNode,
  choiceNode,
  humanWorkflowNode,
  InMemoryApprovalStore,
  ConsoleNotifier,
} from '@cogitator-ai/workflows';

type ExpenseState = { amount: number; approved?: boolean };

const approvalStore = new InMemoryApprovalStore();

const builder = new WorkflowBuilder<ExpenseState>('expense')
  .initialState({ amount: 0 })
  .addNode(
    'manager-approval',
    humanWorkflowNode(
      approvalNode<ExpenseState>('manager-approval', {
        title: 'Approve expense',
        description: (state) => `Approve expense of $${state.amount}?`,
        assignee: 'manager@company.com',
        timeout: 24 * 60 * 60 * 1000,
        timeoutAction: 'reject',
      }),
      {
        approvalStore,
        approvalNotifier: new ConsoleNotifier(),
        stateMapper: (result) => ({ approved: result.approved }),
      }
    )
  )
  .addNode(
    'select-action',
    humanWorkflowNode(
      choiceNode<ExpenseState>('select-action', {
        title: 'How should we handle it?',
        choices: [
          { id: 'fix', label: 'Fix', value: 'fix' },
          { id: 'escalate', label: 'Escalate', value: 'escalate' },
        ],
      }),
      { approvalStore }
    ),
    { after: ['manager-approval'] }
  );

Answer from an API handler or UI — another process works too when both use the same durable store:

import { ApprovalAlreadyAnsweredError } from '@cogitator-ai/workflows';

const [request] = await approvalStore.getPendingForAssignee('manager@company.com');

try {
  await approvalStore.submitResponse({
    requestId: request.id,
    decision: true,
    respondedBy: 'manager@company.com',
    respondedAt: Date.now(),
    comment: 'Within budget',
  });
} catch (error) {
  if (error instanceof ApprovalAlreadyAnsweredError) {
    console.log('Already answered:', error.existing?.decision);
  } else {
    throw error;
  }
}

The onApprovalRequired execute option is called with every request a human node opens — each chain step and escalation included — once the node is waiting for its answer. getPendingRequests(workflowId?) lists open requests of a run; each request carries workflowId, nodeId, type, title, description, choices, assignee, deadline and priority.

The node output is { approved, decision, timedOut, escalated, withdrawn }; escalated is true when the answer came from the escalateTo assignee after a timeout. stateMapper receives the full HumanNodeResult — approved, decision, response, timedOut, escalated and withdrawn — so copy what a following conditional needs into state there.

What counts as approved depends on the type: approve-reject needs true or 'approve', multi-choice any non-null choice, free-form a non-empty string and numeric-rating a number.

• First answer wins. A second submitResponse for the same request — another approver, or a timeout firing just after someone answered — throws ApprovalAlreadyAnsweredError carrying the answer that stands (existing), and the node keeps that answer. submitOrExisting(store, response) submits unless an answer exists and returns whichever answer stands.
• Withdrawn requests. Deleting a request nobody answered yet (deleteRequest, or the in-memory store's expiry) withdraws it: the waiting node finishes with withdrawn: true and approved: false instead of waiting forever. The withdrawn answer has respondedBy: WITHDRAWN.
• Timeouts. With timeout, timeoutAction decides: 'approve' auto-approves, 'reject' auto-rejects, 'fail' (default) answers with an error decision, and 'escalate' opens a new request for escalateTo that waits at least 30 minutes before failing. isUnanswered(response) tells a timeout, withdrawal or expired escalation apart from a person's decision; those never count as approval.
• Waits that survive a restart. A request's id is derived from the run and the question (node, type, title, description, choices, assignee and the workflow state), so a node that runs again after a restart finds its own request: an answer given while the process was down is used at once, an open request is waited on again without a second notification, and the timeout keeps counting from the original deadline. A node visited again in a loop asks about a changed state, which makes a new request. An escalation gets its id from the request it escalates and keeps its own deadline, so it is picked up the same way. After a restart, manager.recoverRuns({ approvalStore }) resumes the runs the stopped process left running or waiting from their last checkpoint (the newest one saved, even if the run record never heard of it); call it once at startup from the one process that runs these workflows, with a durable store (PostgresApprovalStore, RedisApprovalStore, FileApprovalStore).
• Chains. chainNode and managementChain ask each approver in turn; a required step that rejects (or times out) stops the chain with approved: false, and timeoutAction: 'skip' on a step moves on to the next approver.
• Delegation. Wrap a store with withDelegation(store): a response with delegatedTo answers the request with 'delegated' and opens a copy assigned to the delegate.

Timer, Map-Reduce and Subworkflow AdaptersCopy Anchor Link

Timer, map-reduce and subworkflow helpers return configurations. Wrap them with the matching adapter to add them to a builder:

builder
  .addNode('cool-down', timerWorkflowNode(delayNode('cool-down', 5_000)))
  .addNode('analyze', subworkflowWorkflowNode(subConfig), { after: ['cool-down'] });

Timers stop waiting when the run is aborted (signal, pause(), cancel()): the node fails with an AbortError instead of completing, so it is not checkpointed and a resumed run waits again. A persist: true timer interrupted this way is cancelled in its store, and the resumed node waits only until the original firesAt, so a pause does not shorten or extend the wait (other timers start their delay over). A persisted timer cancelled through its store while the node waits (TimerManager.cancel) is not marked fired: the node reports cancelled: true and onCancelled when the wait ends. Subworkflow failures propagate to the parent (or follow onError — with 'catch' the parent continues and the node outputs { error: { name, message } }), and timed-out child runs are cancelled. A timer config with persist: true records its timer in the timerStore (node option or execute option) so it shows up in a TimerManager, and reports the stored entry to the onTimerScheduled execute option; the wait itself still happens in the running process.

To run a swarm as a step, use swarmNode, conditionalSwarmNode or parallelSwarmsNode from @cogitator-ai/swarms (see Swarms).

Wiring Nodes into WorkflowsCopy Anchor Link

All node factories and adapters return WorkflowNode<S> objects. Pass them directly to WorkflowBuilder.addNode (a node's config — timeout, retries, retryDelay, compensation — is merged with the config passed to addNode):

type MyState = { url: string; content?: string; analysis?: string };

const workflow = new WorkflowBuilder<MyState>('pipeline')
  .initialState({ url: '' })
  .addNode(
    'fetch',
    toolNode(fetchUrl, {
      argsMapper: (state) => ({ url: state.url }),
      stateMapper: (result) => ({ content: result as string }),
    })
  )
  .addNode(
    'analyze',
    agentNode(analyst, {
      inputMapper: (state) => state.content!,
      stateMapper: (result) => ({ analysis: result.output }),
    }),
    { after: ['fetch'] }
  )
  .addNode(
    'format',
    functionNode('format', async (state) => {
      return `# Analysis\n\n${state.analysis}`;
    }),
    { after: ['analyze'] }
  )
  .build();