app/docs/concepts/runs.md

# Runs: pipelines that finish

A flow that automates a house never ends: a value arrives, nodes fire, and it
waits for the next one. A research pipeline is the other shape: parameters go
in, stages execute in order, and at some point it is *done* and has produced
something worth keeping. Fluksio calls the second one a **run**, and it is the
same engine either way.

This is what makes Fluksio usable where Kedro, MLflow or ClearML would be: a
run has parameters that identify it, a result, per-step metrics, artifacts and
a place in a queryable history, with no second server and no
project bootstrap on every execution.

## A batch flow

Set `mode: "batch"` on the flow and name the messages its result should hold:

```json
{
  "name": "train_polymer_gnn",
  "mode": "batch",
  "outputs": ["final_loss", "report"],
  "inputs": [
    {"spec": {"name": "lr", "dtype": "float"}, "initial": 0.1},
    {"spec": {"name": "steps", "dtype": "int"}, "initial": 3000}
  ],
  "nodes": [{"id": "train", "timeout": 7200, "device": "gpu", "...": "..."}]
}
```

A batch flow is built and validated like any other (it appears on the canvas,
its ports are type-checked) but it is never *activated*: no subscriptions, no
schedules, no webhooks. It runs when a run asks it to, and not otherwise.

Its `inputs` are its parameters. A run supplies values for them; anything it
does not supply keeps the declared initial value.

One thing a batch flow may not do is rate-limit a port (`interval`). A rate
limit holds a value back for a timer to release, and a run has no timer, so the
value would be dropped rather than delayed, so submitting is refused instead.

## Submitting

**Runs → New run** in the app is the same call with a form in front of it: pick
a flow, fill in its declared inputs, press Run. A comma-separated list in a
number field runs every combination of them as a sweep. **Use draft** runs the
working copy rather than what is published, which is how a change is tried
before it is deployed, and each run keeps its own state, so nothing it computes
reaches the live flow.

```bash
curl -X POST $FLUKSIO/runs/flows/train_polymer_gnn \
     -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
     -d '{"params": {"lr": 0.3, "steps": 4000}, "seed": 7}'
```

The answer is immediate and the run is `queued`; a training run is measured in
hours, so nothing waits for it. Poll `GET /api/v1/runs/{id}` for its status,
result, per-node record and artifacts.

Wrong parameters are refused before anything executes: an undeclared name, or
a value of the wrong type, comes back as a 422 naming the problem.

`"draft": true` on the body runs the unpublished working copy — the flow
document and the node sources the editor is showing. The copy is read again
when the run starts, so a draft published while it waited runs as the
published one.

### Sweeps

An ensemble is the same parameters at different seeds; a grid search is the
parameters spread out. Both are one call, and the caller builds the list:

```bash
curl -X POST $FLUKSIO/runs/flows/train_polymer_gnn/sweep \
     -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' -d '{
  "runs": [{"params": {"lr": 0.1}, "seed": 1}, {"params": {"lr": 0.3}, "seed": 1}]
}'
```

From a terminal that is `fluksio sweep train_polymer_gnn --param lr=0.1,0.3`,
which builds the product of the lists you give it and posts the same call.

They share a `group_id`, so `GET /api/v1/runs?group=…` is the sweep, and they
execute in parallel. That is safe because **each run has a state backend of its
own**: message names are global keys, so two runs of one flow would otherwise
overwrite each other's values. They do not.

## Producing values before you are finished

A training loop has numbers worth keeping long before it has a result. Those
numbers are **outputs**, not logs: a node declares a port for them and produces
them over time, which in Python is a generator.

```python
def process(lr, steps):
    loss = 1.0
    for _ in range(steps):
        loss = train_one_step(lr)
        yield {"loss": loss}          # published now, on the `loss` port
    return {
        "weights": fluksio.save_artifact(dump(model), "weights.npz"),
        "final_loss": loss,
    }
```

Mark the port it streams on, so the flow says what it does:

```json
{"name": "loss", "dtype": "float", "stream": true}
```

Every yield is published the instant it happens: same port, same type check,
same place on the canvas as any other value. Whatever the generator `return`s
is the node's result, and is what downstream nodes read. If you never
`return`, the last thing you yield is the result instead.

