from __future__ import annotations import argparse import asyncio import hashlib import json import secrets import time import uuid from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from threading import Lock from typing import Any from urllib.parse import unquote from fastapi import FastAPI, Header, HTTPException, Query, Request, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from agent import CoreAgent from core_agent.config import ( apply_compat_env_aliases, build_core_agent_config, load_core_agent_env, require_model_config, ) from core_agent.session import ConversationSession from meeting_memory.service import MeetingKnowledgeService from providers.openai_compatible import OpenAICompatibleProvider from providers.rule_based import RuleBasedMeetingProvider from tools.default_tools import build_default_registry WORKSPACE = Path(__file__).resolve().parent DATA_DIR = WORKSPACE / "data" FRONTEND_DIST_DIR = WORKSPACE / "webui" / "dist" SIDEBAR_STATE_PATH = WORKSPACE / ".webui_sidebar_state.json" UPLOADS_DIR = WORKSPACE / "agent_memory" / "uploads" def now_iso() -> str: return datetime.now().isoformat(timespec="seconds") def now_ms() -> int: return int(time.time() * 1000) def _history_to_ui_messages(history: list[dict[str, Any]]) -> list[dict[str, Any]]: messages: list[dict[str, Any]] = [] created_at = now_ms() for item in history: role = str(item.get("role", "")) content = str(item.get("content", "")) if role not in {"user", "assistant"} or not content: continue messages.append( { "id": uuid.uuid4().hex, "role": role, "content": content, "createdAt": created_at, "turnId": f"persisted_{len(messages)}", "turnPhase": "user" if role == "user" else "answer", "turnSeq": 0 if role == "user" else 99, } ) created_at += 1 return messages def split_session_key(key: str) -> tuple[str, str]: if ":" in key: channel, chat_id = key.split(":", 1) return channel, chat_id return "websocket", key def default_sidebar_state() -> dict[str, Any]: return { "schema_version": 1, "project_paths": [], "pinned_keys": [], "archived_keys": [], "title_overrides": {}, "project_name_overrides": {}, "tags_by_key": {}, "collapsed_groups": {}, "view": { "density": "comfortable", "show_previews": False, "show_timestamps": False, "show_archived": False, "sort": "updated_desc", }, "updated_at": now_iso(), } def _sanitize_upload_name(name: str, fallback_stem: str = "attachment") -> str: raw = Path(name or "").name if not raw: raw = f"{fallback_stem}.txt" safe = "".join(ch if ch not in '<>:"/\\|?*' else "_" for ch in raw).strip().strip(".") return safe or f"{fallback_stem}.txt" def _persist_text_attachments(chat_id: str, attachments: list[dict[str, Any]]) -> list[dict[str, Any]]: if not attachments: return [] target_dir = UPLOADS_DIR / chat_id target_dir.mkdir(parents=True, exist_ok=True) saved: list[dict[str, Any]] = [] for index, item in enumerate(attachments, start=1): name = _sanitize_upload_name(str(item.get("name") or f"attachment_{index}.txt")) content = str(item.get("content") or "") kind = str(item.get("kind") or "text/plain") stem = Path(name).stem or f"attachment_{index}" suffix = Path(name).suffix or ".txt" path = target_dir / f"{int(time.time() * 1000)}_{index}_{stem}{suffix}" path.write_text(content, encoding="utf-8") saved.append( { "name": name, "kind": kind, "path": str(path), "size": len(content.encode("utf-8")), } ) return saved def _augment_message_with_saved_attachments(content: str, saved_attachments: list[dict[str, Any]]) -> str: if not saved_attachments: return content lines = [content.rstrip(), "", "附加文本文件已保存，可直接读取或导入："] for item in saved_attachments: lines.append(f"- {item['name']}: {item['path']}") return "\n".join(line for line in lines if line is not None) def _validate_workspace_scope(workspace_scope: dict[str, Any] | None) -> None: project_path = str((workspace_scope or {}).get("project_path") or "").strip() if not project_path or project_path.startswith("knowledge-base://"): return candidate = Path(project_path).expanduser().resolve() if not candidate.exists() or not candidate.is_dir(): raise ValueError("project_path must be an existing directory") @dataclass(slots=True) class ChatRecord: chat_id: str session: ConversationSession title: str = "" preview: str = "" knowledge_base_id: str | None = None created_at: str = field(default_factory=now_iso) updated_at: str = field(default_factory=now_iso) ui_messages: list[dict[str, Any]] = field(default_factory=list) workspace_scope: dict[str, Any] | None = None running: bool = False current_turn_id: str | None = None run_task: asyncio.Task[Any] | None = None @property def key(self) -> str: return f"websocket:{self.chat_id}" def to_summary(self) -> dict[str, Any]: return { "key": self.key, "created_at": self.created_at, "updated_at": self.updated_at, "title": self.title, "preview": self.preview, "run_started_at": int(time.time()) if self.running else None, "workspace_scope": self.workspace_scope, "knowledge_base_id": self.knowledge_base_id, } class ChatStore: def __init__(self, agent: CoreAgent, data_dir: Path) -> None: self._agent = agent self._data_dir = data_dir self._lock = Lock() self._records: dict[str, ChatRecord] = {} def _seed_chat_id(self, knowledge_base_id: str) -> str: digest = hashlib.md5(knowledge_base_id.encode("utf-8")).hexdigest()[:12] return f"kb_{digest}" def _default_preview(self, knowledge_base_id: str | None) -> str: if knowledge_base_id: return f"围绕知识库「{knowledge_base_id}」开始提问" return "" def _new_session(self, knowledge_base_id: str | None = None) -> ConversationSession: session_id = knowledge_base_id or f"chat_{uuid.uuid4().hex[:12]}" session = self._agent.new_session(session_id=session_id) if knowledge_base_id: session.session_state["last_knowledge_base_id"] = knowledge_base_id session.session_state["last_team_id"] = knowledge_base_id return session def _create_record(self, chat_id: str, knowledge_base_id: str | None, title: str = "") -> ChatRecord: session = self._new_session(knowledge_base_id or chat_id) record = ChatRecord( chat_id=chat_id, session=session, title=title or (knowledge_base_id or ""), preview=self._default_preview(knowledge_base_id), knowledge_base_id=knowledge_base_id, workspace_scope=None, ) record.ui_messages = _history_to_ui_messages(session.persisted_dialog_messages()) if record.ui_messages and not record.preview: record.preview = str(record.ui_messages[-1].get("content", ""))[:120] self._records[chat_id] = record return record def _apply_workspace_scope(self, record: ChatRecord, workspace_scope: dict[str, Any] | None) -> None: record.workspace_scope = workspace_scope project_path = str((workspace_scope or {}).get("project_path") or "").strip() if project_path and not project_path.startswith("knowledge-base://"): candidate = Path(project_path).expanduser().resolve() record.session.workspace = candidate record.session.session_state["workspace"] = str(candidate) else: record.session.workspace = WORKSPACE.resolve() record.session.session_state["workspace"] = str(WORKSPACE.resolve()) def ensure_seed_records(self) -> None: meeting_service = MeetingKnowledgeService(self._data_dir, workspace=WORKSPACE) for knowledge_base_id in meeting_service.list_knowledge_bases().get("knowledge_bases", []): chat_id = self._seed_chat_id(knowledge_base_id) with self._lock: if chat_id not in self._records: self._create_record(chat_id, knowledge_base_id, title=knowledge_base_id) def list_records(self) -> list[ChatRecord]: self.ensure_seed_records() with self._lock: return sorted( self._records.values(), key=lambda item: (item.updated_at, item.created_at, item.title), reverse=True, ) def create_chat(self, knowledge_base_id: str | None = None) -> ChatRecord: with self._lock: chat_id = f"chat_{uuid.uuid4().hex[:12]}" return self._create_record(chat_id, knowledge_base_id) def fork_chat(self, source_chat_id: str, before_user_index: int) -> ChatRecord: with self._lock: source = self._records[source_chat_id] fork = self._create_record( f"chat_{uuid.uuid4().hex[:12]}", source.knowledge_base_id, title=source.title, ) user_seen = 0 copied: list[dict[str, Any]] = [] for message in source.ui_messages: if message.get("role") == "user": if user_seen >= before_user_index: break user_seen += 1 copied.append(json.loads(json.dumps(message))) fork.ui_messages = copied fork.preview = source.preview return fork def get_by_chat_id(self, chat_id: str) -> ChatRecord: self.ensure_seed_records() with self._lock: if chat_id not in self._records: return self._create_record(chat_id, knowledge_base_id=None) return self._records[chat_id] def get_by_key(self, key: str) -> ChatRecord: _, chat_id = split_session_key(key) return self.get_by_chat_id(chat_id) def set_workspace_scope(self, chat_id: str, workspace_scope: dict[str, Any] | None) -> ChatRecord: record = self.get_by_chat_id(chat_id) self._apply_workspace_scope(record, workspace_scope) record.updated_at = now_iso() return record def set_knowledge_base(self, chat_id: str, knowledge_base_id: str | None) -> ChatRecord: record = self.get_by_chat_id(chat_id) cleaned = (knowledge_base_id or "").strip() or None record.knowledge_base_id = cleaned if cleaned: record.session.session_state["last_knowledge_base_id"] = cleaned record.session.session_state["last_team_id"] = cleaned else: record.session.session_state.pop("last_knowledge_base_id", None) record.session.session_state.pop("last_team_id", None) record.updated_at = now_iso() return record def append_user_message( self, record: ChatRecord, content: str, turn_id: str | None, *, media: list[dict[str, Any]] | None = None, ) -> None: text = content.strip() record.updated_at = now_iso() if text: record.preview = text[:120] if not record.title and not record.knowledge_base_id: record.title = text[:32] record.ui_messages.append( { "id": uuid.uuid4().hex, "role": "user", "content": content, "createdAt": now_ms(), "turnId": turn_id, "turnPhase": "user", "turnSeq": 0, **({"media": media} if media else {}), } ) def append_assistant_message(self, record: ChatRecord, content: str, turn_id: str | None, latency_ms: int) -> None: record.updated_at = now_iso() record.ui_messages.append( { "id": uuid.uuid4().hex, "role": "assistant", "content": content, "createdAt": now_ms(), "turnId": turn_id, "turnPhase": "answer", "turnSeq": 99, "latencyMs": latency_ms, } ) class GatewayState: def __init__(self, *, agent: CoreAgent, config: Any, mode: str) -> None: self.agent = agent self.config = config self.mode = mode self.chat_store = ChatStore(agent, DATA_DIR) self.token = secrets.token_urlsafe(24) self.sidebar_state = self._load_sidebar_state() self.socket_chats: dict[int, set[str]] = {} self.chat_sockets: dict[str, set[WebSocket]] = {} def _load_sidebar_state(self) -> dict[str, Any]: if SIDEBAR_STATE_PATH.exists(): try: return json.loads(SIDEBAR_STATE_PATH.read_text(encoding="utf-8")) except Exception: pass return default_sidebar_state() def save_sidebar_state(self, state: dict[str, Any]) -> dict[str, Any]: state["updated_at"] = now_iso() self.sidebar_state = state SIDEBAR_STATE_PATH.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8") return state def register_socket_chat(self, websocket: WebSocket, chat_id: str) -> None: socket_key = id(websocket) self.socket_chats.setdefault(socket_key, set()).add(chat_id) self.chat_sockets.setdefault(chat_id, set()).add(websocket) def unregister_socket(self, websocket: WebSocket) -> None: socket_key = id(websocket) chat_ids = self.socket_chats.pop(socket_key, set()) for chat_id in chat_ids: sockets = self.chat_sockets.get(chat_id) if not sockets: continue sockets.discard(websocket) if not sockets: self.chat_sockets.pop(chat_id, None) async def send_json(self, websocket: WebSocket, payload: dict[str, Any]) -> None: await websocket.send_text(json.dumps(payload, ensure_ascii=False)) async def broadcast_chat(self, chat_id: str, payload: dict[str, Any]) -> None: sockets = list(self.chat_sockets.get(chat_id, set())) stale: list[WebSocket] = [] for websocket in sockets: try: await self.send_json(websocket, payload) except Exception: stale.append(websocket) for websocket in stale: self.unregister_socket(websocket) def build_provider(force_offline: bool = False): load_core_agent_env(WORKSPACE) apply_compat_env_aliases() config = build_core_agent_config() if force_offline: return RuleBasedMeetingProvider(), config, "offline" require_model_config(config) return ( OpenAICompatibleProvider( model=config.model, api_key=config.api_key, base_url=config.base_url, timeout=config.timeout, temperature=config.temperature, ), config, "openai-compatible", ) def build_agent(force_offline: bool = False) -> tuple[CoreAgent, Any, str]: provider, config, mode = build_provider(force_offline=force_offline) agent = CoreAgent( provider=provider, workspace=WORKSPACE, skill_dirs=[WORKSPACE / "skills"], tool_registry=build_default_registry(DATA_DIR), max_iterations=config.max_iterations, ) return agent, config, mode def list_knowledge_base_cards(data_dir: Path) -> list[dict[str, Any]]: cards: list[dict[str, Any]] = [] if not data_dir.exists(): return cards for path in sorted(data_dir.iterdir()): if not path.is_dir(): continue state_path = path / "state.json" if not state_path.exists(): continue state = json.loads(state_path.read_text(encoding="utf-8")) meetings = state.get("meetings", []) latest = meetings[-1] if meetings else {} cards.append( { "knowledge_base_id": path.name, "title": path.name, "meeting_count": len(meetings), "last_meeting_date": state.get("last_meeting_date", ""), "updated_at": state.get("last_updated", ""), "latest_summary": latest.get("summary", [])[:3], "counts": { "metrics": len(state.get("current_metrics", [])), "actions": len(state.get("open_actions", [])), "risks": len(state.get("risks", [])), "followups": len(state.get("followups", [])), }, } ) return cards def default_knowledge_base_id(data_dir: Path) -> str | None: cards = list_knowledge_base_cards(data_dir) if len(cards) == 1: return cards[0]["knowledge_base_id"] return None def _read_auth_token(request: Request, authorization: str | None = Header(default=None)) -> str | None: if authorization and authorization.lower().startswith("bearer "): return authorization.split(" ", 1)[1].strip() token = request.query_params.get("token") return token.strip() if token else None def _require_token(request: Request, gateway: GatewayState, authorization: str | None = Header(default=None)) -> None: token = _read_auth_token(request, authorization) if token != gateway.token: raise HTTPException(status_code=401, detail="Unauthorized") def _minimal_settings(gateway: GatewayState) -> dict[str, Any]: model_name = getattr(gateway.config, "model", "") or "" provider_name = "openai-compatible" if gateway.mode != "offline" else "offline" return { "agent": { "model": model_name, "provider": provider_name, "resolved_provider": provider_name, "has_api_key": bool(getattr(gateway.config, "api_key", "") or gateway.mode == "offline"), "model_preset": "default", "max_tokens": 0, "context_window_tokens": 0, "temperature": getattr(gateway.config, "temperature", 0.2), "reasoning_effort": None, "timezone": "Asia/Shanghai", "bot_name": "MeetingAgent", "bot_icon": "MK", "tool_hint_max_length": 160, }, "model_presets": [ { "name": "default", "label": "Default", "active": True, "is_default": True, "model": model_name, "provider": provider_name, "max_tokens": 0, "context_window_tokens": 0, "temperature": getattr(gateway.config, "temperature", 0.2), "reasoning_effort": None, "reasoning_effort_values": [], } ], "providers": [ { "name": provider_name, "label": provider_name, "configured": True, "auth_type": "api_key", "api_key_required": gateway.mode != "offline", "api_key_hint": "OPENAI_API_KEY", "api_base": getattr(gateway.config, "base_url", None), "default_api_base": getattr(gateway.config, "base_url", None), "model_selectable": True, "api_type": "chat_completions", } ], "web_search": {"provider": "none", "max_results": 5, "timeout": 20, "providers": []}, "web": {"enable": False, "search": {"max_results": 5, "timeout": 20}, "fetch": {"use_jina_reader": False}}, "image_generation": { "enabled": False, "provider": "none", "provider_configured": False, "model": "", "default_aspect_ratio": "1:1", "default_image_size": "1024x1024", "max_images_per_turn": 1, "save_dir": str(WORKSPACE / "generated_images"), "providers": [], }, "transcription": { "enabled": False, "provider": "none", "provider_configured": False, "model": "", "language": "zh", "max_duration_sec": 0, "max_upload_mb": 0, "providers": [], }, "runtime": { "config_path": str(WORKSPACE / ".env"), "workspace_path": str(WORKSPACE), "gateway_host": "127.0.0.1", "gateway_port": 8010, "heartbeat": {"enabled": False, "interval_s": 0, "keep_recent_messages": 0}, "dream": {"schedule": ""}, "unified_session": False, }, "advanced": { "restrict_to_workspace": True, "webui_allow_local_service_access": True, "allow_local_preview_access": True, "webui_default_access_mode": "default", "private_service_protection_enabled": False, "ssrf_whitelist_count": 0, "mcp_server_count": 0, "exec_enabled": False, "exec_sandbox": "workspace-write", "exec_path_prepend_set": False, "exec_path_append_set": False, }, "requires_restart": False, "runtime_surface": "browser", "runtime_capabilities": { "can_restart_engine": False, "can_pick_folder": False, "can_open_logs": False, "can_export_diagnostics": False, }, "version": {"current": "0.1.0"}, } def create_app(force_offline: bool = False) -> FastAPI: agent, config, mode = build_agent(force_offline=force_offline) gateway = GatewayState(agent=agent, config=config, mode=mode) app = FastAPI(title="Meeting Knowledge WebUI Gateway", version="0.2.0") app.state.gateway = gateway app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) if FRONTEND_DIST_DIR.exists(): assets_dir = FRONTEND_DIST_DIR / "assets" brand_dir = FRONTEND_DIST_DIR / "brand" if assets_dir.exists(): app.mount("/assets", StaticFiles(directory=assets_dir), name="assets") if brand_dir.exists(): app.mount("/brand", StaticFiles(directory=brand_dir), name="brand") @app.get("/webui/bootstrap") def webui_bootstrap() -> dict[str, Any]: return { "token": gateway.token, "ws_path": "/ws", "expires_in": 24 * 3600, "model_name": getattr(config, "model", "") or None, "runtime_surface": "browser", "runtime_capabilities": { "can_restart_engine": False, "can_pick_folder": False, "can_open_logs": False, "can_export_diagnostics": False, }, } @app.get("/api/runtime") def runtime_info() -> dict[str, Any]: return { "mode": mode, "model_name": getattr(config, "model", "") or None, "default_knowledge_base_id": default_knowledge_base_id(DATA_DIR), "workspace": str(WORKSPACE), } @app.get("/api/knowledge-bases") def list_knowledge_bases() -> dict[str, Any]: cards = list_knowledge_base_cards(DATA_DIR) return { "knowledge_bases": cards, "default_knowledge_base_id": default_knowledge_base_id(DATA_DIR), } @app.get("/api/sessions") def list_sessions(request: Request, authorization: str | None = Header(default=None)) -> dict[str, Any]: _require_token(request, gateway, authorization) rows = [record.to_summary() for record in gateway.chat_store.list_records()] return {"sessions": rows} @app.get("/api/sessions/{session_key}/webui-thread") def fetch_webui_thread(session_key: str, request: Request, authorization: str | None = Header(default=None)) -> dict[str, Any]: _require_token(request, gateway, authorization) record = gateway.chat_store.get_by_key(unquote(session_key)) return { "schemaVersion": 1, "sessionKey": record.key, "savedAt": record.updated_at, "messages": record.ui_messages, "fork_boundary_message_count": None, "has_pending_tool_calls": record.running, "page": { "before_cursor": None, "has_more_before": False, "loaded_message_count": len(record.ui_messages), "total_known_message_count": len(record.ui_messages), "user_message_offset": 0, }, "workspace_scope": record.workspace_scope, } @app.get("/api/sessions/{session_key}/automations") def fetch_session_automations(session_key: str, request: Request, authorization: str | None = Header(default=None)) -> dict[str, Any]: _require_token(request, gateway, authorization) return {"jobs": []} @app.get("/api/webui/automations") def fetch_automations(request: Request, authorization: str | None = Header(default=None)) -> dict[str, Any]: _require_token(request, gateway, authorization) return {"jobs": []} @app.get("/api/webui/skills") def fetch_skills(request: Request, authorization: str | None = Header(default=None)) -> dict[str, Any]: _require_token(request, gateway, authorization) return {"skills": []} @app.get("/api/settings") def fetch_settings(request: Request, authorization: str | None = Header(default=None)) -> dict[str, Any]: _require_token(request, gateway, authorization) return _minimal_settings(gateway) @app.get("/api/settings/usage") def fetch_settings_usage(request: Request, authorization: str | None = Header(default=None)) -> dict[str, Any]: _require_token(request, gateway, authorization) return { "days": [], "total_tokens": 0, "total_tokens_30d": 0, "total_tokens_365d": 0, "peak_day_tokens": 0, "current_streak_days": 0, "longest_streak_days": 0, "active_days_30d": 0, "requests_30d": 0, "updated_at": now_iso(), } @app.get("/api/settings/version-check") def check_version(request: Request, authorization: str | None = Header(default=None)) -> dict[str, Any]: _require_token(request, gateway, authorization) return {"updateAvailable": None} @app.get("/api/settings/provider-models") def provider_models(request: Request, authorization: str | None = Header(default=None), provider: str = Query(default="openai-compatible")) -> dict[str, Any]: _require_token(request, gateway, authorization) model_name = getattr(config, "model", "") or "" return { "provider": provider, "label": provider, "status": "available", "catalog_kind": "local", "models": [{"id": model_name, "label": model_name, "owned_by": provider}], "model_count": 1 if model_name else 0, "fetched_at": int(time.time()), } @app.get("/api/settings/cli-apps") def cli_apps(request: Request, authorization: str | None = Header(default=None)) -> dict[str, Any]: _require_token(request, gateway, authorization) return {"apps": [], "installed_count": 0} @app.get("/api/settings/mcp-presets") def mcp_presets(request: Request, authorization: str | None = Header(default=None)) -> dict[str, Any]: _require_token(request, gateway, authorization) return {"presets": [], "installed_count": 0} @app.get("/api/commands") def list_commands(request: Request, authorization: str | None = Header(default=None)) -> dict[str, Any]: _require_token(request, gateway, authorization) return {"commands": []} @app.get("/api/workspaces") def workspaces(request: Request, authorization: str | None = Header(default=None)) -> dict[str, Any]: _require_token(request, gateway, authorization) return { "schema_version": 1, "default_access_mode": "default", "default_scope": { "project_path": str(WORKSPACE), "project_name": WORKSPACE.name, "access_mode": "restricted", "restrict_to_workspace": True, }, "controls": { "can_change_project": True, "can_use_full_access": False, }, } @app.get("/api/webui/sidebar-state") def fetch_sidebar_state(request: Request, authorization: str | None = Header(default=None)) -> dict[str, Any]: _require_token(request, gateway, authorization) return gateway.sidebar_state @app.get("/api/webui/sidebar-state/update") def update_sidebar_state( request: Request, state: str, authorization: str | None = Header(default=None), ) -> dict[str, Any]: _require_token(request, gateway, authorization) payload = json.loads(state) return gateway.save_sidebar_state(payload) @app.get("/api/sessions/{session_key}/delete") def delete_session(session_key: str, request: Request, authorization: str | None = Header(default=None)) -> dict[str, Any]: _require_token(request, gateway, authorization) record = gateway.chat_store.get_by_key(unquote(session_key)) with gateway.chat_store._lock: gateway.chat_store._records.pop(record.chat_id, None) return {"deleted": True} @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket, token: str = Query(default="")) -> None: if token != gateway.token: await websocket.close(code=4401) return await websocket.accept() client_id = uuid.uuid4().hex await gateway.send_json(websocket, {"event": "ready", "chat_id": "home", "client_id": client_id}) try: while True: raw = await websocket.receive_text() payload = json.loads(raw) frame_type = payload.get("type") if frame_type == "new_chat": workspace_scope = payload.get("workspace_scope") knowledge_base_id = str(payload.get("knowledge_base_id") or "").strip() or None try: _validate_workspace_scope(workspace_scope) except ValueError as exc: await gateway.send_json( websocket, { "event": "error", "chat_id": "home", "detail": "workspace_scope_rejected", "reason": str(exc), }, ) continue record = gateway.chat_store.create_chat(knowledge_base_id=knowledge_base_id) if workspace_scope: gateway.chat_store.set_workspace_scope(record.chat_id, workspace_scope) gateway.register_socket_chat(websocket, record.chat_id) await gateway.send_json(websocket, {"event": "attached", "chat_id": record.chat_id}) await gateway.broadcast_chat( record.chat_id, { "event": "session_updated", "chat_id": record.chat_id, "scope": "metadata", "workspace_scope": record.workspace_scope, "knowledge_base_id": record.knowledge_base_id, }, ) continue if frame_type == "fork_chat": source_chat_id = str(payload.get("source_chat_id") or "") before_user_index = int(payload.get("before_user_index") or 0) record = gateway.chat_store.fork_chat(source_chat_id, before_user_index) gateway.register_socket_chat(websocket, record.chat_id) await gateway.send_json(websocket, {"event": "attached", "chat_id": record.chat_id}) await gateway.broadcast_chat( record.chat_id, { "event": "session_updated", "chat_id": record.chat_id, "scope": "thread", "workspace_scope": record.workspace_scope, "knowledge_base_id": record.knowledge_base_id, }, ) continue if frame_type == "attach": chat_id = str(payload.get("chat_id") or "") record = gateway.chat_store.get_by_chat_id(chat_id) gateway.register_socket_chat(websocket, chat_id) await gateway.send_json(websocket, {"event": "attached", "chat_id": record.chat_id}) continue if frame_type == "set_workspace_scope": chat_id = str(payload.get("chat_id") or "") workspace_scope = payload.get("workspace_scope") try: record = gateway.chat_store.set_workspace_scope(chat_id, workspace_scope) except ValueError as exc: await gateway.broadcast_chat( chat_id, { "event": "error", "chat_id": chat_id, "detail": "workspace_scope_rejected", "reason": str(exc), }, ) continue await gateway.broadcast_chat( chat_id, { "event": "session_updated", "chat_id": chat_id, "scope": "metadata", "workspace_scope": record.workspace_scope, "knowledge_base_id": record.knowledge_base_id, }, ) continue if frame_type == "message": chat_id = str(payload.get("chat_id") or "") record = gateway.chat_store.get_by_chat_id(chat_id) gateway.register_socket_chat(websocket, chat_id) content = str(payload.get("content") or "") if content.strip() == "/stop": if record.running: record.session.request_cancel() else: await gateway.broadcast_chat( chat_id, { "event": "turn_end", "chat_id": chat_id, "latency_ms": 0, }, ) await gateway.broadcast_chat( chat_id, { "event": "goal_status", "chat_id": chat_id, "status": "idle", }, ) continue if record.running: await gateway.broadcast_chat( chat_id, { "event": "error", "chat_id": chat_id, "detail": "A turn is already running for this chat.", }, ) continue knowledge_base_id = str(payload.get("knowledge_base_id") or "").strip() or None attachments = payload.get("attachments") attachment_list = attachments if isinstance(attachments, list) else [] saved_attachments = _persist_text_attachments(chat_id, attachment_list) content_for_agent = _augment_message_with_saved_attachments(content, saved_attachments) user_media = [ { "kind": "file", "name": item.get("name"), } for item in saved_attachments ] if "knowledge_base_id" in payload: record = gateway.chat_store.set_knowledge_base(chat_id, knowledge_base_id) turn_id = payload.get("turn_id") or uuid.uuid4().hex gateway.chat_store.append_user_message( record, content, turn_id, media=user_media or None, ) await gateway.broadcast_chat( chat_id, { "event": "goal_status", "chat_id": chat_id, "status": "running", "started_at": time.time(), }, ) await gateway.broadcast_chat( chat_id, { "event": "session_updated", "chat_id": chat_id, "scope": "thread", "workspace_scope": record.workspace_scope, "knowledge_base_id": record.knowledge_base_id, }, ) record.run_task = asyncio.create_task(run_agent_turn(gateway, record, content_for_agent, turn_id)) continue except WebSocketDisconnect: gateway.unregister_socket(websocket) except Exception: gateway.unregister_socket(websocket) try: await websocket.close(code=1011) except Exception: pass @app.get("/{full_path:path}", response_model=None) def serve_frontend(full_path: str): if not FRONTEND_DIST_DIR.exists(): return JSONResponse( status_code=503, content={ "detail": "Meeting Knowledge WebUI build is missing. Run `npm install` and `npm run build` in frontend_nanobot first." }, ) index_path = FRONTEND_DIST_DIR / "index.html" return FileResponse(index_path) return app async def run_agent_turn(gateway: GatewayState, record: ChatRecord, content: str, turn_id: str) -> None: record.running = True record.current_turn_id = turn_id record.session.clear_cancel() if record.knowledge_base_id: record.session.session_state["last_knowledge_base_id"] = record.knowledge_base_id record.session.session_state["last_team_id"] = record.knowledge_base_id started_at = time.time() try: loop = asyncio.get_running_loop() queue: asyncio.Queue[object] = asyncio.Queue() sentinel = object() def live_event_sink(event: dict[str, Any]) -> None: loop.call_soon_threadsafe(queue.put_nowait, event) record.session.session_state["_live_event_sink"] = live_event_sink def produce_events() -> None: try: for event in record.session.stream_ask(content): loop.call_soon_threadsafe(queue.put_nowait, event) except Exception as exc: loop.call_soon_threadsafe(queue.put_nowait, exc) finally: loop.call_soon_threadsafe(queue.put_nowait, sentinel) producer_task = asyncio.create_task(asyncio.to_thread(produce_events)) tool_seq = 1 final_text = "" stream_open = False content_seen = False while True: item = await queue.get() if item is sentinel: break if isinstance(item, Exception): raise item if isinstance(item, dict) and item.get("kind") == "subagent_progress": payload = item.get("payload") or {} text = str(payload.get("text") or "") if text: await gateway.broadcast_chat( record.chat_id, { "event": "message", "chat_id": record.chat_id, "text": text, "kind": "progress", "agent_ui": { "kind": str(payload.get("kind") or "subagent_status"), "data": payload, }, "turn_id": turn_id, "turn_phase": "activity", "turn_seq": tool_seq, }, ) continue event = item if event.type == "reasoning_delta" and event.delta: await gateway.broadcast_chat( record.chat_id, { "event": "reasoning_delta", "chat_id": record.chat_id, "text": event.delta, "stream_id": f"{turn_id}:{event.iteration}", "turn_id": turn_id, "turn_phase": "reasoning", "turn_seq": max(1, event.iteration * 10 - 2), }, ) stream_open = True if gateway.mode == "offline": await asyncio.sleep(0.012) elif event.type == "content_delta" and event.delta: content_seen = True await gateway.broadcast_chat( record.chat_id, { "event": "delta", "chat_id": record.chat_id, "text": event.delta, "stream_id": f"{turn_id}:{event.iteration}", "turn_id": turn_id, "turn_phase": "answer", "turn_seq": max(1, event.iteration * 10 - 1), }, ) stream_open = True if gateway.mode == "offline": await asyncio.sleep(0.012) elif event.type == "assistant_turn": if stream_open: await gateway.broadcast_chat( record.chat_id, { "event": "reasoning_end", "chat_id": record.chat_id, "stream_id": f"{turn_id}:{event.iteration}", "turn_id": turn_id, "turn_phase": "reasoning", "turn_seq": max(1, event.iteration * 10), }, ) await gateway.broadcast_chat( record.chat_id, { "event": "stream_end", "chat_id": record.chat_id, "stream_id": f"{turn_id}:{event.iteration}", "text": event.turn_content or "", "turn_id": turn_id, "turn_phase": "answer", "turn_seq": max(1, event.iteration * 10 + 1), }, ) stream_open = False elif event.type == "tool_call": await gateway.broadcast_chat( record.chat_id, { "event": "message", "chat_id": record.chat_id, "text": "", "kind": "tool_hint", "tool_events": [ { "phase": "start", "call_id": f"{turn_id}:{tool_seq}", "name": event.tool_name, "arguments": event.tool_args, } ], "turn_id": turn_id, "turn_phase": "activity", "turn_seq": tool_seq, }, ) tool_seq += 1 elif event.type == "tool_result": result_obj: Any try: result_obj = json.loads(event.tool_result) except Exception: result_obj = event.tool_result await gateway.broadcast_chat( record.chat_id, { "event": "message", "chat_id": record.chat_id, "text": "", "kind": "progress", "tool_events": [ { "phase": "end", "call_id": f"{turn_id}:{max(tool_seq - 1, 1)}", "name": event.tool_name, "arguments": event.tool_args, "result": result_obj, } ], "turn_id": turn_id, "turn_phase": "activity", "turn_seq": tool_seq, }, ) elif event.type == "final": final_text = event.final_response or "" await producer_task latency_ms = max(1, int((time.time() - started_at) * 1000)) gateway.chat_store.append_assistant_message(record, final_text, turn_id, latency_ms) if not content_seen: await gateway.broadcast_chat( record.chat_id, { "event": "message", "chat_id": record.chat_id, "text": final_text, "turn_id": turn_id, "turn_phase": "answer", "turn_seq": tool_seq + 1, "latency_ms": latency_ms, }, ) await gateway.broadcast_chat( record.chat_id, { "event": "turn_end", "chat_id": record.chat_id, "turn_id": turn_id, "turn_phase": "complete", "turn_seq": tool_seq + 2, "latency_ms": latency_ms, "goal_state": {"active": False, "ui_summary": ""}, }, ) await gateway.broadcast_chat( record.chat_id, { "event": "goal_status", "chat_id": record.chat_id, "status": "idle", }, ) await gateway.broadcast_chat( record.chat_id, { "event": "session_updated", "chat_id": record.chat_id, "scope": "thread", "workspace_scope": record.workspace_scope, "knowledge_base_id": record.knowledge_base_id, }, ) except asyncio.CancelledError: await gateway.broadcast_chat( record.chat_id, { "event": "turn_end", "chat_id": record.chat_id, "turn_id": turn_id, "turn_phase": "complete", "turn_seq": tool_seq + 2, "latency_ms": 0, "goal_state": {"active": False, "ui_summary": ""}, }, ) await gateway.broadcast_chat( record.chat_id, { "event": "goal_status", "chat_id": record.chat_id, "status": "idle", }, ) await gateway.broadcast_chat( record.chat_id, { "event": "session_updated", "chat_id": record.chat_id, "scope": "thread", "workspace_scope": record.workspace_scope, "knowledge_base_id": record.knowledge_base_id, }, ) raise except Exception as exc: await gateway.broadcast_chat( record.chat_id, { "event": "error", "chat_id": record.chat_id, "detail": str(exc), }, ) await gateway.broadcast_chat( record.chat_id, { "event": "goal_status", "chat_id": record.chat_id, "status": "idle", }, ) finally: record.session.session_state.pop("_live_event_sink", None) record.running = False record.current_turn_id = None record.run_task = None record.updated_at = now_iso() def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Meeting Knowledge WebUI server") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8010) parser.add_argument("--offline", action="store_true") return parser.parse_args() def main() -> None: args = parse_args() import uvicorn uvicorn.run( create_app(force_offline=args.offline), host=args.host, port=args.port, log_level="info", ) if __name__ == "__main__": main()