"""会话与历史路由。 所有 DB 访问都走 ``asyncio.to_thread``:pymysql/sqlite3 都是同步驱动, 直接在协程里跑会阻塞事件循环——单进程下这意味着「一个用户查历史,所有人 的流都卡住」。这是 1.5 踩过的坑,2.0 从一开始就分线程。 """ from __future__ import annotations import asyncio from typing import Any from fastapi import APIRouter, HTTPException from agents import agent_memory from api.models import ModeUpdateRequest, PinUpdateRequest, TitleUpdateRequest from db import chat_store from obs import get_logger logger = get_logger(__name__) router = APIRouter(tags=["会话"]) VALID_MODES = ("plan", "full") def _require_session(session_id: str, user_name: str) -> dict[str, Any]: """取会话并校验归属;不存在或不属于该用户一律 404。""" session = chat_store.get_session(session_id) if session is None or session.get("user_name") != user_name: raise HTTPException( status_code=404, detail={"code": "session_not_found", "message": "会话不存在"} ) return session @router.get("/sessions") async def list_sessions(user_name: str | None = None, limit: int = 50, offset: int = 0): """列出会话(置顶优先、更新时间倒序)。""" user = (user_name or "default").strip() or "default" items = await asyncio.to_thread(chat_store.list_sessions, user, min(limit, 200), max(offset, 0)) return {"user_name": user, "items": items, "count": len(items)} @router.get("/sessions/search") async def search_sessions(q: str, user_name: str | None = None, limit: int = 50): """按标题模糊搜索会话。""" user = (user_name or "default").strip() or "default" items = await asyncio.to_thread(chat_store.search_sessions, user, q, min(limit, 200)) return {"user_name": user, "keyword": q, "items": items, "count": len(items)} @router.get("/chat/history/{session_id}") async def chat_history( session_id: str, user_name: str | None = None, limit: int = 100, offset: int = 0 ): """取会话历史消息(时间正序)。""" user = (user_name or "default").strip() or "default" await asyncio.to_thread(_require_session, session_id, user) items = await asyncio.to_thread( chat_store.get_messages, session_id, min(limit, 500), max(offset, 0) ) from db import business_store name = await asyncio.to_thread(business_store.session_agent, session_id) return {"session_id": session_id, "agent_name": name or "main", "items": items, "count": len(items)} @router.delete("/sessions/{session_id}") async def delete_session(session_id: str, user_name: str | None = None): """删除会话。 必须**同时**清 checkpoint 线程:会话元数据删了但图状态还在的话, 下次若有同 ID 的会话(或用户重建)会读到上一个会话的上下文—— 表现为「新会话里 agent 记得不该记得的事」。 """ user = (user_name or "default").strip() or "default" await asyncio.to_thread(_require_session, session_id, user) from runtime.concurrency import session_guard with session_guard(session_id): await agent_memory.delete_thread(session_id) await asyncio.to_thread(chat_store.delete_session, session_id) return {"deleted": True, "session_id": session_id} @router.delete("/messages/{message_id}") async def delete_message(message_id: str, user_name: str | None = None): """归属校验后删除历史,并清除图状态以便下次按新历史重建。""" from db import db_conn from runtime.concurrency import session_guard user = (user_name or "default").strip() or "default" message = await asyncio.to_thread(db_conn.query_one, "SELECT m.* FROM messages m JOIN sessions s ON s.id = m.session_id WHERE m.id = ? AND s.user_name = ?", (message_id, user)) if not message: raise HTTPException(status_code=404, detail={"code": "message_not_found", "message": "消息不存在"}) with session_guard(message["session_id"]): from agents import agent_registry from runtime.interrupts import pending_interrupts agent = await asyncio.to_thread(agent_registry.get_agent) if await pending_interrupts(agent, {"configurable": {"thread_id": message["session_id"]}}): raise HTTPException(status_code=409, detail={"code": "pending_interrupt", "message": "请先完成或拒绝待确认操作,再删除消息"}) await agent_memory.delete_thread(message["session_id"]) deleted = await asyncio.to_thread(chat_store.delete_message, message_id) return {"deleted": deleted, "message_id": message_id} @router.get("/sessions/{session_id}/runs") async def execution_runs(session_id: str, user_name: str | None = None): from db import run_store await asyncio.to_thread(_require_session, session_id, (user_name or "default").strip() or "default") return {"items": await asyncio.to_thread(run_store.list_runs, session_id)} @router.get("/sessions/{session_id}/mode") async def get_mode(session_id: str, user_name: str | None = None): """读会话的权限模式。""" user = (user_name or "default").strip() or "default" session = await asyncio.to_thread(_require_session, session_id, user) return {"session_id": session_id, "mode": session.get("mode") or "full"} @router.put("/sessions/{session_id}/mode") async def put_mode(session_id: str, payload: ModeUpdateRequest, user_name: str | None = None): """设置会话的权限模式。""" user = (user_name or "default").strip() or "default" await asyncio.to_thread(_require_session, session_id, user) mode = payload.mode.strip().lower() if mode not in VALID_MODES: raise HTTPException( status_code=400, detail={"code": "invalid_mode", "message": f"mode 只能是 {' | '.join(VALID_MODES)}"}, ) await asyncio.to_thread(chat_store.update_session_mode, session_id, mode) return {"session_id": session_id, "mode": mode} @router.put("/sessions/{session_id}/pin") async def put_pin(session_id: str, payload: PinUpdateRequest, user_name: str | None = None): """设置/取消置顶。""" user = (user_name or "default").strip() or "default" await asyncio.to_thread(_require_session, session_id, user) await asyncio.to_thread(chat_store.update_session_pin, session_id, payload.pinned) return {"session_id": session_id, "pinned": payload.pinned} @router.put("/sessions/{session_id}/title") async def put_title(session_id: str, payload: TitleUpdateRequest, user_name: str | None = None): """重命名会话。""" user = (user_name or "default").strip() or "default" await asyncio.to_thread(_require_session, session_id, user) await asyncio.to_thread(chat_store.update_session_title, session_id, payload.title) return {"session_id": session_id, "title": payload.title} @router.get("/sessions/{session_id}/context") async def get_context(session_id: str, user_name: str | None = None, limit: int = 50): """上下文用量明细。 口径以模型返回的真实 usage 为准,``usage_source=estimate`` 的那几条是 provider 没返回用量时的本地估算,可信度较低(前端应区别展示)。 """ user = (user_name or "default").strip() or "default" await asyncio.to_thread(_require_session, session_id, user) usage = await asyncio.to_thread(chat_store.get_context_usage, session_id, min(limit, 200)) archives = await asyncio.to_thread(chat_store.list_context_archives, session_id) from agents import model_factory return { **usage, "context_window": model_factory.context_window(), "archives": archives, } # ============================================================ # 用户偏好 # ============================================================ @router.get("/preferences") async def list_preferences(user_name: str | None = None): """列出用户偏好。""" user = (user_name or "default").strip() or "default" items = await asyncio.to_thread(chat_store.list_preferences, user) return {"user_name": user, "items": items} @router.put("/preferences") async def put_preference( key: str, value: str, user_name: str | None = None ): """写入/覆盖一条偏好。每轮对话会自动注入,所以只放长期有效的设定。""" user = (user_name or "default").strip() or "default" if not key.strip(): raise HTTPException(status_code=400, detail={"code": "invalid_key", "message": "key 不能为空"}) await asyncio.to_thread(chat_store.save_preference, user, key.strip(), value) return {"user_name": user, "key": key.strip(), "value": value} @router.delete("/preferences/{key}") async def delete_preference(key: str, user_name: str | None = None): """删除一条偏好。""" user = (user_name or "default").strip() or "default" deleted = await asyncio.to_thread(chat_store.delete_preference, user, key) if not deleted: raise HTTPException( status_code=404, detail={"code": "preference_not_found", "message": "偏好不存在"} ) return {"deleted": True, "key": key}