From eb17528ce65aaf7ce3f54208928eb1e65ad08bf4 Mon Sep 17 00:00:00 2001
From: stroblme <stroblme@posteo.de>
Date: Wed, 4 Nov 2026 21:20:29 +0100
Subject: [PATCH 1/9] FlowNotFound says whether a published flow or any flow is
 missing; read_all stamps its cache with the revision from before its scan;
 dashboard discard and delete commits name the key; a RedisState ttl expires
 counters and the names set too; bench_startup publishes the version after its
 saves, and only when there is a draft

---
 backend/fluksio/flow/dashboards.py     |  8 +++---
 backend/fluksio/flow/state.py          | 16 +++++++++---
 backend/fluksio/flow/store.py          | 14 +++++++----
 backend/scripts/bench_startup.py       | 14 +++++++----
 backend/tests/flow/test_redis_state.py | 34 ++++++++++++++++++++++++++
 5 files changed, 70 insertions(+), 16 deletions(-)
 create mode 100644 backend/tests/flow/test_redis_state.py

diff --git a/backend/fluksio/flow/dashboards.py b/backend/fluksio/flow/dashboards.py
index 26c7efb..6fd370d 100644
--- a/backend/fluksio/flow/dashboards.py
+++ b/backend/fluksio/flow/dashboards.py
@@ -961,21 +961,23 @@ class DashboardStore:
                         self._dump(shown.model_copy(update={"version": top + 1}))
                     )
                 draft.unlink()
-            self.flows._commit(f"Discard draft of dashboard '{name}'")
-        return self.read(name)
+            shown = self.read(name)
+            self.flows._commit(f"Discard draft of dashboard {named(shown.name, name)}")
+        return shown
 
     def delete(self, name: str) -> None:
         if not self.exists(name):
             raise DashboardNotFound(name)
         path = self._file(name)
         with self.flows._write_lock:
+            key = next((k for k, i in self.key_index().items() if i == name), name)
             path.unlink(missing_ok=True)
             self._draft_file(name).unlink(missing_ok=True)
             try:
                 path.parent.rmdir()
             except OSError:
                 pass
-            self.flows._commit(f"Delete dashboard '{name}'")
+            self.flows._commit(f"Delete dashboard {named(key, name)}")
 
     def duplicate(self, name: str) -> DashboardDef:
         """Copy the working copy into a new dashboard, as a draft only.
diff --git a/backend/fluksio/flow/state.py b/backend/fluksio/flow/state.py
index 74c8d8f..fcd0133 100644
--- a/backend/fluksio/flow/state.py
+++ b/backend/fluksio/flow/state.py
@@ -639,6 +639,14 @@ class RedisState(StateBackend):
         names = [k for k in keys if not k.startswith("__")]
         if names:
             pipe.sadd(self._names_key(), *names)
+            if self._ttl:
+                pipe.expire(self._names_key(), self._ttl)
+
+    def _incr(self, pipe: Any, key: str) -> None:
+        """INCR a counter in the caller's pipeline, expiring with the values."""
+        pipe.incr(self._key(key))
+        if self._ttl:
+            pipe.expire(self._key(key), self._ttl)
 
     def set(self, key: str, value: Any) -> None:
         data = self._serialize(value)
@@ -747,7 +755,9 @@ class RedisState(StateBackend):
 
     def increment(self, key: str) -> int:
         """Atomically increment a counter using Redis INCR."""
-        return cast(int, self._client.incr(self._key(key)))
+        pipe = self._client.pipeline()
+        self._incr(pipe, key)
+        return cast(int, pipe.execute()[0])
 
     def increment_multi(self, keys: list[str]) -> None:
         """Bump every counter in one round trip rather than one INCR each."""
@@ -755,7 +765,7 @@ class RedisState(StateBackend):
             return
         pipe = self._client.pipeline()
         for key in keys:
-            pipe.incr(self._key(key))
+            self._incr(pipe, key)
         pipe.execute()
 
     def get_multi(self, keys: list[str]) -> dict[str, Any]:
@@ -916,7 +926,7 @@ class RedisState(StateBackend):
             if self._ttl:
                 pipe.expire(history_key, self._ttl)
         for key in counters:
-            pipe.incr(self._key(key))
+            self._incr(pipe, key)
         self._note(pipe, values)
         pipe.execute()
 
diff --git a/backend/fluksio/flow/store.py b/backend/fluksio/flow/store.py
index 9f9b58c..f3f117b 100644
--- a/backend/fluksio/flow/store.py
+++ b/backend/fluksio/flow/store.py
@@ -202,12 +202,14 @@ def new_id(kind: Literal["f", "d"], taken: Callable[[str], bool]) -> str:
 
 
 class FlowNotFound(KeyError):
-    def __init__(self, name: str) -> None:
+    def __init__(self, name: str, published: bool = False) -> None:
         super().__init__(name)
         self.name = name
+        self.published = published
 
     def __str__(self) -> str:
-        return f"No flow named '{self.name}'"
+        what = "published flow" if self.published else "flow"
+        return f"No {what} with the id '{self.name}'"
 
 
 class FlowExists(ValueError):
@@ -720,7 +722,7 @@ class FlowStore:
                 return self._load(path)
         path = self._flow_file(name)
         if not path.exists():
-            raise FlowNotFound(name)
+            raise FlowNotFound(name, published=not draft)
         return self._load(path)
 
     def key_index(self) -> dict[str, str]:
@@ -782,7 +784,9 @@ class FlowStore:
         and returned as a copy so a caller sorting it cannot disturb the next.
         """
         with self._read_all_lock:
-            if self._read_all_at != self.revision:
+            # Taken before scanning, so a commit landing meanwhile scans again.
+            at = self.revision
+            if self._read_all_at != at:
                 flows = []
                 for path in sorted(self.root.glob("*/flow.json")):
                     name = path.parent.name
@@ -791,7 +795,7 @@ class FlowStore:
                     except Exception:
                         logger.exception("Skipping unreadable flow '%s'", name)
                 self._read_all = flows
-                self._read_all_at = self.revision
+                self._read_all_at = at
             return list(self._read_all)
 
     def write_flow(self, flow: FlowDef) -> bool:
diff --git a/backend/scripts/bench_startup.py b/backend/scripts/bench_startup.py
index 3610ee8..6bb3a46 100644
--- a/backend/scripts/bench_startup.py
+++ b/backend/scripts/bench_startup.py
@@ -96,11 +96,15 @@ def ensure_flow(api: Api) -> None:
     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"]},
-    )
+    # Read again, since the saves move the version. A flow an interrupted run
+    # left published, and unchanged since, has nothing to publish.
+    current = api.call("GET", f"/api/v1/flows/{BENCH_FLOW}")
+    if current["has_draft"]:
+        api.call(
+            "POST",
+            f"/api/v1/flows/{BENCH_FLOW}/publish",
+            {"version": current["definition"]["version"]},
+        )
 
 
 def time_runs(api: Api, count: int) -> tuple[list[float], list[float]]:
diff --git a/backend/tests/flow/test_redis_state.py b/backend/tests/flow/test_redis_state.py
new file mode 100644
index 0000000..525c090
--- /dev/null
+++ b/backend/tests/flow/test_redis_state.py
@@ -0,0 +1,34 @@
+"""RedisState against a real server, enabled by FLUKSIO_TEST_REDIS_HOST (and
+optional PORT)."""
+
+import os
+import uuid
+
+import pytest
+
+from fluksio.flow.state import RedisState
+
+
+@pytest.fixture
+def state():
+    host = os.environ.get("FLUKSIO_TEST_REDIS_HOST")
+    if not host:
+        pytest.skip("Set FLUKSIO_TEST_REDIS_HOST to run Redis integration checks")
+    port = int(os.environ.get("FLUKSIO_TEST_REDIS_PORT", "6379"))
+    state = RedisState(host, port, namespace=f"test-state-{uuid.uuid4().hex}", ttl=600)
+    try:
+        yield state
+    finally:
+        state.clear()
+        state.close()
+
+
+def test_a_ttl_expires_counters_and_the_names_set(state):
+    state.set("heat.temp", 21)
+    state.increment("__version__:heat.a")
+    state.increment_multi(["__version__:heat.b"])
+    state.record({"heat.v": 1}, {}, ["__version__:heat.c"], 0.0)
+
+    keys = state.keys()
+    assert "__names__" in keys and "__version__:heat.c" in keys
+    assert all(state._client.ttl(state._key(k)) > 0 for k in keys)

From d028bf326a11534b0c69f1371f265a21473a18c5 Mon Sep 17 00:00:00 2001
From: stroblme <stroblme@posteo.de>
Date: Wed, 4 Nov 2026 21:38:21 +0100
Subject: [PATCH 2/9] A batch run skips the nodes waiting on a failed one and
 ends error; a run whose nodes cannot load fails with the load error before
 anything executes; cancelling a run stops its pipeline without announcing a
 pause of the live flow; batch_issues refuses a delay node's interval; the
 client accepts an engine URL with or without /api/v1; RedisState's
 compare-and-swap records the names it writes

---
 backend/fluksio/flow/controller.py     | 24 ++++++-
 backend/fluksio/flow/pipeline.py       | 21 ++++++-
 backend/fluksio/flow/runs.py           | 68 +++++++++++++-------
 backend/fluksio/flow/state.py          |  1 +
 backend/fluksio/sdk/client.py          | 12 +++-
 backend/tests/api/routes/test_runs.py  | 86 +++++++++++++++++++++++++-
 backend/tests/flow/test_redis_state.py |  6 ++
 backend/tests/flow/test_runs.py        | 10 +++
 backend/tests/sdk/test_client.py       | 13 ++++
 docs/code/cli.md                       |  3 +-
 docs/concepts/runs.md                  |  9 ++-
 11 files changed, 221 insertions(+), 32 deletions(-)

diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py
index ee0c0e9..33d74d3 100644
--- a/backend/fluksio/flow/controller.py
+++ b/backend/fluksio/flow/controller.py
@@ -148,6 +148,17 @@ class RebuildBusy(RuntimeError):
     """
 
 
+class NodesNotLoaded(RuntimeError):
+    """A run's flow has nodes that could not be built, so none of it runs."""
+
+    def __init__(self, errors: dict[str, str]) -> None:
+        #: The load error of each node, by its qualified id.
+        self.errors = errors
+        node, error = next(iter(errors.items()))
+        more = f" and {len(errors) - 1} more" if len(errors) > 1 else ""
+        super().__init__(f"Node '{node}'{more} could not be loaded: {error}")
+
+
 class NodeStatus(str, Enum):
     ACTIVE = "active"
     ERROR = "error"
@@ -2010,10 +2021,21 @@ class FlowController:
         and webhooks would only be a second copy of what the live pipeline
         already holds. The state is the run's, which is what keeps two runs of
         one flow from overwriting each other's messages.
