fix: do not kill queen when switching

core/framework/server/routes_queens.py

@@ -25,17 +25,6 @@ from framework.config import QUEENS_DIR
 logger = logging.getLogger(__name__)
 
 
-async def _stop_live_sessions(manager, keep_session_id: str | None = None) -> None:
-    """Stop live sessions so only the selected queen session remains active."""
-    for session in list(manager.list_sessions()):
-        if keep_session_id and session.id == keep_session_id:
-            continue
-        try:
-            await manager.stop_session(session.id)
-        except Exception:
-            logger.debug("Failed to stop session %s during queen switch", session.id)
-
-
 def _read_queen_session_meta(queen_id: str, session_id: str) -> dict[str, Any]:
     """Return persisted metadata for a queen session when available."""
     session_dir = QUEENS_DIR / queen_id / "sessions" / session_id
@@ -229,10 +218,6 @@ async def handle_queen_session(request: web.Request) -> web.Response:
                 }
             )
 
-    # Stop any live sessions bound to a different queen so only one queen
-    # is active at a time.
-    await _stop_live_sessions(manager)
-
     # 2. Find the most recent cold session for this queen and resume it.
     # IMPORTANT: skip sessions that don't belong in the queen DM:
     #   - ``colony_fork: true`` -- duplicates created by handle_colony_spawn
@@ -323,7 +308,6 @@ async def handle_select_queen_session(request: web.Request) -> web.Response:
 
     live_session = manager.get_session(target_session_id)
     if live_session is not None:
-        await _stop_live_sessions(manager, keep_session_id=target_session_id)
         return web.json_response(
             {
                 "session_id": live_session.id,
@@ -332,8 +316,6 @@ async def handle_select_queen_session(request: web.Request) -> web.Response:
             }
         )
 
-    await _stop_live_sessions(manager)
-
     meta = _read_queen_session_meta(queen_id, target_session_id)
     agent_path = meta.get("agent_path")
     initial_phase = None if agent_path else "independent"
@@ -367,7 +349,6 @@ async def handle_new_queen_session(request: web.Request) -> web.Response:
     initial_prompt = body.get("initial_prompt")
     initial_phase = body.get("initial_phase") or "independent"
 
-    await _stop_live_sessions(manager)
     session = await manager.create_session(
         initial_prompt=initial_prompt,
         queen_name=queen_id,

core/framework/server/tests/test_api.py

@@ -638,13 +638,17 @@ class TestQueenSessionSelection:
             )
             assert resp.status == 200
             data = await resp.json()
-
-        assert data == {
-            "session_id": "queen_live",
-            "queen_id": "queen_technology",
-            "status": "live",
-        }
-        assert any(call.args == ("other_live",) for call in manager.stop_session.await_args_list)
+            # Assert inside the async-with so app shutdown (which stops
+            # remaining sessions as cleanup) doesn't pollute the assertions.
+            assert data == {
+                "session_id": "queen_live",
+                "queen_id": "queen_technology",
+                "status": "live",
+            }
+            # Other queen's live session must be left running so multiple
+            # queens can stay active in parallel across navigation.
+            manager.stop_session.assert_not_awaited()
+            assert "other_live" in manager._sessions
 
     @pytest.mark.asyncio
     async def test_select_queen_session_restores_specific_history_session(self, monkeypatch, tmp_path):
@@ -745,18 +749,21 @@ class TestQueenSessionSelection:
             )
             assert resp.status == 200
             data = await resp.json()
-
-        assert data == {
-            "session_id": "fresh_thread",
-            "queen_id": "queen_technology",
-            "status": "created",
-        }
-        manager.stop_session.assert_awaited_once_with("old_live")
-        manager.create_session.assert_awaited_once_with(
-            initial_prompt=None,
-            queen_name="queen_technology",
-            initial_phase="independent",
-        )
+            # Assert inside the async-with so app shutdown (which stops
+            # remaining sessions as cleanup) doesn't pollute the assertions.
+            assert data == {
+                "session_id": "fresh_thread",
+                "queen_id": "queen_technology",
+                "status": "created",
+            }
+            # Other queen's live session must be left running.
+            manager.stop_session.assert_not_awaited()
+            assert "old_live" in manager._sessions
+            manager.create_session.assert_awaited_once_with(
+                initial_prompt=None,
+                queen_name="queen_technology",
+                initial_phase="independent",
+            )
 
 
 class TestExecution: