Stop a quiet producer vetoing a noisy one in the same wave

Both boilers on the house had been unable to switch on since the Node-RED
transition, and the reason was here rather than in their logic: the command
reached `boiler.water_boiler` and stopped, because `dmx.switches` never ran.

A wave orders nodes by a dependency count, and two things decremented that
count only on success:

- a node that published nothing — rate limited, unchanged, or failed — never
  freed its consumers. `dmx.switches` reads both boilers through `rbe` nodes,
  so the kitchen one being unchanged, which it is nearly always, held the main
  one's command back. The encoder ran about four times an hour, and only when
  the lights happened to change in the same wave.
- a node that could not run at all never freed them either, permanently.
  `plugs.pump_run` waits on a watering pulse that only exists at 02:00, so
  every wave it appeared in took its consumers out with it.

Freeing a consumer is not the same as running it: `untouched` already refuses
to run anything whose inputs nothing refreshed, and that is the accurate test.
The dependency count is ordering, not permission.

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

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