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] 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