Benchmark the startup claim, and write down what a run is

The milestone is measured on being lighter than Kedro, so make bench-startup
measures it rather than asserting it: 61 ms from submit to result against
1110 ms for kedro run on a pipeline that does the same nothing. The difference
is not orchestration, it is that nothing is booted per run — on a 510-config
sweep that is about nine minutes of pure startup that never happens.

docs/flows/runs.md is the guide: batch flows, sweeps, reporting from inside a
node, artifacts, and the two sanctioned patterns for objects that cannot be
serialized — keep them in one node, or cross at a checkpoint.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AD8SfVhzXBG2nAfFcVh3iD

Makefile

@@ -3,7 +3,7 @@
 # The workspace root delegates to these (see ../Makefile).
 
 .PHONY: dev-utils dev dev-local up down update install dev-backend dev-frontend \
-        generate-client seed-example test test-backend test-frontend soak lint lint-backend \
+        generate-client seed-example test test-backend test-frontend soak bench-startup lint lint-backend \
         lint-frontend umami clean help
 
 COMPOSE_ROOT := $(CURDIR)
@@ -114,6 +114,9 @@ test-frontend:  ## Run frontend tests (Playwright e2e)
 soak:  ## Run the soak/chaos harness (SOAK_ARGS="--minutes 30 --scenario redis")
 	cd backend && uv run python scripts/soak.py $(SOAK_ARGS)
 
+bench-startup:  ## Time submitting a run (BENCH_ARGS="--kedro ../some/kedro/project")
+	cd backend && uv run python scripts/bench_startup.py $(BENCH_ARGS)
+
 # ── Linting ───────────────────────────────────────────────────────
 
 lint: lint-backend lint-frontend  ## Run all linters

NOTEPAD.md

@@ -217,7 +217,13 @@ Open on purpose. Each names what should bring it back.
 - FEAT/UI: the node-panel and edge trend curves take no range, unlike the health block. They are drawn from a Redis ring of the last 120 values per message, which has no window to ask for — a hover caption names what the curve covers instead of a picker promising a span nothing can serve. Reopen if per-message history ever gains a time window.
 - FEAT/UI: an e-ink rendering profile for a dashboard — motion off, hover-only affordances resolved to something visible, high-contrast palette, thick strokes, and a repaint cadence low enough for a display that takes a second to settle. Reopen when a panel with such a display is actually hung.
 - CHORE/INFRA: Postgres stays. The 2026-08 review rejected YugabyteDB/CockroachDB (multi-node cluster systems, ~4 GB+ RAM per node, against the small-server target — the scaling story is remote workers, not a distributed DB) and found merging Postgres into Redis or vice versa buys little: the stores hold disjoint data and both sit behind abstractions. SQLite would fit the single-instance design and drop a container; reopen if the home-install footprint becomes a product concern.
-- CHORE/INFRA: NATS JetStream as the work-queue backend — durable streams whose consumer semantics match the `WorkQueue` interface, in one small binary. Reopen with M5 remote workers, when the queue crosses hosts.
+- CHORE/INFRA: NATS JetStream as the work-queue backend — durable streams whose consumer semantics match the `WorkQueue` interface, in one small binary. Reopen with M5 remote workers, when the queue crosses hosts. NOTE: remote workers landed without it — a worker dials the engine's own socket and never touches Redis, so the queue still does not cross a host. Reopen if a second engine ever pulls from the same stream.
+- FEAT/RUNS: stage caching. `run_node.cache_key` is written on every run and the artifact store is content-addressed, so the pieces are in place; what is missing is computing the key from the node's source digest plus its input values and skipping a node whose key already has an `ok` row with its artifacts still present. The two research repos want this more than they want resume — neither persists checkpoints, and both re-run unchanged preprocessing every time.
+- FEAT/RUNS: per-label requirements overlays (`requirements-gpu.txt`) synced into a remote worker's venv, with drift surfaced against the engine's manifest. Today a worker's environment is whatever `--python` points at, which is fine for one hand-managed GPU box and not for several. `venv_digest` already arrives at attach and is shown on `/workers`, so the reporting half exists.
+- FEAT/RUNS: a run detail screen. The API answers everything — params, per-node status with logs and tracebacks, artifacts, metrics, and `/runs/series/compare` in the chart widget's own `series` shape — but nothing in the dashboard reads it yet, so a run is inspected over HTTP. Comparing curves is a widget binding once someone builds the page around it.
+- FEAT/RUNS: a thin client CLI (`fluksio run/runs/sweep/worker`) over the same API. The engine being resident is what makes runs cheap; a CLI is ergonomics on top, and `curl` covers it until someone is running sweeps daily.
+- CHORE/RUNS: `run_metric` has no retention. Deliberately outside `OBS_RETENTION_DAYS` — an experiment nobody deleted should not vanish on a rollup window — but a few thousand runs at 3000 steps will want a policy eventually, probably per-flow rather than global.
+- CHORE/RUNS: a run holds one worker slot per node for its whole duration, and `MAX_PARALLEL` run drivers bound how many graphs are in flight. A sweep of 500 therefore queues behind the pool rather than the driver count. Fine — the GPU is the scarce thing — but the two limits are unrelated numbers that read as if they were one.
 
 ## Blocked
 

