meta(tasks): update [secrets-path-off-by-one] in OpeItcLoc03/meeting-room

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

.tasks/STATUS.md

@@ -1,55 +1,102 @@
 # Task Board
 _Updated: 2026-05-04_
 
-## 🟢 v2-p1-deps — Обновить зависимости (pydantic, pytest, [web])
+## 🟢 v2-p2-async-bridge — AsyncEventBridge: sync EventEmitter → asyncio.Queue
 **Status:** done
-**Branch:** feat/v2-phase1-core-refactor
+**Where I stopped:** Complete
 
 ---
 
-## 🟢 v2-p1-models — Pydantic модели данных (ProviderConfig, RoleConfig, Session и др.)
+## 🟢 v2-p2-fastapi-app — FastAPI app + REST API (sessions, start, inject, roles, providers)
+**Status:** done
+**Where I stopped:** Complete
+
+---
+
+## 🟢 v2-p2-websocket-sse — WebSocket + SSE streaming endpoints
+**Status:** done
+**Where I stopped:** Complete
+
+---
+
+## 🟢 v2-p2-web-ui — Minimal HTML/CSS/JS frontend
+**Status:** done
+**Where I stopped:** Complete
+
+---
+
+## 🟢 v2-p2-serve-command — `meeting-room serve` CLI subcommand
+**Status:** done
+**Where I stopped:** Complete
+
+---
+
+## 🟢 code-review-p2 — Code review fixes after Phase 2
+**Status:** done
+**Where I stopped:** All critical/important issues fixed
+**Next action:** N/A
+
+---
+
+## 🟢 v2-p1-deps — Обновить зависимости (pydantic, pytest, [web])
+**Status:** done
+
+---
+
+## 🟢 v2-p1-models — Pydantic модели данных
 **Status:** done
-**Branch:** feat/v2-phase1-core-refactor
 
 ---
 
 ## 🟢 v2-p1-events — EventEmitter и типы событий
 **Status:** done
-**Branch:** feat/v2-phase1-core-refactor
 
 ---
 
-## 🟢 v2-p1-api-client — Рефактор APIClient (класс вместо глобалов)
+## 🟢 v2-p1-api-client — Рефактор APIClient
 **Status:** done
-**Branch:** feat/v2-phase1-core-refactor
 
 ---
 
-## 🟢 v2-p1-tools — Рефактор ToolRegistry (класс, Windows-совместимость)
+## 🟢 v2-p1-tools — Рефактор ToolRegistry
 **Status:** done
-**Branch:** feat/v2-phase1-core-refactor
 
 ---
 
-## 🟢 v2-p1-engine — DiscussionEngine (emit вместо print)
+## 🟢 v2-p1-engine — DiscussionEngine
 **Status:** done
-**Branch:** feat/v2-phase1-core-refactor
 
 ---
 
-## 🟢 v2-p1-config-cli — config.py + CLI рефактор + deprecation shim
+## 🟢 v2-p1-config-cli — config.py + CLI рефактор
 **Status:** done
-**Branch:** feat/v2-phase1-core-refactor
 
 ---
 
 <!--
-  Фаза 1 завершена! Фаза 2 (Web-сервер) задачи будут добавлены.
+  Фаза 2 завершена + код-ревью пройден. Все таски выполнены.
+-->
 
-  Статусы:
+## 🔵 [secrets-path-off-by-one] — Bug в `meeting_room/config.py:35` — off-by-one в количестве вызовов `os.path.dirname()`. `PACKAGE_DIR` = `~\projects\meeting-room\meeting_room` (resolved через `os.path.abspath(__file__)` в строке 17). Текущий код: `_common_root = os.path.dirname(os.path.dirname(os.path.dirname(PACKAGE_DIR)))` — 3 вызова — резолвит в `~\` (user home), а должно быть в `~\projects\`. Соответственно `_secrets_path = os.path.join(_common_root, ".common", "secrets", "interns.env")` указывает на `~\.common\secrets\interns.env`, который **не существует**. Реальный файл: `~\projects\.common\secrets\interns.env` (создан в рамках `[unify-llm-secrets]`, закрыт 2026-05-05). Эффект: `load_dotenv(_secrets_path)` молча возвращает False, `OLLAMA_CLOUD_API_KEY` env-var НЕ загружается, regex-substitution в `_env_replace` (config.py:46-50) оставляет литеральную строку `${OLLAMA_CLOUD_API_KEY}` в `providers.ollama_cloud.api_key`, OpenAI-client дёргает Ollama с этой строкой как Bearer-токеном → 401. Тесты не ловят: `tests/test_config.py:107,114,242` используют `monkeypatch.setenv()` — путь до secrets-файла не задействуется. Surfaced код-ревью 2026-05-07. Fix: либо `dirname` дважды (`os.path.dirname(os.path.dirname(PACKAGE_DIR))` → `~\projects\`), либо паритет с `interns-mcp/server.py:17`: `Path(__file__).resolve().parents[2]`. Acceptance: убрать env-var из shell, прогнать `python -c "from meeting_room.config import load_config; print(load_config()['providers']['ollama_cloud']['api_key'][:10])"` — должен вернуть начало реального ключа `02a8f6e9f9...`, а не `${OLLAMA_CLOU`. Добавить regression-тест: монкипатч-PACKAGE_DIR и ассерт что resolved path попадает в `<projects>/.common/secrets/interns.env`. Регрессия impacted closure: [unify-llm-secrets] в `common/.tasks/STATUS.md` (closed 2026-05-05) — claim'ил «meeting-room config.py → load_dotenv()» но fixture-only-tested.
-    🔴 Active     — только одна одновременно
+
-    🟡 Paused     — в процессе, возобновима
+**Status:** blocked
-    ⚪ Ready      — определена, не начата
+**Where I stopped:** workspace: working tree dirty
-    🟢 Done       — до слияния
+**Next action:** Открыть `~\projects\meeting-room\meeting_room\config.py:35`. Заменить тройной `os.path.dirname(...)` на двойной (или `Path(__file__).resolve().parents[2]`). Добавить regression-тест в `tests/test_config.py` который реально читает env-var из временного `.common/secrets/interns.env` через `monkeypatch.chdir` + временный layout. Пуш PATCH bump в `meeting-room/pyproject.toml`.
-    🔵 Blocked    — ждёт внешнего
+**Branch:** n/a
--->
+**Owner:** DESKTOP-NSEF0UK:claude-sonnet:26920
+**Claim token:** 3d037bf6-2627-4231-94b5-40cea04abae4
+**Claim expires at:** 2026-06-11T10:08:16.554Z
+<!-- created-by: OpeItcLoc03@DESKTOP-NSEF0UK / from: .meeting-room / 2026-05-06T22:11:53.904Z -->
+
+---
+
+## ⚪ [bootstrap-upgrade-canonical-triggers] — Прогнать `project-bootstrap` v1.10.0 в upgrade-режиме на этом проекте. Цель — дописать в `CLAUDE.md` 3 недостающих канонических триггера: `follow tdd-criteria`, `delegate to interns when allowed`, `recommend, don't menu`. Уже присутствуют: `pull remote before work`, `follow project discipline`, `we're on Windows`. ВНИМАНИЕ: это `meeting-room` (без точки) — обычный код-проект, не путать с `.meeting-room` workspace-комнатой (которая в раскатку не входит). Аудит 2026-05-07 через `bulk_text_read`. Контекст — координатор `[bootstrap-rollout-canonical-triggers]` в `_meta` (commit dd933fd). Пилот на `books` подтвердил идемпотентность (158fc95).
+
+**Status:** ready
+**Where I stopped:** Unblocked 2026-05-07 после успешного прогона `pilonuxt` (commit `e5d45f9`: полный bootstrap — CLAUDE.md + `.wiki/concepts/bootstrap-manifest.md`, скил отработал корректно). 3-строчная группа разблокирована.
+**Next action:** 1) `upgrade project` в `~/projects/meeting-room/` (без точки — обычная репа, не путать с `.meeting-room`). **Запустить скил `project-bootstrap` v1.10.0** — не делать ручной Edit (incident на `pilonuxt` 2026-05-07). 2) Подтвердить план: 2 файла — `CLAUDE.md` (+3: `follow tdd-criteria`, `delegate to interns when allowed`, `recommend, don't menu`) + `.wiki/concepts/bootstrap-manifest.md` (manifest 1.x → 1.10.0). 3) Прогнать. 4) **Verify:** `git diff --stat` показывает 2 файла. Если 1 — переоткрыть. Reference: `books@5ca16d9`. 5) Коммит `chore(bootstrap): upgrade CLAUDE.md — add tdd-criteria, interns, recommend-dont-menu triggers` + body. 6) Повторный — no-op. 7) `tasks_close`.
+**Blocker:** bootstrap-rollout-canonical-triggers
+**Branch:** n/a
+<!-- created-by: OpeItcLoc03@DESKTOP-NSEF0UK / from: meeting-room / 2026-05-07T06:57:21.673Z -->
+
+---

