Let a quarantine expire, and stop two tasks spending one budget

A house's inverter broker dropped at 04:27 and the power flow was quarantined
20 seconds later. Quarantine was terminal — the supervised task returned and
only a publish or an engine restart could bring it back — so five hours of
power and battery readings are missing, and what ended it was an unrelated
`git pull` restarting uvicorn.

Two changes, both in that path:

- the failure budget is per task, not per flow. `power` runs an MQTT subscriber
  and a Victron keepalive publisher against the same broker; they died together
  and spent one shared budget in 41s, giving up before the 60s backoff step was
  ever reached.
- quarantine is now a rest. The task sits out 5min, then 15, then an hour, and
  each time gets its budget back and tries again, so a broker that comes back
  is picked up without anyone watching. `quarantined` reads from whichever
  tasks are currently resting.

The alert for it says when it will try again.

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

backend/tests/flow/test_supervision.py

@@ -13,6 +13,9 @@ from fluksio.flow.supervision import FAILURE_BUDGET, Supervisor
 def no_backoff(monkeypatch: pytest.MonkeyPatch):
     """Real backoff would make these tests a minute long."""
     monkeypatch.setattr(supervision, "BACKOFF", (0.0,))
+    # Long enough that a quarantine holds for the length of an assertion, short
+    # enough that the test asking it to expire does not wait five minutes.
+    monkeypatch.setattr(supervision, "QUARANTINE_BACKOFF", (0.05,))
 
 
 async def _settle() -> None:
@@ -61,7 +64,7 @@ def test_a_loop_that_returns_is_left_alone():
     asyncio.run(scenario())
 
 
-def test_a_flow_that_keeps_crashing_is_quarantined():
+def test_a_flow_that_keeps_crashing_is_stood_down():
     attempts = 0
     events: list[dict] = []
     bus = EventBus()
@@ -81,13 +84,14 @@ def test_a_flow_that_keeps_crashing_is_quarantined():
         assert attempts == FAILURE_BUDGET
         assert supervisor.quarantined == {"heating"}
         assert [e["type"] for e in events].count("task_crashed") == FAILURE_BUDGET
-        assert [e["type"] for e in events][-1] == "flow_quarantined"
+        quarantine = [e for e in events if e["type"] == "flow_quarantined"][-1]
+        assert quarantine["retry_in_s"] == 0.05
         await supervisor.cancel_all()
 
     asyncio.run(scenario())
 
 
-def test_the_budget_is_per_flow():
+def test_one_flows_failure_does_not_stand_another_down():
     ran = 0
 
     async def failing() -> None:
@@ -111,6 +115,63 @@ def test_the_budget_is_per_flow():
     asyncio.run(scenario())
 
 
+def test_two_tasks_on_one_flow_do_not_spend_each_others_budget():
+    """The house's power flow: an MQTT subscriber and its keepalive publisher.
+
+    Both die with the same broker. Sharing one budget meant the flow was stood
+    down after half the restarts either task was allowed on its own.
+    """
+    tries = {"sub": 0, "pub": 0}
+
+    def crasher(which: str):
+        async def loop() -> None:
+            tries[which] += 1
+            raise ConnectionError("broker went away")
+
+        return loop
+
+    async def scenario() -> None:
+        supervisor = Supervisor()
+        supervisor.spawn("power.cerbo:mqtt", "power", crasher("sub"))
+        supervisor.spawn("power.keepalive:mqtt-out", "power", crasher("pub"))
+        await _settle()
+
+        assert tries == {"sub": FAILURE_BUDGET, "pub": FAILURE_BUDGET}
+        await supervisor.cancel_all()
+
+    asyncio.run(scenario())
+
+
+def test_a_quarantine_expires_and_the_task_tries_again():
+    """The whole point: a broker that comes back is picked up without a human."""
+    attempts = 0
+
+    async def loop() -> None:
+        nonlocal attempts
+        attempts += 1
+        if attempts <= FAILURE_BUDGET:
+            raise ConnectionError("broker went away")
+        await asyncio.sleep(3600)
+
+    async def scenario() -> None:
+        supervisor = Supervisor()
+        supervisor.spawn("mqtt", "power", loop)
+        await _settle()
+
+        assert attempts == FAILURE_BUDGET
+        assert supervisor.quarantined == {"power"}
+
+        # Sit out the rest, which the fixture has shortened to 50ms.
+        await asyncio.sleep(0.1)
+        await _settle()
+
+        assert attempts == FAILURE_BUDGET + 1
+        assert supervisor.quarantined == set()
+        await supervisor.cancel_all()
+
+    asyncio.run(scenario())
+
+
 def test_cancel_all_is_idempotent():
     async def loop() -> None:
         await asyncio.sleep(3600)
@@ -165,8 +226,8 @@ def test_cancelling_one_flow_lifts_only_its_quarantine():
         assert supervisor.quarantined == {"also-broken"}
         # And its budget, or the flow would be quarantined again on the first
         # crash after the rebuild that was meant to fix it.
-        assert "broken" not in supervisor._failures
-        assert "also-broken" in supervisor._failures
+        assert "a" not in supervisor._failures
+        assert "b" in supervisor._failures
         await supervisor.cancel_all()
 
     asyncio.run(scenario())