server.py

# app/server.py
# ================= 强制清除代理配置 (必须放在最前面) =================
import os
os.environ.pop("http_proxy", None)
os.environ.pop("https_proxy", None)
os.environ.pop("all_proxy", None)
# ===============================================================
import asyncio
import uvicorn
import json
import uuid
from pathlib import Path
from fastapi import FastAPI, HTTPException, Depends, UploadFile, File, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime
from sqlalchemy import Column, Integer, String, Text, desc, UniqueConstraint, text
from sqlalchemy.orm import declarative_base, sessionmaker, Session



# 引入你的 RAG 组件
from app.chains import fallback_chain
from app.graph import app_graph, get_retrievers
from app.retriever import (
    add_single_file,
    delete_documents_by_paper_id,
    ingest_single_file,
    load_vector_index,
    refresh_retriever_cache,
)
from app.runtime import RunContext, get_chat_service
from app.chat_pipeline_trace import (
    emit,
    make_stage,
    reset_stage_sequence,
)
from app.memory_health_stats import collect_memory_storage_stats
from app.session_cache import (
    invalidate_all_session_cache_for_user,
    invalidate_chat_history_cache,
)
from app.skills_config import get_skills
from app.config import (
    AGENT_DATA_VIA_MCP,
    CORS_ALLOW_CREDENTIALS,
    CORS_ALLOWED_ORIGINS,
    DATA_DIR,
    LOG_CONFIG_AT_STARTUP,
    TASK_QUEUE_ENABLED,
    LLM_RATE_LIMIT_ENABLED,
    CHAT_PREFER_ASYNC_HEAVY,
    CHAT_HEAVY_MIN_CHARS,
    log_config_summary,
    validate_runtime_config,
)
from app.db_session import get_session_factory, session_scope
from app.observability import log_event
from app.upload_utils import (
    hash_bytes,
    resolve_upload_path,
    safe_remove_file,
    write_bytes,
)
from app.task_queue import enqueue_task, get_task_result, is_task_queue_enabled
from app.llm_concurrency import acquire_llm_slot, release_llm_slot
from app.rate_limit import check_chat_rate_limit
from app.index_build_queue import enqueue_index_build, is_async_index_build_enabled
from app.backend_stack.redis_streams import is_redis_stream_backend
from app.backend_stack.pipeline_stages import enqueue_full_pipeline
from app.backend_stack.schema_upgrade import ensure_backend_stack_schema
from app.backend_stack.processing_repo import get_processing_logs
from app.backend_stack.hot_queries import record_query, top_queries
from app.backend_stack.usage import increment_request
from app.backend_stack.config import REDIS_REQUIRED
from app.backend_stack.redis_streams import get_redis_client
from app.memory_schema import ensure_memory_schema
from app.paper_schema import ensure_paper_schema
from app.archivist import start_archivist_worker
from app.db_engine import get_sqlalchemy_engine
from app.mcp_data_client import (
    get_data_mcp_client,
    shutdown_data_mcp_client,
)
from app.memory_metrics import get_memory_metrics_snapshot

# [新增] 引入认证模块
from app.auth import (
    get_password_hash, 
    verify_password, 
    create_access_token, 
    get_current_user
)

# ================= 1. MySQL 数据库配置（优化版） =================
engine = get_sqlalchemy_engine()
SessionLocal = get_session_factory()
Base = declarative_base()

from contextlib import asynccontextmanager
from app.graph import get_retrievers  # 引入你的资源加载函数


# 1. 定义生命周期管理器
@asynccontextmanager
async def lifespan(app: FastAPI):
    # --- 启动时执行 ---
    validate_runtime_config()
    if LOG_CONFIG_AT_STARTUP:
        log_config_summary()
    if AGENT_DATA_VIA_MCP:
        print("🔥 [Startup] 数据面 MCP 模式：正在连接 mcp_data_server 子进程…")
        try:
            get_data_mcp_client()
            print("✅ [Startup] 数据面 MCP 客户端已就绪（向量预热在子进程内按需进行）")
        except Exception as e:
            print(f"⚠️  [Startup] 数据面 MCP 启动失败: {e}（/chat 将无法正常走数据工具）")
    else:
        print("🔥 [Startup] 正在检查检索索引状态（只读）...")
        try:
            result = load_vector_index(read_docs_for_bm25=False)
            if result.status == "ready":
                get_retrievers()
                print(f"✅ [Startup] 检索索引已就绪：{result.message}")
            elif result.status == "empty":
                print("⚠️  [Startup] 向量索引为空，服务继续启动")
                print("💡 [Startup] 请显式运行 python scripts/rebuild_index.py 构建索引")
            elif result.status == "corrupted":
                print("⚠️  [Startup] 检测到向量索引损坏，服务继续启动但检索不可用")
                print(f"💡 [Startup] {result.message}")
            else:
                print(f"⚠️  [Startup] 检索索引状态未知: {result.message}")
        except Exception as e:
            print(f"⚠️  [Startup] 索引状态检查失败: {str(e)[:200]}")
            print("⚠️  [Startup] 服务器将继续运行，但向量检索功能可能不可用")

    # 史官：异步消费记忆固化事件（与主请求解耦）
    start_archivist_worker(engine)

    ensure_backend_stack_schema(engine)
    if REDIS_REQUIRED:
        try:
            get_redis_client().ping()
            print("✅ [Startup] Redis 已连接（REDIS_REQUIRED=true）")
        except Exception as e:
            raise RuntimeError(f"Redis 不可用: {e}") from e

    yield  # 服务运行中...

    # --- 关闭时执行 (可选) ---
    if AGENT_DATA_VIA_MCP:
        shutdown_data_mcp_client()
    print("👋 [Shutdown] 服务器正在关闭...")