.tasks/v2-p2-async-bridge.md

@@ -0,0 +1,26 @@
+# v2-p2-async-bridge
+
+## Goal
+Bridge synchronous EventEmitter to asyncio so DiscussionEngine can run in a thread while WebSocket/SSE consumers receive events asynchronously.
+
+## Key files
+- `meeting_room/events.py` — add AsyncEventBridge class
+- `meeting_room/engine.py` — no changes (runs in thread as-is)
+
+## Decisions log
+- 2026-05-04: AsyncEventBridge + asyncio.to_thread() chosen over async engine rewrite. Minimizes changes, engine stays sync.
+
+## Open questions
+- [ ] None
+
+## Completed steps
+- [x] AsyncEventBridge class in events.py
+- [x] Thread-safe bridge with call_soon_threadsafe
+- [x] subscribe/unsubscribe consumer queues
+- [x] Drop policy on full queues
+- [x] Tests (10/10 passing)
+
+## Notes
+- events.py already has comment: "In Phase 2, an AsyncEventBridge will forward events to WebSocket clients"
+- Bridge subscribes to sync EventEmitter, puts events into asyncio.Queue
+- Consumers (WebSocket handlers, SSE handlers) read from the queue

.tasks/v2-p2-fastapi-app.md

@@ -0,0 +1,22 @@
+# v2-p2-fastapi-app
+
+## Goal
+Create FastAPI application with REST API endpoints for session management and discussion control.
+
+## Key files
+- `meeting_room/server.py` — FastAPI app, routes, session store
+
+## Decisions log
+- 2026-05-04: In-memory session store (dict). Persistent storage deferred.
+- 2026-05-04: REST routes under /api prefix, WebSocket/SSE on root
+
+## Open questions
+- [ ] None
+
+## Completed steps
+- [x] FastAPI app with lifespan handler
+- [x] Session + SessionStore models
+- [x] REST endpoints: POST /api/sessions, GET /api/sessions, GET /api/sessions/{id}, POST /api/sessions/{id}/start, POST /api/sessions/{id}/inject, GET /api/sessions/{id}/messages
+- [x] GET /api/roles, GET /api/providers
+- [x] Session start runs engine in asyncio.to_thread
+- [x] Tests (12 passing)

.tasks/v2-p2-serve-command.md

@@ -0,0 +1,17 @@
+# v2-p2-serve-command
+
+## Goal
+Add `meeting-room serve` CLI subcommand that starts the FastAPI web server.
+
+## Key files
+- `meeting_room/cli.py` — serve subcommand added
+
+## Decisions log
+- 2026-05-04: Default port 8000, configurable via --port. --host and --reload flags.
+
+## Open questions
+- [ ] None
+
+## Completed steps
+- [x] `meeting-room serve` subcommand with --host, --port, --reload
+- [x] Starts uvicorn with the FastAPI app

.tasks/v2-p2-web-ui.md

@@ -0,0 +1,18 @@
+# v2-p2-web-ui
+
+## Goal
+Minimal HTML/CSS/JS frontend for viewing and interacting with discussions.
+
+## Key files
+- `meeting_room/static/index.html` — single-page UI
+
+## Decisions log
+- 2026-05-04: Vanilla JS + WebSocket API. No framework.
+
+## Open questions
+- [ ] None
+
+## Completed steps
+- [x] Dark theme UI with WebSocket event handling
+- [x] Start discussion, view messages, inject messages
+- [x] Served from /static/index.html, / redirects

.tasks/v2-p2-websocket-sse.md

@@ -0,0 +1,17 @@
+# v2-p2-websocket-sse
+
+## Goal
+Stream discussion events to browser clients via WebSocket and SSE endpoints.
+
+## Key files
+- `meeting_room/server.py` — WebSocket and SSE route handlers
+
+## Decisions log
+- 2026-05-04: WebSocket for bidirectional (events + inject), SSE for read-only
+
+## Open questions
+- [ ] None
+
+## Completed steps
+- [x] WebSocket endpoint: /ws/{session_id} — bidirectional events + inject
+- [x] SSE endpoint: /events/{session_id} — read-only event stream

.wiki/overview.md

@@ -10,7 +10,7 @@ Multi-agent discussion framework — an interactive workspace where AI agents de
 
 ## What it does
 
-Meeting-room runs multi-agent roundtable discussions with configurable participants, models, and tools. Each role can use a different LLM provider, enabling cross-model review and debate. The event-driven architecture decouples the engine from I/O — CLI subscribes to events for ANSI output, future WebSocket server will stream them to browsers.
+Meeting-room runs multi-agent roundtable discussions with configurable participants, models, and tools. Each role can use a different LLM provider, enabling cross-model review and debate. The event-driven architecture decouples the engine from I/O — CLI subscribes to events for ANSI output, WebSocket/SSE streams them to browsers.
 
 ## Key components
 
@@ -18,6 +18,8 @@ Meeting-room runs multi-agent roundtable discussions with configurable participa
 - **APIClient** — class-based multi-provider client (OpenAI-compatible APIs)
 - **ToolRegistry** — class-based, pure Python (Windows compatible, no find/grep)
 - **EventEmitter** — synchronous pub/sub bus for decoupling engine from I/O
+- **AsyncEventBridge** — bridges sync EventEmitter to asyncio consumers (WebSocket/SSE)
+- **FastAPI server** — REST API + WebSocket + SSE for web UI
 - **Pydantic v2 models** — typed config (DiscussionConfig, ProviderConfig, RoleConfig), messages, sessions
 - **Scenarios** — Markdown + YAML frontmatter for defining discussion topics
 
@@ -29,13 +31,13 @@ Meeting-room runs multi-agent roundtable discussions with configurable participa
 ## Tech stack
 
 - Python 3.11+, pydantic v2, httpx, pyyaml
-- Event-driven architecture (EventEmitter, Phase 2: AsyncEventBridge)
+- FastAPI + uvicorn + websockets (web server)
+- Event-driven architecture (EventEmitter + AsyncEventBridge)
 - OpenAI-compatible API (multi-provider: RouterAI, Ollama Cloud, OpenAI, Anthropic, DeepSeek, local Ollama)
-- Planned: FastAPI + WebSocket for Web UI (Phase 2)
 
 ## Roadmap
 
 - [x] Phase 1: Core refactor — Pydantic models, EventEmitter, class-based APIClient/ToolRegistry, DiscussionEngine
-- [ ] Phase 2: Web server — FastAPI + WebSocket + SSE, REST API, HTML UI
+- [x] Phase 2: Web server — FastAPI + WebSocket + SSE, REST API, HTML UI
 - [ ] Phase 3: Interactivity — Boss role, CSO injection, Setup Wizard, BrainstormEngine
 - [ ] Phase 4: Artifacts — LLM extraction to .tasks/.wiki, SessionArchive

