Let the supervisor stop one flow's tasks and clear its quarantine

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

backend/tests/flow/test_supervision.py

@@ -124,3 +124,49 @@ def test_cancel_all_is_idempotent():
         await supervisor.cancel_all()
 
     asyncio.run(scenario())
+
+
+def test_cancelling_one_flow_leaves_another_flows_tasks_running():
+    """Rebuilding one flow must not take every other flow's loops with it."""
+
+    async def loop() -> None:
+        await asyncio.sleep(3600)
+
+    async def scenario() -> None:
+        supervisor = Supervisor()
+        supervisor.spawn("gone.sub", "gone", loop)
+        supervisor.spawn("fine.sub", "fine", loop)
+        await _settle()
+
+        survivor = supervisor._tasks["fine.sub"]
+        await supervisor.cancel_flow("gone")
+
+        assert list(supervisor._tasks) == ["fine.sub"]
+        assert not survivor.done()
+        await supervisor.cancel_all()
+
+    asyncio.run(scenario())
+
+
+def test_cancelling_one_flow_lifts_only_its_quarantine():
+    async def failing() -> None:
+        raise RuntimeError("nope")
+
+    async def scenario() -> None:
+        supervisor = Supervisor()
+        supervisor.spawn("a", "broken", failing)
+        supervisor.spawn("b", "also-broken", failing)
+        await _settle()
+
+        assert supervisor.quarantined == {"broken", "also-broken"}
+
+        await supervisor.cancel_flow("broken")
+
+        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
+        await supervisor.cancel_all()
+
+    asyncio.run(scenario())