There is no `log_metric()`. A metric leaves through a port, which makes it a
message: a chart binds to it directly, a downstream node can consume it, and the
run keeps its series.

**Where a yield cannot reach** — the value comes from inside somebody else's
callback, and they call you rather than the other way round. `fluksio.emit`
writes the same ports the same way:

```python
import fluksio


def process():
    model.fit(callbacks=[LambdaCallback(
        on_epoch_end=lambda epoch, logs: fluksio.emit(loss=logs["loss"])
    )])
    return {"weights": ...}
```

### What a run does with them

Every number a node emits is kept as the run's series, stepped by the count of
emissions on that message. Read one back with
`GET /api/v1/runs/{id}/metrics?name=<flow>.loss`, or leave `name` off for
every series the run kept, or compare runs:

```
GET /api/v1/runs/series/compare?ids=<a>,<b>,<c>&metric=<flow>.loss
```

That answers in the `series` shape a chart widget already draws, so three
training curves side by side is a widget binding. During a run the values also
arrive live on the flow socket, so a chart bound to the port fills in as the
training goes.

A streaming port may set `interval` to thin out what reaches the canvas. The
run's history still keeps every value, because the interval is asking for the
display not to be flooded, not for the curve to have holes in it.

Emitting has a second effect: **a node's timeout measures silence, not
duration**. A node that yields every few seconds can run for hours under a
`timeout` of 300; one that says nothing for longer than its timeout is killed.
There is no timeout unless one is set, since a training node that reports
nothing is usually working. Set `timeout` where going quiet means stuck, at how long
the node may plausibly be.

In a *live* flow, an emission also wakes whatever is downstream of it, exactly
as a subscriber publishing does. In a run it does not: a run's graph is
scheduled once, and three thousand mid-node cascades would leave "the run has
finished" with nothing to mean.

## Artifacts

Bytes never travel as a message. `save_artifact` writes them to a
content-addressed store and returns a small reference (digest, size, media
type, name) which is what an `artifact`-typed port carries:

```python
def process(weights):          # requires: weights, dtype "artifact"
    path = fluksio.load_artifact(weights)
    ...
```

Because the address is the content's hash, a sweep whose fifty configs share
one preprocessed input stores it once, and a reference stays valid wherever
the store is reachable from. Artifacts a run produced are listed on it and
downloadable at `GET /api/v1/artifacts/{digest}`.

## Stage caching

A run mostly does not redo what an earlier one already did. Before a node
executes it is fingerprinted with a sha256 over its source, its settings, the
ports it declares and the values it is about to read. If some earlier run of that
same fingerprint finished, what that one returned is restored into this run's
state and the node is skipped. It is recorded with the status `cached` and a
duration of zero, and its artifacts are listed on the new run as well, so they
stay downloadable from either.

The settings go into the key raw, so a secret contributes its `{"$secret":
name}` reference and never its value. An artifact input counts as its content
digest: the same bytes under a different filename are the same input. Each node
on a run carries the `cache_key` it was looked up by.

For a [code-defined flow](../getting-started/data-science.md), "its source" is
the generated shim, which imports the real function and does not change when
that function does. So the key carries one thing more: a digest of the project
modules that node's function reaches, worked out by `fluksio sync`, which is
the only side that imports your code and can see what it imports, and read
again from those files when the run starts.

Editing a helper three calls down from the node invalidates it, which is the
point: the alternative is a re-run answering with the previous code's numbers.
Editing something the node does not reach leaves the hit standing, which is
the other half: a notebook two directories away is not a reason to retrain.
The walk follows imports statically and stops at the standard library, at
anything installed, and at Fluksio itself; a module imported under a name the
code computes is not followed. An engine that cannot see the files keeps what
sync recorded instead of nothing, so a worker on another machine no longer
keys every run the same.

The run history *is* the cache; there is no second store. A node's returned
outputs are kept on its run record as canonical JSON, up to 256000 characters.
A node returning more than that is not cacheable that run. An entry
whose artifact bytes have since left the store is a miss, not an error.