# 2. 注入 lifespan
app = FastAPI(lifespan=lifespan)

from app.paper_routes import router as paper_writer_router
from app.a2a_routes import router as a2a_router

app.include_router(paper_writer_router)
app.include_router(a2a_router)

# CORS：allowlist 驱动；禁止 * + credentials 组合
_cors_origins = list(CORS_ALLOWED_ORIGINS)
_cors_credentials = CORS_ALLOW_CREDENTIALS and "*" not in _cors_origins
app.add_middleware(
    CORSMiddleware,
    allow_origins=_cors_origins,
    allow_credentials=_cors_credentials,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 前端页面：优先 Vue 构建产物，回退 legacy web/index.html
FRONTEND_DIST = Path(__file__).resolve().parent / "frontend" / "dist"
WEB_DIR = Path(__file__).resolve().parent / "web"


# --- 根路径（避免访问 / 时 404） ---
@app.get("/")
def root():
    """浏览器访问 http://localhost:8000 时返回说明"""
    return {
        "service": "PaperCopilot · 论文副驾驶 RAG API",
        "docs": "http://localhost:8000/docs",
        "app": "http://localhost:8000/app",
        "frontend_dev": "http://localhost:5173 (npm run dev in frontend/)",
        "status": "running"
    }


@app.get("/app")
def serve_app():
    """返回前端 SPA（Vue 构建产物优先，否则 legacy index.html）"""
    vue_index = FRONTEND_DIST / "index.html"
    if vue_index.is_file():
        return FileResponse(vue_index)
    legacy = WEB_DIR / "index.html"
    if legacy.is_file():
        return FileResponse(legacy)
    raise HTTPException(status_code=404, detail="前端未构建：请在 frontend 目录执行 npm run build")


@app.get("/app/{full_path:path}")
def serve_app_spa(full_path: str):
    """Vue Router history 模式：非静态资源回退 index.html"""
    if full_path.startswith("assets/"):
        asset = FRONTEND_DIST / full_path
        if asset.is_file():
            return FileResponse(asset)
    vue_index = FRONTEND_DIST / "index.html"
    if vue_index.is_file():
        return FileResponse(vue_index)
    raise HTTPException(status_code=404, detail="前端未构建")


_assets_dir = FRONTEND_DIST / "assets"
if _assets_dir.is_dir():
    app.mount("/app/assets", StaticFiles(directory=_assets_dir), name="vue-assets")


# --- 定义数据库模型（优化版） ---

class ChatHistory(Base):
    __tablename__ = "chat_history"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), index=True, default="default")
    session_id = Column(String(64), index=True, default="legacy")
    role = Column(String(10))  # 'user' 或 'ai'
    content = Column(Text)
    time = Column(String(50))


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True)
    password = Column(String(255))  # [修改] 扩容以存储 bcrypt 加密后的密码


class Paper(Base):
    """
    论文清单表：用于业务管理与去重
    - 与向量库中的 chunk 通过 paper_id 等元数据关联
    """
    __tablename__ = "papers"
    __table_args__ = (
        UniqueConstraint("username", "file_hash", name="uk_paper_user_hash"),
    )

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), index=True, nullable=False)
    title = Column(String(512), nullable=False)
    file_hash = Column(String(64), nullable=False)
    file_path = Column(String(512), nullable=False)
    upload_time = Column(String(50), nullable=False)


# [状态与记忆] Agent 记忆表：键值 + category 便于按类查询（如 preference / fact / constraint）
class AgentMemory(Base):
    __tablename__ = "agent_memory"
    __table_args__ = (UniqueConstraint("username", "session_id", "memory_key", name="uk_agent_memory_user_session_key"),)
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), index=True, nullable=False)
    session_id = Column(String(64), index=True, nullable=False)
    memory_key = Column(String(128), nullable=False)
    memory_value = Column(Text, default="")
    category = Column(String(32), default="general", index=True)  # preference / fact / constraint / feedback / general
    updated_at = Column(String(32), nullable=False)
    access_count = Column(Integer, default=0, nullable=False, index=True)
    last_accessed = Column(String(32), nullable=True, index=True)
    expires_at = Column(String(32), nullable=True, index=True)


