vent_agent_2.0/api/session_routes.py

"""会话与历史路由。

所有 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}