Merge branch 'main' of git.stroblme.de:Fluksio/app

The two sides both touched `submit_ready`'s readiness check, for unrelated
reasons, so the conflict is textual rather than semantic and both changes
stand:

- `831a537` completes a node that is not ready instead of passing over it,
  so a producer that can never run stops stranding its consumers.
- the audit branch has `_is_node_ready` return the values it read, so the
  node runs on them instead of asking state for the same keys again.

Merged as: read once, keep the values whether or not the answer is yes, and
take the not-ready branch from `831a537`. Its reasoning holds under the
merge — by the time readiness is consulted, `in_degree` is zero and every
in-wave producer has finished, so the answer cannot change later in the
wave.

Also fixes a fixture this branch added: the module-scoped row cleanup in
`tests/conftest.py` assumed a schema, and `tests/flow` overrides `db` with a
no-op because those tests need no database. It only showed when that
directory ran on its own.

746 tests green, and each directory green alone. Engine throughput is
unchanged by the merge (559 msg/s on the memory backend, against 639 before
it and 262 at the start of the audit).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01M6hPWS6YEbT1P8LxhhFb2T

backend/tests/conftest.py

@@ -7,7 +7,7 @@ from typing import Any
 import pytest
 from cryptography.hazmat.primitives.asymmetric import rsa
 from fastapi.testclient import TestClient
-from sqlalchemy import delete
+from sqlalchemy import delete, inspect
 from sqlalchemy.engine import make_url
 from sqlmodel import Session, select
 
@@ -89,6 +89,11 @@ def _clean_engine_rows() -> Generator[None, None, None]:
     between modules bounds it to where the writes actually come from.
     """
     yield
+    # `tests/flow` overrides `db` with a no-op — the engine holds no database
+    # state, so those tests build no schema — and there is nothing to clear
+    # there. Running that directory on its own is how that shows.
+    if not inspect(engine).has_table("run"):
+        return
     with Session(engine) as session:
         for model in _ENGINE_TABLES:
             session.execute(delete(model))

backend/tests/flow/test_pipeline.py

@@ -206,3 +206,150 @@ def test_only_advisory_codes_are_flagged_advisory():
 
     assert hook.advisory is True
     assert cycle.advisory is False
+
+
+def test_a_producer_that_cannot_run_does_not_strand_its_consumer():
+    """One wave, two producers of one consumer, and one of them never ready.
+
+    The house's shape: a battery reading wakes the boiler arbiter and the
+    watering-pump trigger in the same wave, and the DMX encoder reads both. The
+    pump waits on a pulse that only exists at 02:00, so it is never ready — and
+    passing over it used to leave the encoder unreachable, which put every
+    boiler and plug command into its message and no further.
+    """
+    ran = []
+
+    power = make_node(
+        "cerbo", "power", lambda params: {"batt_v": 50.3}, provides=[spec("batt_v")]
+    )
+
+    def arbitrate(batt_v, params):
+        ran.append("boiler")
+        return {"boiler_on": 1.0}
+
+    boiler = make_node(
+        "arbiter",
+        "boiler",
+        arbitrate,
+        requires=[spec("power.batt_v")],
+        provides=[spec("boiler_on")],
+    )
+
+    # Woken by the same wave; `pump_for` has never been published.
+    pump = make_node(
+        "pump_run",
+        "plugs",
+        lambda batt_v, pump_for, params: {"pump": 1.0},
+        requires=[spec("power.batt_v"), spec("plugs.pump_for")],
+        provides=[spec("pump")],
+    )
+
+    def encode(boiler_on, pump, params):
+        ran.append("encoder")
+        return {"dmx_boiler": boiler_on}
+
+    encoder = make_node(
+        "switches",
+        "dmx",
+        encode,
+        requires=[spec("boiler.boiler_on"), spec("plugs.pump")],
+        provides=[spec("dmx_boiler")],
+    )
+
+    pipeline = Pipeline(nodes=[power, boiler, pump, encoder])
+    pipeline.state["plugs.pump"] = 0.0
+
+    power.inject()
+
+    assert ran == ["boiler", "encoder"]
+    assert pipeline.state["dmx.dmx_boiler"] == 1.0
+
+
+def test_a_quiet_producer_does_not_veto_a_noisy_sibling():
+    """Two report-by-exception producers, one consumer, and only one with news.
+
+    The house reads both boilers into one DMX encoder through `rbe` nodes. The
+    kitchen boiler is unchanged almost always, so it returns nothing — and
+    while that held the encoder's dependency count up, the main boiler's
+    command reached its message and got no further.
+    """
+    ran = []
+
+    power = make_node(
+        "cerbo", "power", lambda params: {"batt_v": 50.3}, provides=[spec("batt_v")]
+    )
+    water = make_node(
+        "water_changed",
+        "boiler",
+        lambda batt_v, params: {"water_boiler": 1.0},
+        requires=[spec("power.batt_v")],
+        provides=[spec("water_boiler")],
+    )
+    # Unchanged, so its rbe publishes nothing this wave.
+    kitchen = make_node(
+        "kitchen_changed",
+        "boiler",
+        lambda batt_v, params: None,
+        requires=[spec("power.batt_v")],
+        provides=[spec("kitchen_boiler")],
+    )
+
+    def encode(water_boiler, kitchen_boiler, params):
+        ran.append("switches")
+        return {"dmx_water_boiler": 255.0 if water_boiler else 0.0}
+
+    encoder = make_node(
+        "switches",
+        "dmx",
+        encode,
+        requires=[spec("boiler.water_boiler"), spec("boiler.kitchen_boiler")],
+        provides=[spec("dmx_water_boiler")],
+    )
+
+    pipeline = Pipeline(nodes=[power, water, kitchen, encoder])
+    pipeline.state["boiler.water_boiler"] = 0.0
+    pipeline.state["boiler.kitchen_boiler"] = 0.0
+
+    power.inject()
+
+    assert ran == ["switches"]
+    assert pipeline.state["dmx.dmx_water_boiler"] == 255.0
+
+
+def test_a_consumer_of_only_quiet_producers_still_does_not_run():
+    """The other half of it: freeing a consumer is not the same as running it.
+
+    Nothing it reads was refreshed, so it must stay put — otherwise every wave
+    would re-run the whole graph on values it has already seen.
+    """
+    ran = []
+
+    power = make_node(
+        "cerbo", "power", lambda params: {"batt_v": 50.3}, provides=[spec("batt_v")]
+    )
+    quiet = make_node(
+        "changed",
+        "boiler",
+        lambda batt_v, params: None,
+        requires=[spec("power.batt_v")],
+        provides=[spec("water_boiler")],
+    )
+
+    def encode(water_boiler, params):
+        ran.append("switches")
+        return {"dmx_water_boiler": 255.0}
+
+    encoder = make_node(
+        "switches",
+        "dmx",
+        encode,
+        requires=[spec("boiler.water_boiler")],
+        provides=[spec("dmx_water_boiler")],
+    )
+
+    pipeline = Pipeline(nodes=[power, quiet, encoder])
+    pipeline.state["boiler.water_boiler"] = 0.0
+
+    power.inject()
+
+    assert ran == []