# [状态与记忆] 会话摘要表：每 N 轮或新会话时异步生成
class AgentSessionSummary(Base):
    __tablename__ = "agent_session_summary"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), index=True, nullable=False)
    session_id = Column(String(64), index=True, nullable=False)
    summary = Column(Text, default="")
    updated_at = Column(String(32), nullable=False)


# 自动创建表结构 (如果不存在)
Base.metadata.create_all(bind=engine)


def _ensure_agent_memory_lifecycle_columns() -> None:
    """兼容已有库：为 agent_memory 补齐生命周期字段。"""
    statements = [
        "ALTER TABLE agent_memory ADD COLUMN access_count INT NOT NULL DEFAULT 0",
        "ALTER TABLE agent_memory ADD COLUMN last_accessed VARCHAR(32) NULL",
        "ALTER TABLE agent_memory ADD COLUMN expires_at VARCHAR(32) NULL",
    ]
    with engine.begin() as conn:
        for sql in statements:
            try:
                conn.execute(text(sql))
            except Exception:
                # 字段已存在或数据库方言不支持时忽略，保证启动可继续。
                pass


_ensure_agent_memory_lifecycle_columns()
ensure_memory_schema(engine)
ensure_paper_schema(engine)


# 并发控制：限制同时执行的 Agent/RAG 请求数，防止 LLM API 被撑爆
_chat_semaphore = asyncio.Semaphore(10)


# 数据库依赖项
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# [状态与记忆] 每满 N 轮用户消息向史官队列投递一次（见 app/memory_scheduler.py）
SUMMARY_ROUNDS = 5


# ================= 2. API 请求/响应结构 =================

class QueryRequest(BaseModel):
    question: str
    username: Optional[str] = "default"
    session_id: Optional[str] = None  # 会话 ID，不传则按日期划分（同日同会话）


class LoginRequest(BaseModel):
    username: str
    password: str


class LogoutRequest(BaseModel):
    username: str


class RegisterRequest(BaseModel):
    username: str
    password: str


class HistoryItem(BaseModel):
    role: str
    content: str
    time: str


class PaperItem(BaseModel):
    """论文列表项，供前端展示"""
    id: int
    title: str
    file_hash: str
    upload_time: str
    file_path: str

    class Config:
        from_attributes = True


class FilterRetrieveRequest(BaseModel):
    """Metadata 过滤检索请求。"""
    question: str
    filters: dict = {}
    k: int = 10
    paper_id: Optional[int] = None
    include_scores: bool = False


class VectorRevisionPromoteRequest(BaseModel):
    min_entities: int = 1


# ================= 3. 接口逻辑 =================

@app.post("/login")
def login_endpoint(request: LoginRequest, db: Session = Depends(get_db)):
    """
    用户登录接口
    - 使用明文密码验证
    - 返回 JWT Token
    """
    try:
        print(f"🔐 [Login] 收到登录请求: {request.username}")

        # 1. 查询用户
        user = db.query(User).filter(User.username == request.username).first()

        # 2. 验证账号密码（明文比较）
        if not user:
            return {"status": "error", "message": "用户不存在"}

        # 直接比较明文密码
        if request.password != user.password:
            return {"status": "error", "message": "密码错误"}

        # 3. [优化] 生成 JWT Token
        access_token = create_access_token(data={"sub": user.username})
        
        print(f"✅ [Login] {request.username} 验证通过")
        return {
            "status": "success",
            "message": "登录成功",
            "token": access_token,
            "username": user.username
        }
    except Exception as e:
        import traceback
        error_msg = f"登录失败: {str(e)}"
        print(f"❌ [Login] 错误: {error_msg}")
        traceback.print_exc()
        return {"status": "error", "message": error_msg}


@app.post("/register")
def register_endpoint(request: RegisterRequest, db: Session = Depends(get_db)):
    """
    用户注册接口
    - 密码明文存储（不加密）
    """
    try:
        print(f"📝 [Register] 收到注册请求: {request.username}")

        # 1. 检查用户是否已存在
        existing_user = db.query(User).filter(User.username == request.username).first()
        if existing_user:
            return {"status": "error", "message": "用户名已存在"}

        # 2. 直接存储明文密码
        new_user = User(username=request.username, password=request.password)
        
        db.add(new_user)
        db.commit()
        db.refresh(new_user)
        print(f"✅ [Register] 用户 {request.username} 注册成功")
        return {"status": "success", "message": "注册成功"}
    except Exception as e:
        db.rollback()
        import traceback
        error_msg = f"注册失败: {str(e)}"
        print(f"❌ [Register] 错误: {error_msg}")
        traceback.print_exc()
        return {"status": "error", "message": error_msg}


@app.post("/logout")
def logout_endpoint(
    request: LogoutRequest,
    current_user: str = Depends(get_current_user)  # [新增] 需要认证
):
    """
    用户登出接口（优化版）
    - 需要 Token 认证
    """
    print(f"👋 [Logout] 用户 {current_user} 已退出登录")
    return {"status": "success", "message": "已退出"}