Only `python` nodes are cached, and by default all of them are. A built-in node
type or a connector node has side effects and no source to fingerprint, so
neither is ever a candidate. Turn it off for one node with
`@node(..., cache=False)` (the flow document carries it as `cache`, so the
canvas and the API can change it too) or for one run with
`fluksio run --no-cache`, `fluksio sweep --no-cache`, or `"no_cache": true` in
the submission body.

A cached node replays no emissions, since those values were the story of an
execution that is not happening this time, so its series is not rewritten
either. The run it was restored from is recorded instead, and that is where the
curve is read back from: asking the reusing run for its metrics answers with
the same points, under its own flow's names. The one way to be left with a
result and no curve is for that earlier run to have been deleted, which
deleting its flow does.

## Objects that cannot be serialized

A live model, a `DataLoader`, a JAX-compiled function: none of these cross a
node boundary, and no framework flag will make them. There are exactly two
patterns, and they are both deliberate:

* **Keep them in one node.** Stages that must share live memory are one node.
  Building the model and training it is one stage; the fact that Kedro would
  make them two nodes is Kedro's problem, not a structure worth reproducing.
* **Cross at a checkpoint.** Save what matters as an artifact and rebuild from
  it on the other side. That is the boundary that also survives the next node
  running on a different machine.

## Running a node somewhere else

A node that needs a GPU declares the label of the machine that has one:

```json
{"id": "train", "device": "gpu", "device_policy": "require", "timeout": 7200}
```

A worker on that machine dials **out** to the engine, because the engine
generally cannot reach it (different network, no inbound route) and because
nothing should expose Redis across hosts. Install it on the box, mint it a
token, and start it:

```bash
curl -X POST $FLUKSIO/workers/tokens -d '{"name": "gpu-dev"}'   # once, as an admin

pip install fluksio-worker

fluksio-worker \
  --url wss://api.fluksio.com/api/v1/workers/attach \
  --token "$FLUKSIO_WORKER_TOKEN" \
  --labels gpu,cuda12 \
  --python /opt/torch-venv/bin/python
```

`fluksio-worker` is its own distribution: the agent, the node runner, and
`websockets`. Nothing of the engine, so a GPU box does not install a database
driver to run a training step. Where pip is not an option, the two files still
work copied into one directory and run with `python agent.py …`; the engine
serves the runner at `GET /api/v1/workers/runtime`.

`--python` is the interpreter node code runs on, which is how the GPU box keeps
its CUDA wheels without the engine ever installing them. The node's source
travels with every call, so nothing has to be deployed there.

A few consequences worth knowing:

* `import fluksio` inside a node is the worker's own reporter — `emit`,
  `save_artifact`, `load_artifact`, installed before the node's code runs, so
  the installed `fluksio` package (if the box has one) never shadows it.
* A node bound to a device is **compiled on that machine**. A node importing
  `torch` is correct on the GPU box and a missing module on the engine, so
  checking it on the engine would fail a node that is fine.
* If nothing carrying the label is attached, the run stays `queued` and says
  what it is waiting for. Submit first, switch the GPU box on later.
* Cancelling a run kills what it is executing, there or here, and leaves other
  runs of the same node alone.
* If the worker disappears mid-call, the run fails in seconds with
  `worker went away mid-call` rather than waiting out its timeout.
* `device_policy: "prefer"` runs locally when no such worker is attached;
  `"require"` (the default) waits for one.

## Durability

Submitting journals the run to a Redis stream of its own, separate from the
one the automations use: a burst of five hundred sweep runs must not stand
between a house and its heating. An engine that is down when a run is
submitted picks it up when it starts.

The row is what makes that true rather than the queue. A run is written before
the work item is added, so an engine reads its own history at startup and
wakes itself for anything still `queued` — which is what a run submitted
seconds before a restart is, and what an in-memory queue would otherwise have
lost with the process.

