# Build a scientific pipeline

Chain calculation, verification and save steps, or fan out across inputs in parallel.

Split multi-stage workflows into **Run → Check → Save** jobs so that if verification or database saving fails, you never have to re-run an expensive calculation.

## 1. Run the 3-stage pipeline example

Run `compute/platform/examples/pipeline/run_pipeline.py` to execute a 3-stage chain (`run` → `check` → `save`):

```bash
compute/platform/.venv/bin/python compute/platform/examples/pipeline/run_pipeline.py \
  --output ./pipeline-output
```

> **Tip:** Using the local dev stack? Add `--config .local-dev/compute/client.json`.

The script prints the three job IDs, waits for the final `save` stage, and downloads its report:

```text
Run 0195f3c0-1111-7000-8000-000000000001
Check 0195f3c0-2222-7000-8000-000000000002
Save 0195f3c0-3333-7000-8000-000000000003
{"value":42,"project_id":"...","source_job_id":"...","check_job_id":"..."}
Outputs: pipeline-output/0195f3c0-3333-7000-8000-000000000003
```

## 2. How the stages connect

First, look at the three worker functions in `compute/platform/examples/pipeline/jobs.py`:


```python
"""Produce, check and save a small artifact-backed report bound to the job's project."""

from __future__ import annotations

import json
from uuid import UUID

from pydantic import BaseModel, ConfigDict

from chemistry_compute import Collection, DataClient, job_context, one_result


class Model(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True, frozen=True, allow_inf_nan=False)


class Input(Model):
    file: str


class Measurement(Model):
    value: int


class Checked(Model):
    value: int
    source_job_id: UUID


class Report(Model):
    value: int
    project_id: UUID
    source_job_id: UUID
    check_job_id: UUID


def run(request: Input) -> Measurement:
    value = int(job_context().input(request.file).read_text()) * 2
    job_context().output("measurement.txt").write_text(str(value))
    return Measurement(value=value)


def check(request: Collection) -> Checked:
    parent = one_result(request, Measurement, artifacts=("results/measurement.txt",))
    if int(parent.artifacts["results/measurement.txt"].read_text()) != parent.value.value:
        raise ValueError("measurement artifact differs from its typed result")
    return Checked(value=parent.value.value, source_job_id=parent.source.job_id)


def save(request: Collection) -> Report:
    parent = one_result(request, Checked)
    project = DataClient().query("SELECT authz.current_project_id()::text")["rows"][0][0]
    report = Report(
        value=parent.value.value,
        project_id=UUID(project),
        source_job_id=parent.value.source_job_id,
        check_job_id=parent.source.job_id,
    )
    job_context().output("report.json").write_text(json.dumps(report.model_dump(mode="json")))
    return report
```


Next, look at how `run_pipeline.py` submits all three jobs up front using `depends_on`:


```python
"""Run a complete dependency chain and fetch its project-bound report."""

from __future__ import annotations

import argparse
from pathlib import Path

import jobs

from chemistry_compute import Client, Collection, Program


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", type=Path)
    parser.add_argument("--version", default="1")
    parser.add_argument("--output", type=Path, default=Path("pipeline-output"))
    args = parser.parse_args()
    client = Client.from_login(args.config)
    source = Path(__file__).resolve().parent
    run = client.publish(Program(jobs.run, source), f"example.pipeline.run@{args.version}")
    check = client.publish(Program(jobs.check, source), f"example.pipeline.check@{args.version}")
    save = client.publish(Program(jobs.save, source), f"example.pipeline.save@{args.version}")
    measured = client.submit(
        run, jobs.Input(file="number.txt"), inputs={"number.txt": source / "number.txt"}
    )
    checked = client.submit(check, Collection(), depends_on=[measured.id])
    saved = client.submit(save, Collection(), depends_on=[checked.id])
    print(f"Run {measured.id}\nCheck {checked.id}\nSave {saved.id}", flush=True)
    print(saved.result(timeout=300).model_dump_json())
    saved.fetch(args.output)
    print(f"Outputs: {args.output / saved.id}")


if __name__ == "__main__":
    main()
```


How data flows between stages:

- **`job_context()`** reads staged input files (`.input("number.txt")`) and writes output files (`.output("measurement.txt")` → `results/measurement.txt`).
- **`depends_on=[measured.id]`** tells the scheduler to hold `check` until `measured` succeeds. Once submitted, your laptop can disconnect and the pipeline still finishes on the cluster.
- **`Collection()` + `one_result()`** gives the child job the parent's typed return value (`parent.value`) and local paths to its output files (`parent.artifacts["results/measurement.txt"]`).

For structured scientific outputs, use Parquet and reuse the producer's accepted artifact. See
[Share and stream scientific datasets](/operate/files/) for a complete file-producing pipeline,
bounded readers and the distinction between full input staging and partial trajectory reads.

## 3. Fan out and reduce with a Flow

When you need to run the same calculation across many inputs in parallel and combine the results, use `flows.prepare(...).map(...).reduce(...).then(...)`:

```bash
compute/platform/.venv/bin/python compute/platform/examples/programs/run_flow.py
```

```text
Flow 0195f3c5-4444-7000-8000-000000000004
{"value":7}
```

Here is the complete driver (`compute/platform/examples/programs/run_flow.py`):


```python
"""Publish and run prepare → map → reduce → follow-up using your saved CLI login."""

from __future__ import annotations

import argparse
from pathlib import Path

import jobs

from chemistry_compute import Client, Program, flows


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", type=Path)
    parser.add_argument("--version", default="1")
    args = parser.parse_args()
    client = Client.from_login(args.config)
    source = Path(__file__).resolve().parent
    prepare = client.publish(Program(jobs.prepare, source), f"example.prepare@{args.version}")
    double = client.publish(Program(jobs.double, source), f"example.double@{args.version}")
    total = client.publish(Program(jobs.total, source), f"example.total@{args.version}")
    increment = client.publish(Program(jobs.increment, source), f"example.increment@{args.version}")
    flow = (
        flows.prepare(prepare, jobs.Count(count=3), concurrency=2)
        .map(double)
        .reduce(total)
        .then(increment)
        .submit(client)
    )
    print(f"Flow {flow.id}", flush=True)
    print(flow.result(timeout=300).model_dump_json())  # (0 + 2 + 4) + 1 = 7


if __name__ == "__main__":
    main()
```


Each stage is type-checked against the next:

1. **`prepare`** takes `Count(count=3)` and returns `Items[Number]` (`0, 1, 2`).
2. **`.map(double)`** runs `double` on each item in parallel (`0, 2, 4`).
3. **`.reduce(total)`** receives `Items[Number]` and sums them (`6`). Use `.collect(reducer)` instead if your reducer also needs parent files or failed item records.
4. **`.then(increment)`** runs a follow-up step on the reduced result (`7`).

## 4. Check status and debug failures

Inspect a Flow or any individual stage from the CLI:

```bash
ct-cli flow status FLOW_UUID
ct-cli status JOB_UUID
ct-cli logs JOB_UUID
ct-cli fetch JOB_UUID --output ./evidence
```

If a stage fails, upstream stages keep their completed outputs. Our built-in [transition state](/science/transition-states/) and [DFT](/science/dft/) workflows use this exact pattern so `ct-cli ts retry` and `ct-cli dft retry` can resume from the failed step.