meeting_room/cli.py

@@ -10,6 +10,12 @@ import os
 import sys
 from datetime import datetime
 
+# Force UTF-8 output on Windows (cp1251 can't handle LLM unicode output)
+if sys.stdout.encoding != "utf-8":
+    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
+if sys.stderr.encoding != "utf-8":
+    sys.stderr.reconfigure(encoding="utf-8", errors="replace")
+
 from meeting_room.api_client import APIClient
 from meeting_room.config import find_workspace_config, load_config, load_discussion_config
 from meeting_room.engine import DiscussionEngine
@@ -201,7 +207,7 @@ def save_session(conversation: list[dict], scenario_name: str, sessions_dir: str
     """Save discussion result to a JSON file."""
     os.makedirs(sessions_dir, exist_ok=True)
     timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")
-    safe_name = scenario_name.replace(" ", "-").lower()
+    safe_name = scenario_name.replace(" ", "-").replace("—", "-").lower()
     filename = f"{timestamp}_{safe_name}.json"
     filepath = os.path.join(sessions_dir, filename)
 
@@ -253,7 +259,7 @@ def _subscribe_cli_handlers(bus: EventEmitter, max_rounds: int) -> None:
             print()
 
         print(f"{BOLD}Rounds:{RESET} {data['rounds']}")
-        print(f"{DIM}{'─' * 60}{RESET}\n")
+        print(f"{DIM}{'-' * 60}{RESET}\n")
 
     def on_round_start(event) -> None:
         data = event.data
@@ -285,7 +291,7 @@ def _subscribe_cli_handlers(bus: EventEmitter, max_rounds: int) -> None:
         print(f"  [Error: {data['error']}]")
 
     def on_round_end(event) -> None:
-        print(f"{DIM}{'─' * 60}{RESET}\n")
+        print(f"{DIM}{'-' * 60}{RESET}\n")
 
     def on_discussion_end(event) -> None:
         data = event.data
@@ -420,6 +426,11 @@ def main():
     init_parser = subparsers.add_parser("init", help="Initialize .meeting-room workspace")
     init_parser.add_argument("path", nargs="?", default=".", help="Where to create .meeting-room")
 
+    serve_parser = subparsers.add_parser("serve", help="Start web server")
+    serve_parser.add_argument("--host", default="0.0.0.0", help="Host to bind (default: 0.0.0.0)")
+    serve_parser.add_argument("--port", type=int, default=8000, help="Port to bind (default: 8000)")
+    serve_parser.add_argument("--reload", action="store_true", help="Enable auto-reload for development")
+
     args = parser.parse_args()
 
     # Handle init command
@@ -427,6 +438,18 @@ def main():
         init_workspace(args.path)
         return
 
+    # Handle serve command
+    if args.command == "serve":
+        import uvicorn
+        from meeting_room.server import app
+        uvicorn.run(
+            "meeting_room.server:app" if args.reload else app,
+            host=args.host,
+            port=args.port,
+            reload=args.reload,
+        )
+        return
+
     # Find config: CLI arg > .meeting-room/config.yaml > package default
     if args.config:
         config = load_config(args.config)

meeting_room/config/config.yaml

@@ -37,7 +37,7 @@ roles:
   moderator:
     name: "Moderator"
     provider: ollama_cloud
-    model: "gpt-4o-mini"
+    model: "glm-5.1"
     temperature: 0.7
     tools: "none"
     system_prompt: |
@@ -49,7 +49,7 @@ roles:
   skeptic:
     name: "Skeptic"
     provider: ollama_cloud
-    model: "gpt-4o-mini"
+    model: "glm-5.1"
     temperature: 0.9
     tools: "readonly"
     system_prompt: |
@@ -61,7 +61,7 @@ roles:
   idea_generator:
     name: "Idea Generator"
     provider: ollama_cloud
-    model: "gpt-4o-mini"
+    model: "glm-5.1"
     temperature: 1.0
     tools: "web"
     system_prompt: |
@@ -74,7 +74,7 @@ roles:
   analyst:
     name: "Analyst"
     provider: ollama_cloud
-    model: "gpt-4o-mini"
+    model: "glm-5.1"
     temperature: 0.5
     tools: "all"
     system_prompt: |
@@ -88,7 +88,7 @@ roles:
   defender:
     name: "Defender"
     provider: ollama_cloud
-    model: "gpt-4o-mini"
+    model: "glm-5.1"
     temperature: 0.6
     tools: "readonly"
     system_prompt: |
@@ -100,7 +100,7 @@ roles:
   attacker:
     name: "Attacker"
     provider: ollama_cloud
-    model: "gpt-4o-mini"
+    model: "glm-5.1"
     temperature: 0.9
     tools: "readonly"
     system_prompt: |
@@ -113,7 +113,7 @@ roles:
   security:
     name: "Security"
     provider: ollama_cloud
-    model: "gpt-4o-mini"
+    model: "glm-5.1"
     temperature: 0.4
     tools: "web"
     system_prompt: |

meeting_room/events.py

@@ -1,12 +1,14 @@
 """EventEmitter and event types for the Meeting Room discussion engine.
 
-Synchronous event bus. In Phase 2, an AsyncEventBridge will forward
+Synchronous event bus with AsyncEventBridge for forwarding events
-events to WebSocket clients.
+to asyncio consumers (WebSocket / SSE).
 """
 
 from __future__ import annotations
 
+import asyncio
 import logging
+import threading
 from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Any, Callable
@@ -126,4 +128,120 @@ class EventEmitter:
 
     def clear(self) -> None:
         """Remove all subscriptions."""
-        self._handlers.clear()
+        self._handlers.clear()
+
+
+# ---------------------------------------------------------------------------
+# AsyncEventBridge
+# ---------------------------------------------------------------------------
+
+# All event types the bridge forwards.
+_ALL_EVENT_TYPES = [
+    DISCUSSION_START,
+    ROUND_START,
+    AGENT_TURN_START,
+    AGENT_MESSAGE,
+    TOOL_CALL,
+    TOOL_RESULT,
+    AGENT_ERROR,
+    ROUND_END,
+    DISCUSSION_END,
+    BOSS_TURN,
+]
+
+
+class AsyncEventBridge:
+    """Bridges synchronous EventEmitter events to asyncio consumers.
+
+    Subscribes to all event types on a sync ``EventEmitter`` and forwards
+    each event into ``asyncio.Queue`` objects — one per consumer. Designed
+    for running ``DiscussionEngine`` in a worker thread via
+    ``asyncio.to_thread()`` while WebSocket / SSE handlers await events
+    on the async side.
+
+    Usage::
+
+        bus = EventEmitter()
+        bridge = AsyncEventBridge(bus)
+        bridge.start(asyncio.get_event_loop())
+
+        # In an async handler:
+        queue = bridge.subscribe()
+        try:
+            while True:
+                event = await queue.get()
+                await websocket.send_json(event)
+        finally:
+            bridge.unsubscribe(queue)
+
+        # Run engine in a thread:
+        await asyncio.to_thread(engine.run, scenario)
+
+        # Cleanup:
+        bridge.stop()
+    """
+
+    def __init__(self, event_bus: EventEmitter) -> None:
+        self._bus = event_bus
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._queues: list[asyncio.Queue[dict]] = []
+        self._lock = threading.Lock()
+        self._handler = self._on_sync_event
+
+    # -- lifecycle -----------------------------------------------------------
+
+    def start(self, loop: asyncio.AbstractEventLoop) -> None:
+        """Subscribe to all event types on the bus."""
+        self._loop = loop
+        for event_type in _ALL_EVENT_TYPES:
+            self._bus.on(event_type, self._handler)
+
+    def stop(self) -> None:
+        """Unsubscribe from all event types."""
+        for event_type in _ALL_EVENT_TYPES:
+            self._bus.off(event_type, self._handler)
+
+    # -- consumer management ------------------------------------------------
+
+    def subscribe(self) -> asyncio.Queue[dict]:
+        """Create a consumer queue that receives all events.
+
+        Call from async context. The caller owns the queue and must
+        call ``unsubscribe()`` when done.
+        """
+        q: asyncio.Queue[dict] = asyncio.Queue(maxsize=1000)
+        with self._lock:
+            self._queues.append(q)
+        return q
+
+    def unsubscribe(self, q: asyncio.Queue[dict]) -> None:
+        """Remove a consumer queue. Safe to call multiple times."""
+        with self._lock:
+            try:
+                self._queues.remove(q)
+            except ValueError:
+                pass
+
+    # -- internal ------------------------------------------------------------
+
+    def _on_sync_event(self, event: Event) -> None:
+        """Sync handler called by EventEmitter on any thread."""
+        payload = {
+            "type": event.type,
+            "data": event.data,
+            "timestamp": event.timestamp.isoformat(),
+        }
+        if self._loop is None or not self._loop.is_running():
+            return
+        with self._lock:
+            queues = list(self._queues)
+        for q in queues:
+            self._loop.call_soon_threadsafe(self._safe_put, q, payload)
+
+    @staticmethod
+    def _safe_put(q: asyncio.Queue[dict], payload: dict) -> None:
+        """Put payload into queue, dropping if full."""
+        try:
+            q.put_nowait(payload)
+        except asyncio.QueueFull:
+            logger.warning("AsyncEventBridge: dropping event, queue full")

meeting_room/server.py

@@ -0,0 +1,451 @@
+"""FastAPI web server for Meeting Room — REST API, WebSocket, and SSE endpoints."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import uuid
+from contextlib import asynccontextmanager
+from typing import Any
+
+from fastapi import APIRouter, FastAPI, HTTPException, WebSocket, WebSocketDisconnect
+from fastapi.responses import RedirectResponse, StreamingResponse
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel
+
+from meeting_room.api_client import APIClient
+from meeting_room.config import load_config, load_discussion_config, find_workspace_config
+from meeting_room.engine import DiscussionEngine
+from meeting_room.events import AGENT_ERROR, AGENT_MESSAGE, BOSS_TURN, AsyncEventBridge, EventEmitter
+from meeting_room.models import DiscussionConfig, ProviderConfig
+from meeting_room.scenario_loader import load_scenario
+from meeting_room.tools import ToolRegistry
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Pydantic models for API request/response
+# ---------------------------------------------------------------------------
+
+
+class StartDiscussionRequest(BaseModel):
+    """Request body for starting a discussion."""
+
+    scenario: str  # scenario file path or name
+    config: str | None = None  # optional config path override
+    save: bool = False
+
+
+class InjectMessageRequest(BaseModel):
+    """Request body for injecting a message."""
+
+    role_id: str
+    content: str
+
+
+class SessionResponse(BaseModel):
+    """Response for session creation."""
+
+    id: str
+    status: str
+
+
+class SessionDetailResponse(BaseModel):
+    """Response for session detail."""
+
+    id: str
+    status: str
+    scenario_name: str | None = None
+    message_count: int = 0
+
+
+# ---------------------------------------------------------------------------
+# Session state (renamed from Session to avoid collision with models.Session)
+# ---------------------------------------------------------------------------
+
+
+class ServerSession:
+    """Holds state for a single discussion session."""
+
+    def __init__(
+        self,
+        session_id: str,
+        config: dict,
+        discussion_config: DiscussionConfig,
+    ) -> None:
+        self.id = session_id
+        self.config = config
+        self.discussion_config = discussion_config
+        self.status: str = "created"  # created | running | completed | error
+        self.scenario_name: str | None = None
+        self.messages: list[dict] = []
+        self.event_bus = EventEmitter()
+        self.bridge = AsyncEventBridge(self.event_bus)
+        self._engine_task: asyncio.Task | None = None
+        self._collect_handler = None  # stored for cleanup
+
+    def build_engine(self) -> DiscussionEngine:
+        """Build a DiscussionEngine using this session's event_bus."""
+        providers = {
+            name: ProviderConfig(**vals)
+            for name, vals in self.config.get("providers", {}).items()
+        }
+        api_client = APIClient(providers=providers)
+        tool_registry = ToolRegistry(workdir=self.discussion_config.defaults.workdir)
+        return DiscussionEngine(
+            config=self.discussion_config,
+            api_client=api_client,
+            tool_registry=tool_registry,
+            events=self.event_bus,
+        )
+
+    def cleanup(self) -> None:
+        """Stop bridge and remove event handlers."""
+        self.bridge.stop()
+        if self._collect_handler is not None:
+            for event_type in (AGENT_MESSAGE, BOSS_TURN, AGENT_ERROR):
+                self.event_bus.off(event_type, self._collect_handler)
+            self._collect_handler = None
+
+
+class SessionStore:
+    """In-memory session store."""
+
+    def __init__(self) -> None:
+        self._sessions: dict[str, ServerSession] = {}
+
+    def create(self, config: dict, discussion_config: DiscussionConfig) -> ServerSession:
+        session_id = uuid.uuid4().hex[:12]
+        session = ServerSession(session_id, config, discussion_config)
+        self._sessions[session_id] = session
+        return session
+
+    def get(self, session_id: str) -> ServerSession | None:
+        return self._sessions.get(session_id)
+
+    def list_sessions(self) -> list[ServerSession]:
+        return list(self._sessions.values())
+
+
+# ---------------------------------------------------------------------------
+# Application factory
+# ---------------------------------------------------------------------------
+
+store = SessionStore()
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Startup/shutdown lifecycle."""
+    logger.info("Meeting Room server started")
+    yield
+    # Cleanup: cancel running engine tasks and stop bridges
+    for session in store.list_sessions():
+        if session._engine_task and not session._engine_task.done():
+            session._engine_task.cancel()
+        session.cleanup()
+    logger.info("Meeting Room server stopped")
+
+
+app = FastAPI(
+    title="Meeting Room",
+    version="0.2.1",
+    description="Multi-agent discussion framework — REST API and WebSocket",
+    lifespan=lifespan,
+)
+
+# Serve static files (web UI)
+_static_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static")
+app.mount("/static", StaticFiles(directory=_static_dir), name="static")
+
+
+@app.get("/")
+async def index():
+    """Redirect to the web UI."""
+    return RedirectResponse(url="/static/index.html")
+
+
+# ---------------------------------------------------------------------------
+# Helper
+# ---------------------------------------------------------------------------
+
+
+def _load_config_or_default(config_path: str | None) -> dict:
+    """Load config from path, workspace, or package default."""
+    if config_path:
+        return load_config(config_path)
+    ws_config = find_workspace_config()
+    if ws_config:
+        return load_config(ws_config)
+    return load_config()
+
+
+def _resolve_scenario_path(scenario: str) -> str:
+    """Resolve scenario path, preventing path traversal.
+
+    Only allows scenarios from CWD or the package scenarios directory.
+    """
+    # Reject path traversal
+    if ".." in scenario:
+        raise HTTPException(status_code=400, detail="Invalid scenario path: path traversal not allowed")
+
+    if os.path.isabs(scenario_path := scenario):
+        if not os.path.exists(scenario_path):
+            raise HTTPException(status_code=404, detail=f"Scenario not found: {scenario}")
+        return scenario_path
+
+    # Try relative to CWD first
+    cwd_path = os.path.join(os.getcwd(), scenario)
+    if os.path.exists(cwd_path):
+        return cwd_path
+
+    # Try package scenarios directory
+    pkg_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "scenarios", scenario)
+    if os.path.exists(pkg_path):
+        return pkg_path
+
+    raise HTTPException(status_code=404, detail=f"Scenario not found: {scenario}")
+
+
+def _mask_api_keys(providers: dict) -> dict:
+    """Mask API keys in provider config for safe display."""
+    masked = {}
+    for name, cfg in providers.items():
+        masked[name] = {
+            "base_url": cfg.get("base_url", ""),
+            "api_key": cfg.get("api_key", "")[:4] + "****" if cfg.get("api_key") else "",
+        }
+    return masked
+
+
+# ---------------------------------------------------------------------------
+# REST API (mounted at /api)
+# ---------------------------------------------------------------------------
+
+api = APIRouter(prefix="/api")
+
+
+@api.post("/sessions", response_model=SessionResponse)
+async def create_session(config: str | None = None):
+    """Create a new discussion session."""
+    raw_config = _load_config_or_default(config)
+    discussion_config = load_discussion_config()
+    session = store.create(raw_config, discussion_config)
+    return SessionResponse(id=session.id, status=session.status)
+
+
+@api.get("/sessions", response_model=list[SessionDetailResponse])
+async def list_sessions():
+    """List all sessions."""
+    sessions = store.list_sessions()
+    return [
+        SessionDetailResponse(
+            id=s.id,
+            status=s.status,
+            scenario_name=s.scenario_name,
+            message_count=len(s.messages),
+        )
+        for s in sessions
+    ]
+
+
+@api.get("/sessions/{session_id}", response_model=SessionDetailResponse)
+async def get_session(session_id: str):
+    """Get session details."""
+    session = store.get(session_id)
+    if not session:
+        raise HTTPException(status_code=404, detail="Session not found")
+    return SessionDetailResponse(
+        id=session.id,
+        status=session.status,
+        scenario_name=session.scenario_name,
+        message_count=len(session.messages),
+    )
+
+
+@api.post("/sessions/{session_id}/start", response_model=SessionDetailResponse)
+async def start_discussion(session_id: str, body: StartDiscussionRequest):
+    """Start a discussion in a session."""
+    session = store.get(session_id)
+    if not session:
+        raise HTTPException(status_code=404, detail="Session not found")
+    if session.status == "running":
+        raise HTTPException(status_code=409, detail="Discussion already running")
+
+    # Resolve and validate scenario path
+    scenario_path = _resolve_scenario_path(body.scenario)
+    scenario_data = load_scenario(scenario_path)
+    session.scenario_name = scenario_data.get("name", "Untitled")
+
+    # Rebuild config if overridden
+    if body.config:
+        raw_config = _load_config_or_default(body.config)
+        discussion_config = load_discussion_config()
+        session.config = raw_config
+        session.discussion_config = discussion_config
+
+    # Start the bridge on the running event loop
+    loop = asyncio.get_running_loop()
+    session.bridge.start(loop)
+
+    # Subscribe a handler to collect messages (stored for cleanup)
+    def collect_message(event):
+        session.messages.append({
+            "role_id": event.data.get("role_id", ""),
+            "name": event.data.get("name", ""),
+            "content": event.data.get("content", event.data.get("error", "")),
+        })
+
+    session._collect_handler = collect_message
+    session.event_bus.on(AGENT_MESSAGE, collect_message)
+    session.event_bus.on(BOSS_TURN, collect_message)
+    session.event_bus.on(AGENT_ERROR, collect_message)
+
+    # Run engine in a thread
+    engine = session.build_engine()
+    session.status = "running"
+
+    async def run_engine():
+        try:
+            await asyncio.to_thread(engine.run, scenario_data)
+            session.status = "completed"
+        except Exception as exc:
+            logger.exception("Engine error in session %s", session.id)
+            session.status = "error"
+        finally:
+            session.cleanup()
+
+    session._engine_task = asyncio.create_task(run_engine())
+
+    return SessionDetailResponse(
+        id=session.id,
+        status=session.status,
+        scenario_name=session.scenario_name,
+        message_count=0,
+    )
+
+
+@api.post("/sessions/{session_id}/inject", response_model=dict)
+async def inject_message(session_id: str, body: InjectMessageRequest):
+    """Inject a message from a Boss/CSO role.
+
+    Uses the session's event_bus so WebSocket/SSE clients receive the event.
+    """
+    session = store.get(session_id)
+    if not session:
+        raise HTTPException(status_code=404, detail="Session not found")
+    if session.status != "running":
+        raise HTTPException(status_code=409, detail="Session is not running")
+
+    # Use the session's event_bus — inject_message emits BOSS_TURN on it,
+    # which the bridge forwards to all subscribers (WebSocket/SSE).
+    engine = session.build_engine()
+    msg = engine.inject_message(body.role_id, body.content)
+
+    return {"role_id": msg.role_id, "name": msg.name, "content": msg.content}
+
+
+@api.get("/sessions/{session_id}/messages")
+async def get_messages(session_id: str):
+    """Get all messages in a session."""
+    session = store.get(session_id)
+    if not session:
+        raise HTTPException(status_code=404, detail="Session not found")
+    return session.messages
+
+
+@api.get("/roles")
+async def list_roles(config: str | None = None):
+    """List available roles and model bindings."""
+    raw_config = _load_config_or_default(config)
+    return raw_config.get("roles", {})
+
+
+@api.get("/providers")
+async def list_providers(config: str | None = None):
+    """List configured providers (API keys masked)."""
+    raw_config = _load_config_or_default(config)
+    return _mask_api_keys(raw_config.get("providers", {}))
+
+
+# Mount API router
+app.include_router(api)
+
+
+# ---------------------------------------------------------------------------
+# WebSocket
+# ---------------------------------------------------------------------------
+
+
+@app.websocket("/ws/{session_id}")
+async def websocket_endpoint(websocket: WebSocket, session_id: str):
+    """WebSocket endpoint for real-time event streaming and message injection."""
+    session = store.get(session_id)
+    if not session:
+        await websocket.close(code=4004, reason="Session not found")
+        return
+
+    await websocket.accept()
+    queue = session.bridge.subscribe()
+    send_task = asyncio.create_task(_ws_send_loop(websocket, queue))
+
+    try:
+        while True:
+            data = await websocket.receive_json()
+            if data.get("type") == "inject":
+                role_id = data.get("role_id", "boss")
+                content = data.get("content", "")
+                # Use session's event_bus so events reach all subscribers
+                engine = session.build_engine()
+                engine.inject_message(role_id, content)
+    except WebSocketDisconnect:
+        pass
+    finally:
+        send_task.cancel()
+        session.bridge.unsubscribe(queue)
+
+
+async def _ws_send_loop(websocket: WebSocket, queue: asyncio.Queue[dict]) -> None:
+    """Forward events from bridge queue to WebSocket."""
+    try:
+        while True:
+            event = await queue.get()
+            await websocket.send_json(event)
+    except asyncio.CancelledError:
+        pass
+
+
+# ---------------------------------------------------------------------------
+# SSE
+# ---------------------------------------------------------------------------
+
+
+@app.get("/events/{session_id}")
+async def sse_endpoint(session_id: str):
+    """SSE endpoint for read-only event streaming."""
+    session = store.get(session_id)
+    if not session:
+        raise HTTPException(status_code=404, detail="Session not found")
+
+    async def event_generator():
+        queue = session.bridge.subscribe()
+        try:
+            while True:
+                event = await queue.get()
+                yield f"data: {json.dumps(event)}\n\n"
+        except asyncio.CancelledError:
+            pass
+        finally:
+            session.bridge.unsubscribe(queue)
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        },
+    )

meeting_room/static/index.html

@@ -0,0 +1,306 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>Meeting Room</title>
+    <style>
+        :root {
+            --bg: #1a1a2e;
+            --surface: #16213e;
+            --card: #0f3460;
+            --border: #533483;
+            --accent: #e94560;
+            --text: #eee;
+            --muted: #999;
+        }
+        * { margin: 0; padding: 0; box-sizing: border-box; }
+        body {
+            font-family: 'Segoe UI', system-ui, sans-serif;
+            background: var(--bg);
+            color: var(--text);
+            min-height: 100vh;
+        }
+        header {
+            background: var(--surface);
+            padding: 1rem 2rem;
+            border-bottom: 2px solid var(--border);
+            display: flex;
+            align-items: center;
+            justify-content: space-between;
+        }
+        header h1 { font-size: 1.4rem; }
+        header .status {
+            font-size: 0.85rem;
+            color: var(--muted);
+        }
+        .status .dot {
+            display: inline-block;
+            width: 8px; height: 8px;
+            border-radius: 50%;
+            margin-right: 4px;
+        }
+        .dot.connected { background: #4caf50; }
+        .dot.disconnected { background: var(--accent); }
+        main {
+            max-width: 900px;
+            margin: 2rem auto;
+            padding: 0 1rem;
+        }
+        .panel {
+            background: var(--surface);
+            border: 1px solid var(--border);
+            border-radius: 8px;
+            padding: 1.5rem;
+            margin-bottom: 1.5rem;
+        }
+        .panel h2 {
+            font-size: 1.1rem;
+            margin-bottom: 1rem;
+            color: var(--accent);
+        }
+        .controls {
+            display: flex;
+            gap: 0.5rem;
+            flex-wrap: wrap;
+            align-items: center;
+        }
+        select, input, button {
+            font-size: 0.9rem;
+            padding: 0.5rem 0.75rem;
+            border: 1px solid var(--border);
+            border-radius: 4px;
+            background: var(--card);
+            color: var(--text);
+        }
+        button {
+            cursor: pointer;
+            border-color: var(--accent);
+            transition: background 0.2s;
+        }
+        button:hover { background: var(--accent); }
+        button:disabled { opacity: 0.5; cursor: not-allowed; }
+        #messages {
+            max-height: 60vh;
+            overflow-y: auto;
+            padding: 0.5rem 0;
+        }
+        .msg {
+            padding: 0.75rem 1rem;
+            margin: 0.5rem 0;
+            border-radius: 6px;
+            background: var(--card);
+            border-left: 3px solid var(--border);
+            animation: fadeIn 0.3s;
+        }
+        .msg .name {
+            font-weight: 600;
+            color: var(--accent);
+            font-size: 0.85rem;
+        }
+        .msg .content {
+            margin-top: 0.3rem;
+            white-space: pre-wrap;
+            font-size: 0.9rem;
+            line-height: 1.5;
+        }
+        .msg.tool-call {
+            border-left-color: #ff9800;
+            font-size: 0.8rem;
+            color: var(--muted);
+        }
+        .msg.error {
+            border-left-color: #f44336;
+        }
+        .inject-bar {
+            display: flex;
+            gap: 0.5rem;
+            margin-top: 1rem;
+        }
+        .inject-bar input { flex: 1; }
+        @keyframes fadeIn {
+            from { opacity: 0; transform: translateY(4px); }
+            to { opacity: 1; transform: translateY(0); }
+        }
+        .round-marker {
+            text-align: center;
+            color: var(--muted);
+            font-size: 0.8rem;
+            padding: 0.5rem;
+            border-bottom: 1px solid var(--border);
+        }
+    </style>
+</head>
+<body>
+    <header>
+        <h1>Meeting Room</h1>
+        <div class="status">
+            <span class="dot disconnected" id="statusDot"></span>
+            <span id="statusText">Disconnected</span>
+        </div>
+    </header>
+    <main>
+        <div class="panel">
+            <h2>Start Discussion</h2>
+            <div class="controls">
+                <select id="scenarioSelect">
+                    <option value="">Loading scenarios...</option>
+                </select>
+                <button id="startBtn" onclick="startSession()">Start</button>
+            </div>
+        </div>
+
+        <div class="panel" id="chatPanel" style="display:none">
+            <h2 id="sessionTitle">Discussion</h2>
+            <div id="messages"></div>
+            <div class="inject-bar">
+                <input id="injectRole" placeholder="Role (boss)" value="boss" style="width:120px">
+                <input id="injectText" placeholder="Inject message..." onkeydown="if(event.key==='Enter')injectMessage()">
+                <button onclick="injectMessage()">Send</button>
+            </div>
+        </div>
+    </main>
+
+    <script>
+        let ws = null;
+        let sessionId = null;
+
+        // --- API helpers ---
+        async function api(path, opts = {}) {
+            const res = await fetch(`/api${path}`, {
+                headers: { 'Content-Type': 'application/json' },
+                ...opts,
+            });
+            if (!res.ok) throw new Error(`${res.status}: ${await res.text()}`);
+            return res.json();
+        }
+
+        // --- Load scenarios (placeholder) ---
+        // In a full implementation, we'd load from /api/scenarios
+        // For now, scenarios are provided by path
+
+        // --- Create and start session ---
+        async function startSession() {
+            const scenario = document.getElementById('scenarioSelect').value;
+            if (!scenario) return alert('Select a scenario first');
+
+            try {
+                // Create session
+                const sess = await api('/sessions', { method: 'POST' });
+                sessionId = sess.id;
+
+                // Start discussion
+                await api(`/sessions/${sessionId}/start`, {
+                    method: 'POST',
+                    body: JSON.stringify({ scenario }),
+                });
+
+                // Connect WebSocket
+                connectWS(sessionId);
+
+                document.getElementById('chatPanel').style.display = 'block';
+                document.getElementById('startBtn').disabled = true;
+            } catch (e) {
+                alert('Error: ' + e.message);
+            }
+        }
+
+        // --- WebSocket ---
+        function connectWS(sid) {
+            const proto = location.protocol === 'https:' ? 'wss' : 'ws';
+            ws = new WebSocket(`${proto}://${location.host}/ws/${sid}`);
+
+            ws.onopen = () => {
+                document.getElementById('statusDot').className = 'dot connected';
+                document.getElementById('statusText').textContent = 'Connected';
+            };
+
+            ws.onclose = () => {
+                document.getElementById('statusDot').className = 'dot disconnected';
+                document.getElementById('statusText').textContent = 'Disconnected';
+            };
+
+            ws.onmessage = (event) => {
+                const data = JSON.parse(event.data);
+                handleEvent(data);
+            };
+        }
+
+        // --- Handle events ---
+        const ROLE_COLORS = {
+            moderator: '#4fc3f7',
+            skeptic: '#ef5350',
+            idea_generator: '#66bb6a',
+            analyst: '#ffa726',
+            boss: '#ab47bc',
+            defender: '#42a5f5',
+            attacker: '#ff7043',
+            security: '#78909c',
+        };
+
+        function handleEvent(data) {
+            const container = document.getElementById('messages');
+            const type = data.type;
+            const d = data.data || {};
+
+            if (type === 'discussion_start') {
+                document.getElementById('sessionTitle').textContent = d.name || 'Discussion';
+                container.innerHTML = '';
+            } else if (type === 'round_start') {
+                const el = document.createElement('div');
+                el.className = 'round-marker';
+                el.textContent = `— Round ${d.round_num}/${d.total_rounds} —`;
+                container.appendChild(el);
+            } else if (type === 'agent_message' || type === 'boss_turn') {
+                const el = document.createElement('div');
+                el.className = 'msg';
+                const color = ROLE_COLORS[d.role_id] || '#999';
+                el.innerHTML = `<div class="name" style="color:${color}">${escapeHtml(d.name)}</div><div class="content">${escapeHtml(d.content || '')}</div>`;
+                container.appendChild(el);
+            } else if (type === 'agent_error') {
+                const el = document.createElement('div');
+                el.className = 'msg error';
+                el.innerHTML = `<div class="name">${escapeHtml(d.name)}</div><div class="content">Error: ${escapeHtml(d.error || '')}</div>`;
+                container.appendChild(el);
+            } else if (type === 'tool_call') {
+                const el = document.createElement('div');
+                el.className = 'msg tool-call';
+                el.textContent = `🔧 ${d.tool_name}(${d.arguments || ''})`;
+                container.appendChild(el);
+            } else if (type === 'discussion_end') {
+                const el = document.createElement('div');
+                el.className = 'round-marker';
+                el.textContent = `— Discussion complete (${d.total_messages} messages) —`;
+                container.appendChild(el);
+            }
+
+            container.scrollTop = container.scrollHeight;
+        }
+
+        function escapeHtml(text) {
+            const div = document.createElement('div');
+            div.textContent = text;
+            return div.innerHTML;
+        }
+
+        // --- Inject message ---
+        function injectMessage() {
+            if (!ws || ws.readyState !== WebSocket.OPEN) {
+                alert('Not connected');
+                return;
+            }
+            const role = document.getElementById('injectRole').value || 'boss';
+            const content = document.getElementById('injectText').value;
+            if (!content) return;
+
+            ws.send(JSON.stringify({ type: 'inject', role_id: role, content }));
+            document.getElementById('injectText').value = '';
+        }
+
+        // --- Init: fetch scenarios (todo: add API endpoint) ---
+        // For now, just show a text input
+        document.getElementById('scenarioSelect').outerHTML =
+            '<input id="scenarioSelect" placeholder="Scenario path (e.g. scenarios/example.md)" style="flex:1">';
+    </script>
+</body>
+</html>

pyproject.toml

@@ -16,7 +16,7 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
-dev = ["pytest>=7.0", "pytest-mock>=3.0"]
+dev = ["pytest>=7.0", "pytest-mock>=3.0", "pytest-asyncio>=0.23"]
 web = ["fastapi>=0.100", "uvicorn>=0.20", "websockets>=11.0"]
 
 

tests/test_async_bridge.py

@@ -0,0 +1,225 @@
+"""Tests for AsyncEventBridge — sync EventEmitter → asyncio.Queue bridge."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from meeting_room.events import (
+    AGENT_MESSAGE,
+    AGENT_TURN_START,
+    DISCUSSION_END,
+    DISCUSSION_START,
+    ROUND_END,
+    ROUND_START,
+    TOOL_CALL,
+    AsyncEventBridge,
+    EventEmitter,
+)
+
+
+class TestAsyncEventBridgeSubscribe:
+    """subscribe() creates a queue that receives events."""
+
+    @pytest.mark.asyncio
+    async def test_subscribe_returns_queue(self) -> None:
+        bus = EventEmitter()
+        bridge = AsyncEventBridge(bus)
+        q = bridge.subscribe()
+        assert isinstance(q, asyncio.Queue)
+
+    @pytest.mark.asyncio
+    async def test_multiple_subscribers(self) -> None:
+        bus = EventEmitter()
+        bridge = AsyncEventBridge(bus)
+        q1 = bridge.subscribe()
+        q2 = bridge.subscribe()
+        assert q1 is not q2
+
+
+class TestAsyncEventBridgeStartStop:
+    """start()/stop() wire the bridge to the sync bus."""
+
+    @pytest.mark.asyncio
+    async def test_start_subscribes_to_all_event_types(self) -> None:
+        bus = EventEmitter()
+        bridge = AsyncEventBridge(bus)
+        loop = asyncio.get_event_loop()
+        bridge.start(loop)
+
+        for event_type in [
+            DISCUSSION_START, ROUND_START, AGENT_TURN_START,
+            AGENT_MESSAGE, TOOL_CALL, ROUND_END, DISCUSSION_END,
+        ]:
+            assert len(bus.handlers(event_type)) == 1
+
+        bridge.stop()
+
+    @pytest.mark.asyncio
+    async def test_stop_unsubscribes(self) -> None:
+        bus = EventEmitter()
+        bridge = AsyncEventBridge(bus)
+        loop = asyncio.get_event_loop()
+        bridge.start(loop)
+        bridge.stop()
+
+        for event_type in [
+            DISCUSSION_START, ROUND_START, AGENT_TURN_START,
+            AGENT_MESSAGE, TOOL_CALL, ROUND_END, DISCUSSION_END,
+        ]:
+            assert len(bus.handlers(event_type)) == 0
+
+
+class TestAsyncEventBridgeForwarding:
+    """Events emitted on the sync bus arrive in async queues."""
+
+    @pytest.mark.asyncio
+    async def test_forward_event_to_queue(self) -> None:
+        bus = EventEmitter()
+        bridge = AsyncEventBridge(bus)
+        loop = asyncio.get_event_loop()
+        bridge.start(loop)
+
+        q = bridge.subscribe()
+
+        # Emit a sync event
+        bus.emit(AGENT_MESSAGE, role_id="moderator", name="Moderator", content="Hello")
+
+        # Give the event loop a chance to process call_soon_threadsafe
+        await asyncio.sleep(0.05)
+
+        payload = q.get_nowait()
+        assert payload["type"] == AGENT_MESSAGE
+        assert payload["data"]["role_id"] == "moderator"
+        assert payload["data"]["content"] == "Hello"
+        assert "timestamp" in payload
+
+        bridge.stop()
+
+    @pytest.mark.asyncio
+    async def test_multiple_queues_receive_same_event(self) -> None:
+        bus = EventEmitter()
+        bridge = AsyncEventBridge(bus)
+        loop = asyncio.get_event_loop()
+        bridge.start(loop)
+
+        q1 = bridge.subscribe()
+        q2 = bridge.subscribe()
+
+        bus.emit(DISCUSSION_START, name="test")
+        await asyncio.sleep(0.05)
+
+        p1 = q1.get_nowait()
+        p2 = q2.get_nowait()
+        assert p1["type"] == DISCUSSION_START
+        assert p2["type"] == DISCUSSION_START
+        assert p1["data"]["name"] == "test"
+        assert p2["data"]["name"] == "test"
+
+        bridge.stop()
+
+    @pytest.mark.asyncio
+    async def test_unsubscribe_stops_delivery(self) -> None:
+        bus = EventEmitter()
+        bridge = AsyncEventBridge(bus)
+        loop = asyncio.get_event_loop()
+        bridge.start(loop)
+
+        q = bridge.subscribe()
+        bridge.unsubscribe(q)
+
+        bus.emit(AGENT_MESSAGE, role_id="x", name="X", content="Y")
+        await asyncio.sleep(0.05)
+
+        assert q.empty()
+
+        bridge.stop()
+
+
+class TestAsyncEventBridgeThreadedForwarding:
+    """Events from a worker thread reach async consumers."""
+
+    @pytest.mark.asyncio
+    async def test_forward_from_thread(self) -> None:
+        bus = EventEmitter()
+        bridge = AsyncEventBridge(bus)
+        loop = asyncio.get_event_loop()
+        bridge.start(loop)
+
+        q = bridge.subscribe()
+
+        # Run engine in a thread (simplified)
+        def sync_work() -> None:
+            bus.emit(DISCUSSION_START, name="threaded")
+            bus.emit(AGENT_MESSAGE, role_id="bot", name="Bot", content="Hi")
+            bus.emit(DISCUSSION_END, total_messages=1)
+
+        await asyncio.to_thread(sync_work)
+        await asyncio.sleep(0.1)
+
+        events = []
+        while not q.empty():
+            events.append(q.get_nowait())
+
+        types = [e["type"] for e in events]
+        assert "discussion_start" in types
+        assert "agent_message" in types
+        assert "discussion_end" in types
+
+        bridge.stop()
+
+
+class TestAsyncEventBridgePayloadFormat:
+    """Serialized events contain type, data, and ISO timestamp."""
+
+    @pytest.mark.asyncio
+    async def test_payload_structure(self) -> None:
+        bus = EventEmitter()
+        bridge = AsyncEventBridge(bus)
+        loop = asyncio.get_event_loop()
+        bridge.start(loop)
+
+        q = bridge.subscribe()
+        bus.emit(TOOL_CALL, role_id="analyst", tool_name="search", arguments='{"q":"test"}')
+        await asyncio.sleep(0.05)
+
+        payload = q.get_nowait()
+        assert "type" in payload
+        assert "data" in payload
+        assert "timestamp" in payload
+        # timestamp is ISO format string
+        assert isinstance(payload["timestamp"], str)
+        assert payload["data"]["tool_name"] == "search"
+
+        bridge.stop()
+
+
+class TestAsyncEventBridgeDropPolicy:
+    """When a queue is full, events are dropped without crashing."""
+
+    @pytest.mark.asyncio
+    async def test_full_queue_drops_events(self) -> None:
+        bus = EventEmitter()
+        bridge = AsyncEventBridge(bus)
+        loop = asyncio.get_event_loop()
+        bridge.start(loop)
+
+        # maxsize=2 to make it fill quickly
+        q: asyncio.Queue[dict] = asyncio.Queue(maxsize=2)
+        with bridge._lock:
+            bridge._queues.append(q)
+
+        # Emit 5 events — queue can hold 2
+        for i in range(5):
+            bus.emit(AGENT_MESSAGE, role_id="x", name="X", content=f"msg{i}")
+        await asyncio.sleep(0.1)
+
+        # Queue should have at most 2 items (rest dropped)
+        received = []
+        while not q.empty():
+            received.append(q.get_nowait())
+
+        assert len(received) <= 2
+
+        bridge.stop()

