feat: room broadcast — say command delivers to all occupants in room

- say <message> now queues room_broadcast events on other sessions
- New GET /bridge/room_events/{user_id} endpoint (drain-on-read)
- WS connections receive real-time room broadcasts
- 5 new tests: broadcast, no-echo, room isolation, drain, 404
- Total tests: 27 (all passing)

tests/test_multi_user_bridge.py

@@ -277,3 +277,90 @@ class TestHTTPEndpoints:
         assert "cats" in d1["response"].lower() or d1["user_id"] == "alice"
         assert "dogs" in d2["response"].lower() or d2["user_id"] == "bob"
         assert d1["user_id"] != d2["user_id"]
+
+
+# ── Room Broadcast Tests ─────────────────────────────────────
+
+class TestRoomBroadcast:
+
+    @pytest.mark.asyncio
+    async def test_say_broadcasts_to_room_occupants(self, client):
+        c, _ = client
+        # Position both users in the same room
+        await c.post("/bridge/chat", json={
+            "user_id": "alice", "username": "Alice", "message": "hi", "room": "Tower"
+        })
+        await c.post("/bridge/chat", json={
+            "user_id": "bob", "username": "Bob", "message": "hi", "room": "Tower"
+        })
+        # Alice says something
+        await c.post("/bridge/chat", json={
+            "user_id": "alice", "username": "Alice", "message": "say Hello everyone!", "room": "Tower"
+        })
+        # Bob should have a pending room event
+        resp = await c.get("/bridge/room_events/bob")
+        data = await resp.json()
+        assert data["count"] >= 1
+        assert any("Alice" in e.get("message", "") for e in data["events"])
+
+    @pytest.mark.asyncio
+    async def test_say_does_not_echo_to_speaker(self, client):
+        c, _ = client
+        await c.post("/bridge/chat", json={
+            "user_id": "alice", "message": "hi", "room": "Tower"
+        })
+        await c.post("/bridge/chat", json={
+            "user_id": "bob", "message": "hi", "room": "Tower"
+        })
+        await c.post("/bridge/chat", json={
+            "user_id": "alice", "message": 'say Hello!', "room": "Tower"
+        })
+        # Alice should NOT have room events from herself
+        resp = await c.get("/bridge/room_events/alice")
+        data = await resp.json()
+        alice_events = [e for e in data["events"] if e.get("from_user") == "alice"]
+        assert len(alice_events) == 0
+
+    @pytest.mark.asyncio
+    async def test_say_no_broadcast_to_different_room(self, client):
+        c, _ = client
+        await c.post("/bridge/chat", json={
+            "user_id": "alice", "message": "hi", "room": "Tower"
+        })
+        await c.post("/bridge/chat", json={
+            "user_id": "bob", "message": "hi", "room": "Chapel"
+        })
+        await c.post("/bridge/chat", json={
+            "user_id": "alice", "message": 'say Hello!', "room": "Tower"
+        })
+        # Bob is in Chapel, shouldn't get Tower broadcasts
+        resp = await c.get("/bridge/room_events/bob")
+        data = await resp.json()
+        assert data["count"] == 0
+
+    @pytest.mark.asyncio
+    async def test_room_events_drain_after_read(self, client):
+        c, _ = client
+        await c.post("/bridge/chat", json={
+            "user_id": "alice", "message": "hi", "room": "Tower"
+        })
+        await c.post("/bridge/chat", json={
+            "user_id": "bob", "message": "hi", "room": "Tower"
+        })
+        await c.post("/bridge/chat", json={
+            "user_id": "alice", "message": 'say First!', "room": "Tower"
+        })
+        # First read drains
+        resp = await c.get("/bridge/room_events/bob")
+        data = await resp.json()
+        assert data["count"] >= 1
+        # Second read is empty
+        resp2 = await c.get("/bridge/room_events/bob")
+        data2 = await resp2.json()
+        assert data2["count"] == 0
+
+    @pytest.mark.asyncio
+    async def test_room_events_404_for_unknown_user(self, client):
+        c, _ = client
+        resp = await c.get("/bridge/room_events/nonexistent")
+        assert resp.status == 404