# --- [优化] 清空历史记录接口（按用户隔离） ---
@app.post("/clear_history")
def clear_history_endpoint(
    current_user: str = Depends(get_current_user),  # [新增] 需要认证
    db: Session = Depends(get_db)
):
    """
    清空当前用户的聊天记录（优化版）
    - 只删除当前用户的历史记录
    - 需要 Token 认证
    """
    try:
        num_deleted = db.query(ChatHistory).filter(
            ChatHistory.username == current_user
        ).delete()
        db.commit()
        invalidate_all_session_cache_for_user(current_user)
        print(f"🧹 [Clear] 用户 {current_user} 清空历史记录，共删除 {num_deleted} 条")
        return {"status": "success", "message": f"已删除 {num_deleted} 条记录"}
    except Exception as e:
        db.rollback()
        return {"status": "error", "message": str(e)}


# --- [优化] 文件上传接口（需要认证） ---
@app.post("/upload")
async def upload_file_endpoint(
    file: UploadFile = File(...),
    current_user: str = Depends(get_current_user),  # [新增] 需要认证
    db: Session = Depends(get_db),
):
    """
    上传文件 -> 保存 -> 索引 -> 返回详细日志（优化版）
    - 需要 Token 认证
    """
    upload_logs = []

    def log(msg):
        print(msg)
        upload_logs.append(msg)

    try:
        content = await file.read()
        if not content:
            return {"status": "error", "message": "空文件", "logs": upload_logs}

        file_hash = hash_bytes(content)
        log(f"🧾 [Upload] 文件 MD5: {file_hash}")

        existing = db.query(Paper).filter(
            Paper.username == current_user,
            Paper.file_hash == file_hash,
        ).first()
        if existing:
            msg = "该文件您已上传过，请勿重复上传。"
            log(f"⚠️ [Upload] {msg}")
            return {
                "status": "error",
                "message": msg,
                "logs": upload_logs,
            }

        try:
            file_path, display_title = resolve_upload_path(current_user, file.filename or "upload.bin")
        except ValueError as e:
            return {"status": "error", "message": str(e), "logs": upload_logs}

        log(f"📂 [Upload] 正在接收文件: {display_title}...")
        loop = asyncio.get_running_loop()
        await loop.run_in_executor(None, lambda: write_bytes(file_path, content))
        log(f"✅ [Upload] 文件已保存至: {file_path}")

        upload_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        paper = Paper(
            username=current_user,
            title=display_title,
            file_hash=file_hash,
            file_path=file_path,
            upload_time=upload_time,
        )
        db.add(paper)
        db.flush()
        log(f"🧾 [Upload] 已创建 Paper 记录，id={paper.id}")

        # 5. 向量索引：Redis Streams 流水线 / 任务队列 / 同步
        if is_redis_stream_backend():
            from app.backend_stack.processing_repo import update_paper_status

            update_paper_status(engine, paper.id, "pending")
            stream_id = enqueue_full_pipeline(
                paper.id,
                current_user,
                file_path,
                metadata={
                    "paper_title": paper.title,
                    "file_hash": file_hash,
                    "upload_time": upload_time,
                },
            )
            db.commit()
            log(f"📤 [Pipeline] 已入队 Redis Streams，entry={stream_id}")
            return {
                "status": "processing",
                "message": "文件已保存，五阶段流水线处理中，可查询 GET /papers/{paper_id}/processing-logs",
                "paper_id": paper.id,
                "stream_entry_id": stream_id,
                "logs": upload_logs,
            }

        if is_async_index_build_enabled():
            task_id = enqueue_index_build(
                {
                    "file_path": file_path,
                    "username": current_user,
                    "paper_id": paper.id,
                    "paper_title": paper.title,
                    "upload_time": upload_time,
                    "file_hash": file_hash,
                }
            )
            db.commit()
            log(f"📤 [Index] 已入队异步构建 task_id={task_id}（在线问答不阻塞）")
            return {
                "status": "processing",
                "message": f"文件已保存，索引构建中，task_id={task_id}，可轮询 GET /chat/result/{task_id}",
                "task_id": task_id,
                "paper_id": paper.id,
                "logs": upload_logs,
            }

        log(f"⚙️ [Index] 开始构建向量索引（同步 Ingestion Pipeline）...")
        try:
            ingest_result = ingest_single_file(
                file_path,
                username=current_user,
                paper_meta={
                    "paper_id": paper.id,
                    "paper_title": paper.title,
                    "uploader": current_user,
                    "upload_time": upload_time,
                    "file_hash": file_hash,
                },
            )
            upload_logs.extend(ingest_result.logs)
            ingestion_validation = ingest_result.validation
        except ValueError as e:
            err_msg = f"❌ [Index] 索引构建失败: {e}"
            log(err_msg)
            safe_remove_file(file_path)
            db.rollback()
            return {"status": "error", "message": str(e), "logs": upload_logs}
        except Exception as e:
            err_msg = f"❌ [Index] 索引构建失败: {e}"
            log(err_msg)
            import traceback
            traceback.print_exc()
            safe_remove_file(file_path)
            
            # 如果是 ChromaDB 错误，提供更友好的提示
            error_str = str(e)
            if "Error in compaction" in error_str or "hnsw" in error_str.lower():
                user_message = "向量数据库内部错误，可能是数据库文件损坏。建议重启服务器或联系管理员。"
            else:
                user_message = str(e)

            db.rollback()
            return {"status": "error", "message": user_message, "logs": upload_logs}

        # 6. 索引成功后提交论文记录
        db.commit()
        log(f"💾 [Upload] Paper 记录已写入数据库，id={paper.id}")

        # 7. 清除检索器缓存，让新数据生效
        log(f"🔄 [Cache] 刷新检索缓存与知识库版本...")
        refresh_retriever_cache()

        return {
            "status": "success",
            "message": f"文件 {file.filename} 上传成功",
            "paper_id": paper.id,
            "logs": upload_logs,
            "ingestion": ingestion_validation,
        }

    except Exception as e:
        return {"status": "error", "message": str(e), "logs": upload_logs}