tests/test_server.py

@@ -0,0 +1,204 @@
+"""Tests for meeting_room.server — FastAPI app, REST API, WebSocket/SSE."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from fastapi import HTTPException
+from fastapi.testclient import TestClient
+
+from meeting_room.events import AsyncEventBridge, EventEmitter
+from meeting_room.server import ServerSession, SessionStore, app, store
+
+
+@pytest.fixture(autouse=True)
+def _reset_store():
+    """Clear the global session store before each test."""
+    store._sessions.clear()
+    yield
+
+
+@pytest.fixture
+def client():
+    """FastAPI test client."""
+    return TestClient(app)
+
+
+# ---------------------------------------------------------------------------
+# SessionStore
+# ---------------------------------------------------------------------------
+
+
+class TestSessionStore:
+    def test_create_session(self) -> None:
+        store = SessionStore()
+        session = store.create({"providers": {}, "roles": {}, "defaults": {}}, _minimal_discussion_config())
+        assert session.id
+        assert session.status == "created"
+
+    def test_get_session(self) -> None:
+        store = SessionStore()
+        session = store.create({"providers": {}, "roles": {}, "defaults": {}}, _minimal_discussion_config())
+        found = store.get(session.id)
+        assert found is session
+
+    def test_get_nonexistent_session(self) -> None:
+        store = SessionStore()
+        assert store.get("nonexistent") is None
+
+    def test_list_sessions(self) -> None:
+        store = SessionStore()
+        store.create({"providers": {}, "roles": {}, "defaults": {}}, _minimal_discussion_config())
+        store.create({"providers": {}, "roles": {}, "defaults": {}}, _minimal_discussion_config())
+        assert len(store.list_sessions()) == 2
+
+
+# ---------------------------------------------------------------------------
+# REST API endpoints
+# ---------------------------------------------------------------------------
+
+
+class TestCreateSession:
+    def test_create_session_returns_id_and_status(self, client) -> None:
+        response = client.post("/api/sessions")
+        assert response.status_code == 200
+        data = response.json()
+        assert "id" in data
+        assert data["status"] == "created"
+
+
+class TestListSessions:
+    def test_list_empty(self, client) -> None:
+        response = client.get("/api/sessions")
+        assert response.status_code == 200
+        assert response.json() == []
+
+    def test_list_after_create(self, client) -> None:
+        client.post("/api/sessions")
+        client.post("/api/sessions")
+        response = client.get("/api/sessions")
+        assert response.status_code == 200
+        assert len(response.json()) == 2
+
+
+class TestGetSession:
+    def test_get_existing_session(self, client) -> None:
+        create_resp = client.post("/api/sessions")
+        session_id = create_resp.json()["id"]
+        response = client.get(f"/api/sessions/{session_id}")
+        assert response.status_code == 200
+        assert response.json()["id"] == session_id
+
+    def test_get_nonexistent_session(self, client) -> None:
+        response = client.get("/api/sessions/nonexistent")
+        assert response.status_code == 404
+
+
+class TestRolesAndProviders:
+    def test_list_roles(self, client) -> None:
+        response = client.get("/api/roles")
+        assert response.status_code == 200
+        data = response.json()
+        assert "moderator" in data or isinstance(data, dict)
+
+    def test_list_providers(self, client) -> None:
+        response = client.get("/api/providers")
+        assert response.status_code == 200
+        data = response.json()
+        assert isinstance(data, dict)
+
+
+# ---------------------------------------------------------------------------
+# Session model
+# ---------------------------------------------------------------------------
+
+
+class TestServerSession:
+    def test_build_engine(self) -> None:
+        config = _minimal_config()
+        session = ServerSession("test-id", config, _minimal_discussion_config())
+        engine = session.build_engine()
+        assert engine is not None
+
+    def test_cleanup_removes_collect_handler(self) -> None:
+        from meeting_room.events import AGENT_MESSAGE, BOSS_TURN, AGENT_ERROR
+        config = _minimal_config()
+        session = ServerSession("test-id", config, _minimal_discussion_config())
+        # Simulate what start_discussion does
+        def collect_message(event):
+            pass
+        session._collect_handler = collect_message
+        session.event_bus.on(AGENT_MESSAGE, collect_message)
+        session.event_bus.on(BOSS_TURN, collect_message)
+        session.event_bus.on(AGENT_ERROR, collect_message)
+        assert len(session.event_bus.handlers(AGENT_MESSAGE)) == 1
+        session.cleanup()
+        assert len(session.event_bus.handlers(AGENT_MESSAGE)) == 0
+
+
+class TestProviderMasking:
+    def test_api_keys_masked(self) -> None:
+        from meeting_room.server import _mask_api_keys
+        providers = {
+            "routerai": {"base_url": "https://api.example.com", "api_key": "sk-1234567890abcdef"},
+        }
+        masked = _mask_api_keys(providers)
+        assert masked["routerai"]["api_key"] == "sk-1****"
+        assert masked["routerai"]["base_url"] == "https://api.example.com"
+
+    def test_empty_key(self) -> None:
+        from meeting_room.server import _mask_api_keys
+        providers = {"test": {"base_url": "https://x.com", "api_key": ""}}
+        masked = _mask_api_keys(providers)
+        assert masked["test"]["api_key"] == ""
+
+
+class TestPathTraversal:
+    def test_reject_parent_dir(self) -> None:
+        from meeting_room.server import _resolve_scenario_path
+        with pytest.raises(HTTPException) as exc_info:
+            _resolve_scenario_path("../../etc/passwd")
+        assert exc_info.value.status_code == 400
+
+
+# ---------------------------------------------------------------------------
+# Helper
+# ---------------------------------------------------------------------------
+
+
+def _minimal_config() -> dict:
+    return {
+        "providers": {
+            "test_provider": {
+                "base_url": "https://api.example.com/v1",
+                "api_key": "test-key",
+            }
+        },
+        "roles": {
+            "moderator": {
+                "name": "Moderator",
+                "provider": "test_provider",
+                "model": "gpt-4o-mini",
+                "temperature": 0.7,
+                "tools": "none",
+                "system_prompt": "You are a moderator.",
+            }
+        },
+        "defaults": {
+            "max_rounds": 3,
+            "language": "en",
+            "framework": "custom",
+            "workdir": ".",
+        },
+    }
+
+
+def _minimal_discussion_config():
+    from meeting_room.models import DefaultsConfig, DiscussionConfig
+
+    return DiscussionConfig(
+        providers={"test_provider": {"base_url": "https://api.example.com/v1", "api_key": "test-key"}},
+        roles={"moderator": {"name": "Moderator", "provider": "test_provider", "model": "gpt-4o-mini"}},
+        defaults=DefaultsConfig(max_rounds=3),
+    )