# Node Types

URL: https://cogitator.app/docs/workflows/nodes

> Built-in workflow node factories for agents, tools, functions, custom logic, and human-in-the-loop interactions.

## Overview

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.

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

## Node Context

Every node function receives a `NodeContext<S>`:

```typescript
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 Result

Every node function returns a `NodeResult<S>`:

```typescript
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)

***

## agentNode

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

```typescript
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 },
});
```

### AgentNodeOptions

| Option        | Type                                | Description                                                                                                                                                    |
| ------------- | ----------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `inputMapper` | `(state, input?) => string`         | Build the agent input. Without it: `ctx.input` if it is a string, otherwise `JSON.stringify(ctx.input)`, or `JSON.stringify(state)` when the node has no input |
| `stateMapper` | `(result: RunResult) => Partial<S>` | Map agent output to state updates                                                                                                                              |
| `runOptions`  | `Partial<RunOptions>`               | Extra run options (`timeout`, `threadId`, `context`, `onToken`, ...). Sampling settings such as `temperature` belong on the `Agent`                            |

### Behavior

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`.

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

***

## toolNode

Runs a single `Tool` directly without an agent wrapper.

```typescript
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 }),
});
```

### ToolNodeOptions

| Option        | Type                       | Required | Description                                             |
| ------------- | -------------------------- | -------- | ------------------------------------------------------- |
| `argsMapper`  | `(state, input?) => TArgs` | Yes      | Map state to tool arguments                             |
| `stateMapper` | `(result) => Partial<S>`   | No       | Map tool result to state updates                        |
| `signal`      | `AbortSignal`              | No       | Signal passed to the tool; defaults to the run's signal |

### Behavior

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

***

## functionNode

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

```typescript
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 }),
  }
);
```

### Signature

```typescript
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`.

### Examples

Validation:

```typescript
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:

```typescript
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[] }),
  }
);
```

***

## customNode

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.

```typescript
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',
  };
});
```

### Signature

```typescript
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:

```typescript
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 Nodes

`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.

```typescript
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:

```typescript
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 Adapters

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

| Configuration helpers                                                                 | Adapter                    |
| ------------------------------------------------------------------------------------- | -------------------------- |
| `delayNode`, `dynamicDelayNode`, `cronWaitNode`, `untilNode`                          | `timerWorkflowNode`        |
| `mapNode`                                                                             | `mapWorkflowNode`          |
| `mapReduceNode`                                                                       | `mapReduceWorkflowNode`    |
| `subworkflowNode`, `simpleSubworkflow`, `nestedSubworkflow`, `conditionalSubworkflow` | `subworkflowWorkflowNode`  |
| `parallelSubworkflows`, `fanOutFanIn`, `scatterGather`                                | `parallelSubworkflowsNode` |

```typescript
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](/docs/swarms#swarms-inside-workflows)).

***

## Wiring Nodes into Workflows

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`):

```typescript
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();
```