@app.post("/chat")
async def chat_endpoint(
    request: QueryRequest,
    background_tasks: BackgroundTasks,
    current_user: str = Depends(get_current_user),
):
    """
    智能问答接口（优化版）
    - 需要 Token 认证
    - 按用户隔离历史记录
    - 异步执行，不阻塞事件循环
    """
    raw_question = (request.question or "").strip()
    if not raw_question:
        from fastapi.responses import JSONResponse
        return JSONResponse({"error": "question 不能为空"}, status_code=400)

    if (
        CHAT_PREFER_ASYNC_HEAVY
        and is_task_queue_enabled()
        and len(raw_question) >= CHAT_HEAVY_MIN_CHARS
    ):
        task_id = enqueue_task(
            "rag_query",
            {
                "question": raw_question,
                "username": current_user,
                "session_id": request.session_id,
                "chat_history": "",
            },
        )
        return {
            "status": "processing",
            "task_id": task_id,
            "message": "长问题已入队异步处理，请轮询 GET /chat/result/{task_id}",
        }

    async with _chat_semaphore:
        try:
            session_id = request.session_id
            answer, server_logs, request_id, research_trace = await _run_chat_and_get_answer_async(
                raw_question, current_user, background_tasks, session_id=session_id
            )
            return {
                "answer": answer,
                "logs": server_logs,
                "request_id": request_id,
                "research_trace": research_trace or {},
            }
        except Exception as e:
            import traceback
            traceback.print_exc()
            return {
                "answer": f"服务器内部错误: {str(e)}",
                "logs": [f"❌ Error: {str(e)}"]
            }


async def _run_chat_and_get_answer_async(
    raw_question: str,
    current_user: str,
    background_tasks: BackgroundTasks,
    session_id: Optional[str] = None,
    on_stage=None,
):
    """在线程池中执行 chat；worker 内独立创建 DB Session，避免跨线程共享。"""
    loop = asyncio.get_running_loop()

    def _worker():
        with session_scope() as db:
            return _run_chat_and_get_answer(
                raw_question,
                current_user,
                db,
                background_tasks,
                session_id,
                on_stage=on_stage,
            )

    return await loop.run_in_executor(None, _worker)


def _run_chat_and_get_answer(
    raw_question: str,
    current_user: str,
    db: Session,
    background_tasks: BackgroundTasks,
    session_id: Optional[str] = None,
    on_stage=None,
):
    """执行一轮对话并返回 (answer, server_logs, request_id, research_trace)。用于 /chat 与 /chat/stream 复用。"""
    reset_stage_sequence()
    request_id = uuid.uuid4().hex
    server_logs: list = []

    def log(msg):
        print(msg)
        server_logs.append(msg)

    log(f"🚀 [Chat] request_id={request_id} 使用问题: '{raw_question[:80]}...' 进入主链路")
    log_event(
        "chat.start",
        request_id=request_id,
        runtime="chat",
        username=current_user,
        session_id=(session_id or "").strip() or datetime.now().strftime("%Y-%m-%d"),
    )
    emit(on_stage, make_stage("request", "接收请求", "running", agent="gateway"))
    emit(on_stage, make_stage("request", "接收请求", "completed", agent="gateway", detail=f"id={request_id[:8]}"))

    allowed, rl_reason = check_chat_rate_limit(current_user)
    if not allowed:
        log(f"⚠️ [RateLimit] Redis 限流拒绝: {rl_reason}")
        emit(on_stage, make_stage("rate_limit", "限流保护", "error", agent="gateway", detail=rl_reason))
        answer = fallback_chain.invoke({"question": raw_question, "chat_history": ""})
        return answer, server_logs + [f"[RateLimit] redis:{rl_reason}"], request_id, {}

    try:
        increment_request(current_user)
        record_query(raw_question, current_user)
    except Exception:
        pass

    _llm_slot_acquired = False
    if LLM_RATE_LIMIT_ENABLED:
        allowed = asyncio.run(acquire_llm_slot())
        if not allowed:
            log("⚠️ [RateLimit] 令牌桶已满，走 fallback 降级")
            emit(on_stage, make_stage("rate_limit", "LLM 令牌桶", "error", agent="gateway", detail="busy"))
            answer = fallback_chain.invoke({"question": raw_question, "chat_history": ""})
            return answer, server_logs + ["[RateLimit] busy fallback"], request_id, {}
        _llm_slot_acquired = True

    try:
        ctx = RunContext.for_chat(
            current_user,
            (session_id or "").strip() or datetime.now().strftime("%Y-%m-%d"),
            request_id=request_id,
        )
        chat_svc = get_chat_service(
            engine,
            graph=app_graph,
            chat_model=ChatHistory,
            summary_rounds=SUMMARY_ROUNDS,
        )
        result = chat_svc.answer(
            raw_question,
            ctx,
            db=db,
            background_tasks=background_tasks,
            log=log,
            on_stage=on_stage,
            chat_model=ChatHistory,
        )
        merged_logs = server_logs + [l for l in result.logs if l not in server_logs]
        return result.answer, merged_logs, result.request_id, result.research_trace
    finally:
        if _llm_slot_acquired:
            release_llm_slot()


