Pipelines

A pipeline is one YAML file, kept in git with the actions it uses. Steps read streams and write streams, every item is persisted, and the engine validates the whole graph before anything runs.

Steps only know streams

A step names its input streams with in and its outputs with out. It never refers to another step, so the graph is whatever the streams connect:

• several steps writing one stream merge their outputs, such as several crawlers feeding listings.raw;
• several steps reading one stream each receive every item;
• inserting a step means pointing it at an existing stream and re-pointing one consumer to its output.

pipeline: property-sourcing
description: Daily sourcing of renovation opportunities

vars: { city: Lyon, maxPrice: 400000, minScore: 70, minMarginPct: 15 }
secrets: [sourceToken]

triggers:
  daily: { cron: "0 6 * * *", tz: Europe/Paris }

streams:
  listings.raw: { retention: 90d, redact: [/seller/phone] }

steps:
  crawl-a:
    action: lab/crawl-source-a@1.2
    in: trigger.daily
    out: listings.raw
    config:
      filters: { city: "${vars.city}", maxPrice: "${vars.maxPrice}" }
      token: "${secrets.sourceToken}"
  crawl-b:
    action: lab/crawl-source-b@0.9
    in: trigger.daily
    out: listings.raw
  dedupe:
    action: dedupe
    in: listings.raw
    out: listings.unique
    config: { key: "item.addressNorm + ':' + string(item.surfaceM2)", window: 30d }
  score:
    action: lab/score-listing@2
    in: listings.unique
    out: listings.scored
    concurrency: 8
  keep-good:
    action: filter
    in: listings.scored
    out: listings.shortlist
    config: { expr: "item.score >= vars.minScore" }

Thresholds live in filter steps, not inside actions, so one scoring action serves every pipeline and every threshold is visible in the file.

Expressions, variables and secrets

Every expression is CEL: typed, free of side effects and bounded in cost. millstream check compiles and type-checks each one before the pipeline runs.

• Variables declare their defaults, which also give their types. A deployment overrides one with MILLSTREAM_VAR_<NAME>, a run with --var, and the resolved set is stored on the run.
• Secrets are declared by name only. The environment file maps each pipeline's names to a source, and that mapping is the grant: a pipeline resolves only the names mapped for it. Values are redacted from logs, events, stored config, transcripts and the UI.
• Config interpolation (${vars.maxPrice}) is evaluated once per run; item expressions in builtins and routes are evaluated per item.

Routing and dead letters

out takes one stream, a list of streams, or route objects where the first match wins:

out:
  - { when: "item.city == 'Lyon'", out: listings.lyon }
  - { when: "item.city == 'Paris'", out: listings.paris }
  - { when: "true", out: listings.other }

An output that matches no route goes to the step's dead-letter stream as an unrouted letter, so a gap in the routes is visible and replayable. A delivery that exhausts its retries goes there too. Dead letters are grouped by error in the UI, and fixing one is a change and a command:

millstream apply .
millstream replay --pipeline property-sourcing --from dlq --run run_20261020T060000Z-7559f3e7_daily --migrate

Triggers

A trigger publishes one item per run to its own stream, trigger.<name>, which steps read like any other. Firing ids are deterministic, so a cron time fired twice after a restart opens one run, and a webhook sent again with the same Idempotency-Key returns the run it already started.

Revisions and replay

A pipeline has no version field. Each distinct package content is a revision, identified by its hash, and every run records the revision it started on and finishes on it, even when a new one is applied meanwhile. A run waiting three days for a review keeps its prompt. millstream rollback reactivates the previous revision.

Replay runs one step again over past items, in a new generation with outputs of its own. It back-fills a step you just inserted, re-scores last week with a new prompt, or shows what a threshold change would do before it ships:

millstream replay --pipeline property-sourcing --step keep-good --since 7d --var minScore=75
millstream run report property-sourcing run_replay_20261021T091500Z-3c9f2a10

Packages and shared streams

A pipeline ships as a package: the pipeline file, its local actions, lessons and assets. Registry actions are OCI artifacts pinned by digest in millstream.lock. Each package is isolated, with its own secrets, limits and lifecycle, and several share one deployment, with round-robin dispatch so a backlog in one never starves another.

A pipeline reads another's stream by declaring it, without the producer changing:

streams:
  approved: { pipeline: property-sourcing, stream: listings.approved }
steps:
  notify-buyers:
    action: lab/notify-buyers@1
    in: approved

The producer's output schema is the contract, checked across both packages. Lineage crosses pipelines, while data, retention and erasure stay per pipeline.

Checked before it runs

millstream check runs before every run and apply. It fails on unknown actions or streams, cycles, schema mismatches between a producer and its consumers, expressions that do not compile and unmapped secrets, and warns when agent output can reach a step with external effects without a person in between. millstream test pipeline runs package tests with stubbed actions on a temporary store, so a pipeline is tested in CI like any other code.

millstream check
millstream graph
millstream test pipeline

Next

How agents run Human review Read the quick start