ROADMAP.md

@@ -32,6 +32,11 @@ existing flow engine reachable and persistent precedes new feature breadth.
       stack's own containers. Two caveats: the redis scenario stops the container the
       whole stack shares, and a cascade finishes fast enough that the engine kill
       proves redelivery without stressing it
+- [x] Startup benchmark: `backend/scripts/bench_startup.py` behind `make
+      bench-startup` times submitting a run against a Kedro project doing the same
+      nothing — 61 ms against 1110 ms, because nothing is booted per run. The
+      claim the ML-pipeline milestone is measured on, kept checkable rather than
+      asserted
 
 ## Phase 1 — Backend: management
 
@@ -120,6 +125,32 @@ external interfaces. See `docs/architecture/structure.canvas` → *Backend – M
       cascade — including the manual runs and previews that never went through the
       queue. Read back through `/observability/*`, which always answers 200 so a
       degraded engine still renders, and pruned on a retention window
+- [x] Batch runs: a `mode: batch` flow taken from its declared inputs to its
+      declared outputs once, with parameters that identify it and a result kept.
+      Journaled to a Redis stream of its own and Postgres-authoritative from the
+      claim onwards, so a stale lease — not an unacked entry — marks a run whose
+      engine died. Each run executes an isolated pipeline over its own state
+      namespace, so a sweep's configs run in parallel without overwriting each
+      other's messages. `run`, `run_node`, `run_metric` and `run_artifact` are
+      separate from the cascade rollups, which are pruned on a retention window
+      and an experiment must not be. `/runs`, `/runs/{id}`, `/runs/flows/{name}`,
+      `/sweep`, `/cancel`, `/metrics` and `/series/compare`
+- [x] Reporting from inside a running node: node code imports `fluksio` and calls
+      `log_metric` / `progress` / `save_artifact` mid-call. The worker protocol
+      carries event lines before the reply, so the metrics of a two-hour training
+      arrive while it trains rather than with its result — and each event resets
+      the deadline, which turns `NodeDef.timeout` into an idle timeout for a node
+      that reports
+- [x] Artifacts: `DType.ARTIFACT` carries a reference (digest, size, media type,
+      name) into a content-addressed store on the data volume, so bytes never
+      enter a message, Redis or the queue. The digest is the future stage-cache
+      key, which is why it is content-addressed now rather than per-run
+- [x] Remote workers: a worker dials out to `WS /workers/attach` with an RS256
+      worker-scope token, advertises labels, and answers the same JSON protocol
+      the local pool speaks. `NodeDef.device` routes a node to one, resolved per
+      call; a run whose labels nothing carries waits saying so. A device-bound
+      node is compiled on that machine. The agent is one file plus `worker_main`,
+      with `websockets` as its only dependency
 - [ ] Test nodes: a small node dragged onto an existing one, smoke or unit, blocking
       deployment on failure
 - [ ] User management scoped per flow and per data set

backend/scripts/bench_startup.py