@app.post("/chat/async")
async def chat_async_endpoint(
    request: QueryRequest,
    current_user: str = Depends(get_current_user),
):
    """复杂查询异步版：入队后立即返回 task_id（需 TASK_QUEUE_ENABLED 且 Redis 可用）。"""
    if not is_task_queue_enabled():
        from fastapi.responses import JSONResponse

        return JSONResponse(
            {"error": "任务队列未启用或 Redis 不可用", "hint": "设置 TASK_QUEUE_ENABLED=true 并配置 REDIS_URL"},
            status_code=503,
        )
    raw_question = (request.question or "").strip()
    if not raw_question:
        from fastapi.responses import JSONResponse

        return JSONResponse({"error": "question 不能为空"}, status_code=400)
    task_id = enqueue_task(
        "rag_query",
        {
            "question": raw_question,
            "username": current_user,
            "session_id": request.session_id,
            "chat_history": "",
        },
    )
    return {"status": "processing", "task_id": task_id}


@app.get("/chat/result/{task_id}")
async def chat_result_endpoint(
    task_id: str,
    current_user: str = Depends(get_current_user),
):
    """轮询异步任务结果。"""
    _ = current_user
    result = get_task_result(task_id)
    if result is None:
        return {"status": "processing"}
    return {
        "status": "done",
        "answer": result.get("answer", ""),
        "logs": result.get("logs", []),
        "error": result.get("error"),
    }


@app.post("/chat/stream")
async def chat_stream_endpoint(
    request: QueryRequest,
    background_tasks: BackgroundTasks,
    current_user: str = Depends(get_current_user),
):
    """流式返回回答：先异步执行 Agent，再将答案按小块通过 SSE 推送，前端可逐字显示。"""
    import json
    raw_question = (request.question or "").strip()
    if not raw_question:
        from fastapi.responses import JSONResponse
        return JSONResponse({"error": "question 不能为空"}, status_code=400)

    async def event_stream():
        async with _chat_semaphore:
            loop = asyncio.get_running_loop()
            q: asyncio.Queue = asyncio.Queue()
            pipeline_stages: list = []

            def on_stage(stage: dict):
                pipeline_stages.append(stage)
                loop.call_soon_threadsafe(q.put_nowait, ("stage", stage))

            async def run_pipeline():
                try:
                    result = await _run_chat_and_get_answer_async(
                        raw_question,
                        current_user,
                        background_tasks,
                        session_id=getattr(request, "session_id", None),
                        on_stage=on_stage,
                    )
                    await q.put(("done", result))
                except Exception as e:
                    import traceback
                    traceback.print_exc()
                    await q.put(("error", e))

            yield f"data: {json.dumps({'type': 'start'}, ensure_ascii=False)}\n\n"
            task = asyncio.create_task(run_pipeline())
            answer = ""
            server_logs: list = []
            request_id = ""
            research_trace: dict = {}

            while True:
                kind, payload = await q.get()
                if kind == "stage":
                    yield f"data: {json.dumps({'type': 'stage', 'stage': payload}, ensure_ascii=False)}\n\n"
                elif kind == "done":
                    answer, server_logs, request_id, research_trace = payload
                    break
                elif kind == "error":
                    yield f"data: {json.dumps({'type': 'error', 'message': str(payload)}, ensure_ascii=False)}\n\n"
                    await task
                    return

            chunk_size = 2
            for i in range(0, len(answer), chunk_size):
                chunk = answer[i : i + chunk_size]
                yield f"data: {json.dumps({'type': 'chunk', 'content': chunk}, ensure_ascii=False)}\n\n"
            yield f"data: {json.dumps({'type': 'done', 'logs': server_logs, 'request_id': request_id, 'pipeline': pipeline_stages, 'research_trace': research_trace or {}}, ensure_ascii=False)}\n\n"
            await task

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