From the moment a run is claimed, its database row is the record and the queue
is finished with it. Redelivering two hours of training because an
acknowledgement was late is not recovery; instead a running run refreshes a
lease, and one whose lease goes stale is marked `abandoned`, which is what a
run whose engine was killed mid-training becomes. Starting it over is a
decision rather than something that happens: [`fluksio
retry`](../code/cli.md#fluksio-retry) submits it again, and `--group` does
that for the runs of a sweep that did not finish.

## Looking at what ran

**New run** is where one is started, and the **Runs** screen is the experiment
log: every run newest-first, filtered by flow, by status, or down to one
sweep. A sweep is worth filtering to, since the table then draws a column per
parameter that actually varied, which is what makes fifty runs of one flow
readable.

Tick two or more and their curves go side by side. Shift-click to take a
range, or the header box to take everything on screen. That comparison is the
address, so a link to it is a link someone else can open.

The curves are drawn against the step by default. They can also be drawn
against elapsed seconds, which answers "which one got there sooner" and is
measured from each run's own first reading so that runs started hours apart
still lie on top of each other; or against another metric of the same runs (an
epoch, or samples seen) joined on the step the two share.

One run in full is params, the per-node record with its logs and traceback,
the artifacts it made, its metrics and its result. **Retry** submits it again
as a run of its own, keeping the flow, the inputs, the seed and the sweep it
belonged to.

### Taking it into a dataframe

An analysis wants a table rather than a screen, and there are two it usually
wants. `fluksio export metrics` is the long one, a row per run, metric and
step; `fluksio export runs` is the wide one, a row per run with the
inputs that varied as columns beside its final numbers:

```python
import pandas as pd
from fluksio.sdk.client import Client

client = Client()
curves = pd.DataFrame(client.export_metrics(flow="train"))
arms = pd.DataFrame(client.export_runs(flow="train", status="ok"))
```

The run id is on every row of both, so a curve joins to the arm it came from
and to the run page it was recorded on, and the wide table carries the commit
and the code digest, so an exported file says what produced its numbers.
[The CLI](../code/cli.md#fluksio-export) writes the same rows as csv, jsonl or
parquet, which is where an export belongs: in the script beside the analysis.

A node's numbers usually arrive inside a record, and the wide table follows
them in: every number a result carries is a column of its own, named by its
path (`metric.final_metrics.train_loss`) and `metrics="final_metrics.train_loss"`
selects one. Inputs work the same way, and are compared leaf by leaf, so two
configurations differing in one field give that field as the axis.

### A dashboard, read against runs

A run records values under the same names a dashboard binds to (a run of
`study` writes `study.loss`) so a dashboard is already a way of looking at
one. **Open in dashboard** from a run or a comparison opens any dashboard with
`?runs=a,b,c` on it, and the widgets resolve from those runs instead of from
the live engine: each chart draws a line per run, the tiles that show one
number show the first, and the controls go quiet because there is nothing left
to publish to.

Nothing about the dashboard is specific to runs. The page built to watch a
training run happen is the page that shows the finished ones.

If a flow has no dashboard yet, the same menu offers to build one from what
the flow declares: a chart per streaming port, a tile per output:

```
POST /api/v1/dashboards/from-flow/study
```

It publishes `study_results` and is an ordinary dashboard afterwards; editing
it is how it stops being generic.

### A tile that always shows the last few

The other direction is a chart pinned to runs rather than a page opened
against them, which is what a panel over a bench wants. Set a chart's source
to **Runs**, name the metric, and pick either the latest N of a flow, one
sweep, or specific runs. It re-reads on its own and whenever a run finishes.

### When a run draws nothing

A node restored from the [stage cache](#stage-caching) has its curve read back
from the run that recorded it. Delete that run, which deleting its flow does,
and the reusing run is left with a result and an empty curve, and the chart says so
rather than looking broken.

## What this costs, compared

The repository ships a benchmark that measures submitting a run against a
Kedro project doing the same nothing:

```
fluksio — submit accepted     median   15.3 ms
          submit -> result    median   60.8 ms
kedro   — kedro run           median 1109.5 ms
```

The difference is not the orchestration; it is that Fluksio does not boot a
project per run. The engine is already up, and the workers already have the
node's code compiled. On a 510-run sweep, that gap is about nine minutes of
pure startup that never happens.

## See also

- [Getting started: data science](../getting-started/data-science.md) — the same
  material as a first setup
- [Writing node code](../code/nodes.md) — generators, `fluksio.emit`, artifacts
- [Remote workers](../code/workers.md) — attaching the machine with the GPU
- [The HTTP API](../code/api.md) — every endpoint used above