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

@@ -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:

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.