@app.get("/skills")
def list_skills():
    """返回当前启用的技能列表（id、name、description），供前端或调试查看。"""
    return get_skills(enabled_only=True)


@app.get("/ingestion/metrics")
def ingestion_metrics(current_user: str = Depends(get_current_user)):
    """论文索引 Ingestion 治理指标（进程级内存计数 + 最近一次 validation）。"""
    from app.ingestion.metrics import get_ingestion_metrics_snapshot

    snap = get_ingestion_metrics_snapshot()
    counters = snap.get("counters") or {}
    docs = max(counters.get("documents_ingested", 0), 1)
    return {
        "metrics": snap,
        "derived": {
            "failure_rate": round(counters.get("ingestion_failures", 0) / docs, 4),
            "avg_warnings_per_doc": round(counters.get("validation_warnings", 0) / docs, 4),
        },
        "user": current_user,
    }


@app.get("/retrieve/schema")
def retrieve_metadata_schema(current_user: str = Depends(get_current_user)):
    """返回允许用于 filter 检索的 metadata 字段列表。"""
    from app.filter_retrieval import FILTERABLE_METADATA_FIELDS
    from app.collection_revision import get_revision_status

    rev = get_revision_status()
    return {
        "filterable_fields": list(FILTERABLE_METADATA_FIELDS),
        "active_collection": rev.get("active_collection"),
        "user": current_user,
    }


@app.post("/retrieve/filter")
def retrieve_with_metadata_filter(
    request: FilterRetrieveRequest,
    current_user: str = Depends(get_current_user),
):
    """语义检索 + metadata 过滤（Milvus expr / Chroma where）。"""
    from app.filter_retrieval import (
        metadata_filter_search,
        metadata_filter_search_with_scores,
    )

    q = (request.question or "").strip()
    if not q:
        from fastapi.responses import JSONResponse

        return JSONResponse({"error": "question 不能为空"}, status_code=400)

    filters = dict(request.filters or {})
    if request.paper_id is not None:
        filters["paper_id"] = request.paper_id
    filters.setdefault("username", current_user)

    try:
        if request.include_scores:
            hits = metadata_filter_search_with_scores(q, filters, k=request.k)
            return {"question": q, "filters": filters, "results": hits, "count": len(hits)}
        docs = metadata_filter_search(q, filters, k=request.k)
        results = [
            {"content": d.page_content[:2000], "metadata": dict(d.metadata or {})}
            for d in docs
        ]
        return {"question": q, "filters": filters, "results": results, "count": len(results)}
    except Exception as ex:
        from fastapi.responses import JSONResponse

        return JSONResponse({"error": str(ex)}, status_code=500)


@app.get("/vector/revision")
def vector_revision_status(current_user: str = Depends(get_current_user)):
    """Collection revision 灰度状态（active / staging）。"""
    from app.collection_revision import get_revision_status

    return {"user": current_user, **get_revision_status()}


@app.post("/vector/revision/promote")
def vector_revision_promote(
    request: VectorRevisionPromoteRequest,
    current_user: str = Depends(get_current_user),
):
    """将 staging collection 提升为 active（灰度切换）。"""
    from app.collection_revision import promote_staging_to_active

    result = promote_staging_to_active(min_entities=request.min_entities)
    code = 200 if result.get("ok") else 400
    from fastapi.responses import JSONResponse

    return JSONResponse({"user": current_user, **result}, status_code=code)


@app.post("/vector/revision/init")
def vector_revision_init(current_user: str = Depends(get_current_user)):
    """初始化 revision 双 collection 指针（首次启用时）。"""
    from app.collection_revision import init_revision_pair

    pair = init_revision_pair()
    return {"user": current_user, "initialized": pair}


@app.get("/research/metrics")
def research_metrics(current_user: str = Depends(get_current_user)):
    """Deep Research 治理指标快照（内存计数 + 派生比率，进程级）。"""
    from app.research_metrics import get_research_metrics_snapshot

    m = get_research_metrics_snapshot()
    triggered = max(m.get("external_research_triggered", 0), 1)
    hits = (
        m.get("external_arxiv_hits", 0)
        + m.get("external_github_hits", 0)
        + m.get("external_web_hits", 0)
    )
    return {
        "metrics": m,
        "derived": {
            "external_hit_rate": round(hits / triggered, 4),
            "empty_rate": round(m.get("external_research_empty", 0) / triggered, 4),
            "parallel_rate": round(m.get("external_research_parallel", 0) / triggered, 4),
            "citation_mode_rate": round(
                m.get("citation_mode_generations", 0) / max(m.get("research_plans", 0), 1),
                4,
            ),
        },
        "user": current_user,
    }


