Write a pipeline

You will build acme/tidy: it cleans a JSONL file, stops if nothing is left, and waits for a person to accept the result. Along the way you will see where pipeline files live, how legs are wired, and how to check a pipeline before it runs.

1. Create a pack

Pipelines and flows live in a pack: a directory with a pack.toml, a pipelines/ directory and a flows/ directory. cairn init pack creates one:

cairn init pack acme

The pack name must be a valid Python identifier (snake_case). The scaffold contains, among other files:

acme/
  __init__.py
  pack.toml
  pyproject.toml
  flows/hello.md
  pipelines/hello.yaml
  policies/default.yaml

policies/default.yaml is a starter policy that asks for approval on every action. Delete it for this guide, or keep it and expect an extra approval before each step.

2. Make cairn find the pack

cairn only sees pipelines that belong to a registered pack. There are two ways to register one:

• Point CAIRN_PACK_PATH at the pack, or at a directory that contains packs. Several entries are separated by :.

  export CAIRN_PACK_PATH=$PWD     # the directory that contains acme/

• Or install it as a package (pip install -e acme), which registers it through its cairn.packs entry point.

Start (or restart) cairn dev or the server with the variable set, and check:

$ cairn pipeline ls
  acme/hello  1 legs (template)  sha256:a7040e93b0b05eba
  …

3. Add the flow it runs

A pipeline can only use flows from its own pack. Add acme/flows/scrub.yaml, a small data flow (Write a flow explains it):

name: scrub
kind: data
description: Normalize a JSONL file, drop exact duplicates, redact PII, export JSONL.

stages:
  - stage: ingest.normalize
    config:
      path: $config.path
  - stage: dedup.exact
  - stage: pii.redact
  - stage: export.jsonl

4. Write the pipeline

Create acme/pipelines/tidy.yaml. The file name must match name:.

name: tidy
kind: pipeline
description: Clean a JSONL dataset, stop if nothing is left, and hold the result for a person.

inputs:
  path: {type: Dataset, description: "JSONL file to clean."}
  min_rows: {type: int, default: 1, description: "Fewest rows the cleaned file may have."}

governance:
  budget: {usd: 1}
  approvers: data-team

steps:
  - id: clean
    flow: scrub
    with:
      path: $input.path
    consumes: {path: Dataset}
    produces: {export: Dataset}

  - gate: has-rows
    depends_on: [clean]
    predicate: steps.clean.findings.pii_redact_rows >= input.min_rows

  - gate: accept
    depends_on: [has-rows]
    show: [clean.pii_redact_rows, clean.dedup_exact_dropped, clean.pii_redact_redacted_rows, clean.export]

Piece by piece:

• inputs are what the person launching the pipeline provides. An input with no default is required.
• governance sets the pipeline’s budget and names who is expected to approve its gates.
• clean is a flow leg. with: fills the flow’s $config.path placeholder from the pipeline input.
• has-rows is a data gate. Its predicate: is a JMESPath expression over input and steps. A data flow’s numbers sit under findings, so the predicate reads steps.clean.findings.pii_redact_rows. If it is false, the pipeline stops.
• accept is a person gate. show: lists the fields the approver sees, as <leg>.<field>; for these, cairn looks in the leg’s findings for you.

To find a leg’s field names, run the pipeline once and read cairn remote findings <pipeline-id>.

Name pipelines and agentic flows differently

In this version, the console and POST /start refuse to launch a pipeline whose name matches an agentic flow (flows/<name>.md) in the same pack, with “names BOTH a pipeline and a flow/template”. Give the pipeline its own name. A data flow with the same name does not cause this.

5. Check it before it runs

cairn pipeline show prints the legs and gates. cairn pipeline explain resolves everything the launch will resolve, without running anything:

$ cairn pipeline explain acme/tidy -e sandbox -i path=/data/chats.jsonl
…
  [gate] has-rows  gate:has-rows   after clean
       predicate: steps.clean.findings.pii_redact_rows >= input.min_rows  → STOP (gate_blocked) when false
…
  [gate] accept  gate:accept   after has-rows
       gate[this leg] accept                       human (HITL)             authority=pipeline def — approvers: data-team
…
  PROBLEMS
    ! step 'clean': external_write leg has no approval gate on its path — nothing pauses before it writes (add a `gate:` step upstream, or an environment policy that injects one)

The warning is fair: the clean leg writes a file before any gate. For this guide that is fine. For a leg that writes somewhere that matters, put a gate before it.

Mistakes fail when the file is read, with the file and field named. Unknown fields are rejected, and so is a gate that has both kinds of answer:

pipelines/t.yaml: gate 'g' declares both `predicate:` and `approvers:` — a gate is answered by data OR a human, never both. Split into two gates if you need both verdicts.

6. Run it

cairn remote launch acme/tidy -i path=/data/chats.jsonl --environment sandbox --watch

  paused
    clean              done                 run=run-flow-acme--scrub-d11ff2
    has-rows           done                 run=None
    accept             awaiting_approval    run=None
⏸ paused — 1 step(s) waiting on a human

The data gate passed on its own. Approve accept in the console’s Approvals, or from the terminal:

cairn remote advance pl_f70a69c319de4f0e accept

advanced — pipeline status=done

The console records your signed-in name as the approver; prefer it when the approver matters.

Next

• Write a flow: stages, placeholders and findings.
• Pipeline and flow YAML: every field.
• Gates and approvals: policy gates and durable pauses.