vitya
f0500502a7
- inject_message now uses session's event_bus (BOSS_TURN reaches WebSocket/SSE) - asyncio.get_running_loop() replaces deprecated get_event_loop() - API keys masked in /api/providers response - ServerSession (renamed from Session) with cleanup() for handler teardown - Path traversal protection in _resolve_scenario_path - XSS fix: escapeHtml(d.name) in web UI - Duplicate imports removed, os/json moved to module level - WebSocket handler restructured for reliable cleanup - New tests: cleanup, provider masking, path traversal Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
451 lines
15 KiB
Python
451 lines
15 KiB
Python
"""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",
|
|
},
|
|
) |