+
+        :raises NodesNotLoaded: When a node could not be built. The live engine
+            leaves such a node out, but a run without it would finish as if
+            the flow had done its work.
         """
-        nodes, _loaded, initial_values, _inputs = self._build_flows(
+        nodes, loaded, initial_values, _inputs = self._build_flows(
             [(flow, draft)], run=run
         )
+        errors = {
+            entry.id: entry.error or "unknown error"
+            for entry in loaded.values()
+            if entry.status is NodeStatus.ERROR
+        }
+        if errors:
+            raise NodesNotLoaded(errors)
         pipeline = Pipeline(
             nodes=nodes,
             state=state,
diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py
index 8cd66bd..0817f29 100644
--- a/backend/fluksio/flow/pipeline.py
+++ b/backend/fluksio/flow/pipeline.py
@@ -147,6 +147,8 @@ class NodeOutcome(BaseModel):
     artifacts: dict[str, dict[str, Any]] = {}
     #: Restored from an earlier run rather than executed.
     cached: bool = False
+    #: Not executed, because a node it waits on failed or was skipped itself.
+    skipped: bool = False
     #: Which run the restored values came from — and, because a restored row
     #: holds no series of its own, which run this node's metrics live in.
     cached_from: str = ""
@@ -1071,7 +1073,8 @@ class Pipeline:
         overrides: dict[str, Any] | None = None,
         inputs: dict[str, Any] | None = None,
     ) -> dict[str, Any] | None:
-        """Run one node and record its outputs. Never raises.
+        """Run one node and record its outputs. Never raises: None means it
+        failed, and a node that published nothing answers ``{}``.
 
         ``inputs`` is what the readiness check already read — the same keys, a
         moment earlier. Reading them again here was the second of two MGETs
@@ -1106,7 +1109,7 @@ class Pipeline:
             if key:
                 hit, restored = self._from_cache(node, key, state, entry_id)
                 if hit:
-                    return restored
+                    return restored or {}
 
             with logs.capture(collected):
                 result = node.execute(inputs)
@@ -1169,7 +1172,7 @@ class Pipeline:
                         output_values=result,
                     )
                 )
-            return result
+            return result or {}
         except Exception as exc:
             # One failing node must not take the rest of the graph down.
             error = self.publish_error(node, exc, collected, entry_id)
@@ -1431,6 +1434,11 @@ class Pipeline:
 
         ready_nodes = deque(n for n, count in in_degree.items() if count == 0)
         blocked: set[Node] = set()
+        # A batch run starts from an empty state, so a node whose producer
+        # failed would be called without that input. It is skipped instead,
+        # and so is everything waiting on it. A live wave still has the last
+        # values to run on.
+        failed: set[Node] | None = set() if self.run_id else None
         gate_revision = self._gate_revision
         node_futures: dict[Future[dict[str, Any] | None], Node] = {}
         finished: SimpleQueue[Future[dict[str, Any] | None]] = SimpleQueue()
@@ -1469,6 +1477,11 @@ class Pipeline:
                 if n.passive:
                     complete(n)
                     continue
+                if failed is not None and not failed.isdisjoint(deps[n]):
+                    failed.add(n)
+                    self._observe(NodeOutcome(node=n.id, ok=False, skipped=True))
+                    complete(n)
+                    continue
                 if fresh is not None and untouched(n):
                     complete(n)
                     continue
@@ -1497,6 +1510,8 @@ class Pipeline:
                 future = finished.get()
                 n = node_futures.pop(future)
                 result = future.result()
+                if result is None and failed is not None:
+                    failed.add(n)
                 if result is not None and fresh is not None:
                     fresh.update(result)
                 # Silent, failed and throttled producers still release their
diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py
index 555d4a8..65aa1df 100644
--- a/backend/fluksio/flow/runs.py
+++ b/backend/fluksio/flow/runs.py
@@ -51,7 +51,12 @@ from sqlmodel import Session, col, select
 from fluksio.core.db import engine as db_engine
 from fluksio.flow import tombstones
 from fluksio.flow.artifacts import ArtifactStore, is_reference, valid_digest
-from fluksio.flow.controller import NODE_TYPES, FlowController, RunContext
+from fluksio.flow.controller import (
+    NODE_TYPES,
+    FlowController,
+    NodesNotLoaded,
+    RunContext,
+)
 from fluksio.flow.messages import qualify, requalify
 from fluksio.flow.pipeline import CACHE_SHAPE_SINCE, CacheHit, NodeOutcome, Pipeline
 from fluksio.flow.queue import WorkItem, WorkQueue
@@ -137,8 +142,9 @@ def batch_issues(
     history keeps every value, and the interval only thins what is published.
     Anywhere else it is a message quietly going missing.
 
-    A delay node is fine; without a queue to defer into it simply sleeps,
-    which in a run is what was asked for.
+    A delay node's ``interval`` is the same limit, put on every output it
+    has. Its ``delay`` is fine: without a queue to defer into it simply
+    sleeps, which in a run is what was asked for.
 
     The other is a node with no body. The store answers a new node's template
     when nothing was ever written for one, so such a node runs — and returns
@@ -148,8 +154,11 @@ def batch_issues(
     """
     issues: list[str] = []
     for node in flow.nodes:
