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

# Your Own Workflow Shape

> Define, validate and inspect a pure graph type the runtime doesn't ship.

Write a workflow shape of your own as pure code, and the runtime records
and runs it like the shapes it ships. Use it when a pipeline, a review loop
or a dag doesn't fit the way your team moves work. For the shapes that
ship, start from the [built-in pipeline](/docs/graphs/pipeline). The contract
gives your graph type frozen data and graph lookups, and never a file, a
model, a process, a clock or storage.

The smallest graph type is two nodes and one edge, defined as JSON data:

```ts examples/custom-graph.ts (excerpt) {2-3,5-6,9} theme={null}
const definition: GraphDefinition = {
  id: 'draft-review',
  definitionVersion: 1,
  data: {},
  nodes: [
    { id: 'draft', data: {} },
    { id: 'review', data: {} },
  ],
  edges: [
    { id: 'draft-to-review', source: 'draft', target: 'review', data: {} },
  ],
};
```

`compileGraph(graphType, definition)` validates the definition before it
calls the graph type, and the compiled graph's `definition` field holds a
frozen copy with its canonical JSON and a SHA-256 digest. A definition
needs:

* **A stable `id` and a positive `definitionVersion`.**
* **Graph `data`**, whatever your type reads.
* **Nodes with stable `id` values and `data`**, and edges with stable `id`,
  `source`, `target` and `data`.

## Implement the graph type

A graph type has a stable `kind`, a positive `version` and one `compile`
method, which receives the validated, frozen definition and a `GraphKernel`
for node and neighbour lookups:

```ts examples/custom-graph.ts (excerpt) {7-8,12,13,17} theme={null}
const graphType: GraphType<
  GraphDefinition,
  State,
  Event,
  { readonly memory: 'unused' }
> = {
  kind: 'draft-review',
  version: 1,
  compile(value) {
    return {
      requirements: { memory: 'unused' },
      initialState: () => ({ next: 'draft' }),
      reduce: (state, event) =>
        event.payload.nodeId === state.next
          ? { next: state.next === 'draft' ? 'review' : 'done' }
          : state,
      decide: (state) =>
        state.next === 'done'
          ? [{ kind: 'complete', output: { approved: true } }]
          : [{
              kind: 'dispatch',
              nodeId: state.next,
              input: {},
              position: `work/${state.next}`,
            }],
```

`compile` returns pure operations:

| Operation | Result |
| - | - |
| `initialState()` | The state before events. |
| `reduce(state, event)` | The next state for one recorded event. |
| `decide(state)` | Ordered dispatch, pause, complete or fail commands. |
| `describe()` | The stable graph description for a host. |

`reduce` and `decide` receive frozen copies of their inputs and must return
JSON data; the same definition and events must give the same state and
commands. `decide` returns zero or more dispatch commands, or exactly one
`pause`, `complete` or `fail`. An empty decision means start nothing new,
and the executor accepts it only while a recorded attempt is in flight;
without in-flight work it fails instead of asking forever. `position` is the
stable identity of one requested node occurrence, unique within a decision;
once the history records a dispatch, `decide` must not return it again, and
a later event can make the same node dispatchable as a new occurrence with
its own position.

An optional `validateNodeResult(nodeId, result)` runs before the executor
saves a completion, with a frozen copy of the result. `null` accepts it; an
issue with string `code`, `path` and `message` records `node-failed` with
`RESULT_INVALID`. A malformed return or an exception propagates as an error.

`GraphBindings<Requirements>` maps a graph requirement to a host binding
type: a `required` memory requirement to a required `Memory` binding, an
`unused` one to a type with no memory binding. This layer never creates,
injects or executes a binding.

## Describe the graph

`describe()` declares what a host inspects before anything runs: phases,
nodes, edges, input and output contracts, policies, execution lanes,
requested permissions and bounds:

```ts examples/custom-graph.ts (excerpt) {4-8,26-33} theme={null}
      describe: () => ({
        inputContract: {},
        outputContract: {},
        phases: [{
          id: 'work',
          name: 'Draft and review',
          nodeIds: value.nodes.map((node) => node.id),
        }],
        nodes: value.nodes.map((node) => ({
          id: node.id,
          phaseId: 'work',
          inputContract: {},
          outputContract: {},
          laneId: null,
        })),
        policies: {
          retry: null,
          stop: null,
          concurrency: null,
          write: null,
          budget: null,
          action: null,
        },
        executionLanes: [],
        requestedPermissions: [],
        bounds: {
          dispatches: {
            min: { kind: 'known', value: 2 },
            max: { kind: 'known', value: 2 },
          },
          maxConcurrency: { kind: 'known', value: 1 },
          maxFanOut: { kind: 'known', value: 1 },
        },
      }),
```

The description names every definition node once in a phase and keeps the
definition's node order; `compileGraph` supplies the `edges` list in the
definition's edge order. Each node's lane must exist in `executionLanes`,
and a node's `phaseId` must name the phase that lists it. Each bound is
known or unknown: use a known bound only when the graph can calculate the
value, and an unknown bound with a reason when it can't. Don't invent a
maximum or a concurrency value. `validateGraphDescription(unknown)` validates
a description that didn't come from `compileGraph` and returns it frozen;
invalid data, including JSON nested more than 256 levels, throws
`GraphValidationError`.