@app.get("/memory/health")
def memory_health(
    session_id: Optional[str] = None,
    current_user: str = Depends(get_current_user),
):
    """
    记忆健康诊断：返回质量看板关键指标与存储分布。
    """
    sid = (session_id or "").strip() or datetime.now().strftime("%Y-%m-%d")
    storage = collect_memory_storage_stats(engine, current_user, sid)
    metrics = get_memory_metrics_snapshot()
    return {
        "user": current_user,
        "session_id": sid,
        "storage": storage,
        "quality_dashboard": {
            "conflict_rate": metrics.get("conflict_rate", 0.0),
            "mis_hit_rate": metrics.get("mis_hit_rate", 0.0),
            "read_memory_hit_rate": metrics.get("read_memory_hit_rate", 0.0),
            "episodic_recall_hit_rate": metrics.get("episodic_recall_hit_rate", 0.0),
            "graph_recall_hit_rate": metrics.get("graph_recall_hit_rate", 0.0),
            "consolidation_skip_rate": metrics.get("consolidation_skip_rate", 0.0),
            "inject_budget_chars_utilization": metrics.get("inject_budget_chars_utilization", 0.0),
            "inject_budget_items_utilization": metrics.get("inject_budget_items_utilization", 0.0),
            "raw_counters": metrics,
        },
    }


@app.get("/papers", response_model=List[PaperItem])
def list_papers(
    current_user: str = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    """返回当前用户已上传的论文列表，按上传时间倒序。"""
    rows = (
        db.query(Paper)
        .filter(Paper.username == current_user)
        .order_by(desc(Paper.upload_time))
        .all()
    )
    return [PaperItem(id=r.id, title=r.title, file_hash=r.file_hash, upload_time=r.upload_time, file_path=r.file_path) for r in rows]


@app.get("/papers/{paper_id}/processing-logs")
def paper_processing_logs(
    paper_id: int,
    current_user: str = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    """查询论文流水线各阶段处理日志（Redis Streams 模式）。"""
    paper = db.query(Paper).filter(Paper.id == paper_id, Paper.username == current_user).first()
    if not paper:
        raise HTTPException(status_code=404, detail="论文不存在或无权查看")
    logs = get_processing_logs(engine, paper_id)
    status_row = None
    try:
        with engine.connect() as conn:
            row = conn.execute(
                text("SELECT status, error_message, processed_at, chunk_count FROM papers WHERE id = :pid"),
                {"pid": paper_id},
            ).mappings().first()
            status_row = dict(row) if row else None
    except Exception:
        status_row = None
    return {"paper_id": paper_id, "status": status_row, "logs": logs}


@app.get("/admin/hot-queries")
def admin_hot_queries(
    limit: int = 20,
    current_user: str = Depends(get_current_user),
):
    """热点查询排行榜（ZSet）。"""
    items = top_queries(limit=min(limit, 100))
    return {"username": current_user, "items": [{"query": q, "score": s} for q, s in items]}


@app.delete("/papers/{paper_id}")
def delete_paper(
    paper_id: int,
    current_user: str = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    """删除当前用户下的指定论文：先删向量库中该 paper_id 的 chunk，再删 MySQL 记录。"""
    paper = db.query(Paper).filter(Paper.id == paper_id, Paper.username == current_user).first()
    if not paper:
        raise HTTPException(status_code=404, detail="论文不存在或无权删除")
    file_path = paper.file_path
    try:
        delete_documents_by_paper_id(paper_id)
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=500, detail=f"向量库删除失败: {e}")
    db.delete(paper)
    db.commit()
    removed = safe_remove_file(file_path)
    refresh_retriever_cache()
    return {
        "status": "success",
        "message": "已删除该论文",
        "file_removed": removed,
    }


@app.get("/get_history", response_model=List[HistoryItem])
def get_history(
    current_user: str = Depends(get_current_user),  # [新增] 需要认证
    db: Session = Depends(get_db)
):
    """
    获取历史记录接口（优化版）
    - 只返回当前用户的历史记录
    - 需要 Token 认证
    """
    records = db.query(ChatHistory).filter(
        ChatHistory.username == current_user
    ).order_by(ChatHistory.id).all()
    return records


if __name__ == "__main__":
    import os as _os

    _host = (_os.getenv("SERVER_HOST") or "127.0.0.1").strip()
    _port = int(_os.getenv("SERVER_PORT", "8000"))
    # 浏览器勿访问 http://0.0.0.0:8000（Chrome 会报 iframe/跨域错误）；局域网暴露请 SERVER_HOST=0.0.0.0 并用本机 IP 访问
    _open = "127.0.0.1" if _host in ("0.0.0.0", "::") else _host
    print(f"🌐 浏览器打开: http://{_open}:{_port}/app")
    if _host == "0.0.0.0":
        print("   监听 0.0.0.0 时请勿点 http://0.0.0.0:...，请用 http://127.0.0.1 或本机局域网 IP")
    uvicorn.run(app, host=_host, port=_port)