-        for spec in list(node.requires) + list(node.provides):
-            if spec.interval > 0 and not spec.stream:
+        held = float(node.params.get("interval") or 0) if node.type == "delay" else 0
+        limits = [(spec, spec.interval) for spec in node.requires]
+        limits += [(spec, max(spec.interval, held)) for spec in node.provides]
+        for spec, interval in limits:
+            if interval > 0 and not spec.stream:
                 issues.append(
                     f"Node '{node.id}' rate-limits '{spec.port or spec.name}'. "
                     "A run has no timer to release what that holds back, so "
@@ -1236,8 +1245,10 @@ class RunService:
                 return True
         # The gate first, so nothing new is submitted while the running nodes
         # are being killed; a gated node is never handed to the executor, so
-        # the graph drains instead of going further.
-        pipeline.pause(self._key_of(run_id))
+        # the graph drains instead of going further. Stopped rather than
+        # paused: a pause announces itself on the engine's bus, under the key
+        # the live flow shares, so every open editor showed that one paused.
+        pipeline.set_disabled({self._key_of(run_id)})
         workers = self.controller.workers
         if workers is not None:
             # Keyed by run, so a sweep cancelling one config leaves the others
@@ -1502,7 +1513,7 @@ class RunService:
 
         def observe(outcome: NodeOutcome) -> None:
             nonlocal errors
-            if not outcome.ok:
+            if not outcome.ok and not outcome.skipped:
                 errors += 1
             self._record_node(run, key, outcome)
 
@@ -1537,19 +1548,26 @@ class RunService:
             digests = node_digests(flow)
             run.code_digest = self._restamp(run, run_digest(flow))
             state = self._state_factory(f"{RUN_NAMESPACE}:{run_id}")
-            pipeline = self.controller.build_run_pipeline(
-                flow,
-                state=state,
-                draft=run.draft,
-                observer=observe,
-                emission_observer=sink.handle,
-                run=RunContext(
-                    run_id=run_id,
-                    code_digest=run.code_digest,
-                    node_digests=digests,
-                ),
-                run_cache=None if run.no_cache else self._cache,
-            )
+            try:
+                pipeline = self.controller.build_run_pipeline(
+                    flow,
+                    state=state,
+                    draft=run.draft,
+                    observer=observe,
+                    emission_observer=sink.handle,
+                    run=RunContext(
+                        run_id=run_id,
+                        code_digest=run.code_digest,
+                        node_digests=digests,
+                    ),
+                    run_cache=None if run.no_cache else self._cache,
+                )
+            except NodesNotLoaded as exc:
+                # Nothing of it runs; the finally below records it.
+                for node, error in exc.errors.items():
+                    observe(NodeOutcome(node=node, ok=False, error=error))
+                status, reason = "error", str(exc)[:1024]
+                return
             with self._lock:
                 if run_id in self._cancelled:
                     # Cancelled while it was being built, so nothing of it
@@ -1627,10 +1645,16 @@ class RunService:
         run_id = run.id
         node = requalify(outcome.node, key, run.flow_id)[:255]
         outputs = _cacheable(outcome)
+        if outcome.cached:
+            status = "cached"
+        elif outcome.ok:
+            status = "ok"
+        else:
+            status = "skipped" if outcome.skipped else "error"
         row = RunNode(
             run_id=run_id,
             node=node,
-            status="cached" if outcome.cached else ("ok" if outcome.ok else "error"),
+            status=status,
             started_at=datetime.now(UTC),
             duration_ms=outcome.duration_ms,
             error=outcome.error[:ERROR_CAP],
diff --git a/backend/fluksio/flow/state.py b/backend/fluksio/flow/state.py
index fcd0133..cf2a70f 100644
--- a/backend/fluksio/flow/state.py
+++ b/backend/fluksio/flow/state.py
@@ -855,6 +855,7 @@ class RedisState(StateBackend):
                     pipe.setex(self._key(key), self._ttl, data)
                 else:
                     pipe.set(self._key(key), data)
+            self._note(pipe, updates)
 
             pipe.execute()
             return True
diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py
index 83589e9..1a5c3b0 100644
--- a/backend/fluksio/sdk/client.py
+++ b/backend/fluksio/sdk/client.py
@@ -170,6 +170,11 @@ class Conflict(ApiError):
             self.current_version = int(detail.get("current_version") or 0)
 
 
+def engine_url(url: str) -> str:
+    """An engine's address, whether it was given with the API's path or not."""
+    return url.rstrip("/").removesuffix(API).rstrip("/")
+
+
 class Client:
     """An authenticated engine, addressed over its HTTP API."""
 
@@ -182,7 +187,9 @@ class Client:
         retries: int = RETRIES,
     ) -> None:
         stored = _stored()
-        self.url = url or os.environ.get("FLUKSIO_URL") or stored.get("url") or ""
+        self.url = engine_url(
+            url or os.environ.get("FLUKSIO_URL") or stored.get("url") or ""
+        )
         self.token = (
             token or os.environ.get("FLUKSIO_TOKEN") or stored.get("token") or ""
         )
@@ -766,8 +773,9 @@ def login(
     """Exchange credentials for a token and remember the engine."""
     import httpx
 
+    url = engine_url(url)
     response = httpx.post(
-        f"{url.rstrip('/')}{API}/login/access-token",
+        f"{url}{API}/login/access-token",
         data={"username": email, "password": password},
         timeout=timeout,
     )
diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py
index 3dd4261..894989c 100644
--- a/backend/tests/api/routes/test_runs.py
+++ b/backend/tests/api/routes/test_runs.py
@@ -19,7 +19,7 @@ from fluksio.core.db import engine as db_engine
 from fluksio.flow.artifacts import ArtifactStore
 from fluksio.flow.controller import FlowController
 from fluksio.flow.messages import DType, MessageSpec
-from fluksio.flow.pipeline import CACHE_SHAPE_SINCE, NodeOutcome
+from fluksio.flow.pipeline import CACHE_SHAPE_SINCE, NodeOutcome, Pipeline
 from fluksio.flow.runs import (
     LEASE_STALE_S,
     OUTPUT_CAP,
@@ -729,6 +729,57 @@ def test_a_draft_run_executes_the_draft(tmp_path):
             session.commit()
 
 
+def test_a_failed_stage_skips_what_waits_on_it_and_fails_the_run(tmp_path):
+    """Stages after a failure were called without their input and raised
+    TypeError, and a run whose only node could not load finished `ok`."""
+    model, score = (MessageSpec(name=n, dtype=DType.FLOAT) for n in ("model", "score"))
+    staged = FlowDef(
+        name="staged",
+        mode="batch",
+        nodes=[
+            NodeDef(id="fit", provides=[model]),
+            NodeDef(id="readout", requires=[model], provides=[score]),
+            NodeDef(id="evaluate", requires=[score]),
+        ],
+    )
+    broken = FlowDef(name="broken", mode="batch", nodes=[NodeDef(id="fit")])
+    store = FlowStore(tmp_path / "flows")
+    store.write_flow(staged)
+    store.write_flow(broken)
+    store.write_node_source("staged", "fit", "def process():\n    raise ValueError\n")
+    store.write_node_source(
+        "staged", "readout", "def process(model):\n    return {'score': model}\n"
+    )
+    store.write_node_source("staged", "evaluate", "def process(score):\n    pass\n")
+    store.write_node_source("broken", "fit", "def process(:\n")
+
+    service = RunService(controller=FlowController(store), queue=_Collect())
+    made = [service.submit(name, {}).id for name in ("staged", "broken")]
+    try:
+        for run_id in made:
+            service._drive(run_id)
+        with Session(db_engine) as session:
+            ran, failed = (session.get(Run, run_id) for run_id in made)
+            assert ran.status == failed.status == "error"
+            assert failed.status_reason.startswith("Node 'broken.fit' could not")
+            rows = session.exec(select(RunNode).where(col(RunNode.run_id).in_(made)))
+            assert {row.node: row.status for row in rows} == {
+                "staged.fit": "error",
+                "staged.readout": "skipped",
+                "staged.evaluate": "skipped",
+                "broken.fit": "error",
+            }
+    finally:
+        with Session(db_engine) as session:
+            for node in session.exec(
+                select(RunNode).where(col(RunNode.run_id).in_(made))
+            ).all():
+                session.delete(node)
+            for run in session.exec(select(Run).where(col(Run.id).in_(made))).all():
+                session.delete(run)
+            session.commit()
+
+
 def test_a_retry_is_a_new_run_that_names_the_one_it_repeats():
     """The way back from a run an engine restart interrupted.
 
@@ -905,6 +956,39 @@ def test_a_run_waiting_for_a_worker_keeps_its_lease_and_can_be_cancelled(
             session.commit()
 
 
+class _Bus:
+    def __init__(self):
+        self.events = []
+
+    def publish(self, event):
+        self.events.append(event)
+
+
+def test_cancelling_a_run_does_not_pause_the_live_flow_it_shares_a_key_with():
+    """A run's pipeline publishes on the engine's bus under the flow's key, so
+    its pause marked the live flow paused in every open editor."""
+
+    class _Idle(_OneFlow):
+        workers = None
+        remote = None
+
+    service = RunService(
+        controller=_Idle(FlowDef(name="study", mode="batch")), queue=_Collect()
+    )
+    run = service.submit("study", {})
+    bus = _Bus()
+    pipeline = Pipeline(events=bus)
+    service._active[run.id] = pipeline
+    try:
+        assert service.cancel(run.id)
+        assert pipeline.is_disabled("study")
+        assert not [e for e in bus.events if e["type"] == "flow_paused"]
+    finally:
+        with Session(db_engine) as session:
+            session.delete(session.get(Run, run.id))
+            session.commit()
+
+
 def test_a_key_nobody_used_submits_normally(
     client, superuser_token_headers, monkeypatch
 ):
diff --git a/backend/tests/flow/test_redis_state.py b/backend/tests/flow/test_redis_state.py
index 525c090..6868c4b 100644
--- a/backend/tests/flow/test_redis_state.py
+++ b/backend/tests/flow/test_redis_state.py
@@ -32,3 +32,9 @@ def test_a_ttl_expires_counters_and_the_names_set(state):
     keys = state.keys()
     assert "__names__" in keys and "__version__:heat.c" in keys
     assert all(state._client.ttl(state._key(k)) > 0 for k in keys)
+
+
+def test_a_compare_and_swap_names_what_it_wrote(state):
+    state.set("heat.a", 1)
+    assert state.compare_and_swap_multi({"heat.a": 1}, {"heat.b": 2})
+    assert "heat.b" in state.message_names()
diff --git a/backend/tests/flow/test_runs.py b/backend/tests/flow/test_runs.py
index f1abf65..b1f56db 100644
--- a/backend/tests/flow/test_runs.py
+++ b/backend/tests/flow/test_runs.py
@@ -346,6 +346,16 @@ def test_a_rate_limited_port_cannot_be_run_as_a_batch():
     assert not batch_issues(double_flow())
 
 
+def test_a_delay_nodes_interval_cannot_be_run_as_a_batch():
+    flow = double_flow()
+    flow.nodes[0].type = "delay"
+    flow.nodes[0].params = {"interval": 5}
+    # The node puts it on every output, where it holds values back the same way.
+    assert batch_issues(flow)
+    flow.nodes[0].params = {"delay": 5}
+    assert not batch_issues(flow)
+
+
 def test_a_node_with_no_stored_code_cannot_be_run():
     """The store answers a template for one, and a template publishes nothing."""
 
diff --git a/backend/tests/sdk/test_client.py b/backend/tests/sdk/test_client.py
index aaf7c1a..3a1b0f5 100644
--- a/backend/tests/sdk/test_client.py
+++ b/backend/tests/sdk/test_client.py
@@ -33,6 +33,19 @@ def a_client(handler, monkeypatch, **kwargs):
     return Client(url="http://engine", token="t", http=http, **kwargs)
 
 
+@pytest.mark.parametrize(
+    "url",
+    [
+        "http://e:8000",
+        "http://e:8000/",
+        "http://e:8000/api/v1",
+        "http://e:8000/api/v1/",
+    ],
+)
+def test_an_engine_url_may_carry_the_api_path(url):
+    assert Client(url=url, token="t").url == "http://e:8000"
+
+
 def test_idempotent_get_is_tried_again(monkeypatch):
     calls = []
 
diff --git a/docs/code/cli.md b/docs/code/cli.md
index 89ad14e..1cbd006 100644
--- a/docs/code/cli.md
+++ b/docs/code/cli.md
@@ -151,7 +151,8 @@ Client commands use the nearest `.fluksio/client.json`, or the global
 configuration when there is no project instance. Legacy client configuration
 is a fallback.
 `FLUKSIO_URL` and `FLUKSIO_TOKEN` override saved values; command options
-`--url` and `--token` override both.
+`--url` and `--token` override both. An engine URL works with or without
+the `/api/v1` path.
 
 A client needs an engine at least as new as itself. An engine from before
 flow ids cannot create flows the way this client does, and the client says
diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md
index 351e814..2c711e1 100644
--- a/docs/concepts/runs.md
+++ b/docs/concepts/runs.md
@@ -18,9 +18,14 @@ A batch flow does not need activation and does not open subscriptions,
 schedules or webhooks. A run executes each node once after its dependencies
 finish. Missing parameters use their declared initial values.
 
+When a node fails, the nodes that wait on it are recorded as `skipped` and
+the run ends `error`. A node that cannot be loaded fails the run before
+anything executes, with the node's load error as the reason.
+
 Non-streaming ports cannot use `interval` in a batch flow: there is no timer
-to deliver a delayed value. Streaming ports can use it to reduce display
-updates while keeping every metric point.
+to deliver a delayed value. The same goes for a delay node's `interval`
+setting. Streaming ports can use it to reduce display updates while keeping
+every metric point.
 
 ## Submitting
 

From 1a622edb1fd5c0cb26bfb6b37fe29a043025cdd3 Mon Sep 17 00:00:00 2001
From: stroblme <stroblme@posteo.de>
Date: Wed, 4 Nov 2026 21:53:41 +0100
Subject: [PATCH 3/9] RedisState's compare-and-swap watches through its
 pipeline, so a write between the compare and EXEC fails the swap;
 pause_intake waits out a claim already under way, so nothing dispatches after
 it returns

---
 backend/fluksio/flow/executor.py       | 15 ++++++--
 backend/fluksio/flow/state.py          | 52 ++++++++++++--------------
 backend/tests/flow/test_engine_hold.py | 25 ++++++++++++-
 backend/tests/flow/test_redis_state.py | 14 +++++++
 4 files changed, 74 insertions(+), 32 deletions(-)

diff --git a/backend/fluksio/flow/executor.py b/backend/fluksio/flow/executor.py
index 89f15d0..a77baac 100644
--- a/backend/fluksio/flow/executor.py
+++ b/backend/fluksio/flow/executor.py
@@ -259,10 +259,19 @@ class ExecutionService:
         """Stop claiming, and wait for what is already claimed to finish.
 
         Called around a rebuild: items claimed against the old pipeline should
-        finish there rather than half-run against the new one.
+        finish there rather than half-run against the new one. That includes
+        a claim still under way: the consumer holds its lock from the claim to
+        the dispatch, so once that lock is free here the claim is counted, and
+        every later pass finds the intake closed.
         """
         self._intake.clear()
+        # So a consumer waiting for capacity sees the intake closed.
+        self._capacity_changed()
         deadline = time.monotonic() + DRAIN_TIMEOUT_S
+        if not self._consumer_lock.acquire(timeout=DRAIN_TIMEOUT_S):
+            logger.warning("Rebuild did not wait out a claim")
+            return
+        self._consumer_lock.release()
         with self._inflight_lock:
             while self._running:
                 remaining = deadline - time.monotonic()
@@ -284,8 +293,8 @@ class ExecutionService:
         """Stop claiming, promoting, touching and reclaiming, once every
         claimed cascade has finished. For a rename, which rewrites the queue.
 
-        Unlike `pause_intake` this also waits out a claim already blocking
-        when intake clears, and a timer pass. It waits without a limit and
+        Unlike `pause_intake` this also waits out a timer pass, and keeps the
+        consumer lock until `release`. It waits without a limit and
         tells ``report`` the nodes whose cascades are still running each time
         they change. Setting ``abort`` gives up: claiming resumes and it
         returns False. True means held until `release`.
diff --git a/backend/fluksio/flow/state.py b/backend/fluksio/flow/state.py
index cf2a70f..d6fc341 100644
--- a/backend/fluksio/flow/state.py
+++ b/backend/fluksio/flow/state.py
@@ -834,35 +834,31 @@ class RedisState(StateBackend):
 
         watch_keys = [self._key(k) for k in expected.keys()]
 
-        try:
-            # Start watching the keys for changes
-            self._client.watch(*watch_keys)
+        # WATCH has to go through the pipeline: it pins one connection, and
+        # the client-level call is a no-op in redis-py 7. Leaving the block
+        # resets the pipeline, which unwatches.
+        with self._client.pipeline() as pipe:
+            try:
+                pipe.watch(*watch_keys)  # type: ignore[no-untyped-call]
+                # A watching pipeline runs commands immediately.
+                current = cast(list[bytes | None], pipe.mget(watch_keys))
+                for raw, exp_value in zip(current, expected.values(), strict=True):
+                    if self._deserialize(raw) != exp_value:
+                        return False
 
-            # Check current values
-            current = self.get_multi(list(expected.keys()))
-
-            for key, exp_value in expected.items():
-                if current.get(key) != exp_value:
-                    self._client.unwatch()
-                    return False
-
-            # All values match - execute the transaction
-            pipe = self._client.pipeline(True)  # True = use MULTI/EXEC
-
-            for key, value in updates.items():
-                data = self._serialize(value)
-                if self._ttl:
-                    pipe.setex(self._key(key), self._ttl, data)
-                else:
-                    pipe.set(self._key(key), data)
-            self._note(pipe, updates)
-
-            pipe.execute()
-            return True
-
-        except redis.WatchError:
-            # Another client modified one of the watched keys
-            return False
+                pipe.multi()
+                for key, value in updates.items():
+                    data = self._serialize(value)
+                    if self._ttl:
+                        pipe.setex(self._key(key), self._ttl, data)
+                    else:
+                        pipe.set(self._key(key), data)
+                self._note(pipe, updates)
+                pipe.execute()
+                return True
+            except redis.WatchError:
+                # Another client modified one of the watched keys
+                return False
 
     def close(self) -> None:
         """Give the connection pool back."""
diff --git a/backend/tests/flow/test_engine_hold.py b/backend/tests/flow/test_engine_hold.py
index c796d4c..a815df4 100644
--- a/backend/tests/flow/test_engine_hold.py
+++ b/backend/tests/flow/test_engine_hold.py
@@ -125,8 +125,31 @@ class SlowClaim(MemoryWorkQueue):
         return items
 
 
+def test_pause_intake_waits_out_a_claim_already_under_way(monkeypatch):
+    seen: list = []
+    queue = SlowClaim()
+    _, service, source = _engine(seen.append, queue)
+    wake = service._capacity_changed
+
+    def closing() -> None:
+        wake()
+        # The claim lands only once the intake has closed.
+        queue.go.set()
+
+    monkeypatch.setattr(service, "_capacity_changed", closing)
+    service.start()
+    try:
+        source.inject({"reading": 1.0})
+        assert queue.claiming.wait(5)
+        service.pause_intake()
+        assert seen == [1.0]
+        assert service.inflight == 0
+    finally:
+        queue.go.set()
+        service.stop()
+
+
 def test_a_hold_waits_out_a_claim_already_under_way():
-    """`pause_intake` returns here and the item runs after it."""
     seen: list = []
     queue = SlowClaim()
     _, service, source = _engine(seen.append, queue)
diff --git a/backend/tests/flow/test_redis_state.py b/backend/tests/flow/test_redis_state.py
index 6868c4b..ae84d53 100644
--- a/backend/tests/flow/test_redis_state.py
+++ b/backend/tests/flow/test_redis_state.py
@@ -38,3 +38,17 @@ def test_a_compare_and_swap_names_what_it_wrote(state):
     state.set("heat.a", 1)
     assert state.compare_and_swap_multi({"heat.a": 1}, {"heat.b": 2})
     assert "heat.b" in state.message_names()
+
+
+def test_a_compare_and_swap_loses_to_a_write_after_its_read(state, monkeypatch):
+    state.set("heat.a", 1)
+    note = RedisState._note
+
+    def racing_note(self, pipe, keys):
+        # Another writer lands between the compare and EXEC.
+        self._client.set(self._key("heat.a"), b"2")
+        note(self, pipe, keys)
+
+    monkeypatch.setattr(RedisState, "_note", racing_note)
+    assert not state.compare_and_swap_multi({"heat.a": 1}, {"heat.a": 3})
+    assert state.get("heat.a") == 2

From 78b2c1438ca8f4923df74ee858d4f99bf5b553ce Mon Sep 17 00:00:00 2001
From: stroblme <stroblme@posteo.de>
Date: Thu, 5 Nov 2026 06:43:33 +0100
Subject: [PATCH 4/9] pause_intake answers False when claimed work outlasts its
 timeout, and reload then resumes intake and raises RebuildBusy before tearing
 anything down

---
 backend/fluksio/flow/controller.py     |  9 +++++++--
 backend/fluksio/flow/executor.py       | 10 +++++++---
 backend/tests/flow/test_engine_hold.py | 26 +++++++++++++++++++++++++-
 3 files changed, 39 insertions(+), 6 deletions(-)

diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py
index 33d74d3..eaffc4f 100644
--- a/backend/fluksio/flow/controller.py
+++ b/backend/fluksio/flow/controller.py
@@ -741,8 +741,13 @@ class FlowController:
         try:
             # Work already claimed belongs to the pipeline it was claimed
             # against; let it finish there before swapping the graph out.
-            if self.execution is not None:
-                await run_in_threadpool(self.execution.pause_intake)
+            # Work that does not finish in time keeps the old graph, which
+            # nothing has touched yet.
+            if self.execution is not None and not await run_in_threadpool(
+                self.execution.pause_intake
+            ):
+                self.execution.resume_intake()
+                raise RebuildBusy("Claimed work is still running on the pipeline")
             await self._teardown()
             # A fresh supervisor per build, so a flow quarantined by the last
             # one gets another chance once its author has changed something.
diff --git a/backend/fluksio/flow/executor.py b/backend/fluksio/flow/executor.py
index a77baac..6b98c78 100644
--- a/backend/fluksio/flow/executor.py
+++ b/backend/fluksio/flow/executor.py
@@ -255,7 +255,7 @@ class ExecutionService:
         with self._inflight_lock:
             self._inflight_lock.notify_all()
 
-    def pause_intake(self) -> None:
+    def pause_intake(self) -> bool:
         """Stop claiming, and wait for what is already claimed to finish.
 
         Called around a rebuild: items claimed against the old pipeline should
@@ -263,6 +263,9 @@ class ExecutionService:
         a claim still under way: the consumer holds its lock from the claim to
         the dispatch, so once that lock is free here the claim is counted, and
         every later pass finds the intake closed.
+
+        False when that took longer than `DRAIN_TIMEOUT_S`. The intake stays
+        closed either way; the caller decides whether to go on.
         """
         self._intake.clear()
         # So a consumer waiting for capacity sees the intake closed.
@@ -270,7 +273,7 @@ class ExecutionService:
         deadline = time.monotonic() + DRAIN_TIMEOUT_S
         if not self._consumer_lock.acquire(timeout=DRAIN_TIMEOUT_S):
             logger.warning("Rebuild did not wait out a claim")
-            return
+            return False
         self._consumer_lock.release()
         with self._inflight_lock:
             while self._running:
@@ -279,8 +282,9 @@ class ExecutionService:
                     logger.warning(
                         "Rebuild did not wait out %d cascades", self._running.total()
                     )
-                    return
+                    return False
                 self._inflight_lock.wait(remaining)
+        return True
 
     def resume_intake(self) -> None:
         self._intake.set()
diff --git a/backend/tests/flow/test_engine_hold.py b/backend/tests/flow/test_engine_hold.py
index a815df4..6a5704e 100644
--- a/backend/tests/flow/test_engine_hold.py
+++ b/backend/tests/flow/test_engine_hold.py
@@ -5,13 +5,14 @@ service stops claiming and its timers once claimed work has finished. Both
 wait without a limit and can be aborted.
 """
 
+import asyncio
 import threading
 import time
 
 import pytest
 
 from fluksio.flow import executor
-from fluksio.flow.controller import FlowController, LoadedNode
+from fluksio.flow.controller import FlowController, LoadedNode, RebuildBusy
 from fluksio.flow.executor import ExecutionService
 from fluksio.flow.messages import DType, MessageSpec
 from fluksio.flow.namespace import Rename
@@ -149,6 +150,29 @@ def test_pause_intake_waits_out_a_claim_already_under_way(monkeypatch):
         service.stop()
 
 
+def test_a_reload_that_cannot_drain_leaves_the_pipeline_as_it_was(
+    tmp_path, monkeypatch
+):
+    monkeypatch.setattr(executor, "DRAIN_TIMEOUT_S", 0.05)
+    seen: list = []
+    running, finish, consume = _blocking(seen)
+    pipeline, service, source = _engine(consume)
+    controller = FlowController(FlowStore(tmp_path / "flows"), execution=service)
+    controller.pipeline = pipeline
+    service.start()
+    try:
+        source.inject({"reading": 1.0})
+        assert running.wait(5)
+        with pytest.raises(RebuildBusy):
+            asyncio.run(controller.reload())
+        assert controller.pipeline is pipeline
+        assert service._pipeline is pipeline
+        assert service._intake.is_set()
+    finally:
+        finish.set()
+        service.stop()
+
+
 def test_a_hold_waits_out_a_claim_already_under_way():
     seen: list = []
     queue = SlowClaim()

From c6ac7f80a9f52eeab9ee3158cc03ec72ce1a7066 Mon Sep 17 00:00:00 2001
From: stroblme <stroblme@posteo.de>
Date: Thu, 5 Nov 2026 07:35:02 +0100
Subject: [PATCH 5/9] A batch run fails a node with no value on a triggering
 input instead of calling it, naming the input and its producer, and skips
 what waits on it

---
 backend/fluksio/flow/pipeline.py      | 23 +++++++++++++
 backend/tests/api/routes/test_runs.py | 48 +++++++++++++++++++++++++++
 docs/concepts/runs.md                 |  3 ++
 3 files changed, 74 insertions(+)

diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py
index 0817f29..22da5ff 100644
--- a/backend/fluksio/flow/pipeline.py
+++ b/backend/fluksio/flow/pipeline.py
@@ -1486,6 +1486,17 @@ class Pipeline:
                     complete(n)
                     continue
                 ready_inputs: dict[str, Any] | None = None
+                if failed is not None:
+                    # A live wave's readiness rule: a node runs once every
+                    # triggering input has a value. A run has nothing to wait
+                    # for, so the node fails instead.
+                    ready_inputs = state.get_present(list(n.requires))
+                    missing = self._no_value(n, ready_inputs)
+                    if missing:
+                        failed.add(n)
+                        self._observe(NodeOutcome(node=n.id, ok=False, error=missing))
+                        complete(n)
+                        continue
                 if check_ready:
                     ready, ready_inputs = self._is_node_ready(n, state, entry_id)
                     if not ready:
@@ -1530,6 +1541,18 @@ class Pipeline:
 
         return state
 
+    def _no_value(self, node: Node, values: dict[str, Any]) -> str:
+        """The first triggering input with nothing in ``values``, as an error,
+        or "" when there is none."""
+        for name, spec in node.requires.items():
+            if spec.trigger and name not in values:
+                producers = ", ".join(
+                    f"'{p.local_id}'" for p in self.produces.get(name, ())
+                )
+                where = f" from {producers}" if producers else ""
+                return f"No value on input '{spec.port or name}'{where}"
+        return ""
+
     def run(
         self, inputs: dict[str, Any] | None = None, nodes: set[Node] | None = None
     ) -> StateBackend:
diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py
index 894989c..1061161 100644
--- a/backend/tests/api/routes/test_runs.py
+++ b/backend/tests/api/routes/test_runs.py
@@ -780,6 +780,54 @@ def test_a_failed_stage_skips_what_waits_on_it_and_fails_the_run(tmp_path):
             session.commit()
 
 
+def test_a_stage_with_nothing_on_an_input_fails_instead_of_being_called(tmp_path):
+    """The producer succeeded but published nothing on the port it waits on."""
+    model, score, report = (
+        MessageSpec(name=n, dtype=DType.FLOAT) for n in ("model", "score", "report")
+    )
+    flow = FlowDef(
+        name="split",
+        mode="batch",
+        nodes=[
+            NodeDef(id="fit", provides=[model, score]),
+            NodeDef(id="readout", requires=[score], provides=[report]),
+            NodeDef(id="evaluate", requires=[report]),
+        ],
+    )
+    store = FlowStore(tmp_path / "flows")
+    store.write_flow(flow)
+    store.write_node_source(
+        "split", "fit", "def process():\n    return {'model': 1.0}\n"
+    )
+    store.write_node_source(
+        "split", "readout", "def process(score):\n    return {'report': score}\n"
+    )
+    store.write_node_source("split", "evaluate", "def process(report):\n    pass\n")
+
+    service = RunService(controller=FlowController(store), queue=_Collect())
+    run_id = service.submit("split", {}).id
+    try:
+        service._drive(run_id)
+        with Session(db_engine) as session:
+            assert session.get(Run, run_id).status == "error"
+            rows = {
+                row.node: row
+                for row in session.exec(select(RunNode).where(RunNode.run_id == run_id))
+            }
+            assert {node: row.status for node, row in rows.items()} == {
+                "split.fit": "ok",
+                "split.readout": "error",
+                "split.evaluate": "skipped",
+            }
+            assert rows["split.readout"].error == "No value on input 'score' from 'fit'"
+    finally:
+        with Session(db_engine) as session:
+            for node in session.exec(select(RunNode).where(RunNode.run_id == run_id)):
+                session.delete(node)
+            session.delete(session.get(Run, run_id))
+            session.commit()
+
+
 def test_a_retry_is_a_new_run_that_names_the_one_it_repeats():
     """The way back from a run an engine restart interrupted.
 
diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md
index 2c711e1..333b9a8 100644
--- a/docs/concepts/runs.md
+++ b/docs/concepts/runs.md
@@ -21,6 +21,9 @@ finish. Missing parameters use their declared initial values.
 When a node fails, the nodes that wait on it are recorded as `skipped` and
 the run ends `error`. A node that cannot be loaded fails the run before
 anything executes, with the node's load error as the reason.
+A node with no value on a triggering input fails without being called, as a
+live flow would wait for it, so a canvas port with a default needs
+`trigger: false` to be optional.
 
 Non-streaming ports cannot use `interval` in a batch flow: there is no timer
 to deliver a delayed value. The same goes for a delay node's `interval`

From 4c15f4abe1759cf3be4033f4ac0a7ff791e540fb Mon Sep 17 00:00:00 2001
From: stroblme <stroblme@posteo.de>
Date: Thu, 5 Nov 2026 08:03:56 +0100
Subject: [PATCH 6/9] Test hygiene: empty the flow store between modules,
 unbind demand on shutdown, poll the hook state in test_flow_ids, count only
 the watcher under test in the revoked-link test, delete runs.spec's runs

---
 backend/fluksio/main.py                   |  4 +++-
 backend/tests/api/routes/test_flow_ids.py | 11 +++++++++--
 backend/tests/conftest.py                 | 14 ++++++++++++++
 backend/tests/test_cloud.py               |  4 +++-
 frontend/tests/runs.spec.ts               |  6 +++++-
 5 files changed, 34 insertions(+), 5 deletions(-)

diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py
index 842bfe2..595ab26 100644
--- a/backend/fluksio/main.py
+++ b/backend/fluksio/main.py
@@ -428,8 +428,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         app.state.device_store = devices
         controller.devices = devices
         # What the sockets want and the flows read, mapped onto the devices'
-        # on-demand ports.
+        # on-demand ports. It is process-wide, so it lets go of this
+        # controller when the engine stops.
         demand.bind(controller)
+        started.append(lambda: demand.bind(None))
         # Charts need a deeper series than the default; tell the engine
         # before it starts recording.
         controller.set_history_limits(dashboards.history_requirements())
diff --git a/backend/tests/api/routes/test_flow_ids.py b/backend/tests/api/routes/test_flow_ids.py
index a3b3050..a7550f9 100644
--- a/backend/tests/api/routes/test_flow_ids.py
+++ b/backend/tests/api/routes/test_flow_ids.py
@@ -20,6 +20,7 @@ from fluksio.flow.dashboards import DashboardDef, WidgetDef
 from fluksio.flow.store import ISSUED_ID
 from fluksio.flow.tombstones import names_of
 from fluksio.models import FlowTombstone, Run
+from tests.api.routes.test_rename import _until
 
 FLOWS = f"{settings.API_V1_STR}/flows"
 DASHBOARDS = f"{settings.API_V1_STR}/dashboards"
@@ -180,8 +181,14 @@ def test_a_flow_runs_under_its_key_and_is_reached_by_its_id(
     hooked = client.post(f"/hooks/{flow_id}/tick", json={"tick": 7.0})
     assert hooked.status_code == 200, hooked.text
     assert client.post(f"/hooks/{key}/tick", json={"tick": 8.0}).status_code == 404
-    state = client.get(f"{FLOWS}/{flow_id}/state", headers=superuser_token_headers)
-    assert state.json()["values"][f"{key}.tick"]["value"] == 7.0
+
+    def values() -> dict[str, Any]:
+        state = client.get(f"{FLOWS}/{flow_id}/state", headers=superuser_token_headers)
+        return dict(state.json()["values"])
+
+    # Once the cascade it queued has run.
+    _until(lambda: f"{key}.tick" in values())
+    assert values()[f"{key}.tick"]["value"] == 7.0
 
     paused = client.post(f"{FLOWS}/{flow_id}/pause", headers=superuser_token_headers)
     assert paused.status_code == 200
diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
index ef3f4d7..3237d2d 100644
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -1,3 +1,4 @@
+import shutil
 import uuid
 from collections.abc import Generator
 from datetime import UTC, datetime
@@ -107,6 +108,19 @@ def _clean_engine_rows() -> Generator[None, None, None]:
         session.commit()
 
 
+@pytest.fixture(scope="module", autouse=True)
+def _clean_flow_store() -> Generator[None, None, None]:
+    """Start the next module on an empty flow store.
+
+    The store is session-wide like the database, so the flows, dashboards and
+    devices one module left were loaded by the next module's app at startup.
+    Removed whole, its git with it, after the module's app has stopped.
+    """
+    yield
+    shutil.rmtree(settings.FLOWS_DIR, ignore_errors=True)
+    settings.DEVICES_FILE.unlink(missing_ok=True)
+
+
 @pytest.fixture(scope="module")
 def client() -> Generator[TestClient, None, None]:
     with TestClient(app) as c:
diff --git a/backend/tests/test_cloud.py b/backend/tests/test_cloud.py
index 84df2be..55d7a51 100644
--- a/backend/tests/test_cloud.py
+++ b/backend/tests/test_cloud.py
@@ -788,7 +788,9 @@ async def test_a_revoked_link_stands_down(enrolled: User) -> None:  # noqa: ARG0
         watcher = asyncio.create_task(connector.watch_enrolment(app))
         await asyncio.sleep(0.05)
         watcher.cancel()
-    assert not started
+    # The module's app runs a watcher of its own on the same patched names and
+    # may dial the enrolment above inside the window, so only this one counts.
+    assert app not in started
 
 
 def test_a_member_removed_at_the_portal_loses_their_user(db: Session) -> None:
diff --git a/frontend/tests/runs.spec.ts b/frontend/tests/runs.spec.ts
index 2cef62e..327bcd7 100644
--- a/frontend/tests/runs.spec.ts
+++ b/frontend/tests/runs.spec.ts
@@ -23,7 +23,11 @@ const TRAIN = `def process(epochs):
 `
 
 test.afterAll(async ({ browser }) => {
-  await deleteAll(browser, [`/flows/${flowName}`])
+  // The runs after the flow, as the editor does it. Its pinned id is its key.
+  await deleteAll(browser, [
+    `/flows/${flowName}`,
+    `/runs?flow=${flowName}&history=true`,
+  ])
 })
 
 test.beforeAll(async ({ browser }) => {

From 261cf1ec2a7f654e56b95d6fe356e54425b31237 Mon Sep 17 00:00:00 2001
From: Melvin Strobl <lc3267@kit.edu>
Date: Fri, 6 Nov 2026 07:27:44 +0100
Subject: [PATCH 7/9] Assistant answers general questions and stays silent for
 overheard talk ("Ignored." is not said); stt prompt setting sent as the
 transcription prompt

---
 backend/fluksio/ai/client.py            | 24 +++++++++++++++--------
 backend/fluksio/flow/nodes/assistant.py | 25 ++++++++++++++++--------
 backend/fluksio/flow/nodes/speech.py    | 26 ++++++++++++++++++++++---
 backend/tests/flow/test_assistant.py    | 18 ++++++++++++++++-
 backend/tests/flow/test_speech.py       |  5 +++--
 docs/reference/node-types.md            | 19 ++++++++++++++++--
 6 files changed, 93 insertions(+), 24 deletions(-)

diff --git a/backend/fluksio/ai/client.py b/backend/fluksio/ai/client.py
index 544d328..da9b553 100644
--- a/backend/fluksio/ai/client.py
+++ b/backend/fluksio/ai/client.py
@@ -194,12 +194,19 @@ async def transcribe(
     *,
     model: str = "whisper-1",
     timeout: float = 30.0,
+    prompt: str = "",
 ) -> str:
-    """``POST {api_base}/audio/transcriptions``: a whole clip, and what it says."""
+    """``POST {api_base}/audio/transcriptions``: a whole clip, and what it says.
+
+    ``prompt`` is words the clip is likely to hold, such as room names.
+    """
+    data = {"model": model, "response_format": "json"}
+    if prompt:
+        data["prompt"] = prompt
     response = await _client(api_base).post(
         api_base.rstrip("/") + "/audio/transcriptions",
         files={"file": (filename, audio, media_type)},
-        data={"model": model, "response_format": "json"},
+        data=data,
         timeout=timeout,
     )
     if response.is_error:
@@ -236,25 +243,26 @@ async def transcribe_stream(
     *,
     rate: int = 16000,
     idle_s: float = 10.0,
+    prompt: str = "",
 ) -> AsyncIterator[tuple[bool, str]]:
     """``WS {api_base}/realtime``: one utterance, transcribed while it is spoken.
 
     Sends each chunk of PCM16 taken off ``audio`` and commits at ``None``.
     Yields ``(False, delta)`` as words are agreed on, then ``(True,
     transcript)`` once. A chunk that takes longer than ``idle_s`` to come
-    raises ``TimeoutError``, so an utterance nobody commits ends.
+    raises ``TimeoutError``, so an utterance nobody commits ends. ``prompt``
+    is words the speech is likely to hold, as for ``transcribe``.
     """
     url = "ws" + api_base.rstrip("/").removeprefix("http") + "/realtime"
     async with connect(url, open_timeout=idle_s) as ws:
-        update = {"type": "audio/pcm", "rate": rate}
+        given: dict[str, Any] = {"format": {"type": "audio/pcm", "rate": rate}}
+        if prompt:
+            given["transcription"] = {"prompt": prompt}
         await ws.send(
             json.dumps(
                 {
                     "type": "session.update",
-                    "session": {
-                        "type": "transcription",
-                        "audio": {"input": {"format": update}},
-                    },
+                    "session": {"type": "transcription", "audio": {"input": given}},
                 }
             )
         )
diff --git a/backend/fluksio/flow/nodes/assistant.py b/backend/fluksio/flow/nodes/assistant.py
index b89a6d7..c6af800 100644
--- a/backend/fluksio/flow/nodes/assistant.py
+++ b/backend/fluksio/flow/nodes/assistant.py
@@ -9,7 +9,7 @@ offered. Every other input is read as context ("outside_temperature: 7.5").
 An answer goes out on ``reply`` a sentence at a time while the model writes
 it, so a ``tts`` node speaks the first before the last is written. A change is
 confirmed from the values set ("Kitchen light off."), without asking the model
-a second time.
+a second time. Talk the panel overheard, such as a TV, gets no answer.
 """
 
 from __future__ import annotations
@@ -47,6 +47,10 @@ LISTENING = "listening"
 UNAVAILABLE = "The assistant isn't available right now."
 #: Said when the model ends a turn without a word or a change.
 CANT = "Sorry, I can't do that."
+#: The model's whole answer to talk not meant for the panel, which is not said.
+#: Asked to answer with nothing instead, Gemma writes tokens nobody sees until
+#: the length limit.
+IGNORED = "Ignored."
 
 #: A server that is not there is given up on after a few seconds, while a
 #: model that has to be loaded first takes about ten.
@@ -60,14 +64,18 @@ _SCHEMAS: dict[DType, dict[str, str]] = {
     DType.STR: {"type": "string"},
 }
 
-_SYSTEM = """You are the voice of a home: someone spoke to a panel on the wall, \
+_SYSTEM = f"""You are the voice of a home: someone spoke to a panel on the wall, \
 and you answer out loud.
 
-- To change something, call its tool without a word: the change is confirmed \
-for you. The tools are all you can change.
-- If no tool can do what was asked, or you did not understand it, say so in a \
-few words, such as "Sorry, I can't do that."
-- Answer questions from the readings below.
+- To change something in the home, call its tool without a word: the change \
+is confirmed for you. The tools are all you can change.
+- If a change was asked for that no tool can make, say so in a few words, such \
+as "Sorry, I can't do that."
+- Answer questions about the home from the readings below, and other questions \
+from what you know.
+- The panel also hears talk that is not meant for it, such as a TV or people \
+talking to each other. If the words are neither a request nor a question for \
+you, answer only "{IGNORED}"
 - Speak plainly: one or two short sentences, no lists, no markdown, no emoji.
 """
 
@@ -308,7 +316,8 @@ class AssistantNode(Node):
         async def say(text: str) -> None:
             nonlocal said
             said = True
-            await asyncio.to_thread(self._say, text)
+            if text.rstrip(".") != IGNORED.rstrip("."):
+                await asyncio.to_thread(self._say, text)
 
         if config is None:
             await say(UNAVAILABLE)
diff --git a/backend/fluksio/flow/nodes/speech.py b/backend/fluksio/flow/nodes/speech.py
index b8c1f8f..02ed592 100644
--- a/backend/fluksio/flow/nodes/speech.py
+++ b/backend/fluksio/flow/nodes/speech.py
@@ -186,7 +186,18 @@ class SttNode(_Speech):
     alone.
     """
 
-    __slots__ = ("_heard", "_listening", "_lock", "_session", "_tasks")
+    class Params(BaseModel):
+        model_config = ConfigDict(extra="allow")
+
+        prompt: str = Field(
+            default="",
+            description=(
+                "Words the speech is likely to hold, such as the names of rooms "
+                "and devices, given to the server as its transcription prompt."
+            ),
+        )
+
+    __slots__ = ("cfg", "_heard", "_listening", "_lock", "_session", "_tasks")
 
     def __init__(
         self,
@@ -195,6 +206,7 @@ class SttNode(_Speech):
         params: dict[str, Any] | None = None,
         name: str | None = None,
     ):
+        self.cfg = self.Params.model_validate(params or {})
         self._lock = threading.Lock()
         #: The chunks of the utterance being spoken, None between utterances.
         self._session: asyncio.Queue[bytes | None] | None = None
@@ -293,7 +305,11 @@ class SttNode(_Speech):
         try:
             async with asyncio.timeout(UTTERANCE_S):
                 async for final, text in client.transcribe_stream(
-                    url + "/v1", chunks, rate=RATE, idle_s=IDLE_S
+                    url + "/v1",
+                    chunks,
+                    rate=RATE,
+                    idle_s=IDLE_S,
+                    prompt=self.cfg.prompt,
                 ):
                     words = text.strip() if final else words + text
                     if final and words:
@@ -317,7 +333,11 @@ class SttNode(_Speech):
     async def _clip(self, url: str, audio: dict[str, Any], data: bytes) -> None:
         try:
             text = await client.transcribe(
-                url + "/v1", data, "speech.wav", str(audio.get("media_type"))
+                url + "/v1",
+                data,
+                "speech.wav",
+                str(audio.get("media_type")),
+                prompt=self.cfg.prompt,
             )
         except Exception as exc:
             logger.warning("'%s' lost a clip: %s: %s", self.id, type(exc).__name__, exc)
diff --git a/backend/tests/flow/test_assistant.py b/backend/tests/flow/test_assistant.py
index 8240d26..7e17479 100644
--- a/backend/tests/flow/test_assistant.py
+++ b/backend/tests/flow/test_assistant.py
@@ -11,7 +11,7 @@ import pytest
 from fluksio.ai import client
 from fluksio.ai import config as ai_config
 from fluksio.flow.messages import DType, MessageSpec
-from fluksio.flow.nodes.assistant import UNAVAILABLE, AssistantNode, cut
+from fluksio.flow.nodes.assistant import IGNORED, UNAVAILABLE, AssistantNode, cut
 from fluksio.flow.pipeline import Pipeline
 
 PROVIDER = ai_config.AIConfig(api_base="http://llm:11434/v1", api_key="k", model="m")
@@ -140,6 +140,22 @@ def test_no_provider_or_one_that_does_not_answer_is_said(
         assert node._health == "degraded"
 
 
+def test_overheard_talk_gets_no_answer(monkeypatch: pytest.MonkeyPatch):
+    said: list[dict[str, Any]] = []
+
+    async def completion(**kwargs: Any) -> AsyncIterator[str | dict[str, Any]]:
+        yield IGNORED
+        yield {"role": "assistant", "content": IGNORED}
+
+    monkeypatch.setattr(ai_config, "load", lambda: PROVIDER)
+    monkeypatch.setattr(client, "stream_completion", completion)
+    monkeypatch.setattr(AssistantNode, "emit", lambda self, values: said.append(values))
+
+    _ask(_node(), "I don't know what to get, Francis.")
+
+    assert said == []
+
+
 def test_a_target_with_values_takes_only_those_and_is_confirmed_in_words(
     monkeypatch: pytest.MonkeyPatch,
 ):
diff --git a/backend/tests/flow/test_speech.py b/backend/tests/flow/test_speech.py
index 3c9c25c..c327191 100644
--- a/backend/tests/flow/test_speech.py
+++ b/backend/tests/flow/test_speech.py
@@ -52,8 +52,8 @@ def test_units_and_clock_times_are_said_as_words():
 def test_an_utterance_is_one_session_committed_when_listening_ends(monkeypatch):
     heard: list[bytes] = []
 
-    async def session(api_base, audio, *, rate, idle_s):
-        assert (api_base, rate) == (f"{URL}/v1", 16000)
+    async def session(api_base, audio, *, rate, idle_s, prompt):
+        assert (api_base, rate, prompt) == (f"{URL}/v1", 16000, "Driveway.")
         while (chunk := await audio.get()) is not None:
             heard.append(chunk)
         yield False, "turn on"
@@ -79,6 +79,7 @@ def test_an_utterance_is_one_session_committed_when_listening_ends(monkeypatch):
                 MessageSpec(name="partial", dtype=DType.STR),
                 MessageSpec(name="text", dtype=DType.STR),
             ],
+            params={"prompt": "Driveway."},
         )
         node.assign_flow("voice", "stt")
         pipeline = Pipeline([node])
diff --git a/docs/reference/node-types.md b/docs/reference/node-types.md
index 5716f29..1580509 100644
--- a/docs/reference/node-types.md
+++ b/docs/reference/node-types.md
@@ -509,10 +509,16 @@ living room setpoint to 22."), so the model is asked once per request. A
 target with `values` is offered only those, each explained to the model, and
 confirmed in its word ("Canopy open."). Any other value is refused.
 
-Every request ends in a reply:
+Questions about the home are answered from the context inputs, other
+questions from what the model knows.
 
-- When no tool fits or the request makes no sense, the model says so, as
+Every request ends in a reply, except talk the panel overheard:
+
+- When no tool can make the change asked for, the model says so, as
   "Sorry, I can't do that.".
+- Words that are neither a request nor a question, such as a TV or people
+  talking to each other, get no reply. The model answers them "Ignored.",
+  which is not said.
 - When no provider is set up, the connection is refused, it takes over 3
   seconds to connect or 15 seconds to answer, the reply is "The assistant
   isn't available right now." and the node reads degraded. A model that has to
@@ -543,6 +549,10 @@ the assistant's prompt, both `reply` outputs go to `tts`, and the device's
 | `partial` | `str` | the words agreed on so far, as they come, then the final text |
 | `text` | `str` | what was said, once per utterance |
 
+| Setting | Default | Notes |
+|---|---|---|
+| `prompt` | empty | words the speech is likely to hold, such as `Living room, kitchen, driveway, appliances.`, sent as the transcription prompt |
+
 The server is any that speaks OpenAI's audio API, such as `parrot serve`. Each
 utterance is one realtime transcription session (`{url}/v1/realtime`), opened
 when `listening` turns true or at the first chunk and committed when it turns
@@ -552,6 +562,11 @@ chunk waits for it. An utterance that comes back empty, fails, or gets no
 chunk for 10 seconds publishes "Didn't catch that." on `partial` alone, so a
 caption shows it and nothing speaks it. A lost one is logged.
 
+A `prompt` naming the installation's rooms and devices helps Whisper with
+words it rarely hears. On the recordings of one wall panel, small.en with such
+a prompt heard 30 of 42 hard commands right instead of 8, as "driveway" for
+"drive bay" and "bathroom" for "best room".
+
 `listening` has to hold a value before a chunk wakes the node. Declare it as a
 flow input with the initial value `false` when a device may never send it,
 such as one that sends whole clips.

From 40e60b422bed2c99019050d0290a72d974c29919 Mon Sep 17 00:00:00 2001
From: stroblme <stroblme@posteo.de>
Date: Fri, 6 Nov 2026 14:33:25 +0100
Subject: [PATCH 8/9] A querying chart offers its own windows: an optional
 ranges list of {label, range_s, interval_s}, up to 8 and checked by the
 server, where range_s 0 asks for the whole history; the editor lists and
 picks them, and the refresh is capped at what a timer can hold

---
 backend/fluksio/flow/dashboards.py            |  33 ++++-
 backend/tests/flow/test_dashboards.py         |  21 +++
 docs/interface/dashboards.md                  |  15 +++
 frontend/src/client/schemas.gen.ts            |   7 +-
 frontend/src/client/types.gen.ts              |   7 +-
 .../src/components/Dashboard/ChartWidget.tsx  |  98 ++++++++++----
 frontend/src/components/Dashboard/panels.tsx  | 120 ++++++++++++++++--
 7 files changed, 259 insertions(+), 42 deletions(-)

diff --git a/backend/fluksio/flow/dashboards.py b/backend/fluksio/flow/dashboards.py
index 6fd370d..85510d2 100644
--- a/backend/fluksio/flow/dashboards.py
+++ b/backend/fluksio/flow/dashboards.py
@@ -64,6 +64,10 @@ BAR_ROWS = 8
 #: client (``ui/core/counter.ts``, ``MAX_COUNTERS``).
 COUNTERS = 3
 
+#: How many windows one querying chart offers, stacked down the tile's edge.
+#: Mirrored in the client (``ChartWidget.tsx``, ``MAX_RANGES``).
+CHART_RANGES = 8
+
 #: Resolved out here on purpose: the store has a ``list`` method, which
 #: shadows the builtin for any annotation written inside the class.
 Bindings = list[dict[str, Any]]
@@ -259,10 +263,15 @@ class WidgetDef(BaseModel):
     A chart comes in two kinds. The default reads what the engine kept for a
     message. One with ``source: "query"`` asks instead, and its config is
     ``{source, request, request_dtype: "record", message, dtype: "series",
-    refresh_s, range_s}``: it publishes ``{range_s, interval_s}`` to
+    refresh_s, range_s, ranges}``: it publishes ``{range_s, interval_s}`` to
     ``request`` exactly as a slider publishes a value, and draws the ``series``
     a flow answers with on ``message``.
 
+    ``ranges`` lists the windows its rail offers, up to ``CHART_RANGES`` of
+    ``{label, range_s, interval_s}``, and ``range_s`` names the one it opens on.
+    Without them it offers 1h, 6h, 24h and 7d. ``range_s: 0`` asks for the
+    whole history, and the flow answering decides how far back that reaches.
+
     A colour widget picks what it publishes with ``format``, because fixtures
     differ and a change node per tile is not the answer:
 
@@ -448,6 +457,28 @@ class WidgetDef(BaseModel):
                     raise ValueError(
                         f"a querying chart's {key} must be '{want}', not '{bound}'"
                     )
+            ranges = self.config.get("ranges") or []
+            if not isinstance(ranges, list) or len(ranges) > CHART_RANGES:
+                raise ValueError(
+                    f"a querying chart offers a list of at most {CHART_RANGES} ranges"
+                )
+            for entry in ranges:
+                window = entry if isinstance(entry, dict) else {}
+                label = window.get("label")
+                span, step = window.get("range_s"), window.get("interval_s")
+                # `type() is int`, because a bool is an int to isinstance.
+                if not (
+                    isinstance(label, str)
+                    and label.strip()
+                    and type(span) is int
+                    and span >= 0
+                    and type(step) is int
+                    and step > 0
+                ):
+                    raise ValueError(
+                        "a querying chart's range needs a label, a whole range_s "
+                        f"of 0 or more and a whole interval_s above 0, not {entry!r}"
+                    )
             return self
 
         if self.type == "bar" and len(self.bar_rows) > BAR_ROWS:
diff --git a/backend/tests/flow/test_dashboards.py b/backend/tests/flow/test_dashboards.py
index 9495276..afe6ca1 100644
--- a/backend/tests/flow/test_dashboards.py
+++ b/backend/tests/flow/test_dashboards.py
@@ -5,6 +5,7 @@ from datetime import UTC, datetime
 import pytest
 
 from fluksio.flow.dashboards import (
+    CHART_RANGES,
     DashboardDef,
     DashboardNotFound,
     DashboardStore,
@@ -354,6 +355,26 @@ def test_a_querying_chart_asks_with_a_record_and_draws_a_series():
         WidgetDef(id="c", type="chart", config=query_chart(request_dtype="json"))
 
 
+def test_a_querying_chart_offers_its_own_ranges():
+    """A few labelled windows, where ``range_s: 0`` is the whole history."""
+    daily = [
+        {"label": "1M", "range_s": 30 * 86400, "interval_s": 86400},
+        {"label": "All", "range_s": 0, "interval_s": 86400},
+    ]
+    WidgetDef(id="c", type="chart", config=query_chart(ranges=daily))
+
+    for bad in (
+        [{"label": "", "range_s": 3600, "interval_s": 60}],
+        [{"label": "1h", "range_s": -1, "interval_s": 60}],
+        [{"label": "1h", "range_s": 3600, "interval_s": 0}],
+        [{"label": "1h", "range_s": 3600.5, "interval_s": 60}],
+        daily * CHART_RANGES,
+        "1M",
+    ):
+        with pytest.raises(ValueError, match="range"):
+            WidgetDef(id="c", type="chart", config=query_chart(ranges=bad))
+
+
 def test_a_colour_widget_publishes_the_shape_its_format_names():
     """The format picks the payload, so the binding is held to that one."""
     wheel = WidgetDef(
diff --git a/docs/interface/dashboards.md b/docs/interface/dashboards.md
index 1b00597..f9c7aba 100644
--- a/docs/interface/dashboards.md
+++ b/docs/interface/dashboards.md
@@ -188,6 +188,21 @@ Choose the chart's data source:
 
 Multi-line charts show a legend as well as different colours.
 
+A Query chart offers 1h, 6h, 24h and 7d down its right-hand edge, and opens on
+24h. Under **Ranges** you can give it windows of its own instead, up to eight,
+and it opens on the one its `range_s` names or else the first. Each window
+becomes the request's `range_s` and `interval_s` when picked. A `range_s` of
+`0` asks for the whole history, and the answering flow decides how far back
+that reaches.
+
+```json
+"ranges": [
+  {"label": "1M", "range_s": 2592000, "interval_s": 86400},
+  {"label": "1Y", "range_s": 31536000, "interval_s": 86400},
+  {"label": "All", "range_s": 0, "interval_s": 86400}
+]
+```
+
 !!! tip "Skip the unfinished bucket"
 
     If your query aggregates into time buckets, leave out the bucket still
diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts
index cadd163..8ee7b12 100644
--- a/frontend/src/client/schemas.gen.ts
+++ b/frontend/src/client/schemas.gen.ts
@@ -5713,10 +5713,15 @@ by a schema per class, because the whole set is small and closed.
 A chart comes in two kinds. The default reads what the engine kept for a
 message. One with \`\`source: "query"\`\` asks instead, and its config is
 \`\`{source, request, request_dtype: "record", message, dtype: "series",
-refresh_s, range_s}\`\`: it publishes \`\`{range_s, interval_s}\`\` to
+refresh_s, range_s, ranges}\`\`: it publishes \`\`{range_s, interval_s}\`\` to
 \`\`request\`\` exactly as a slider publishes a value, and draws the \`\`series\`\`
 a flow answers with on \`\`message\`\`.
 
+\`\`ranges\`\` lists the windows its rail offers, up to \`\`CHART_RANGES\`\` of
+\`\`{label, range_s, interval_s}\`\`, and \`\`range_s\`\` names the one it opens on.
+Without them it offers 1h, 6h, 24h and 7d. \`\`range_s: 0\`\` asks for the
+whole history, and the flow answering decides how far back that reaches.
+
 A colour widget picks what it publishes with \`\`format\`\`, because fixtures
 differ and a change node per tile is not the answer:
 
diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts
index afcfa0b..0f17b31 100644
--- a/frontend/src/client/types.gen.ts
+++ b/frontend/src/client/types.gen.ts
@@ -1775,10 +1775,15 @@ export type WebPushKey = {
  * A chart comes in two kinds. The default reads what the engine kept for a
  * message. One with ``source: "query"`` asks instead, and its config is
  * ``{source, request, request_dtype: "record", message, dtype: "series",
- * refresh_s, range_s}``: it publishes ``{range_s, interval_s}`` to
+ * refresh_s, range_s, ranges}``: it publishes ``{range_s, interval_s}`` to
  * ``request`` exactly as a slider publishes a value, and draws the ``series``
  * a flow answers with on ``message``.
  *
+ * ``ranges`` lists the windows its rail offers, up to ``CHART_RANGES`` of
+ * ``{label, range_s, interval_s}``, and ``range_s`` names the one it opens on.
+ * Without them it offers 1h, 6h, 24h and 7d. ``range_s: 0`` asks for the
+ * whole history, and the flow answering decides how far back that reaches.
+ *
  * A colour widget picks what it publishes with ``format``, because fixtures
  * differ and a change node per tile is not the answer:
  *
diff --git a/frontend/src/components/Dashboard/ChartWidget.tsx b/frontend/src/components/Dashboard/ChartWidget.tsx
index a8ee6ca..b8fb6e6 100644
--- a/frontend/src/components/Dashboard/ChartWidget.tsx
+++ b/frontend/src/components/Dashboard/ChartWidget.tsx
@@ -2,11 +2,7 @@ import { useQueries, useQuery } from "@tanstack/react-query"
 import { useEffect, useRef, useState } from "react"
 
 import type { HistoryPoint } from "@/client"
-import {
-  DEFAULT_RANGE,
-  RANGES,
-  type Range,
-} from "@/components/Common/RangePicker"
+import { DEFAULT_RANGE, RANGES } from "@/components/Common/RangePicker"
 import { MAX_SERIES, UplotChart } from "@/components/Common/UplotChart"
 import {
   type LiveValue,
@@ -40,9 +36,59 @@ const DEFAULT_POINTS = 300
  * quarter-hour buckets asks four times an hour instead of twice a minute, for
  * the same picture. A slower refresh than that is honoured; a faster one only
  * buys the same answer again, so it is clamped.
+ *
+ * Capped at what a timer can hold, about 24 days: `setInterval` keeps its
+ * delay in 32 bits of milliseconds, and a longer one wraps round and fires at
+ * once, which would ask again as fast as the flow answers.
  */
-export const refreshFor = (range: Range, configured: unknown) =>
-  Math.max(range.bucketS, Number(configured) || 0)
+export const refreshFor = (range: ChartRange, configured: unknown) =>
+  Math.min(Math.max(range.interval_s, Number(configured) || 0), 2_147_483)
+
+/**
+ * One window a querying chart offers, in the shape its request takes.
+ *
+ * `range_s: 0` asks for the whole history, and the flow answering decides how
+ * far back that reaches.
+ */
+export type ChartRange = { label: string; range_s: number; interval_s: number }
+
+/** How many windows one chart offers, stacked down the tile's edge. Mirrored
+ *  server-side as `CHART_RANGES` (`app/flow/dashboards.py`). */
+export const MAX_RANGES = 8
+
+/** What a chart that names no windows of its own offers. */
+const PRESETS: ChartRange[] = RANGES.map((range) => ({
+  label: range.label,
+  range_s: range.hours * 3600,
+  interval_s: range.bucketS,
+}))
+
+/**
+ * The windows a querying chart offers: its own `ranges`, else the presets.
+ *
+ * A row still being typed into in the editor is not a window yet. Drawn, it
+ * would ask for nothing and pace its timer at zero.
+ */
+export function rangesOf(cfg: Record<string, unknown>): ChartRange[] {
+  const own = ((cfg.ranges ?? []) as Partial<ChartRange>[]).filter(
+    (range): range is ChartRange =>
+      Boolean(range.label) &&
+      (range.range_s ?? -1) >= 0 &&
+      (range.interval_s ?? 0) > 0,
+  )
+  return own.length ? own : PRESETS
+}
+
+/** Which of its windows a chart opens on: the one `range_s` names, else its
+ *  first, or the presets' day. */
+export function startOf(cfg: Record<string, unknown>): number {
+  const ranges = rangesOf(cfg)
+  const named = ranges.findIndex(
+    (range) => range.range_s === Number(cfg.range_s),
+  )
+  if (named >= 0) return named
+  return ranges === PRESETS ? RANGES.indexOf(DEFAULT_RANGE) : 0
+}
 
 /** How long an unanswered request blocks an identical one. */
 const INFLIGHT_MS = 15_000
@@ -353,13 +399,14 @@ function QueryChart({ widget, dashboard }: WidgetProps) {
   const cfg = config(widget)
   const request = String(cfg.request ?? "")
   const message = String(cfg.message ?? "")
-  const [range, setRange] = useState<Range>(
-    () =>
-      RANGES.find((r) => r.hours * 3600 === Number(cfg.range_s)) ??
-      DEFAULT_RANGE,
-  )
-  const rangeS = range.hours * 3600
-  const intervalS = range.bucketS
+  const ranges = rangesOf(cfg)
+  // By position, so two windows alike are still two buttons. One the editor
+  // has since removed falls back to the first.
+  const [picked, setPicked] = useState(() => startOf(cfg))
+  const shown = picked < ranges.length ? picked : 0
+  const range = ranges[shown]
+  const rangeS = range.range_s
+  const intervalS = range.interval_s
   // Follows the window: a wider one is drawn coarser, so it is worth asking
   // about less often.
   const refreshS = refreshFor(range, cfg.refresh_s)
@@ -433,18 +480,20 @@ function QueryChart({ widget, dashboard }: WidgetProps) {
         palette={palette}
         {...presentation(cfg)}
       />
-      <RangeRail range={range} onChange={setRange} />
+      <RangeRail ranges={ranges} picked={shown} onChange={setPicked} />
     </div>
   )
 }
 
 /** The window a chart is showing, as a column beside it. */
 function RangeRail({
-  range,
+  ranges,
+  picked,
   onChange,
 }: {
-  range: Range
-  onChange: (range: Range) => void
+  ranges: ChartRange[]
+  picked: number
+  onChange: (index: number) => void
 }) {
   const { Segmented } = useUi()
   return (
@@ -452,16 +501,11 @@ function RangeRail({
       orientation="vertical"
       testId="range-picker"
       label="Time range"
-      value={String(range.hours)}
-      options={RANGES.map(
-        (entry) => [String(entry.hours), entry.label] as const,
+      value={String(picked)}
+      options={ranges.map(
+        (range, index) => [String(index), range.label] as const,
       )}
-      onChange={(hours) =>
-        onChange(
-          RANGES.find((entry) => String(entry.hours) === hours) ??
-            DEFAULT_RANGE,
-        )
-      }
+      onChange={(index) => onChange(Number(index))}
     />
   )
 }
diff --git a/frontend/src/components/Dashboard/panels.tsx b/frontend/src/components/Dashboard/panels.tsx
index b93b451..d47b04a 100644
--- a/frontend/src/components/Dashboard/panels.tsx
+++ b/frontend/src/components/Dashboard/panels.tsx
@@ -3,7 +3,6 @@ import { Ban, Check, ChevronDown, Plus, RotateCw, X } from "lucide-react"
 import { useState } from "react"
 
 import type { MessageInfo, SettingDef, WidgetDef } from "@/client"
-import { DEFAULT_RANGE, RANGES } from "@/components/Common/RangePicker"
 import { screensOf } from "@/components/Devices/ports"
 import { devicesQueryOptions } from "@/components/Devices/queries"
 import { RenameDialog } from "@/components/Flow/RenameDialog"
@@ -52,7 +51,14 @@ import {
 import { Switch } from "@/components/ui/switch"
 import useAuth from "@/hooks/useAuth"
 import { cn } from "@/lib/utils"
-import { MAX_SERIES, refreshFor } from "./ChartWidget"
+import {
+  type ChartRange,
+  MAX_RANGES,
+  MAX_SERIES,
+  rangesOf,
+  refreshFor,
+  startOf,
+} from "./ChartWidget"
 import { COLOR_DTYPES, COLOR_FORMATS, colorFormatOf } from "./ColorWidget"
 import {
   CANVAS_PRESETS,
@@ -569,13 +575,20 @@ export function WidgetPanel({
     })
   const querying = cfg.source === "query"
   const pinnedRuns = cfg.source === "runs"
+  // A querying chart's windows, read raw like a grid's presses: `rangesOf`
+  // drops a row until it is complete, and the rail offers what it keeps.
+  const ranges = (cfg.ranges ?? []) as Partial<ChartRange>[]
+  const setRange = (index: number, changes: Partial<ChartRange>) =>
+    set({
+      ranges: ranges.map((other, at) =>
+        at === index ? { ...other, ...changes } : other,
+      ),
+    })
+  const offered = rangesOf(cfg)
+  const start = startOf(cfg)
   // What this chart would refresh at with nothing configured. The viewer can
   // pick another window on the widget, which moves it.
-  const paced = refreshFor(
-    RANGES.find((range) => range.hours * 3600 === Number(cfg.range_s)) ??
-      DEFAULT_RANGE,
-    0,
-  )
+  const paced = refreshFor(offered[start], 0)
 
   return (
     <SidePanel
@@ -986,22 +999,105 @@ export function WidgetPanel({
                 }
               />
             </div>
+            <div className="grid gap-2">
+              <FieldLabel help="Each a label, how far back in seconds (0 for the whole history) and the bucket width in seconds. Without any, the widget offers 1h to 7d.">
+                Ranges
+              </FieldLabel>
+              {ranges.map((range, index) => (
+                <div
+                  // Position is the only identity a range row has.
+                  key={`range-${index}`}
+                  className="flex items-end gap-1.5"
+                >
+                  <Input
+                    className="min-w-0 flex-1"
+                    value={range.label ?? ""}
+                    placeholder="Label"
+                    aria-label="Range label"
+                    onChange={(event) =>
+                      setRange(index, { label: event.target.value })
+                    }
+                  />
+                  <Input
+                    type="number"
+                    className="w-24"
+                    min={0}
+                    placeholder="Seconds"
+                    aria-label="Range, seconds"
+                    value={str(range.range_s)}
+                    onChange={(event) =>
+                      setRange(index, {
+                        range_s: numberOrNone(event.target.value),
+                      })
+                    }
+                  />
+                  <Input
+                    type="number"
+                    className="w-24"
+                    min={1}
+                    placeholder="Bucket"
+                    aria-label="Bucket, seconds"
+                    value={str(range.interval_s)}
+                    onChange={(event) =>
+                      setRange(index, {
+                        interval_s: numberOrNone(event.target.value),
+                      })
+                    }
+                  />
+                  <Button
+                    variant="ghost"
+                    size="icon-sm"
+                    className="text-muted-foreground"
+                    aria-label="Remove range"
+                    onClick={() =>
+                      set({ ranges: ranges.filter((_, at) => at !== index) })
+                    }
+                  >
+                    <X />
+                  </Button>
+                </div>
+              ))}
+              {ranges.length < MAX_RANGES ? (
+                <Button
+                  variant="outline"
+                  size="sm"
+                  className="h-8 justify-self-start"
+                  // A copy of the last, so the document stays one the server
+                  // accepts while the copy is edited.
+                  onClick={() =>
+                    set({
+                      ranges: [
+                        ...ranges,
+                        { ...(ranges[ranges.length - 1] ?? offered[start]) },
+                      ],
+                    })
+                  }
+                  data-testid="add-range"
+                >
+                  <Plus />
+                  Add range
+                </Button>
+              ) : null}
+            </div>
             <div className="grid gap-1.5">
               <FieldLabel help="A viewer can pick another on the widget itself.">
                 Range shown first
               </FieldLabel>
               <Select
-                value={str(cfg.range_s ?? DEFAULT_RANGE.hours * 3600)}
-                onValueChange={(value) => set({ range_s: Number(value) })}
+                value={String(start)}
+                onValueChange={(value) =>
+                  set({ range_s: offered[Number(value)].range_s })
+                }
               >
                 <SelectTrigger>
                   <SelectValue />
                 </SelectTrigger>
                 <SelectContent>
-                  {RANGES.map((range) => (
+                  {offered.map((range, index) => (
                     <SelectItem
-                      key={range.label}
-                      value={String(range.hours * 3600)}
+                      // Position, as on the widget: two alike are still two.
+                      key={`offered-${index}`}
+                      value={String(index)}
                     >
                       {range.label}
                     </SelectItem>

From 5eec267a7ed6f6bda9d2507d292f5792d2b6e400 Mon Sep 17 00:00:00 2001
From: stroblme <stroblme@posteo.de>
Date: Fri, 6 Nov 2026 14:35:12 +0100
Subject: [PATCH 9/9] A code-declared node's shim names only its triggering
 ports, so a non-triggering input with no value yet reaches the function as
 its own default instead of failing the call

---
 backend/fluksio/sdk/__init__.py |  6 +++++-
 backend/tests/sdk/test_build.py | 22 ++++++++++++++++++++++
 2 files changed, 27 insertions(+), 1 deletion(-)

diff --git a/backend/fluksio/sdk/__init__.py b/backend/fluksio/sdk/__init__.py
index c41fb5e..d841631 100644
--- a/backend/fluksio/sdk/__init__.py
+++ b/backend/fluksio/sdk/__init__.py
@@ -877,12 +877,16 @@ def _shim(spec: NodeSpec) -> str:
     The import is aliased because ``process``'s parameters are the node's port
     names, and a port may be named after the very function it feeds — which
     would otherwise shadow the import inside the body.
+
+    Only the triggering ports are named. The engine leaves out an input that
+    has no value yet, which only a port that does not trigger can lack, so
+    those travel in ``**settings`` and the function's own default applies.
     """
     _refuse_main(spec)
     fn = spec.fn
     where = inspect.getsourcefile(fn) or fn.__module__
     repo = import_root(fn)
-    ports = [port.port for port in spec.requires]
+    ports = [port.port for port in spec.requires if port.trigger]
     signature = ", ".join(ports + ["**settings"])
     arguments = ", ".join([f"{port}={port}" for port in ports] + ["**settings"])
     call = f"_impl({arguments})"
diff --git a/backend/tests/sdk/test_build.py b/backend/tests/sdk/test_build.py
index 380540d..4bb4b6c 100644
--- a/backend/tests/sdk/test_build.py
+++ b/backend/tests/sdk/test_build.py
@@ -240,6 +240,28 @@ def test_a_port_named_like_its_function_still_calls_the_function():
     assert namespace["process"](scale=2.0) == {"scaled": 4.0}
 
 
+@node(
+    requires=["value", Port("factor", "float", trigger=False)],
+    provides=Port("scaled", "float"),
+)
+def scale_by(value, factor=3.0):
+    return value * factor
+
+
+def test_a_port_that_does_not_trigger_may_be_missing_from_the_call():
+    """The engine leaves out an input with no value yet, so the default applies."""
+    flow = Flow(
+        "optional",
+        nodes=[scale_by],
+        inputs=[Port("value", "float", initial=1.0), Port("factor", "float")],
+    )
+    namespace: dict = {}
+    exec(compile(flow.shims()["scale_by"], "<shim>", "exec"), namespace)
+
+    assert namespace["process"](value=2.0) == {"scaled": 6.0}
+    assert namespace["process"](value=2.0, factor=0.5) == {"scaled": 1.0}
+
+
 def reads_settings(settings):
     return settings