## Prove conformance

`runGraphTypeConformance` checks a fixture without a test framework. The
fixture declares the expected initial state and command, then the expected
state and command after every event prefix:

```ts examples/custom-graph.ts (excerpt) {1,4-6} theme={null}
const conformance = runGraphTypeConformance(fixture);
if (!conformance.ok) throw new Error(JSON.stringify(conformance.failures));

const compiled = compileGraph(graphType, definition);
const state = events.reduce(compiled.reduce, compiled.initialState());
const plan = resolveGraphPlan(compiled.describe(), resolution);
```

The kit compiles the graph type in separate instances and calls each
operation more than once. It checks that the initial state, every
event-prefix state, every ordered command and the declared bounds stay the
same, that invalid definitions fail, and that your code leaves the caller's
definition and events unchanged. It counts every dispatch in the trace and
the largest dispatch set in one decision, and a declared dispatch or fan-out
maximum can't be below those. It doesn't infer `maxConcurrency`, since the
trace doesn't show which attempts overlap; the graph declares that cap or
reports it unknown. Each expected decision must hold only new requests, and
when the final decision is exactly `complete`, the observed total must meet
a known dispatch minimum. Run the file with `npx tsx custom-graph.ts`:

```json Output theme={null}
{
  "conformance": true,
  "cases": 6,
  "state": "done",
  "decision": "complete",
  "planDigest": "sha256:0b17551b9f4274dca832c040922d71251f9bf52bbd5e9462c6ed781506cd367b",
  "dispatches": {
    "min": {
      "kind": "known",
      "value": 2
    },
    "max": {
      "kind": "known",
      "value": 2
    }
  },
  "maxConcurrency": {
    "kind": "known",
    "value": 1
  },
  "maxFanOut": {
    "kind": "known",
    "value": 1
  }
}
```

The kit passed, the folded state is `done`, the next decision is
`complete`, and the plan's bounds are the two dispatches the description
declared.

## Build a callable team

`team(config)` returns a `Job`: callable work for several agents with one
task and separate answers. It doesn't use the stored graph executor, and
passing it to `run()` doesn't give it replay. For saved room messages and
turns requested by mentions, compile `teamGraphType` as
[Team conversation](/docs/patterns/team-conversation) shows. The config has:

| Field | Type | Description |
| - | - | - |
| `task` | `string` | The one task every member works on. |
| `agents` | `TeamAgent[]` | Each with a `name`, a `role`, a `brief` and an `engine` (`EngineRef`). |
| `review` | `{ kind: 'panel', config: ReviewPanelConfig }` or `{ kind: 'callback', definition: CallbackGateDefinition }` | Optional. What judges the team's result. |

The result, `TeamResult` in the outcome's `data`, holds the task, each
member's `name`, `role` and `outcome`, whether the work was `integrated`,
and the review's outcome where there was one. While a callback review
waits, it also holds the question's `requestId`.

In a Git workspace, each agent gets the task and its standing brief in its
own worktree through `isolated()`, and when an agent passes its changes
merge into your branch under the merge lock all `isolated()` jobs share.
Outside a Git repository, `isolated()` warns and runs in the shared
workspace, so a passing team result doesn't prove a merge happened; `team()`
adds no merge or lease of its own. `review` takes a review panel or a
callback gate. A panel receives the completed team result as its previous
outcome, so its reviewers see the task and each member's outcome. A callback
gate posts its question to the run's callbacks client and pauses the team
until someone answers. The question's `input` holds the gate's own `input`,
where the team sits in the graph, and the task with each member's outcome.
So a team that runs again, for example in the next turn of a loop, asks a
new question, and two teams never share an answer. Start the run again with
`recordTo` and `resume: true` over the same stored callbacks client. Members
that finished don't run again, and the team reads the answer. An answer with `approved` set to `false`
fails the review, with its `note` as the summary when it has one. Any other
answer passes the review.

The team returns the usual `Outcome`, with a `TeamResult` in its `data`.
The status is `pass` only when every agent passes and the configured review
passes; a member or review failure returns `fail` with all member results
preserved, a pause returns `paused` with its reason, and a merge conflict
returns `fail`. Before it runs, the team checks that `task` is a non-empty
string, that `agents` has at least one member with a unique `name`, that
every member has a non-empty `role`, `brief` and engine binding, and that a
configured review follows the panel or callback contract.

## Failure

* **`GraphValidationError`** from `compileGraph`: an empty identifier, a
  duplicate node or edge id, or an edge that names a node that doesn't
  exist. No node can run from this layer.
* **`RESULT_INVALID`** from `validateNodeResult`: the executor records
  `node-failed` and saves no completion.

## Limits

* **An outside graph type is trusted package code.** It can import modules
  and cause side effects on its own; the conformance kit checks public
  behaviour and doesn't stop side effects.
* **This contract doesn't schedule nodes, store runs or execute work.** The
  [graph executor](/docs/graphs/executor) runs nodes and stores runs.

