fix(routing): escape hatch out of mix mode — clear-all overrides + pins warning (#663)

tests/integration/test_board_review_gate_timeline_real_db.py

@@ -110,11 +110,15 @@ async def board_gate_setup(
     # the SAME test database, so the gate's real writes land where we read them.
     saved_engine = db_base._DbHolder.engine
     saved_factory = db_base._DbHolder.session_factory
+    saved_loop = db_base._DbHolder.loop
     handoff_engine = create_async_engine(_test_database_url, future=True)
     db_base._DbHolder.engine = handoff_engine
     db_base._DbHolder.session_factory = async_sessionmaker(
         bind=handoff_engine, class_=AsyncSession, expire_on_commit=False
     )
+    # Clear the loop stamp so the per-loop rebind guard claims the injected
+    # engine for this test's loop instead of discarding it as foreign.
+    db_base._DbHolder.loop = None
 
     # The dispatcher receives tasks from the HTTP API, which serializes
     # assigned_to as the agent UUID; _resolve_agent_slug maps it back to a slug.
@@ -134,6 +138,7 @@ async def board_gate_setup(
         await handoff_engine.dispose()
         db_base._DbHolder.engine = saved_engine
         db_base._DbHolder.session_factory = saved_factory
+        db_base._DbHolder.loop = saved_loop
 
 
 def _make_orch() -> AgentOrchestrator:

tests/integration/test_db_base.py

@@ -8,6 +8,7 @@ and drop/close.
 
 from __future__ import annotations
 
+import asyncio
 import time
 from typing import TYPE_CHECKING, cast
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -41,10 +42,12 @@ def _reset_holder() -> Generator[None]:
     """Snapshot/restore the singleton so tests don't poison the live engine."""
     saved_engine = _DbHolder.engine
     saved_factory = _DbHolder.session_factory
+    saved_loop = _DbHolder.loop
     _InitState.completed_url = None
     yield
     _DbHolder.engine = saved_engine
     _DbHolder.session_factory = saved_factory
+    _DbHolder.loop = saved_loop
     _InitState.completed_url = None
 
 
@@ -65,6 +68,44 @@ def test_get_engine_creates_and_caches() -> None:
     assert ce.call_count == 1
 
 
+def test_get_engine_rebinds_on_a_different_event_loop() -> None:
+    """The cached engine is per-loop: an access from a second loop discards
+    the first loop's engine (whose pooled connections are loop-bound) and
+    builds a fresh one, instead of dying later with 'Future attached to a
+    different loop' — the e2e/eval-bench multi-loop failure mode."""
+    _DbHolder.engine = None
+    _DbHolder.session_factory = None
+    _DbHolder.loop = None
+    engines = [MagicMock(), MagicMock()]
+    with patch("roboco.db.base.create_async_engine", side_effect=engines) as ce:
+
+        async def _grab() -> object:
+            return get_engine()
+
+        loop_a_engine = asyncio.run(_grab())
+        loop_b_engine = asyncio.run(_grab())
+    assert loop_a_engine is engines[0]
+    # Loop B must NOT reuse loop A's engine.
+    assert loop_b_engine is engines[1]
+    assert ce.call_count == len(engines)
+
+
+def test_get_engine_same_loop_keeps_the_cache() -> None:
+    _DbHolder.engine = None
+    _DbHolder.session_factory = None
+    _DbHolder.loop = None
+    fake_engine = MagicMock()
+    with patch("roboco.db.base.create_async_engine", return_value=fake_engine) as ce:
+
+        async def _grab_twice() -> tuple[object, object]:
+            return get_engine(), get_engine()
+
+        e1, e2 = asyncio.run(_grab_twice())
+    assert e1 is fake_engine
+    assert e2 is fake_engine
+    assert ce.call_count == 1
+
+
 def test_get_session_factory_creates_and_caches() -> None:
     _DbHolder.engine = None
     _DbHolder.session_factory = None

tests/integration/test_llm_routing.py

@@ -426,6 +426,26 @@ async def test_apply_mode_mix_requires_per_agent(llm_setup: dict) -> None:
         await svc.apply_mode(mode="mix")
 
 
+@pytest.mark.asyncio
+async def test_apply_mode_mix_empty_map_clears_all_pins(llm_setup: dict) -> None:
+    """An EMPTY per_agent map is the explicit clear-all — the only way out of
+    a fully-pinned fleet, since mode switches deliberately spare pins (the
+    2026-07-23 live incident: 25/25 pins made every mode button a no-op)."""
+    svc = llm_setup["svc"]
+    model = _first_model_for_type(ModelProvider.ANTHROPIC)
+    await svc.apply_mode(mode="mix", per_agent={"be-dev-1": model, "be-qa": model})
+    assert any(
+        r.scope == AssignmentScope.AGENT_SLUG for r in await svc.list_assignments()
+    )
+
+    await svc.apply_mode(mode="mix", per_agent={})
+
+    rows = await svc.list_assignments()
+    assert not any(r.scope == AssignmentScope.AGENT_SLUG for r in rows)
+    # Nothing left at all -> the mode label escapes "mix".
+    assert await svc.derive_mode() == "anthropic"
+
+
 @pytest.mark.asyncio
 async def test_apply_mode_mix_writes_overrides(llm_setup: dict) -> None:
     svc = llm_setup["svc"]