test(multi_user_bridge): add 10 unit tests for session isolation and routing

- covers UserSession message isolation
- covers SessionManager get_or_create, max_sessions eviction,
  room updates, and list_sessions
- covers BridgeHandler health and sessions endpoints
- uses monkeypatch fixture to stub heavy runtime deps

Refs: #1503

tests/test_multi_user_bridge.py

@@ -0,0 +1,255 @@
+"""Unit tests for multi_user_bridge.py — session isolation and routing.
+
+Refs: #1503
+"""
+
+from __future__ import annotations
+
+import sys
+import time
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+# ── Constants ───────────────────────────────────────────────────────────────
+
+PROJECT_ROOT = Path(__file__).parent.parent
+
+
+# ── Fixture: patch heavy runtime dependencies BEFORE module import ──────────
+
+@pytest.fixture(autouse=True)
+def patch_runtime_deps(monkeypatch, tmp_path):
+    """Prevent module-level side-effects by stubbing external deps before any import."""
+    tmp = tmp_path
+
+    # 1. Stub run_agent module with a fake AIAgent
+    fake_run_agent = MagicMock()
+    fake_agent = MagicMock()
+    fake_agent.model = "test-mock"
+    fake_agent.quiet_mode = True
+    fake_run_agent.AIAgent = MagicMock(return_value=fake_agent)
+    monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)
+
+    # 2. Set environment that multi_user_bridge reads at import time
+    fake_hermes = tmp / "hermes"
+    fake_hermes.mkdir()
+    fake_world = tmp / "world"
+    fake_world.mkdir()
+    (fake_world / "world_state.json").write_text(
+        '{"tick": 0, "rooms": {}, "time_of_day": "dawn"}'
+    )
+    monkeypatch.setenv("HERMES_PATH", str(fake_hermes))
+
+    # 3. Patch WORLD_DIR attribute that module computes at import
+    #    We'll patch it after import via monkeypatch.setattr
+    monkeypatch.setattr("multi_user_bridge.WORLD_DIR", fake_world, raising=False)
+
+    # 4. Prevent world_tick_system.start() from spawning a daemon thread
+    #    Patch at the class level before instantiation
+    from threading import Thread
+    monkeypatch.setattr(Thread, "start", lambda self: None)
+
+    # Ensure multi_user_bridge is cleared so next import is fresh with patches
+    if "multi_user_bridge" in sys.modules:
+        del sys.modules["multi_user_bridge"]
+
+    yield
+
+
+# ── UserSession tests ───────────────────────────────────────────────────────
+
+def test_user_session_creates_with_unique_id_and_room():
+    """UserSession must initialise with provided identifiers."""
+    from multi_user_bridge import UserSession
+
+    session = UserSession(user_id="user-123", username="Alice", room="The Chapel")
+
+    assert session.user_id == "user-123"
+    assert session.username == "Alice"
+    assert session.room == "The Chapel"
+    assert isinstance(session.created_at, str)
+    assert isinstance(session.last_active, float)
+    assert session.messages == []
+    assert session.agent is not None  # mocked agent created
+
+
+def test_user_session_message_history_is_isolated():
+    """Messages stored in one UserSession must not leak to another."""
+    from multi_user_bridge import UserSession
+
+    alice = UserSession(user_id="alice", username="Alice", room="The Threshold")
+    bob = UserSession(user_id="bob", username="Bob", room="The Threshold")
+
+    # Alice sends two messages
+    alice.messages.append({"role": "user", "content": "Hello from Alice"})
+    alice.messages.append({"role": "assistant", "content": "Hi Alice"})
+
+    # Bob sends one message
+    bob.messages.append({"role": "user", "content": "Greetings from Bob"})
+
+    # Isolation assertions
+    assert len(alice.messages) == 2
+    assert len(bob.messages) == 1
+    assert "Alice" in alice.messages[0]["content"]
+    assert "Bob" in bob.messages[0]["content"]
+    # Cross-check — Alice's history must not contain Bob's message
+    assert all("Bob" not in msg["content"] for msg in alice.messages)
+    assert all("Alice" not in msg["content"] for msg in bob.messages)
+
+
+def test_user_session_multiple_instances_are_independent():
+    """Multiple UserSession instances must not share mutable state."""
+    from multi_user_bridge import UserSession
+
+    sessions = [
+        UserSession(user_id=f"user-{i}", username=f"User{i}", room="The Threshold")
+        for i in range(5)
+    ]
+
+    for i, s in enumerate(sessions):
+        s.messages.append({"role": "user", "content": f"msg-{i}"})
+
+    assert [len(s.messages) for s in sessions] == [1, 1, 1, 1, 1]
+    assert all(s.messages[0]["content"] == f"msg-{i}" for i, s in enumerate(sessions))
+
+
+# ── SessionManager tests ────────────────────────────────────────────────────
+
+def test_session_manager_get_or_create_returns_same_instance():
+    """get_or_create must return the same UserSession for identical user_id."""
+    from multi_user_bridge import SessionManager
+
+    sm = SessionManager(max_sessions=10)
+
+    session1 = sm.get_or_create(user_id="alexander", username="Alexander")
+    session2 = sm.get_or_create(user_id="alexander", username="Alexander")
+    session3 = sm.get_or_create(user_id="alexander", username="DifferentName")
+
+    assert session1 is session2  # same object
+    assert session2 is session3  # same object
+    assert sm.get_session_count() == 1
+
+
+def test_session_manager_respects_max_sessions():
+    """When max_sessions is reached, the least-recently-active session must be evicted."""
+    from multi_user_bridge import SessionManager
+
+    sm = SessionManager(max_sessions=3)
+
+    # Create 3 sessions with deliberate timing
+    sm.get_or_create("user1", "User1")
+    time.sleep(0.01)
+    sm.get_or_create("user2", "User2")
+    time.sleep(0.01)
+    sm.get_or_create("user3", "User3")
+    assert sm.get_session_count() == 3
+
+    # Create 4th — should evict user1 (oldest)
+    sm.get_or_create("user4", "User4")
+    assert sm.get_session_count() == 3
+    # Re-adding user1 creates a fresh session (old was evicted)
+    new_user1 = sm.get_or_create("user1", "User1")
+    assert new_user1 is not None
+    sessions = sm.list_sessions()
+    user_ids = {s["user"] for s in sessions}
+    assert "User1" in user_ids  # new one exists (username)
+
+
+def test_session_manager_update_room_on_get_or_create():
+    """Calling get_or_create with a different room must update session.room."""
+    from multi_user_bridge import SessionManager
+
+    sm = SessionManager()
+    session = sm.get_or_create("alice", "Alice", room="The Threshold")
+    assert session.room == "The Threshold"
+
+    # Later, Alice moves to the Chapel
+    session = sm.get_or_create("alice", "Alice", room="The Chapel")
+    assert session.room == "The Chapel"
+
+
+def test_session_manager_list_sessions_returns_active():
+    """list_sessions must return all currently active UserSession summaries."""
+    from multi_user_bridge import SessionManager
+
+    sm = SessionManager()
+    sm.get_or_create("user1", "User1", room="Room A")
+    sm.get_or_create("user2", "User2", room="Room B")
+
+    sessions = sm.list_sessions()
+    user_ids = {s["user"] for s in sessions}
+    assert user_ids == {"User1", "User2"}
+
+
+# ── BridgeHandler request routing tests ────────────────────────────────────
+
+def test_bridge_handler_health_endpoint_returns_status():
+    """GET /bridge/health must return JSON with active_sessions count."""
+    from multi_user_bridge import BridgeHandler
+
+    class MockRequest:
+        def __init__(self, path):
+            self.path = path
+        def makefile(self, *args, **kwargs):
+            import io
+            return io.BytesIO(b"")
+
+    handler = BridgeHandler(
+        request=MockRequest("/bridge/health"),
+        client_address=("127.0.0.1", 0),
+        server=MagicMock(),
+    )
+    handler.path = "/bridge/health"
+    captured = {}
+
+    def mock_json_response(data):
+        captured["data"] = data
+
+    handler._json_response = mock_json_response
+    handler.do_GET()
+
+    assert captured["data"]["status"] == "ok"
+    assert "active_sessions" in captured["data"]
+
+
+def test_bridge_handler_sessions_endpoint_lists_active():
+    """GET /bridge/sessions must list active sessions."""
+    from multi_user_bridge import BridgeHandler
+
+    class MockRequest:
+        def __init__(self, path):
+            self.path = path
+        def makefile(self, *args, **kwargs):
+            import io
+            return io.BytesIO(b"")
+
+    handler = BridgeHandler(
+        request=MockRequest("/bridge/sessions"),
+        client_address=("127.0.0.1", 0),
+        server=MagicMock(),
+    )
+    handler.path = "/bridge/sessions"
+    captured = {}
+
+    def mock_json_response(data):
+        captured["data"] = data
+
+    handler._json_response = mock_json_response
+    handler.do_GET()
+
+    assert "sessions" in captured["data"]
+
+
+# ── Smoke / sanity test ─────────────────────────────────────────────────────
+
+def test_multi_user_bridge_module_imports_cleanly():
+    """multi_user_bridge must be importable with all runtime deps patched."""
+    from multi_user_bridge import UserSession, SessionManager, BridgeHandler, session_manager
+
+    assert UserSession is not None
+    assert SessionManager is not None
+    assert BridgeHandler is not None
+    assert session_manager is not None