<Accordion title="Full file">
  ```ts examples/custom-graph.ts theme={null}
  import {
    compileGraph,
    resolveGraphPlan,
    type GraphDefinition,
    type GraphEvent,
    type GraphType,
    type PlanResolution,
  } from '@obversa/runtime';
  import {
    runGraphTypeConformance,
    type GraphTypeConformanceFixture,
  } from '@obversa/runtime/testing';

  const definition: GraphDefinition = {
    id: 'draft-review',
    definitionVersion: 1,
    data: {},
    nodes: [
      { id: 'draft', data: {} },
      { id: 'review', data: {} },
    ],
    edges: [
      { id: 'draft-to-review', source: 'draft', target: 'review', data: {} },
    ],
  };

  type State = { readonly next: 'draft' | 'review' | 'done' };
  type Event = GraphEvent<
    'node-completed',
    { readonly nodeId: 'draft' | 'review' }
  >;

  const graphType: GraphType<
    GraphDefinition,
    State,
    Event,
    { readonly memory: 'unused' }
  > = {
    kind: 'draft-review',
    version: 1,
    compile(value) {
      return {
        requirements: { memory: 'unused' },
        initialState: () => ({ next: 'draft' }),
        reduce: (state, event) =>
          event.payload.nodeId === state.next
            ? { next: state.next === 'draft' ? 'review' : 'done' }
            : state,
        decide: (state) =>
          state.next === 'done'
            ? [{ kind: 'complete', output: { approved: true } }]
            : [{
                kind: 'dispatch',
                nodeId: state.next,
                input: {},
                position: `work/${state.next}`,
              }],
        describe: () => ({
          inputContract: {},
          outputContract: {},
          phases: [{
            id: 'work',
            name: 'Draft and review',
            nodeIds: value.nodes.map((node) => node.id),
          }],
          nodes: value.nodes.map((node) => ({
            id: node.id,
            phaseId: 'work',
            inputContract: {},
            outputContract: {},
            laneId: null,
          })),
          policies: {
            retry: null,
            stop: null,
            concurrency: null,
            write: null,
            budget: null,
            action: null,
          },
          executionLanes: [],
          requestedPermissions: [],
          bounds: {
            dispatches: {
              min: { kind: 'known', value: 2 },
              max: { kind: 'known', value: 2 },
            },
            maxConcurrency: { kind: 'known', value: 1 },
            maxFanOut: { kind: 'known', value: 1 },
          },
        }),
      };
    },
  };

  const events: readonly Event[] = [
    { type: 'node-completed', version: 1, payload: { nodeId: 'draft' } },
    { type: 'node-completed', version: 1, payload: { nodeId: 'review' } },
  ];
  const identity = {
    source: 'npm:@example/draft-review',
    version: '1.0.0',
    digest: 'sha256:1111111111111111111111111111111111111111111111111111111111111111',
  } as const;
  const resolution: PlanResolution = {
    package: identity,
    admission: { package: identity, permissions: [] },
    executionLanes: [],
  };
  const fixture = {
    graphType,
    definition,
    events,
    invalidDefinitions: [{
      ...definition,
      nodes: [...definition.nodes, { id: 'draft', data: {} }],
    }] as const,
    planResolution: resolution,
    expected: {
      states: [
        { next: 'draft' },
        { next: 'review' },
        { next: 'done' },
      ],
      commands: [
        [{ kind: 'dispatch', nodeId: 'draft', input: {}, position: 'work/draft' }],
        [{ kind: 'dispatch', nodeId: 'review', input: {}, position: 'work/review' }],
        [{ kind: 'complete', output: { approved: true } }],
      ],
      bounds: {
        dispatches: {
          min: { kind: 'known', value: 2 },
          max: { kind: 'known', value: 2 },
        },
        maxConcurrency: { kind: 'known', value: 1 },
        maxFanOut: { kind: 'known', value: 1 },
      },
    },
  } satisfies GraphTypeConformanceFixture<
    GraphDefinition,
    State,
    Event,
    { readonly memory: 'unused' }
  >;

  const conformance = runGraphTypeConformance(fixture);
  if (!conformance.ok) throw new Error(JSON.stringify(conformance.failures));

  const compiled = compileGraph(graphType, definition);
  const state = events.reduce(compiled.reduce, compiled.initialState());
  const plan = resolveGraphPlan(compiled.describe(), resolution);

  console.log(JSON.stringify({
    conformance: conformance.ok,
    cases: conformance.cases,
    state: state.next,
    decision: compiled.decide(state)[0]?.kind,
    planDigest: plan.digest,
    dispatches: plan.plan.bounds.dispatches,
    maxConcurrency: plan.plan.bounds.maxConcurrency,
    maxFanOut: plan.plan.bounds.maxFanOut,
  }, null, 2));
  ```
</Accordion>

## Next steps

* [Built-in pipeline](/docs/graphs/pipeline): the dag form that ships, as a
  stored pipeline.
* [Plan admission](/docs/graphs/plan-admission): binding a description to a
  host's admission record before a run.
* [Graph executor](/docs/graphs/executor): how a dispatch runs and how a run
  resumes.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.