@@ -0,0 +1,198 @@
+"""How long it takes to get a run started, which is the thing Kedro is slow at.
+
+A pipeline framework that boots the project per run pays that cost every time:
+``kedro run`` on a pipeline that does nothing takes about a second, and a sweep
+of five hundred configs therefore spends ten minutes doing nothing. Fluksio
+answers that by not booting anything — the engine is already up and its workers
+already have the code loaded, so submitting is one request.
+
+This measures that claim against a running stack, so it can be checked rather
+than asserted. Run it with the dev stack up::
+
+    make bench-startup
+    make bench-startup BENCH_ARGS="--runs 50 --kedro ../path/to/kedro/project"
+
+The Kedro figure is optional and measured the same way — a null pipeline, timed
+end to end — so the two numbers mean the same thing.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import statistics
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.request
+from typing import Any
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parents[1]))
+
+from app.core.config import settings  # noqa: E402
+
+log = logging.getLogger("bench")
+
+#: A flow that computes nothing, so what is timed is the getting-started.
+BENCH_FLOW = "bench_startup"
+NODE_SOURCE = '"""Does nothing, on purpose."""\n\n\ndef process(n, params):\n    return {"out": n}\n'
+
+
+class Api:
+    def __init__(self, base: str, token: str) -> None:
+        self.base = base.rstrip("/")
+        self.token = token
+
+    def call(self, method: str, path: str, body: Any = None) -> Any:
+        request = urllib.request.Request(
+            f"{self.base}{path}",
+            method=method,
+            data=json.dumps(body).encode() if body is not None else None,
+            headers={
+                "Authorization": f"Bearer {self.token}",
+                "Content-Type": "application/json",
+            },
+        )
+        with urllib.request.urlopen(request, timeout=60) as response:
+            return json.load(response) if response.status != 204 else None
+
+
+def login(base: str) -> Api:
+    data = (
+        f"username={settings.FIRST_SUPERUSER}&password={settings.FIRST_SUPERUSER_PASSWORD}"
+    ).encode()
+    request = urllib.request.Request(
+        f"{base.rstrip('/')}/api/v1/login/access-token",
+        data=data,
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+    )
+    with urllib.request.urlopen(request, timeout=30) as response:
+        return Api(base, json.load(response)["access_token"])
+
+
+def ensure_flow(api: Api) -> None:
+    """A published batch flow with one node that returns what it was given."""
+    definition = {
+        "name": BENCH_FLOW,
+        "title": "Startup benchmark",
+        "version": 1,
+        "mode": "batch",
+        "outputs": ["out"],
+        "inputs": [{"spec": {"name": "n", "dtype": "int"}, "initial": 1}],
+        "nodes": [
+            {
+                "id": "noop",
+                "type": "python",
+                "requires": [{"name": "n", "dtype": "int"}],
+                "provides": [{"name": "out", "dtype": "int"}],
+            }
+        ],
+    }
+    try:
+        current = api.call("GET", f"/api/v1/flows/{BENCH_FLOW}")
+        definition["version"] = current["definition"]["version"]
+    except urllib.error.HTTPError:
+        pass
+    api.call("PUT", f"/api/v1/flows/{BENCH_FLOW}", definition)
+    api.call(
+        "PUT", f"/api/v1/flows/{BENCH_FLOW}/nodes/noop/source", {"code": NODE_SOURCE}
+    )
+    api.call(
+        "POST",
+        f"/api/v1/flows/{BENCH_FLOW}/publish",
+        {"version": definition["version"]},
+    )
+
+
+def time_runs(api: Api, count: int) -> tuple[list[float], list[float]]:
+    submits: list[float] = []
+    totals: list[float] = []
+    for index in range(count):
+        start = time.perf_counter()
+        run = api.call(
+            "POST", f"/api/v1/runs/flows/{BENCH_FLOW}", {"params": {"n": index}}
+        )
+        submits.append((time.perf_counter() - start) * 1000)
+        while True:
+            got = api.call("GET", f"/api/v1/runs/{run['id']}")
+            if got["status"] not in ("queued", "running"):
+                break
+            time.sleep(0.005)
+        totals.append((time.perf_counter() - start) * 1000)
+        if got["status"] != "ok":
+            raise SystemExit(
+                f"run {run['id']} ended {got['status']}: {got['status_reason']}"
+            )
+    return submits, totals
+
+
+def time_kedro(project: str, pipeline: str, count: int) -> list[float]:
+    """The same measurement for a Kedro project: a whole run, end to end."""
+    timings: list[float] = []
+    for _ in range(count):
+        start = time.perf_counter()
+        result = subprocess.run(
+            ["kedro", "run", "--pipeline", pipeline],
+            cwd=project,
+            capture_output=True,
+            check=False,
+        )
+        if result.returncode != 0:
+            raise SystemExit(f"kedro run failed: {result.stderr.decode()[-400:]}")
+        timings.append((time.perf_counter() - start) * 1000)
+    return timings
+
+
+def report(label: str, timings: list[float]) -> None:
+    ordered = sorted(timings)
+    log.info(
+        f"  {label:<28} median {statistics.median(ordered):7.1f} ms"
+        f"   min {ordered[0]:7.1f}   max {ordered[-1]:7.1f}"
+    )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--base-url", default="http://api.localhost")
+    parser.add_argument("--runs", type=int, default=20)
+    parser.add_argument(
+        "--kedro", default="", help="a Kedro project to compare against"
+    )
+    parser.add_argument("--kedro-pipeline", default="__default__")
+    parser.add_argument("--kedro-runs", type=int, default=3)
+    parser.add_argument(
+        "--keep", action="store_true", help="leave the benchmark flow behind"
+    )
+    args = parser.parse_args()
+    logging.basicConfig(level=logging.INFO, format="%(message)s", stream=sys.stdout)
+
+    api = login(args.base_url)
+    ensure_flow(api)
+    # One throwaway run first: the worker compiles the node on its first call,
+    # and that cost belongs to the deployment rather than to a run.
+    time_runs(api, 1)
+
+    submits, totals = time_runs(api, args.runs)
+    log.info(f"\nfluksio — {args.runs} runs of a flow that computes nothing")
+    report("submit accepted", submits)
+    report("submit -> result", totals)
+
+    if args.kedro:
+        kedro = time_kedro(args.kedro, args.kedro_pipeline, args.kedro_runs)
+        log.info(
+            f"\nkedro — {args.kedro_runs} runs of a pipeline that computes nothing"
+        )
+        report("kedro run", kedro)
+        ratio = statistics.median(kedro) / statistics.median(totals)
+        log.info(f"\n  a run costs {ratio:.0f}x less here, per run")
+
+    if not args.keep:
+        api.call("DELETE", f"/api/v1/flows/{BENCH_FLOW}")
+        log.info(f"\nremoved the '{BENCH_FLOW}' flow")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())