> ## Documentation Index
> Fetch the complete documentation index at: https://docs.obversa.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Built-in pipeline

> Run ordered stages through the durable graph executor.

`dagGraphType` is the built-in dependency-graph form. A pipeline is one use of
that form: each stage is a node, and each edge says which stage must finish
before another can start.

The graph form only works with frozen data. It folds recorded events into state
and decides what runs next. `createGraphExecutor` stores each dispatch before it
runs the matching stage.

## Run the example

From an Obversa checkout, run:

```bash theme={null}
pnpm example:pipeline
```

The command stores and runs a three-stage pipeline through the public executor.
It prints one JSON report:

```json theme={null}
{"executor":"complete","output":{"nodes":{"draft":{"article":"ready"},"review":{"approved":true},"publish":{"published":true}}},"dispatches":3,"order":["draft","review","publish"],"planDigest":"sha256:ebbf122994643028c5222a0afaf8d7a99a257716e00f0e255b15f315e4b862ed","bounds":{"dispatches":{"min":{"kind":"known","value":3},"max":{"kind":"known","value":3}},"maxConcurrency":{"kind":"known","value":1},"maxFanOut":{"kind":"known","value":1}}}
```

The clean-consumer check compiles and runs the same source from the packed
package with TypeScript 6 and TypeScript 7. Its source is
`examples/packages/pipeline.ts`.

## Define a pipeline

```ts theme={null}
import {
  compileGraph,
  dagGraphType,
  type DagDefinition,
} from '@obversa/runtime';
import { defineGraphDefinition } from '@obversa/runtime/testing';

const definition: DagDefinition = defineGraphDefinition({
  id: 'release-pipeline',
  definitionVersion: 1,
  data: {
    globalConcurrency: 1,
    keyedConcurrency: {},
    stopOnError: true,
    retryCapPerNode: 0,
  },
  nodes: [
    { id: 'draft', data: { kind: 'required', key: null } },
    { id: 'review', data: { kind: 'required', key: null } },
    { id: 'publish', data: { kind: 'required', key: null } },
  ],
  edges: [
    { id: 'draft-to-review', source: 'draft', target: 'review', data: {} },
    { id: 'review-to-publish', source: 'review', target: 'publish', data: {} },
  ],
});

const graph = compileGraph(dagGraphType, definition);
```

The example stores `graph.definition` with its resolved plan. It gives the
executor one data function for each node. The review function reads the draft
result, and the publish function reads the review result.

## Restart and resume

If the executor process stops after it records a stage dispatch and before that
stage completes, a new executor over the same store returns `waiting` with that
stage's position. Call `resume` with that position after the old process has
stopped.

Resume does not write a second dispatch event. Completed earlier stages stay
complete. The pipeline then continues from that stage.

The executor does not lock the run. Do not call `resume` while the previous
executor still runs.

This example pipeline sets `globalConcurrency` to 1, so at most one stage is in
flight and `waiting.positions` has one entry.

## Node behaviour

* **Required nodes** block their dependants after failure and fail the pipeline
  after finalizers run.
* **Optional nodes** can fail without blocking their dependants.
* **Expected skips** use a completed result with `skipped: true` and do not
  block dependants.
* **Finalizer nodes** have no edges and run after the other nodes settle.
* **Paused nodes** stop new dispatches and keep the recorded pause reason.

A node with a `lane` object uses the engine route stored in the plan. A node
without one uses its data function and is data-only.
Nodes can share a lane id when every copy of its declaration is identical; a
conflicting declaration is rejected. A failed finalizer fails the graph even
when every worker completed successfully.

`stopOnError: true` stops new work after a required failure. Set it to `false`
when independent branches and declared retries must continue.

## Limits and output

`globalConcurrency` caps all running nodes. A node with a non-null `key` also
uses the matching cap in `keyedConcurrency`.

Each dispatch receives a stable position such as `dag/review/1`. Its input has
the completed results of its direct predecessors, keyed by node name. The final
output maps every completed node name to its returned payload. A failed optional
node is omitted because it has no completed payload. The example's plan declares
exactly three dispatches and a maximum concurrency of one.