diff --git a/meeting_room/server.py b/meeting_room/server.py index 210cfef..047f366 100644 --- a/meeting_room/server.py +++ b/meeting_room/server.py @@ -3,7 +3,9 @@ from __future__ import annotations import asyncio +import json import logging +import os import uuid from contextlib import asynccontextmanager from typing import Any @@ -16,7 +18,7 @@ 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 AsyncEventBridge, EventEmitter +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 @@ -60,11 +62,11 @@ class SessionDetailResponse(BaseModel): # --------------------------------------------------------------------------- -# Session store +# Session state (renamed from Session to avoid collision with models.Session) # --------------------------------------------------------------------------- -class Session: +class ServerSession: """Holds state for a single discussion session.""" def __init__( @@ -82,8 +84,10 @@ class Session: 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() @@ -97,23 +101,31 @@ class Session: 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, Session] = {} + self._sessions: dict[str, ServerSession] = {} - def create(self, config: dict, discussion_config: DiscussionConfig) -> Session: + def create(self, config: dict, discussion_config: DiscussionConfig) -> ServerSession: session_id = uuid.uuid4().hex[:12] - session = Session(session_id, config, discussion_config) + session = ServerSession(session_id, config, discussion_config) self._sessions[session_id] = session return session - def get(self, session_id: str) -> Session | None: + def get(self, session_id: str) -> ServerSession | None: return self._sessions.get(session_id) - def list_sessions(self) -> list[Session]: + def list_sessions(self) -> list[ServerSession]: return list(self._sessions.values()) @@ -129,22 +141,22 @@ async def lifespan(app: FastAPI): """Startup/shutdown lifecycle.""" logger.info("Meeting Room server started") yield - # Cleanup: cancel any running engine tasks + # 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.0", + version="0.2.1", description="Multi-agent discussion framework — REST API and WebSocket", lifespan=lifespan, ) # Serve static files (web UI) -import os _static_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static") app.mount("/static", StaticFiles(directory=_static_dir), name="static") @@ -152,7 +164,6 @@ app.mount("/static", StaticFiles(directory=_static_dir), name="static") @app.get("/") async def index(): """Redirect to the web UI.""" - from fastapi.responses import RedirectResponse return RedirectResponse(url="/static/index.html") @@ -171,6 +182,44 @@ def _load_config_or_default(config_path: str | None) -> dict: 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) # --------------------------------------------------------------------------- @@ -225,18 +274,8 @@ async def start_discussion(session_id: str, body: StartDiscussionRequest): if session.status == "running": raise HTTPException(status_code=409, detail="Discussion already running") - # Load scenario - import os - scenario_path = body.scenario - if not os.path.isabs(scenario_path): - scenario_path = os.path.join(os.getcwd(), scenario_path) - if not os.path.exists(scenario_path): - scenario_path = os.path.join( - os.path.dirname(os.path.abspath(__file__)), "scenarios", body.scenario - ) - if not os.path.exists(scenario_path): - raise HTTPException(status_code=404, detail=f"Scenario not found: {body.scenario}") - + # 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") @@ -247,13 +286,11 @@ async def start_discussion(session_id: str, body: StartDiscussionRequest): session.config = raw_config session.discussion_config = discussion_config - # Start the bridge - loop = asyncio.get_event_loop() + # Start the bridge on the running event loop + loop = asyncio.get_running_loop() session.bridge.start(loop) - # Subscribe a handler to collect messages - from meeting_room.events import AGENT_MESSAGE, BOSS_TURN, AGENT_ERROR - + # Subscribe a handler to collect messages (stored for cleanup) def collect_message(event): session.messages.append({ "role_id": event.data.get("role_id", ""), @@ -261,6 +298,7 @@ async def start_discussion(session_id: str, body: StartDiscussionRequest): "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) @@ -277,7 +315,7 @@ async def start_discussion(session_id: str, body: StartDiscussionRequest): logger.exception("Engine error in session %s", session.id) session.status = "error" finally: - session.bridge.stop() + session.cleanup() session._engine_task = asyncio.create_task(run_engine()) @@ -291,20 +329,20 @@ async def start_discussion(session_id: str, body: StartDiscussionRequest): @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.""" + """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) - session.messages.append({ - "role_id": msg.role_id, - "name": msg.name, - "content": msg.content, - }) return {"role_id": msg.role_id, "name": msg.name, "content": msg.content} @@ -327,9 +365,9 @@ async def list_roles(config: str | None = None): @api.get("/providers") async def list_providers(config: str | None = None): - """List configured providers.""" + """List configured providers (API keys masked).""" raw_config = _load_config_or_default(config) - return raw_config.get("providers", {}) + return _mask_api_keys(raw_config.get("providers", {})) # Mount API router @@ -351,42 +389,34 @@ async def websocket_endpoint(websocket: WebSocket, session_id: str): await websocket.accept() queue = session.bridge.subscribe() + send_task = asyncio.create_task(_ws_send_loop(websocket, queue)) try: - # Task: forward events from queue to WebSocket - async def send_events(): - try: - while True: - event = await queue.get() - await websocket.send_json(event) - except asyncio.CancelledError: - pass - - send_task = asyncio.create_task(send_events()) - - # Listen for incoming messages (inject) - try: - while True: - data = await websocket.receive_json() - if data.get("type") == "inject": - role_id = data.get("role_id", "boss") - content = data.get("content", "") - engine = session.build_engine() - msg = engine.inject_message(role_id, content) - session.messages.append({ - "role_id": msg.role_id, - "name": msg.name, - "content": msg.content, - }) - except WebSocketDisconnect: - pass - finally: - send_task.cancel() - session.bridge.unsubscribe(queue) - except Exception: + 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 # --------------------------------------------------------------------------- @@ -404,8 +434,6 @@ async def sse_endpoint(session_id: str): try: while True: event = await queue.get() - # SSE format - import json yield f"data: {json.dumps(event)}\n\n" except asyncio.CancelledError: pass diff --git a/meeting_room/static/index.html b/meeting_room/static/index.html index bc82388..c0cb2aa 100644 --- a/meeting_room/static/index.html +++ b/meeting_room/static/index.html @@ -255,12 +255,12 @@ const el = document.createElement('div'); el.className = 'msg'; const color = ROLE_COLORS[d.role_id] || '#999'; - el.innerHTML = `