"""Admin-only window into another user's activity.

Lets an administrator inspect any user's conversation history (threads + the
Q&A messages inside them) and their accumulated memory (global + per-agent
buckets). All endpoints require ``system_role == "admin"``; they are strictly
read-only.

The underlying stores normally auto-scope every query to the *calling* user
via a contextvar. These endpoints pass an explicit ``user_id`` so the admin
reads the *target* user's data instead — which is exactly what those store
methods' ``user_id`` parameter is designed for.
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
import re
from collections.abc import AsyncIterator
from difflib import SequenceMatcher
from hashlib import sha256

from datetime import UTC, datetime, timedelta, time as datetime_time

from fastapi import APIRouter, Depends, HTTPException, Query, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from sqlalchemy import and_, case, desc, func, not_, or_, select, text
from langchain_core.messages import HumanMessage, SystemMessage

from app.gateway.deps import (
    get_config,
    get_agent_store,
    get_checkpointer,
    get_current_user_from_request,
    get_thread_store,
)
from deerflow.agents.memory import BuiltinFileProvider
from deerflow.config.app_config import AppConfig
from deerflow.config.paths import get_paths
from deerflow.config.system_settings import load_system_settings
from deerflow.models import create_chat_model
from deerflow.runtime import serialize_channel_values
from app.gateway.routers.memory import _build_v1_provider, _is_v1
from deerflow.persistence.engine import get_session_factory
from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
from deerflow.persistence.llm_metrics.model import LlmCallMetricRow
from deerflow.persistence.run.model import RunRow
from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRunRow
from deerflow.persistence.thread_meta.model import ThreadMetaRow
from deerflow.persistence.tool_metrics.model import ToolCallMetricRow
from deerflow.persistence.types import BEIJING_TZ
from deerflow.persistence.user.model import UserRow

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/admin/users", tags=["admin-users"])

_LEADERBOARD_MAX_CONCURRENT = max(
    1,
    int(os.getenv("ADMIN_LEADERBOARD_MAX_CONCURRENT", "1")),
)
_LEADERBOARD_TIMEOUT_SECONDS = max(
    5.0,
    float(os.getenv("ADMIN_LEADERBOARD_TIMEOUT_SECONDS", "45")),
)
_LEADERBOARD_QUERY_PAUSE_SECONDS = max(
    0.0,
    float(os.getenv("ADMIN_LEADERBOARD_QUERY_PAUSE_SECONDS", "0.05")),
)
_LEADERBOARD_START_DELAY_SECONDS = max(
    0.0,
    float(os.getenv("ADMIN_LEADERBOARD_START_DELAY_SECONDS", "0.2")),
)
_LEADERBOARD_DAILY_SNAPSHOT_LIMIT = max(
    100,
    int(os.getenv("ADMIN_LEADERBOARD_DAILY_SNAPSHOT_LIMIT", "1000")),
)
_LEADERBOARD_TODAY_TTL_SECONDS = max(
    30.0,
    float(os.getenv("ADMIN_LEADERBOARD_TODAY_TTL_SECONDS", "300")),
)
_LEADERBOARD_QUERY_TIMEOUT_MS = max(
    500,
    int(os.getenv("ADMIN_LEADERBOARD_QUERY_TIMEOUT_MS", "8000")),
)
_LEADERBOARD_LOCK_WAIT_SECONDS = max(
    1,
    int(os.getenv("ADMIN_LEADERBOARD_LOCK_WAIT_SECONDS", "5")),
)
_LEADERBOARD_QUEUE_ON_REQUEST_ENABLED = os.getenv(
    "ADMIN_LEADERBOARD_QUEUE_ON_REQUEST_ENABLED",
    os.getenv("ADMIN_LEADERBOARD_WEB_QUEUE_ENABLED", "true"),
).strip().lower() not in {
    "0",
    "false",
    "no",
    "off",
}
_leaderboard_semaphore = asyncio.Semaphore(_LEADERBOARD_MAX_CONCURRENT)


async def _require_admin(request: Request) -> None:
    """Reject non-admin callers. In no-auth mode there is no user → allow."""
    try:
        user = await get_current_user_from_request(request)
    except HTTPException:
        raise
    if user is None:
        return
    if getattr(user, "system_role", None) != "admin":
        raise HTTPException(status_code=403, detail="Admin only")


# ── Leaderboard / analytics ─────────────────────────────────────────────────


class AdminLeaderboardUser(BaseModel):
    user_id: str
    email: str = ""
    system_role: str = "user"
    thread_count: int = 0
    run_count: int = 0
    qa_count: int = 0
    llm_call_count: int = 0
    tool_call_count: int = 0
    skill_call_count: int = 0
    total_tokens: int = 0
    input_tokens: int = 0
    output_tokens: int = 0
    last_active_at: str = ""


class AdminLeaderboardSkill(BaseModel):
    name: str
    call_count: int = 0
    user_count: int = 0
    success_count: int = 0
    error_count: int = 0
    avg_duration_ms: int = 0
    last_used_at: str = ""


class AdminLeaderboardDimension(BaseModel):
    name: str
    count: int = 0
    user_count: int = 0
    success_count: int = 0
    error_count: int = 0
    total_tokens: int = 0
    last_seen_at: str = ""


class AdminLeaderboardQuestion(BaseModel):
    text: str
    user_id: str = ""
    email: str = ""
    thread_id: str = ""
    run_id: str = ""
    asked_at: str = ""


class AdminLeaderboardTrendPoint(BaseModel):
    date: str
    runs: int = 0
    questions: int = 0
    tokens: int = 0
    users: int = 0


class AdminLeaderboardOverview(BaseModel):
    total_users: int = 0
    active_users: int = 0
    total_threads: int = 0
    total_runs: int = 0
    total_questions: int = 0
    total_tokens: int = 0
    total_skill_calls: int = 0
    total_tool_calls: int = 0
    error_runs: int = 0


class AdminLeaderboardResponse(BaseModel):
    range_days: int
    since: str
    until: str
    status: str = "ready"
    generated_at: str = ""
    message: str = ""
    overview: AdminLeaderboardOverview
    users_by_activity: list[AdminLeaderboardUser] = Field(default_factory=list)
    users_by_questions: list[AdminLeaderboardUser] = Field(default_factory=list)
    users_by_tokens: list[AdminLeaderboardUser] = Field(default_factory=list)
    skills: list[AdminLeaderboardSkill] = Field(default_factory=list)
    tools: list[AdminLeaderboardSkill] = Field(default_factory=list)
    models: list[AdminLeaderboardDimension] = Field(default_factory=list)
    agents: list[AdminLeaderboardDimension] = Field(default_factory=list)
    run_statuses: list[AdminLeaderboardDimension] = Field(default_factory=list)
    recent_questions: list[AdminLeaderboardQuestion] = Field(default_factory=list)
    frequent_questions: list[AdminLeaderboardQuestion] = Field(default_factory=list)
    trends: list[AdminLeaderboardTrendPoint] = Field(default_factory=list)


class AdminUserAnalyticsResponse(BaseModel):
    range_days: int
    since: str
    until: str
    user: AdminLeaderboardUser
    overview: AdminLeaderboardOverview
    skills: list[AdminLeaderboardSkill] = Field(default_factory=list)
    tools: list[AdminLeaderboardSkill] = Field(default_factory=list)
    models: list[AdminLeaderboardDimension] = Field(default_factory=list)
    agents: list[AdminLeaderboardDimension] = Field(default_factory=list)
    run_statuses: list[AdminLeaderboardDimension] = Field(default_factory=list)
    recent_questions: list[AdminLeaderboardQuestion] = Field(default_factory=list)
    trends: list[AdminLeaderboardTrendPoint] = Field(default_factory=list)


class AdminLeaderboardReportResponse(BaseModel):
    report: str
    generated_at: str
    cached: bool = False
    model_name: str | None = None


class AdminLeaderboardSnapshotStatusItem(BaseModel):
    stat_date: str
    status: str = "missing"
    has_payload: bool = False
    generated_at: str = ""
    updated_at: str = ""
    message: str = ""
    error: str = ""


class AdminLeaderboardSnapshotStatusResponse(BaseModel):
    settings_hash: str
    since: str
    until: str
    total_days: int
    ready_days: int
    queued_days: int
    running_days: int
    error_days: int
    missing_days: int
    items: list[AdminLeaderboardSnapshotStatusItem] = Field(default_factory=list)


class AdminLeaderboardSnapshotBackfillRequest(BaseModel):
    days: int = Field(default=30, ge=1, le=365)
    since: str | None = Field(default=None, description="Inclusive range start, ISO datetime")
    until: str | None = Field(default=None, description="Inclusive range end, ISO datetime")
    force: bool = Field(default=False, description="Queue dates even when a ready snapshot already exists")


class AdminLeaderboardSnapshotBackfillResponse(BaseModel):
    settings_hash: str
    since: str
    until: str
    total_days: int
    queued_dates: list[str] = Field(default_factory=list)
    message: str = ""


def _now() -> datetime:
    return datetime.now()


def _iso(value) -> str:
    return value.isoformat() if isinstance(value, datetime) else str(value or "")


def _parse_datetime(value: str | None, fallback: datetime) -> datetime:
    if not value:
        return fallback
    text = value.strip()
    if not text:
        return fallback
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    try:
        return datetime.fromisoformat(text)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=f"Invalid datetime: {value}") from exc


def _resolve_range(days: int, since: str | None, until: str | None) -> tuple[datetime, datetime, int]:
    end = _parse_datetime(until, _now())
    start = _parse_datetime(since, end - timedelta(days=max(1, min(days, 365))))
    if start > end:
        raise HTTPException(status_code=400, detail="since must be before until")
    span_days = max(1, min(365, (end - start).days + 1))
    return start, end, span_days


def _as_beijing(value: datetime) -> datetime:
    if value.tzinfo is None:
        value = value.replace(tzinfo=UTC)
    return value.astimezone(BEIJING_TZ)


def _today_stat_date() -> str:
    return datetime.now(BEIJING_TZ).date().isoformat()


def _stat_dates_for_range(since: datetime, until: datetime) -> list[str]:
    start_date = _as_beijing(since).date()
    end_date = _as_beijing(until).date()
    if start_date > end_date:
        raise HTTPException(status_code=400, detail="since must be before until")
    count = min(366, (end_date - start_date).days + 1)
    return [(start_date + timedelta(days=i)).isoformat() for i in range(count)]


def _bounds_for_stat_date(stat_date: str) -> tuple[datetime, datetime]:
    try:
        day = datetime.strptime(stat_date, "%Y-%m-%d").date()
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=f"Invalid stat date: {stat_date}") from exc
    start = datetime.combine(day, datetime_time.min).replace(tzinfo=BEIJING_TZ)
    end = datetime.combine(day, datetime_time.max).replace(tzinfo=BEIJING_TZ)
    return start, end


async def _try_set_session_value(session, sql: str) -> None:
    try:
        await session.execute(text(sql))
    except Exception:
        logger.debug("Failed to apply analytics session setting: %s", sql, exc_info=True)
        try:
            await session.rollback()
        except Exception:
            logger.debug("Failed to rollback after analytics session setting failure", exc_info=True)


async def _set_analytics_read_timeout(session) -> None:
    conn = await session.connection()
    dialect = conn.dialect.name
    if dialect == "mysql":
        await _try_set_session_value(session, f"SET SESSION max_execution_time = {_LEADERBOARD_QUERY_TIMEOUT_MS}")
        await _try_set_session_value(session, f"SET SESSION innodb_lock_wait_timeout = {_LEADERBOARD_LOCK_WAIT_SECONDS}")
    elif dialect == "postgresql":
        await _try_set_session_value(session, f"SET LOCAL statement_timeout = {_LEADERBOARD_QUERY_TIMEOUT_MS}")


def _question_count_expr():
    return func.sum(case((RunRow.first_human_message.is_not(None), 1), else_=0))


_SYSTEM_MARKER_RE = re.compile(r"<(?:uploaded_files|attachments|files|system|metadata)>.*?</(?:uploaded_files|attachments|files|system|metadata)>", re.I | re.S)
_TAG_RE = re.compile(r"</?[^>]+>")


def _clean_question_text(value: str | None) -> str:
    text = str(value or "").strip()
    if not text:
        return ""
    text = _SYSTEM_MARKER_RE.sub(" ", text)
    text = _TAG_RE.sub(" ", text)
    text = re.sub(r"\s+", " ", text).strip()
    return text


def _is_displayable_question(value: str | None) -> bool:
    text = _clean_question_text(value)
    if not text:
        return False
    lowered = text.lower()
    return lowered not in {"uploaded_files", "attachments", "files"}


def _question_similarity_key(value: str) -> str:
    return re.sub(r"[\W_]+", "", value.lower())


def _message_content_to_text(content) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for item in content:
            if isinstance(item, dict):
                parts.append(str(item.get("text") or item.get("content") or ""))
            else:
                parts.append(str(item))
        return "\n".join(part for part in parts if part.strip())
    return str(content or "")


def _sse(event: str, data: dict) -> str:
    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"


def _extract_chunk_parts(chunk: object) -> tuple[str, str]:
    """Return (thinking_delta, text_delta) from a LangChain streaming chunk."""
    thinking = ""
    text = ""
    content = getattr(chunk, "content", "")
    if isinstance(content, str):
        text = content
    elif isinstance(content, list):
        for block in content:
            if isinstance(block, str):
                text += block
            elif isinstance(block, dict):
                block_type = block.get("type", "")
                if block_type in {"thinking", "thinking_delta"}:
                    thinking += block.get("thinking", "")
                elif block_type in {"text", "text_delta"}:
                    text += block.get("text", "")
                elif block_type == "reasoning":
                    thinking += block.get("content", "")
    if not thinking:
        kwargs = getattr(chunk, "additional_kwargs", {}) or {}
        reasoning = kwargs.get("reasoning_content") or ""
        if isinstance(reasoning, str):
            thinking = reasoning
    return thinking, text


def _report_cache_file():
    return get_paths().base_dir / "admin_leaderboard_reports.json"


def _load_report_cache() -> dict:
    path = _report_cache_file()
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        logger.warning("Failed to read leaderboard report cache: %s", path, exc_info=True)
        return {}
    return data if isinstance(data, dict) else {}


def _save_report_cache(cache: dict) -> None:
    path = _report_cache_file()
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(cache, ensure_ascii=False, indent=2), encoding="utf-8")


def _resolve_report_model_name(config: AppConfig, model_name: str | None) -> str | None:
    cleaned = (model_name or "").strip()
    if cleaned:
        return cleaned
    return config.models[0].name if config.models else None


def _report_cache_key(
    *,
    since: datetime,
    until: datetime,
    limit: int,
    model_name: str | None,
    settings,
) -> str:
    payload = {
        "version": 5,
        "since": _iso(since),
        "until": _iso(until),
        "limit": limit,
        "model_name": model_name,
        "settings": {
            "exclude_user_ids": sorted(uid for uid in settings.exclude_user_ids if uid),
            "include_scheduled": settings.include_scheduled,
            "include_admins": settings.include_admins,
            "include_failed": settings.include_failed,
            "cleanup_system_questions": settings.cleanup_system_questions,
        },
    }
    raw = json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str)
    return sha256(raw.encode("utf-8")).hexdigest()


def _cached_report(cache_key: str) -> dict | None:
    item = _load_report_cache().get(cache_key)
    if isinstance(item, dict) and isinstance(item.get("report"), str):
        return item
    return None


def _store_report_cache(cache_key: str, item: dict) -> None:
    cache = _load_report_cache()
    cache[cache_key] = item
    if len(cache) > 80:
        ordered = sorted(cache.items(), key=lambda kv: str(kv[1].get("generated_at") or ""))
        cache = dict(ordered[-80:])
    _save_report_cache(cache)


def _run_time_filter(since: datetime, until: datetime):
    return and_(RunRow.created_at >= since, RunRow.created_at <= until)


def _non_scheduled_run_filter():
    """Exclude scheduler-generated runs that predate scheduled_task_runs links."""
    return or_(
        RunRow.first_human_message.is_(None),
        not_(RunRow.first_human_message.like("【定时任务触发】%")),
    )


def _non_scheduled_tool_filter():
    return or_(
        ToolCallMetricRow.run_id.is_(None),
        RunRow.run_id.is_(None),
        _non_scheduled_run_filter(),
    )


def _tool_time_filter(since: datetime, until: datetime):
    return and_(ToolCallMetricRow.created_at >= since, ToolCallMetricRow.created_at <= until)


def _llm_time_filter(since: datetime, until: datetime):
    return and_(LlmCallMetricRow.created_at >= since, LlmCallMetricRow.created_at <= until)


def _exclude_user_filter(column, exclude_user_ids: set[str]):
    if not exclude_user_ids:
        return None
    return or_(column.is_(None), column.not_in(list(exclude_user_ids)))


def _apply_exclude(stmt, column, exclude_user_ids: set[str]):
    condition = _exclude_user_filter(column, exclude_user_ids)
    return stmt.where(condition) if condition is not None else stmt


def _admin_user_ids_stmt():
    return select(UserRow.id).where(UserRow.system_role == "admin")


def _scheduled_run_ids_stmt():
    return select(ScheduledTaskRunRow.agent_run_id).where(ScheduledTaskRunRow.agent_run_id.is_not(None))


def _apply_user_scope(stmt, column, *, exclude_user_ids: set[str], include_admins: bool):
    stmt = _apply_exclude(stmt, column, exclude_user_ids)
    if not include_admins:
        stmt = stmt.where(or_(column.is_(None), column.not_in(_admin_user_ids_stmt())))
    return stmt


def _apply_run_scope(
    stmt,
    *,
    user_column,
    run_id_column,
    exclude_user_ids: set[str],
    include_admins: bool,
    include_failed: bool,
    include_scheduled: bool,
):
    stmt = _apply_user_scope(stmt, user_column, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
    if not include_failed:
        stmt = stmt.where(RunRow.status != "error")
    if not include_scheduled:
        stmt = stmt.where(or_(run_id_column.is_(None), run_id_column.not_in(_scheduled_run_ids_stmt())))
        stmt = stmt.where(_non_scheduled_run_filter())
    return stmt


def _apply_tool_scope(
    stmt,
    *,
    exclude_user_ids: set[str],
    include_admins: bool,
    include_scheduled: bool,
):
    stmt = _apply_user_scope(stmt, ToolCallMetricRow.user_id, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
    if not include_scheduled:
        stmt = stmt.where(or_(ToolCallMetricRow.run_id.is_(None), ToolCallMetricRow.run_id.not_in(_scheduled_run_ids_stmt())))
        stmt = stmt.outerjoin(RunRow, RunRow.run_id == ToolCallMetricRow.run_id).where(_non_scheduled_tool_filter())
    return stmt


async def _query_user_leaderboard(
    session,
    since: datetime,
    until: datetime,
    user_id: str | None = None,
    exclude_user_ids: set[str] | None = None,
    include_scheduled: bool = True,
    include_admins: bool = True,
    include_failed: bool = True,
) -> list[AdminLeaderboardUser]:
    exclude_user_ids = exclude_user_ids or set()
    run_agg = (
        select(
            RunRow.user_id.label("user_id"),
            func.count(RunRow.run_id).label("run_count"),
            _question_count_expr().label("qa_count"),
            func.coalesce(func.sum(RunRow.llm_call_count), 0).label("llm_call_count"),
            func.coalesce(func.sum(RunRow.total_tokens), 0).label("total_tokens"),
            func.coalesce(func.sum(RunRow.total_input_tokens), 0).label("input_tokens"),
            func.coalesce(func.sum(RunRow.total_output_tokens), 0).label("output_tokens"),
            func.max(RunRow.updated_at).label("last_active_at"),
        )
        .where(_run_time_filter(since, until))
        .group_by(RunRow.user_id)
    )
    run_agg = _apply_run_scope(
        run_agg,
        user_column=RunRow.user_id,
        run_id_column=RunRow.run_id,
        exclude_user_ids=exclude_user_ids,
        include_admins=include_admins,
        include_failed=include_failed,
        include_scheduled=include_scheduled,
    ).subquery()
    thread_agg = (
        select(ThreadMetaRow.user_id.label("user_id"), func.count(ThreadMetaRow.thread_id).label("thread_count"))
        .where(and_(ThreadMetaRow.created_at >= since, ThreadMetaRow.created_at <= until))
        .group_by(ThreadMetaRow.user_id)
    )
    thread_agg = _apply_user_scope(
        thread_agg,
        ThreadMetaRow.user_id,
        exclude_user_ids=exclude_user_ids,
        include_admins=include_admins,
    ).subquery()
    tool_agg = (
        select(
            ToolCallMetricRow.user_id.label("user_id"),
            func.count(ToolCallMetricRow.id).label("tool_call_count"),
            func.sum(case((func.nullif(ToolCallMetricRow.skill_name, "").is_not(None), 1), else_=0)).label("skill_call_count"),
        )
        .where(_tool_time_filter(since, until))
        .group_by(ToolCallMetricRow.user_id)
    )
    tool_agg = _apply_tool_scope(
        tool_agg,
        exclude_user_ids=exclude_user_ids,
        include_admins=include_admins,
        include_scheduled=include_scheduled,
    ).subquery()

    stmt = (
        select(
            UserRow.id,
            UserRow.email,
            UserRow.system_role,
            func.coalesce(thread_agg.c.thread_count, 0),
            func.coalesce(run_agg.c.run_count, 0),
            func.coalesce(run_agg.c.qa_count, 0),
            func.coalesce(run_agg.c.llm_call_count, 0),
            func.coalesce(tool_agg.c.tool_call_count, 0),
            func.coalesce(tool_agg.c.skill_call_count, 0),
            func.coalesce(run_agg.c.total_tokens, 0),
            func.coalesce(run_agg.c.input_tokens, 0),
            func.coalesce(run_agg.c.output_tokens, 0),
            run_agg.c.last_active_at,
        )
        .outerjoin(run_agg, run_agg.c.user_id == UserRow.id)
        .outerjoin(thread_agg, thread_agg.c.user_id == UserRow.id)
        .outerjoin(tool_agg, tool_agg.c.user_id == UserRow.id)
    )
    if user_id:
        stmt = stmt.where(UserRow.id == user_id)
    stmt = _apply_user_scope(stmt, UserRow.id, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
    stmt = stmt.order_by(
        desc(func.coalesce(run_agg.c.run_count, 0) + func.coalesce(tool_agg.c.tool_call_count, 0)),
        desc(func.coalesce(run_agg.c.total_tokens, 0)),
    )
    rows = (await session.execute(stmt)).all()
    return [
        AdminLeaderboardUser(
            user_id=r[0],
            email=r[1] or "",
            system_role=r[2] or "user",
            thread_count=int(r[3] or 0),
            run_count=int(r[4] or 0),
            qa_count=int(r[5] or 0),
            llm_call_count=int(r[6] or 0),
            tool_call_count=int(r[7] or 0),
            skill_call_count=int(r[8] or 0),
            total_tokens=int(r[9] or 0),
            input_tokens=int(r[10] or 0),
            output_tokens=int(r[11] or 0),
            last_active_at=_iso(r[12]),
        )
        for r in rows
    ]


async def _query_call_rankings(
    session,
    since: datetime,
    until: datetime,
    *,
    user_id: str | None,
    field: str,
    limit: int,
    exclude_user_ids: set[str] | None = None,
    include_scheduled: bool = True,
    include_admins: bool = True,
) -> list[AdminLeaderboardSkill]:
    exclude_user_ids = exclude_user_ids or set()
    name_col = ToolCallMetricRow.skill_name if field == "skill" else ToolCallMetricRow.tool_name
    call_count = func.count(ToolCallMetricRow.id).label("call_count")
    last_used_at = func.max(ToolCallMetricRow.created_at).label("last_used_at")
    stmt = (
        select(
            name_col.label("name"),
            call_count,
            func.count(func.distinct(ToolCallMetricRow.user_id)).label("user_count"),
            func.sum(case((ToolCallMetricRow.status == "success", 1), else_=0)).label("success_count"),
            func.sum(case((ToolCallMetricRow.status != "success", 1), else_=0)).label("error_count"),
            func.coalesce(func.avg(ToolCallMetricRow.duration_ms), 0).label("avg_duration_ms"),
            last_used_at,
        )
        .where(_tool_time_filter(since, until), name_col.is_not(None), name_col != "")
        .group_by(name_col)
        .order_by(call_count.desc(), last_used_at.desc())
        .limit(limit)
    )
    if user_id:
        stmt = stmt.where(ToolCallMetricRow.user_id == user_id)
    stmt = _apply_tool_scope(
        stmt,
        exclude_user_ids=exclude_user_ids,
        include_admins=include_admins,
        include_scheduled=include_scheduled,
    )
    rows = (await session.execute(stmt)).all()
    return [
        AdminLeaderboardSkill(
            name=r[0] or "",
            call_count=int(r[1] or 0),
            user_count=int(r[2] or 0),
            success_count=int(r[3] or 0),
            error_count=int(r[4] or 0),
            avg_duration_ms=int(r[5] or 0),
            last_used_at=_iso(r[6]),
        )
        for r in rows
    ]


async def _query_run_dimensions(
    session,
    since: datetime,
    until: datetime,
    *,
    user_id: str | None,
    field: str,
    limit: int,
    exclude_user_ids: set[str] | None = None,
    include_scheduled: bool = True,
    include_admins: bool = True,
    include_failed: bool = True,
) -> list[AdminLeaderboardDimension]:
    exclude_user_ids = exclude_user_ids or set()
    if field == "model":
        name_col = RunRow.model_name
    elif field == "agent":
        name_col = RunRow.assistant_id
    elif field == "status":
        name_col = RunRow.status
    else:
        raise ValueError(f"Unknown run dimension: {field}")

    count = func.count(RunRow.run_id).label("count")
    last_seen_at = func.max(RunRow.updated_at).label("last_seen_at")
    stmt = (
        select(
            name_col.label("name"),
            count,
            func.count(func.distinct(RunRow.user_id)).label("user_count"),
            func.sum(case((RunRow.status == "success", 1), else_=0)).label("success_count"),
            func.sum(case((RunRow.status == "error", 1), else_=0)).label("error_count"),
            func.coalesce(func.sum(RunRow.total_tokens), 0).label("total_tokens"),
            last_seen_at,
        )
        .where(_run_time_filter(since, until), name_col.is_not(None), name_col != "")
        .group_by(name_col)
        .order_by(count.desc(), last_seen_at.desc())
        .limit(limit)
    )
    if user_id:
        stmt = stmt.where(RunRow.user_id == user_id)
    stmt = _apply_run_scope(
        stmt,
        user_column=RunRow.user_id,
        run_id_column=RunRow.run_id,
        exclude_user_ids=exclude_user_ids,
        include_admins=include_admins,
        include_failed=include_failed,
        include_scheduled=include_scheduled,
    )
    rows = (await session.execute(stmt)).all()
    return [
        AdminLeaderboardDimension(
            name=r[0] or "",
            count=int(r[1] or 0),
            user_count=int(r[2] or 0),
            success_count=int(r[3] or 0),
            error_count=int(r[4] or 0),
            total_tokens=int(r[5] or 0),
            last_seen_at=_iso(r[6]),
        )
        for r in rows
    ]


async def _query_model_rankings(
    session,
    since: datetime,
    until: datetime,
    *,
    user_id: str | None,
    limit: int,
    exclude_user_ids: set[str] | None = None,
    include_scheduled: bool = True,
    include_admins: bool = True,
    include_failed: bool = True,
) -> list[AdminLeaderboardDimension]:
    exclude_user_ids = exclude_user_ids or set()
    count = func.count(LlmCallMetricRow.id).label("count")
    last_seen_at = func.max(LlmCallMetricRow.created_at).label("last_seen_at")
    stmt = (
        select(
            LlmCallMetricRow.model_name.label("name"),
            count,
            func.count(func.distinct(LlmCallMetricRow.user_id)).label("user_count"),
            func.sum(case((LlmCallMetricRow.status == "success", 1), else_=0)).label("success_count"),
            func.sum(case((LlmCallMetricRow.status == "error", 1), else_=0)).label("error_count"),
            func.coalesce(func.sum(LlmCallMetricRow.total_tokens), 0).label("total_tokens"),
            last_seen_at,
        )
        .where(_llm_time_filter(since, until), LlmCallMetricRow.model_name.is_not(None), LlmCallMetricRow.model_name != "")
        .group_by(LlmCallMetricRow.model_name)
        .order_by(count.desc(), last_seen_at.desc())
        .limit(limit)
    )
    if user_id:
        stmt = stmt.where(LlmCallMetricRow.user_id == user_id)
    stmt = _apply_user_scope(
        stmt,
        LlmCallMetricRow.user_id,
        exclude_user_ids=exclude_user_ids,
        include_admins=include_admins,
    )
    if not include_failed:
        stmt = stmt.where(LlmCallMetricRow.status != "error")
    if not include_scheduled:
        stmt = stmt.where(or_(LlmCallMetricRow.run_id.is_(None), LlmCallMetricRow.run_id.not_in(_scheduled_run_ids_stmt())))
    rows = (await session.execute(stmt)).all()
    return [
        AdminLeaderboardDimension(
            name=r[0] or "",
            count=int(r[1] or 0),
            user_count=int(r[2] or 0),
            success_count=int(r[3] or 0),
            error_count=int(r[4] or 0),
            total_tokens=int(r[5] or 0),
            last_seen_at=_iso(r[6]),
        )
        for r in rows
    ]


async def _query_recent_questions(
    session,
    since: datetime,
    until: datetime,
    *,
    user_id: str | None,
    limit: int,
    exclude_user_ids: set[str] | None = None,
    include_scheduled: bool = True,
    include_admins: bool = True,
    include_failed: bool = True,
    cleanup_system_questions: bool = True,
) -> list[AdminLeaderboardQuestion]:
    exclude_user_ids = exclude_user_ids or set()
    stmt = (
        select(RunRow.first_human_message, RunRow.user_id, UserRow.email, RunRow.thread_id, RunRow.run_id, RunRow.created_at)
        .outerjoin(UserRow, UserRow.id == RunRow.user_id)
        .where(_run_time_filter(since, until), RunRow.first_human_message.is_not(None), RunRow.first_human_message != "")
        .order_by(RunRow.created_at.desc())
        .limit(limit)
    )
    if user_id:
        stmt = stmt.where(RunRow.user_id == user_id)
    stmt = _apply_run_scope(
        stmt,
        user_column=RunRow.user_id,
        run_id_column=RunRow.run_id,
        exclude_user_ids=exclude_user_ids,
        include_admins=include_admins,
        include_failed=include_failed,
        include_scheduled=include_scheduled,
    )
    rows = (await session.execute(stmt)).all()
    questions: list[AdminLeaderboardQuestion] = []
    for r in rows:
        text = _clean_question_text(r[0]) if cleanup_system_questions else str(r[0] or "").strip()
        if cleanup_system_questions and not _is_displayable_question(text):
            continue
        questions.append(
            AdminLeaderboardQuestion(
                text=text[:500],
                user_id=r[1] or "",
                email=r[2] or "",
                thread_id=r[3] or "",
                run_id=r[4] or "",
                asked_at=_iso(r[5]),
            )
        )
    return questions


async def _query_frequent_questions(
    session,
    since: datetime,
    until: datetime,
    limit: int,
    exclude_user_ids: set[str] | None = None,
    include_scheduled: bool = True,
    include_admins: bool = True,
    include_failed: bool = True,
    cleanup_system_questions: bool = True,
) -> list[AdminLeaderboardQuestion]:
    exclude_user_ids = exclude_user_ids or set()
    stmt = (
        select(
            RunRow.first_human_message,
            RunRow.user_id,
            UserRow.email,
            RunRow.created_at,
        )
        .outerjoin(UserRow, UserRow.id == RunRow.user_id)
        .where(_run_time_filter(since, until), RunRow.first_human_message.is_not(None), RunRow.first_human_message != "")
        .order_by(RunRow.created_at.desc())
        .limit(min(max(limit * 20, 200), 1000))
    )
    stmt = _apply_run_scope(
        stmt,
        user_column=RunRow.user_id,
        run_id_column=RunRow.run_id,
        exclude_user_ids=exclude_user_ids,
        include_admins=include_admins,
        include_failed=include_failed,
        include_scheduled=include_scheduled,
    )
    rows = (await session.execute(stmt)).all()
    groups: list[dict] = []
    for r in rows:
        text = _clean_question_text(r[0]) if cleanup_system_questions else str(r[0] or "").strip()
        if cleanup_system_questions and not _is_displayable_question(text):
            continue
        key = _question_similarity_key(text)
        if not key:
            continue
        target = None
        for group in groups:
            score = SequenceMatcher(None, key, group["key"]).ratio()
            if score >= 0.72 or key in group["key"] or group["key"] in key:
                target = group
                break
        if target is None:
            groups.append({"key": key, "text": text, "count": 1, "user_id": r[1] or "", "email": r[2] or "", "asked_at": r[3]})
        else:
            target["count"] += 1
            if r[3] and (target.get("asked_at") is None or r[3] > target["asked_at"]):
                target.update({"text": text, "user_id": r[1] or "", "email": r[2] or "", "asked_at": r[3]})

    groups.sort(key=lambda item: (item["count"], _iso(item.get("asked_at"))), reverse=True)
    questions: list[AdminLeaderboardQuestion] = []
    for group in groups[:limit]:
        text = group["text"][:460]
        if group["count"] > 1:
            text = f"{text}（相似问题 {group['count']} 次）"
        questions.append(
            AdminLeaderboardQuestion(
                text=text,
                user_id=group["user_id"],
                email=group["email"],
                asked_at=_iso(group.get("asked_at")),
            )
        )
    return questions


async def _query_trends(
    session,
    since: datetime,
    until: datetime,
    *,
    user_id: str | None = None,
    exclude_user_ids: set[str] | None = None,
    include_scheduled: bool = True,
    include_admins: bool = True,
    include_failed: bool = True,
) -> list[AdminLeaderboardTrendPoint]:
    exclude_user_ids = exclude_user_ids or set()
    bucket = func.date(RunRow.created_at)
    stmt = (
        select(
            bucket.label("date"),
            func.count(RunRow.run_id).label("runs"),
            _question_count_expr().label("questions"),
            func.coalesce(func.sum(RunRow.total_tokens), 0).label("tokens"),
            func.count(func.distinct(RunRow.user_id)).label("users"),
        )
        .where(_run_time_filter(since, until))
        .group_by(bucket)
        .order_by(bucket.asc())
    )
    if user_id:
        stmt = stmt.where(RunRow.user_id == user_id)
    stmt = _apply_run_scope(
        stmt,
        user_column=RunRow.user_id,
        run_id_column=RunRow.run_id,
        exclude_user_ids=exclude_user_ids,
        include_admins=include_admins,
        include_failed=include_failed,
        include_scheduled=include_scheduled,
    )
    rows = (await session.execute(stmt)).all()
    return [
        AdminLeaderboardTrendPoint(
            date=str(r[0]),
            runs=int(r[1] or 0),
            questions=int(r[2] or 0),
            tokens=int(r[3] or 0),
            users=int(r[4] or 0),
        )
        for r in rows
    ]


async def _query_overview(
    session,
    since: datetime,
    until: datetime,
    *,
    user_id: str | None = None,
    exclude_user_ids: set[str] | None = None,
    include_scheduled: bool = True,
    include_admins: bool = True,
    include_failed: bool = True,
) -> AdminLeaderboardOverview:
    exclude_user_ids = exclude_user_ids or set()
    run_where = [_run_time_filter(since, until)]
    thread_where = [and_(ThreadMetaRow.created_at >= since, ThreadMetaRow.created_at <= until)]
    tool_where = [_tool_time_filter(since, until)]
    if user_id:
        run_where.append(RunRow.user_id == user_id)
        thread_where.append(ThreadMetaRow.user_id == user_id)
        tool_where.append(ToolCallMetricRow.user_id == user_id)
    if exclude_user_ids:
        run_where.append(_exclude_user_filter(RunRow.user_id, exclude_user_ids))
        thread_where.append(_exclude_user_filter(ThreadMetaRow.user_id, exclude_user_ids))
        tool_where.append(_exclude_user_filter(ToolCallMetricRow.user_id, exclude_user_ids))
    if not include_admins:
        run_where.append(or_(RunRow.user_id.is_(None), RunRow.user_id.not_in(_admin_user_ids_stmt())))
        thread_where.append(or_(ThreadMetaRow.user_id.is_(None), ThreadMetaRow.user_id.not_in(_admin_user_ids_stmt())))
        tool_where.append(or_(ToolCallMetricRow.user_id.is_(None), ToolCallMetricRow.user_id.not_in(_admin_user_ids_stmt())))
    if not include_failed:
        run_where.append(RunRow.status != "error")
    if not include_scheduled:
        run_where.append(or_(RunRow.run_id.is_(None), RunRow.run_id.not_in(_scheduled_run_ids_stmt())))
        tool_where.append(or_(ToolCallMetricRow.run_id.is_(None), ToolCallMetricRow.run_id.not_in(_scheduled_run_ids_stmt())))
        tool_where.append(_non_scheduled_tool_filter())

    total_users_stmt = select(func.count(UserRow.id))
    total_users_stmt = _apply_user_scope(total_users_stmt, UserRow.id, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
    total_users = 1 if user_id else (await session.execute(total_users_stmt)).scalar_one()
    run_row = (await session.execute(
        select(
            func.count(RunRow.run_id),
            _question_count_expr(),
            func.coalesce(func.sum(RunRow.total_tokens), 0),
            func.count(func.distinct(RunRow.user_id)),
            func.sum(case((RunRow.status == "error", 1), else_=0)),
        ).where(*run_where)
    )).one()
    thread_count = (await session.execute(select(func.count(ThreadMetaRow.thread_id)).where(*thread_where))).scalar_one()
    tool_row = (await session.execute(
        select(
            func.count(ToolCallMetricRow.id),
            func.sum(case((func.nullif(ToolCallMetricRow.skill_name, "").is_not(None), 1), else_=0)),
        )
        .outerjoin(RunRow, RunRow.run_id == ToolCallMetricRow.run_id)
        .where(*tool_where)
    )).one()

    return AdminLeaderboardOverview(
        total_users=int(total_users or 0),
        active_users=int(run_row[3] or 0),
        total_threads=int(thread_count or 0),
        total_runs=int(run_row[0] or 0),
        total_questions=int(run_row[1] or 0),
        total_tokens=int(run_row[2] or 0),
        total_tool_calls=int(tool_row[0] or 0),
        total_skill_calls=int(tool_row[1] or 0),
        error_runs=int(run_row[4] or 0),
    )


def _empty_leaderboard_response(
    since_dt: datetime,
    until_dt: datetime,
    range_days: int,
    *,
    status: str,
    message: str,
) -> AdminLeaderboardResponse:
    return AdminLeaderboardResponse(
        range_days=range_days,
        since=_iso(since_dt),
        until=_iso(until_dt),
        status=status,
        message=message,
        overview=AdminLeaderboardOverview(),
    )


async def _leaderboard_pause() -> None:
    if _LEADERBOARD_QUERY_PAUSE_SECONDS > 0:
        await asyncio.sleep(_LEADERBOARD_QUERY_PAUSE_SECONDS)


async def _build_user_leaderboard_response(
    session_factory,
    settings,
    since_dt: datetime,
    until_dt: datetime,
    range_days: int,
    limit: int,
    excluded: set[str],
    include_scheduled: bool,
    include_admins: bool,
    include_failed: bool,
) -> AdminLeaderboardResponse:
    async with session_factory() as session:
        await _set_analytics_read_timeout(session)
        common = {
            "exclude_user_ids": excluded,
            "include_scheduled": include_scheduled,
            "include_admins": include_admins,
            "include_failed": include_failed,
        }
        tool_common = {
            "exclude_user_ids": excluded,
            "include_scheduled": include_scheduled,
            "include_admins": include_admins,
        }
        question_common = {
            **common,
            "cleanup_system_questions": settings.cleanup_system_questions,
        }
        overview = await _query_overview(session, since_dt, until_dt, **common)
        await _leaderboard_pause()
        users = await _query_user_leaderboard(session, since_dt, until_dt, **common)
        await _leaderboard_pause()
        skills = await _query_call_rankings(session, since_dt, until_dt, user_id=None, field="skill", limit=limit, **tool_common)
        await _leaderboard_pause()
        tools = await _query_call_rankings(session, since_dt, until_dt, user_id=None, field="tool", limit=limit, **tool_common)
        await _leaderboard_pause()
        models = await _query_model_rankings(session, since_dt, until_dt, user_id=None, limit=limit, **common)
        if not models:
            models = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="model", limit=limit, **common)
        await _leaderboard_pause()
        agents = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="agent", limit=limit, **common)
        await _leaderboard_pause()
        run_statuses = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="status", limit=limit, **common)
        await _leaderboard_pause()
        recent_questions = await _query_recent_questions(session, since_dt, until_dt, user_id=None, limit=limit, **question_common)
        await _leaderboard_pause()
        frequent_questions = await _query_frequent_questions(session, since_dt, until_dt, limit=limit, **question_common)
        await _leaderboard_pause()
        trends = await _query_trends(session, since_dt, until_dt, **common)

    return AdminLeaderboardResponse(
        range_days=range_days,
        since=_iso(since_dt),
        until=_iso(until_dt),
        status="ready",
        generated_at=_iso(datetime.utcnow()),
        overview=overview,
        users_by_activity=sorted(users, key=lambda u: (u.run_count + u.tool_call_count, u.total_tokens), reverse=True)[:limit],
        users_by_questions=sorted(users, key=lambda u: (u.qa_count, u.run_count), reverse=True)[:limit],
        users_by_tokens=sorted(users, key=lambda u: u.total_tokens, reverse=True)[:limit],
        skills=skills,
        tools=tools,
        models=models,
        agents=agents,
        run_statuses=run_statuses,
        recent_questions=recent_questions,
        frequent_questions=frequent_questions,
        trends=trends,
    )


def _leaderboard_daily_settings_hash(
    *,
    excluded: set[str],
    include_scheduled: bool,
    include_admins: bool,
    include_failed: bool,
    cleanup_system_questions: bool,
) -> str:
    payload = {
        "version": 2,
        "snapshot_limit": _LEADERBOARD_DAILY_SNAPSHOT_LIMIT,
        "exclude_user_ids": sorted(excluded),
        "include_scheduled": include_scheduled,
        "include_admins": include_admins,
        "include_failed": include_failed,
        "cleanup_system_questions": cleanup_system_questions,
    }
    return sha256(json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest()


def _resolve_leaderboard_snapshot_scope(
    settings,
    *,
    exclude_user_ids: list[str] | None = None,
    include_scheduled: bool | None = None,
    include_admins: bool | None = None,
    include_failed: bool | None = None,
) -> tuple[set[str], bool, bool, bool, str]:
    excluded = {uid for uid in settings.exclude_user_ids if uid}
    excluded.update({uid for uid in (exclude_user_ids or []) if uid})
    resolved_include_scheduled = settings.include_scheduled if include_scheduled is None else include_scheduled
    resolved_include_admins = settings.include_admins if include_admins is None else include_admins
    resolved_include_failed = settings.include_failed if include_failed is None else include_failed
    settings_hash = _leaderboard_daily_settings_hash(
        excluded=excluded,
        include_scheduled=resolved_include_scheduled,
        include_admins=resolved_include_admins,
        include_failed=resolved_include_failed,
        cleanup_system_questions=settings.cleanup_system_questions,
    )
    return excluded, resolved_include_scheduled, resolved_include_admins, resolved_include_failed, settings_hash


def _daily_job_key(stat_date: str, settings_hash: str) -> str:
    return f"{stat_date}:{settings_hash}"


def _sort_datetime(value: str | None) -> datetime:
    if not value:
        return datetime.min
    text_value = str(value).strip()
    if text_value.endswith("Z"):
        text_value = text_value[:-1] + "+00:00"
    try:
        parsed = datetime.fromisoformat(text_value)
    except ValueError:
        return datetime.min
    if parsed.tzinfo is not None:
        return parsed.astimezone(UTC).replace(tzinfo=None)
    return parsed


def _latest_iso(left: str | None, right: str | None) -> str:
    if not left:
        return str(right or "")
    if not right:
        return str(left or "")
    return str(right if _sort_datetime(right) > _sort_datetime(left) else left)


def _snapshot_is_fresh(row: AdminLeaderboardDailyStatRow | None, stat_date: str) -> bool:
    if row is None or not row.payload_json or row.status != "ready":
        return False
    if stat_date != _today_stat_date():
        return True
    generated_at = row.generated_at
    if not isinstance(generated_at, datetime):
        return False
    age = (datetime.now(BEIJING_TZ) - _as_beijing(generated_at)).total_seconds()
    return age <= _LEADERBOARD_TODAY_TTL_SECONDS


def _snapshot_response_from_row(row: AdminLeaderboardDailyStatRow | None) -> AdminLeaderboardResponse | None:
    if row is None or not row.payload_json:
        return None
    payload = row.payload_json
    if isinstance(payload, str):
        try:
            payload = json.loads(payload)
        except json.JSONDecodeError:
            return None
    try:
        return AdminLeaderboardResponse.model_validate(payload)
    except Exception:
        logger.warning("Invalid leaderboard snapshot payload: %s/%s", row.stat_date, row.settings_hash, exc_info=True)
        return None


def _merge_users_from_daily(responses: list[AdminLeaderboardResponse], limit: int) -> list[AdminLeaderboardUser]:
    users_by_id: dict[str, dict] = {}
    for response in responses:
        day_users: dict[str, AdminLeaderboardUser] = {}
        for ranking in (response.users_by_activity, response.users_by_questions, response.users_by_tokens):
            for user in ranking:
                if user.user_id:
                    day_users[user.user_id] = user
        for user in day_users.values():
            item = users_by_id.setdefault(
                user.user_id,
                {
                    "user_id": user.user_id,
                    "email": user.email,
                    "system_role": user.system_role or "user",
                    "thread_count": 0,
                    "run_count": 0,
                    "qa_count": 0,
                    "llm_call_count": 0,
                    "tool_call_count": 0,
                    "skill_call_count": 0,
                    "total_tokens": 0,
                    "input_tokens": 0,
                    "output_tokens": 0,
                    "last_active_at": "",
                },
            )
            if user.email:
                item["email"] = user.email
            if user.system_role:
                item["system_role"] = user.system_role
            for field in (
                "thread_count",
                "run_count",
                "qa_count",
                "llm_call_count",
                "tool_call_count",
                "skill_call_count",
                "total_tokens",
                "input_tokens",
                "output_tokens",
            ):
                item[field] += int(getattr(user, field) or 0)
            item["last_active_at"] = _latest_iso(item["last_active_at"], user.last_active_at)
    users = [AdminLeaderboardUser(**item) for item in users_by_id.values()]
    return sorted(users, key=lambda u: (u.run_count + u.tool_call_count, u.total_tokens), reverse=True)[:limit]


def _merge_call_rankings_from_daily(
    responses: list[AdminLeaderboardResponse],
    field_name: str,
    limit: int,
) -> list[AdminLeaderboardSkill]:
    items_by_name: dict[str, dict] = {}
    for response in responses:
        for item in getattr(response, field_name):
            if not item.name:
                continue
            merged = items_by_name.setdefault(
                item.name,
                {
                    "name": item.name,
                    "call_count": 0,
                    "user_count": 0,
                    "success_count": 0,
                    "error_count": 0,
                    "duration_weight": 0,
                    "last_used_at": "",
                },
            )
            calls = int(item.call_count or 0)
            merged["call_count"] += calls
            merged["user_count"] += int(item.user_count or 0)
            merged["success_count"] += int(item.success_count or 0)
            merged["error_count"] += int(item.error_count or 0)
            merged["duration_weight"] += int(item.avg_duration_ms or 0) * calls
            merged["last_used_at"] = _latest_iso(merged["last_used_at"], item.last_used_at)
    merged_items = [
        AdminLeaderboardSkill(
            name=item["name"],
            call_count=item["call_count"],
            user_count=item["user_count"],
            success_count=item["success_count"],
            error_count=item["error_count"],
            avg_duration_ms=int(item["duration_weight"] / item["call_count"]) if item["call_count"] else 0,
            last_used_at=item["last_used_at"],
        )
        for item in items_by_name.values()
    ]
    return sorted(merged_items, key=lambda item: (item.call_count, _sort_datetime(item.last_used_at)), reverse=True)[:limit]


def _merge_dimensions_from_daily(
    responses: list[AdminLeaderboardResponse],
    field_name: str,
    limit: int,
) -> list[AdminLeaderboardDimension]:
    items_by_name: dict[str, dict] = {}
    for response in responses:
        for item in getattr(response, field_name):
            if not item.name:
                continue
            merged = items_by_name.setdefault(
                item.name,
                {
                    "name": item.name,
                    "count": 0,
                    "user_count": 0,
                    "success_count": 0,
                    "error_count": 0,
                    "total_tokens": 0,
                    "last_seen_at": "",
                },
            )
            for field in ("count", "user_count", "success_count", "error_count", "total_tokens"):
                merged[field] += int(getattr(item, field) or 0)
            merged["last_seen_at"] = _latest_iso(merged["last_seen_at"], item.last_seen_at)
    merged_items = [AdminLeaderboardDimension(**item) for item in items_by_name.values()]
    return sorted(merged_items, key=lambda item: (item.count, _sort_datetime(item.last_seen_at)), reverse=True)[:limit]


def _merge_recent_questions_from_daily(
    responses: list[AdminLeaderboardResponse],
    limit: int,
) -> list[AdminLeaderboardQuestion]:
    questions: list[AdminLeaderboardQuestion] = []
    seen: set[str] = set()
    for response in responses:
        for question in response.recent_questions:
            key = question.run_id or f"{question.user_id}:{question.asked_at}:{question.text}"
            if key in seen:
                continue
            seen.add(key)
            questions.append(question)
    questions.sort(key=lambda item: _sort_datetime(item.asked_at), reverse=True)
    return questions[:limit]


_FREQUENT_QUESTION_COUNT_RE = re.compile(r"（相似问题\s+(\d+)\s+次）$")


def _split_frequent_question_count(text: str) -> tuple[str, int]:
    match = _FREQUENT_QUESTION_COUNT_RE.search(text)
    if not match:
        return text, 1
    return _FREQUENT_QUESTION_COUNT_RE.sub("", text).strip(), max(1, int(match.group(1)))


def _merge_frequent_questions_from_daily(
    responses: list[AdminLeaderboardResponse],
    limit: int,
) -> list[AdminLeaderboardQuestion]:
    groups: list[dict] = []
    for response in responses:
        for question in response.frequent_questions:
            base_text, count = _split_frequent_question_count(question.text)
            key = _question_similarity_key(base_text)
            if not key:
                continue
            target = None
            for group in groups:
                score = SequenceMatcher(None, key, group["key"]).ratio()
                if score >= 0.72 or key in group["key"] or group["key"] in key:
                    target = group
                    break
            if target is None:
                groups.append(
                    {
                        "key": key,
                        "text": base_text,
                        "count": count,
                        "user_id": question.user_id,
                        "email": question.email,
                        "thread_id": question.thread_id,
                        "run_id": question.run_id,
                        "asked_at": question.asked_at,
                    }
                )
            else:
                target["count"] += count
                if _sort_datetime(question.asked_at) > _sort_datetime(target.get("asked_at")):
                    target.update(
                        {
                            "text": base_text,
                            "user_id": question.user_id,
                            "email": question.email,
                            "thread_id": question.thread_id,
                            "run_id": question.run_id,
                            "asked_at": question.asked_at,
                        }
                    )
    groups.sort(key=lambda item: (item["count"], _sort_datetime(item.get("asked_at"))), reverse=True)
    questions: list[AdminLeaderboardQuestion] = []
    for group in groups[:limit]:
        text_value = group["text"][:460]
        if group["count"] > 1:
            text_value = f"{text_value}（相似问题 {group['count']} 次）"
        questions.append(
            AdminLeaderboardQuestion(
                text=text_value,
                user_id=group["user_id"],
                email=group["email"],
                thread_id=group["thread_id"],
                run_id=group["run_id"],
                asked_at=group["asked_at"],
            )
        )
    return questions


def _merge_trends_from_daily(responses: list[AdminLeaderboardResponse]) -> list[AdminLeaderboardTrendPoint]:
    trends_by_date: dict[str, AdminLeaderboardTrendPoint] = {}
    for response in responses:
        for point in response.trends:
            item = trends_by_date.setdefault(point.date, AdminLeaderboardTrendPoint(date=point.date))
            item.runs += int(point.runs or 0)
            item.questions += int(point.questions or 0)
            item.tokens += int(point.tokens or 0)
            item.users += int(point.users or 0)
    return sorted(trends_by_date.values(), key=lambda item: item.date)


def _merge_daily_leaderboard_payloads(
    responses: list[AdminLeaderboardResponse],
    *,
    since_dt: datetime,
    until_dt: datetime,
    range_days: int,
    limit: int,
    status: str,
    message: str,
) -> AdminLeaderboardResponse:
    users = _merge_users_from_daily(responses, _LEADERBOARD_DAILY_SNAPSHOT_LIMIT)
    active_user_ids = {
        user.user_id
        for user in users
        if user.user_id and (user.run_count or user.qa_count or user.tool_call_count or user.thread_count)
    }
    overview = AdminLeaderboardOverview(
        total_users=max((item.overview.total_users for item in responses), default=0),
        active_users=len(active_user_ids) if active_user_ids else sum(item.overview.active_users for item in responses),
        total_threads=sum(item.overview.total_threads for item in responses),
        total_runs=sum(item.overview.total_runs for item in responses),
        total_questions=sum(item.overview.total_questions for item in responses),
        total_tokens=sum(item.overview.total_tokens for item in responses),
        total_skill_calls=sum(item.overview.total_skill_calls for item in responses),
        total_tool_calls=sum(item.overview.total_tool_calls for item in responses),
        error_runs=sum(item.overview.error_runs for item in responses),
    )
    generated_at = ""
    for response in responses:
        generated_at = _latest_iso(generated_at, response.generated_at)
    return AdminLeaderboardResponse(
        range_days=range_days,
        since=_iso(since_dt),
        until=_iso(until_dt),
        status=status,
        generated_at=generated_at or _iso(datetime.now(BEIJING_TZ)),
        message=message,
        overview=overview,
        users_by_activity=sorted(users, key=lambda u: (u.run_count + u.tool_call_count, u.total_tokens), reverse=True)[:limit],
        users_by_questions=sorted(users, key=lambda u: (u.qa_count, u.run_count), reverse=True)[:limit],
        users_by_tokens=sorted(users, key=lambda u: u.total_tokens, reverse=True)[:limit],
        skills=_merge_call_rankings_from_daily(responses, "skills", limit),
        tools=_merge_call_rankings_from_daily(responses, "tools", limit),
        models=_merge_dimensions_from_daily(responses, "models", limit),
        agents=_merge_dimensions_from_daily(responses, "agents", limit),
        run_statuses=_merge_dimensions_from_daily(responses, "run_statuses", limit),
        recent_questions=_merge_recent_questions_from_daily(responses, limit),
        frequent_questions=_merge_frequent_questions_from_daily(responses, limit),
        trends=_merge_trends_from_daily(responses),
    )


async def _load_daily_snapshot_rows(
    session_factory,
    stat_dates: list[str],
    settings_hash: str,
) -> dict[str, AdminLeaderboardDailyStatRow]:
    if not stat_dates:
        return {}
    async with session_factory() as session:
        rows = (
            await session.execute(
                select(AdminLeaderboardDailyStatRow).where(
                    AdminLeaderboardDailyStatRow.stat_date.in_(stat_dates),
                    AdminLeaderboardDailyStatRow.settings_hash == settings_hash,
                )
            )
        ).scalars().all()
    return {row.stat_date: row for row in rows}


async def _store_daily_snapshot(
    session_factory,
    stat_date: str,
    settings_hash: str,
    *,
    status: str,
    payload: dict | None = None,
    message: str | None = None,
    error: str | None = None,
    generated_at: datetime | None = None,
) -> None:
    async with session_factory() as session:
        now = datetime.now(UTC)
        row = await session.get(
            AdminLeaderboardDailyStatRow,
            {"stat_date": stat_date, "settings_hash": settings_hash},
        )
        if row is None:
            row = AdminLeaderboardDailyStatRow(
                stat_date=stat_date,
                settings_hash=settings_hash,
                status=status,
                payload_json=payload or {},
                message=message,
                error=error,
                generated_at=generated_at,
                created_at=now,
                updated_at=now,
            )
            session.add(row)
        else:
            row.status = status
            if payload is not None:
                row.payload_json = payload
            if message is not None:
                row.message = message
            row.error = error
            if generated_at is not None:
                row.generated_at = generated_at
            row.updated_at = now
        await session.commit()


async def _queue_daily_leaderboard_snapshots(
    session_factory,
    stat_dates: list[str],
    settings_hash: str,
    *,
    force: bool = False,
    message: str = "等待排行榜后台统计任务处理。",
) -> list[str]:
    queued: list[str] = []
    if not stat_dates:
        return queued
    async with session_factory() as session:
        now = datetime.now(UTC)
        rows = (
            await session.execute(
                select(AdminLeaderboardDailyStatRow).where(
                    AdminLeaderboardDailyStatRow.stat_date.in_(stat_dates),
                    AdminLeaderboardDailyStatRow.settings_hash == settings_hash,
                )
            )
        ).scalars().all()
        rows_by_date = {row.stat_date: row for row in rows}
        for stat_date in stat_dates:
            row = rows_by_date.get(stat_date)
            if row is None:
                row = AdminLeaderboardDailyStatRow(
                    stat_date=stat_date,
                    settings_hash=settings_hash,
                    status="queued",
                    payload_json={},
                    message=message,
                    created_at=now,
                    updated_at=now,
                )
                session.add(row)
                queued.append(stat_date)
                continue
            if not force and _snapshot_is_fresh(row, stat_date):
                continue
            if row.status == "running":
                continue
            row.status = "queued"
            row.message = message
            row.error = None
            row.updated_at = now
            queued.append(stat_date)
        if queued:
            await session.commit()
    return queued


async def _build_daily_leaderboard_snapshot(
    session_factory,
    settings,
    stat_date: str,
    *,
    excluded: set[str],
    include_scheduled: bool,
    include_admins: bool,
    include_failed: bool,
) -> AdminLeaderboardResponse:
    since_dt, until_dt = _bounds_for_stat_date(stat_date)
    return await _build_user_leaderboard_response(
        session_factory,
        settings,
        since_dt,
        until_dt,
        1,
        _LEADERBOARD_DAILY_SNAPSHOT_LIMIT,
        excluded,
        include_scheduled,
        include_admins,
        include_failed,
    )


async def _refresh_daily_leaderboard_snapshot(
    job_key: str,
    *,
    session_factory,
    settings,
    stat_date: str,
    settings_hash: str,
    excluded: set[str],
    include_scheduled: bool,
    include_admins: bool,
    include_failed: bool,
) -> None:
    acquired = False
    try:
        if _LEADERBOARD_START_DELAY_SECONDS > 0:
            await asyncio.sleep(_LEADERBOARD_START_DELAY_SECONDS)
        await _store_daily_snapshot(
            session_factory,
            stat_date,
            settings_hash,
            status="running",
            message="正在生成当天排行榜快照。",
        )
        await _leaderboard_semaphore.acquire()
        acquired = True
        response = await asyncio.wait_for(
            _build_daily_leaderboard_snapshot(
                session_factory,
                settings,
                stat_date,
                excluded=excluded,
                include_scheduled=include_scheduled,
                include_admins=include_admins,
                include_failed=include_failed,
            ),
            timeout=_LEADERBOARD_TIMEOUT_SECONDS,
        )
        generated_at = datetime.now(BEIJING_TZ)
        response.generated_at = _iso(generated_at)
        await _store_daily_snapshot(
            session_factory,
            stat_date,
            settings_hash,
            status="ready",
            payload=response.model_dump(mode="json"),
            message="",
            error=None,
            generated_at=generated_at,
        )
    except asyncio.TimeoutError:
        logger.warning("Daily leaderboard snapshot timed out: %s", job_key)
        await _store_daily_snapshot(
            session_factory,
            stat_date,
            settings_hash,
            status="error",
            message="当天排行榜快照统计超时，后台会在下次访问时重试。",
            error="timeout",
        )
    except Exception as exc:
        logger.exception("Daily leaderboard snapshot failed: %s", job_key)
        await _store_daily_snapshot(
            session_factory,
            stat_date,
            settings_hash,
            status="error",
            message="当天排行榜快照统计失败，后台会在下次访问时重试。",
            error=str(exc),
        )
    finally:
        if acquired:
            _leaderboard_semaphore.release()


async def _load_or_schedule_daily_leaderboard(
    session_factory,
    settings,
    since_dt: datetime,
    until_dt: datetime,
    *,
    limit: int,
    exclude_user_ids: list[str] | None = None,
    include_scheduled: bool | None = None,
    include_admins: bool | None = None,
    include_failed: bool | None = None,
) -> AdminLeaderboardResponse:
    stat_dates = _stat_dates_for_range(since_dt, until_dt)
    range_days = len(stat_dates)
    (
        _excluded,
        _resolved_include_scheduled,
        _resolved_include_admins,
        _resolved_include_failed,
        settings_hash,
    ) = _resolve_leaderboard_snapshot_scope(
        settings,
        exclude_user_ids=exclude_user_ids,
        include_scheduled=include_scheduled,
        include_admins=include_admins,
        include_failed=include_failed,
    )
    rows = await _load_daily_snapshot_rows(session_factory, stat_dates, settings_hash)
    responses: list[AdminLeaderboardResponse] = []
    refresh_dates: list[str] = []
    for stat_date in stat_dates:
        row = rows.get(stat_date)
        response = _snapshot_response_from_row(row)
        if response is not None:
            responses.append(response)
        if not _snapshot_is_fresh(row, stat_date):
            refresh_dates.append(stat_date)

    queued_dates: list[str] = []
    if refresh_dates and _LEADERBOARD_QUEUE_ON_REQUEST_ENABLED:
        queued_dates = await _queue_daily_leaderboard_snapshots(
            session_factory,
            refresh_dates,
            settings_hash,
            message="排行榜页面请求发现该日快照缺失或过期，已加入后台统计队列。",
        )

    if responses:
        status = "refreshing" if refresh_dates else "ready"
        message = ""
        if refresh_dates:
            if _LEADERBOARD_QUEUE_ON_REQUEST_ENABLED:
                message = f"缺失/过期的 {len(refresh_dates)} 天快照已加入队列，当前展示已完成的 {len(responses)} 天数据。后台统计任务会继续补齐。"
            else:
                message = f"缺失/过期的 {len(refresh_dates)} 天快照尚未生成，当前展示已完成的 {len(responses)} 天数据。"
        return _merge_daily_leaderboard_payloads(
            responses,
            since_dt=since_dt,
            until_dt=until_dt,
            range_days=range_days,
            limit=limit,
            status=status,
            message=message,
        )
    return _empty_leaderboard_response(
        since_dt,
        until_dt,
        range_days,
        status="running",
        message=(
            f"该范围暂无可用快照，已加入 {len(queued_dates)} 天统计队列。后台统计任务会继续补齐。"
            if _LEADERBOARD_QUEUE_ON_REQUEST_ENABLED
            else "该范围暂无可用快照，请开启排行榜后台统计任务或手动补算对应日期。"
        ),
    )


@router.get("/leaderboard", response_model=AdminLeaderboardResponse)
async def get_user_leaderboard(
    request: Request,
    days: int = Query(default=30, ge=1, le=365),
    since: str | None = Query(default=None, description="Inclusive range start, ISO datetime"),
    until: str | None = Query(default=None, description="Inclusive range end, ISO datetime"),
    exclude_user_ids: list[str] | None = Query(default=None, description="User ids excluded from this leaderboard aggregation"),
    include_scheduled: bool | None = Query(default=None, description="Include runs created by scheduled tasks"),
    include_admins: bool | None = Query(default=None, description="Include admin accounts in leaderboard aggregation"),
    include_failed: bool | None = Query(default=None, description="Include failed runs and LLM calls"),
    limit: int = Query(default=20, ge=5, le=100),
) -> AdminLeaderboardResponse:
    """Return admin analytics for user activity, Q&A, skills, questions, and trends."""
    await _require_admin(request)
    session_factory = get_session_factory()
    if session_factory is None:
        raise HTTPException(status_code=503, detail="Analytics requires database persistence")

    settings = load_system_settings().leaderboard
    since_dt, until_dt, _range_days = _resolve_range(days, since, until)
    return await _load_or_schedule_daily_leaderboard(
        session_factory,
        settings,
        since_dt,
        until_dt,
        limit=limit,
        exclude_user_ids=exclude_user_ids,
        include_scheduled=include_scheduled,
        include_admins=include_admins,
        include_failed=include_failed,
    )


@router.get("/leaderboard/snapshots", response_model=AdminLeaderboardSnapshotStatusResponse)
async def get_leaderboard_snapshot_status(
    request: Request,
    days: int = Query(default=30, ge=1, le=365),
    since: str | None = Query(default=None, description="Inclusive range start, ISO datetime"),
    until: str | None = Query(default=None, description="Inclusive range end, ISO datetime"),
) -> AdminLeaderboardSnapshotStatusResponse:
    """Return per-day snapshot status for the current leaderboard settings."""
    await _require_admin(request)
    session_factory = get_session_factory()
    if session_factory is None:
        raise HTTPException(status_code=503, detail="Analytics requires database persistence")

    settings = load_system_settings().leaderboard
    since_dt, until_dt, _range_days = _resolve_range(days, since, until)
    stat_dates = _stat_dates_for_range(since_dt, until_dt)
    *_scope, settings_hash = _resolve_leaderboard_snapshot_scope(settings)
    rows = await _load_daily_snapshot_rows(session_factory, stat_dates, settings_hash)
    items: list[AdminLeaderboardSnapshotStatusItem] = []
    counts = {"ready": 0, "queued": 0, "running": 0, "error": 0, "missing": 0}
    for stat_date in stat_dates:
        row = rows.get(stat_date)
        if row is None:
            counts["missing"] += 1
            items.append(AdminLeaderboardSnapshotStatusItem(stat_date=stat_date))
            continue
        status = row.status or "missing"
        counts[status if status in counts else "missing"] += 1
        items.append(
            AdminLeaderboardSnapshotStatusItem(
                stat_date=stat_date,
                status=status,
                has_payload=bool(row.payload_json),
                generated_at=_iso(row.generated_at),
                updated_at=_iso(row.updated_at),
                message=row.message or "",
                error=row.error or "",
            )
        )
    return AdminLeaderboardSnapshotStatusResponse(
        settings_hash=settings_hash,
        since=_iso(since_dt),
        until=_iso(until_dt),
        total_days=len(stat_dates),
        ready_days=counts["ready"],
        queued_days=counts["queued"],
        running_days=counts["running"],
        error_days=counts["error"],
        missing_days=counts["missing"],
        items=items,
    )


@router.post("/leaderboard/snapshots/backfill", response_model=AdminLeaderboardSnapshotBackfillResponse)
async def queue_leaderboard_snapshot_backfill(
    body: AdminLeaderboardSnapshotBackfillRequest,
    request: Request,
) -> AdminLeaderboardSnapshotBackfillResponse:
    """Queue a date range for the leaderboard background snapshot scheduler."""
    await _require_admin(request)
    session_factory = get_session_factory()
    if session_factory is None:
        raise HTTPException(status_code=503, detail="Analytics requires database persistence")

    settings = load_system_settings().leaderboard
    since_dt, until_dt, _range_days = _resolve_range(body.days, body.since, body.until)
    stat_dates = _stat_dates_for_range(since_dt, until_dt)
    *_scope, settings_hash = _resolve_leaderboard_snapshot_scope(settings)
    queued_dates = await _queue_daily_leaderboard_snapshots(
        session_factory,
        stat_dates,
        settings_hash,
        force=body.force,
        message="管理员手动补算请求已加入排行榜后台统计队列。",
    )
    return AdminLeaderboardSnapshotBackfillResponse(
        settings_hash=settings_hash,
        since=_iso(since_dt),
        until=_iso(until_dt),
        total_days=len(stat_dates),
        queued_dates=queued_dates,
        message=f"已加入 {len(queued_dates)} 天快照队列，后台统计任务会继续补齐。",
    )


async def _build_leaderboard_report_prompt(
    session_factory,
    settings,
    since_dt: datetime,
    until_dt: datetime,
    limit: int,
) -> str:
    snapshot = await _load_or_schedule_daily_leaderboard(
        session_factory,
        settings,
        since_dt,
        until_dt,
        limit=max(limit, 30),
    )
    if snapshot.status == "running" and snapshot.overview.total_runs == 0:
        raise HTTPException(status_code=202, detail="排行榜快照正在后台生成，请稍后再生成报告。")

    overview = snapshot.overview
    skills = snapshot.skills
    tools = snapshot.tools
    models = snapshot.models
    agents = snapshot.agents
    run_statuses = snapshot.run_statuses
    recent_questions = snapshot.recent_questions
    frequent_questions = snapshot.frequent_questions
    trends = snapshot.trends
    top_users = snapshot.users_by_activity[:12]
    context = {
        "range": {"since": _iso(since_dt), "until": _iso(until_dt)},
        "overview": overview.model_dump(),
        "top_users": [u.model_dump() for u in top_users],
        "top_skills": [s.model_dump() for s in skills[:12]],
        "top_tools": [t.model_dump() for t in tools[:12]],
        "top_models": [m.model_dump() for m in models[:12]],
        "agents": [a.model_dump() for a in agents[:12]],
        "run_statuses": [s.model_dump() for s in run_statuses[:12]],
        "frequent_questions": [q.model_dump() for q in frequent_questions[:30]],
        "recent_questions": [q.model_dump() for q in recent_questions[:40]],
        "trends": [t.model_dump() for t in trends[-21:]],
    }
    return (
        "请基于以下管理员排行榜统计数据，生成一份中文用户行为分析报告。"
        "报告需要有结论，不要只罗列数字；请使用 Markdown 标题、列表和表格，让前端可以直接渲染。"
        "重点要求："
        "1) 你是专门负责分析用户行为的专家，不要自称企业 AI 应用运营分析师；"
        "2) 优先突出用户最近常用问题、最近关注的实时问题、事件、主题和业务方向，必须引用 recent_questions 中的近期问题作为依据；"
        "3) 单独分析 frequent_questions，归纳高频关注簇，说明这些高频问题反映出的需求强度和重复频率；"
        "4) 结合 top_users、top_skills、top_tools、top_models 和 trends，说明使用频率、使用增长、常用能力和重点用户行为画像；"
        "5) 可以提及使用强度、关注变化和行为机会，但不要把报告重点放在系统本身的问题、故障、失败率或回答质量缺陷上；"
        "6) 给管理员 3-5 条可执行建议，建议应围绕用户关注主题、资源配置、内容供给、培训引导和高频能力优化。"
        "输出结构：摘要、最近关注、高频问题与频率、重点用户行为、常用能力、趋势观察、行动建议。"
        "\n\n统计数据如下：\n"
        f"{json.dumps(context, ensure_ascii=False, default=str)}"
    )


@router.post("/leaderboard/report", response_model=AdminLeaderboardReportResponse)
async def generate_user_leaderboard_report(
    request: Request,
    days: int = Query(default=30, ge=1, le=365),
    since: str | None = Query(default=None, description="Inclusive range start, ISO datetime"),
    until: str | None = Query(default=None, description="Inclusive range end, ISO datetime"),
    limit: int = Query(default=30, ge=10, le=100),
    model_name: str | None = Query(default=None, description="Model name used to generate the report"),
    config: AppConfig = Depends(get_config),
) -> AdminLeaderboardReportResponse:
    """Generate an LLM-written admin report for the current leaderboard window."""
    await _require_admin(request)
    session_factory = get_session_factory()
    if session_factory is None:
        raise HTTPException(status_code=503, detail="Analytics requires database persistence")

    settings = load_system_settings().leaderboard
    since_dt, until_dt, _range_days = _resolve_range(days, since, until)
    resolved_model_name = _resolve_report_model_name(config, model_name)
    cache_key = _report_cache_key(since=since_dt, until=until_dt, limit=limit, model_name=resolved_model_name, settings=settings)
    cached = _cached_report(cache_key)
    if cached:
        return AdminLeaderboardReportResponse(
            report=cached["report"],
            generated_at=str(cached.get("generated_at") or ""),
            cached=True,
            model_name=str(cached.get("model_name") or resolved_model_name or ""),
        )

    prompt = await _build_leaderboard_report_prompt(session_factory, settings, since_dt, until_dt, limit)
    try:
        model_cfg = config.get_model_config(resolved_model_name) if resolved_model_name else None
        thinking_enabled = bool(model_cfg and getattr(model_cfg, "supports_thinking", False))
        model = create_chat_model(name=resolved_model_name, thinking_enabled=thinking_enabled, app_config=config)
        response = await model.ainvoke(
            [
                SystemMessage(content="你是专门负责分析用户行为的专家，擅长从用户最近常用问题、关注主题、使用频率和调用指标中生成管理报告；报告聚焦用户行为，不讨论系统本身问题。"),
                HumanMessage(content=prompt),
            ],
            config={"run_name": "admin_user_leaderboard_report"},
        )
    except Exception as exc:
        logger.exception("Failed to generate leaderboard report")
        raise HTTPException(status_code=503, detail=f"Report generation failed: {exc}") from exc

    report = _message_content_to_text(getattr(response, "content", response)).strip()
    generated_at = _iso(_now())
    _store_report_cache(cache_key, {"report": report, "generated_at": generated_at, "model_name": resolved_model_name})
    return AdminLeaderboardReportResponse(
        report=report,
        generated_at=generated_at,
        cached=False,
        model_name=resolved_model_name,
    )


@router.post("/leaderboard/report/stream")
async def stream_user_leaderboard_report(
    request: Request,
    days: int = Query(default=30, ge=1, le=365),
    since: str | None = Query(default=None, description="Inclusive range start, ISO datetime"),
    until: str | None = Query(default=None, description="Inclusive range end, ISO datetime"),
    limit: int = Query(default=30, ge=10, le=100),
    model_name: str | None = Query(default=None, description="Model name used to generate the report"),
    config: AppConfig = Depends(get_config),
) -> StreamingResponse:
    """Stream an LLM-written admin report, returning cached reports when available."""
    await _require_admin(request)
    session_factory = get_session_factory()
    if session_factory is None:
        raise HTTPException(status_code=503, detail="Analytics requires database persistence")

    settings = load_system_settings().leaderboard
    since_dt, until_dt, _range_days = _resolve_range(days, since, until)
    resolved_model_name = _resolve_report_model_name(config, model_name)
    cache_key = _report_cache_key(since=since_dt, until=until_dt, limit=limit, model_name=resolved_model_name, settings=settings)

    async def generate() -> AsyncIterator[str]:
        cached = _cached_report(cache_key)
        if cached:
            yield _sse(
                "cached",
                {
                    "report": cached["report"],
                    "generated_at": cached.get("generated_at"),
                    "model_name": cached.get("model_name") or resolved_model_name,
                },
            )
            yield _sse("done", {"cached": True, "generated_at": cached.get("generated_at"), "model_name": cached.get("model_name") or resolved_model_name})
            return

        report_parts: list[str] = []
        generated_at = ""
        try:
            yield _sse("meta", {"cached": False, "model_name": resolved_model_name})
            prompt = await _build_leaderboard_report_prompt(session_factory, settings, since_dt, until_dt, limit)
            model_cfg = config.get_model_config(resolved_model_name) if resolved_model_name else None
            thinking_enabled = bool(model_cfg and getattr(model_cfg, "supports_thinking", False))
            model = create_chat_model(name=resolved_model_name, thinking_enabled=thinking_enabled, app_config=config)
            async for chunk in model.astream(
                [
                    SystemMessage(content="你是专门负责分析用户行为的专家，擅长从用户最近常用问题、关注主题、使用频率和调用指标中生成管理报告；报告聚焦用户行为，不讨论系统本身问题。"),
                    HumanMessage(content=prompt),
                ],
                config={"run_name": "admin_user_leaderboard_report"},
            ):
                thinking_delta, text_delta = _extract_chunk_parts(chunk)
                if thinking_delta:
                    yield _sse("thinking", {"chunk": thinking_delta})
                if text_delta:
                    report_parts.append(text_delta)
                    yield _sse("text", {"chunk": text_delta})
            report = "".join(report_parts).strip()
            generated_at = _iso(_now())
            _store_report_cache(cache_key, {"report": report, "generated_at": generated_at, "model_name": resolved_model_name})
            yield _sse("done", {"cached": False, "generated_at": generated_at, "model_name": resolved_model_name, "report": report})
        except Exception as exc:
            logger.exception("Failed to stream leaderboard report")
            yield _sse("error", {"message": str(exc), "generated_at": generated_at, "model_name": resolved_model_name})

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


@router.get("/{user_id}/analytics", response_model=AdminUserAnalyticsResponse)
async def get_user_analytics(
    user_id: str,
    request: Request,
    days: int = Query(default=30, ge=1, le=365),
    since: str | None = Query(default=None, description="Inclusive range start, ISO datetime"),
    until: str | None = Query(default=None, description="Inclusive range end, ISO datetime"),
    limit: int = Query(default=20, ge=5, le=100),
) -> AdminUserAnalyticsResponse:
    """Return one user's admin analytics profile."""
    await _require_admin(request)
    session_factory = get_session_factory()
    if session_factory is None:
        raise HTTPException(status_code=503, detail="Analytics requires database persistence")

    settings = load_system_settings().leaderboard
    common = {
        "include_scheduled": settings.include_scheduled,
        "include_admins": settings.include_admins,
        "include_failed": settings.include_failed,
    }
    tool_common = {
        "include_scheduled": settings.include_scheduled,
        "include_admins": settings.include_admins,
    }
    question_common = {
        **common,
        "cleanup_system_questions": settings.cleanup_system_questions,
    }
    since_dt, until_dt, range_days = _resolve_range(days, since, until)
    async with session_factory() as session:
        await _set_analytics_read_timeout(session)
        users = await _query_user_leaderboard(session, since_dt, until_dt, user_id=user_id, **common)
        if not users:
            raise HTTPException(status_code=404, detail="User not found")
        overview = await _query_overview(session, since_dt, until_dt, user_id=user_id, **common)
        skills = await _query_call_rankings(session, since_dt, until_dt, user_id=user_id, field="skill", limit=limit, **tool_common)
        tools = await _query_call_rankings(session, since_dt, until_dt, user_id=user_id, field="tool", limit=limit, **tool_common)
        models = await _query_model_rankings(session, since_dt, until_dt, user_id=user_id, limit=limit, **common)
        if not models:
            models = await _query_run_dimensions(session, since_dt, until_dt, user_id=user_id, field="model", limit=limit, **common)
        agents = await _query_run_dimensions(session, since_dt, until_dt, user_id=user_id, field="agent", limit=limit, **common)
        run_statuses = await _query_run_dimensions(session, since_dt, until_dt, user_id=user_id, field="status", limit=limit, **common)
        recent_questions = await _query_recent_questions(session, since_dt, until_dt, user_id=user_id, limit=limit, **question_common)
        trends = await _query_trends(session, since_dt, until_dt, user_id=user_id, **common)

    return AdminUserAnalyticsResponse(
        range_days=range_days,
        since=_iso(since_dt),
        until=_iso(until_dt),
        user=users[0],
        overview=overview,
        skills=skills,
        tools=tools,
        models=models,
        agents=agents,
        run_statuses=run_statuses,
        recent_questions=recent_questions,
        trends=trends,
    )


# ── Q&A audit list ─────────────────────────────────────────────────────────


class AdminUserQaItem(BaseModel):
    run_id: str
    thread_id: str = ""
    thread_title: str = ""
    assistant_id: str = ""
    status: str = ""
    model_name: str = ""
    question: str = ""
    answer: str = ""
    total_tokens: int = 0
    created_at: str = ""
    updated_at: str = ""


class AdminUserQaPageResponse(BaseModel):
    user_id: str
    page: int
    page_size: int
    total: int
    items: list[AdminUserQaItem] = Field(default_factory=list)


@router.get("/{user_id}/qa", response_model=AdminUserQaPageResponse)
async def list_user_qa(
    user_id: str,
    request: Request,
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=100),
    q: str | None = Query(default=None, description="Search question, answer, run id, or thread id"),
    since: str | None = Query(default=None, description="Inclusive range start, ISO datetime"),
    until: str | None = Query(default=None, description="Inclusive range end, ISO datetime"),
) -> AdminUserQaPageResponse:
    """Return one user's Q&A runs from lightweight run summary fields."""
    await _require_admin(request)
    session_factory = get_session_factory()
    if session_factory is None:
        raise HTTPException(status_code=503, detail="Q&A audit requires database persistence")

    filters = [
        RunRow.user_id == user_id,
        RunRow.first_human_message.is_not(None),
        RunRow.first_human_message != "",
    ]
    if since or until:
        since_dt, until_dt, _range_days = _resolve_range(365, since, until)
        filters.append(_run_time_filter(since_dt, until_dt))
    needle = (q or "").strip()
    if needle:
        like = f"%{needle}%"
        filters.append(
            or_(
                RunRow.first_human_message.like(like),
                RunRow.last_ai_message.like(like),
                RunRow.run_id.like(like),
                RunRow.thread_id.like(like),
            )
        )

    offset = (page - 1) * page_size
    async with session_factory() as session:
        await _set_analytics_read_timeout(session)
        total = (
            await session.execute(
                select(func.count(RunRow.run_id)).where(*filters)
            )
        ).scalar_one()
        rows = (
            await session.execute(
                select(
                    RunRow.run_id,
                    RunRow.thread_id,
                    ThreadMetaRow.display_name,
                    RunRow.assistant_id,
                    RunRow.status,
                    RunRow.model_name,
                    RunRow.first_human_message,
                    RunRow.last_ai_message,
                    RunRow.total_tokens,
                    RunRow.created_at,
                    RunRow.updated_at,
                )
                .outerjoin(
                    ThreadMetaRow,
                    and_(
                        ThreadMetaRow.thread_id == RunRow.thread_id,
                        ThreadMetaRow.user_id == RunRow.user_id,
                    ),
                )
                .where(*filters)
                .order_by(RunRow.created_at.desc(), RunRow.run_id.desc())
                .limit(page_size)
                .offset(offset)
            )
        ).all()

    return AdminUserQaPageResponse(
        user_id=user_id,
        page=page,
        page_size=page_size,
        total=int(total or 0),
        items=[
            AdminUserQaItem(
                run_id=r[0] or "",
                thread_id=r[1] or "",
                thread_title=r[2] or "",
                assistant_id=r[3] or "",
                status=r[4] or "",
                model_name=r[5] or "",
                question=_clean_question_text(r[6])[:2000],
                answer=str(r[7] or "").strip()[:6000],
                total_tokens=int(r[8] or 0),
                created_at=_iso(r[9]),
                updated_at=_iso(r[10]),
            )
            for r in rows
        ],
    )


# ── Threads ────────────────────────────────────────────────────────────────


class AdminThreadItem(BaseModel):
    thread_id: str
    display_name: str = ""
    status: str = "idle"
    created_at: str = ""
    updated_at: str = ""


@router.get("/{user_id}/threads", response_model=list[AdminThreadItem])
async def list_user_threads(
    user_id: str,
    request: Request,
    limit: int = Query(default=100, ge=1, le=500),
    offset: int = Query(default=0, ge=0),
) -> list[AdminThreadItem]:
    """List one user's conversation threads, newest first."""
    await _require_admin(request)
    repo = get_thread_store(request)
    rows = await repo.search(limit=limit, offset=offset, user_id=user_id)
    return [
        AdminThreadItem(
            thread_id=r["thread_id"],
            display_name=r.get("display_name") or "",
            status=r.get("status", "idle"),
            created_at=str(r.get("created_at") or ""),
            updated_at=str(r.get("updated_at") or ""),
        )
        for r in rows
    ]


@router.get("/{user_id}/threads/{thread_id}/messages", response_model=list[dict])
async def list_user_thread_messages(
    user_id: str,
    thread_id: str,
    request: Request,
    limit: int = Query(default=400, ge=1, le=1000),
) -> list[dict]:
    """Return the conversation messages of one of the user's threads.

    Read straight from the LangGraph checkpoint — the same source the chat UI
    renders — rather than the optional ``run_events`` audit log, which may be
    unconfigured (``run_events.backend=memory``) or empty for older threads.
    """
    await _require_admin(request)

    repo = get_thread_store(request)
    # ``user_id=None`` bypasses the owner filter so the admin can fetch any row;
    # we then assert the thread really belongs to the target user.
    thread = await repo.get(thread_id, user_id=None)
    if thread is None:
        raise HTTPException(status_code=404, detail="Thread not found")
    owner = thread.get("user_id")
    if owner is not None and owner != user_id:
        raise HTTPException(status_code=404, detail="Thread does not belong to this user")

    checkpointer = get_checkpointer(request)
    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
    try:
        checkpoint_tuple = await checkpointer.aget_tuple(config)
    except Exception as exc:  # noqa: BLE001 - surface a clean error
        logger.error("Failed to read checkpoint for thread %s: %s", thread_id, exc, exc_info=True)
        raise HTTPException(status_code=500, detail="Failed to read thread state") from exc
    if checkpoint_tuple is None:
        return []

    checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {}
    channel_values = checkpoint.get("channel_values", {}) if isinstance(checkpoint, dict) else {}
    values = serialize_channel_values(channel_values)
    messages = values.get("messages", [])
    if not isinstance(messages, list):
        return []
    # Keep the most recent ``limit`` messages — long threads stay bounded.
    return messages[-limit:]


# ── Memory ─────────────────────────────────────────────────────────────────


class AdminMemoryBucket(BaseModel):
    name: str = Field(..., description="Bucket name: 'global' or the agent id (used for CRUD)")
    display_name: str = Field("", description="Human-readable name shown in the UI")
    data: dict = Field(default_factory=dict, description="Raw memory data for the bucket")


class AdminMemoryEntryAddRequest(BaseModel):
    content: str = Field(..., min_length=1, description="Entry content to add")
    agent: str | None = Field(None, description="Agent name; None = global USER.md bucket")


class AdminMemoryEntryReplaceRequest(BaseModel):
    old_text: str = Field(..., min_length=1, description="Short unique substring to locate the entry")
    content: str = Field(..., min_length=1, description="New content to replace with")
    agent: str | None = Field(None, description="Agent name; None = global USER.md bucket")


class AdminMemoryResponse(BaseModel):
    user_id: str
    version: str = Field(default="v2", description="记忆版本：v1 或 v2")
    buckets: list[AdminMemoryBucket] = Field(default_factory=list)


def _read_memory_bucket(user_id: str, agent_id: str | None) -> dict:
    """Read one Memory V2 file into a JSON-able ``{"entries": [...]}`` dict."""
    provider = BuiltinFileProvider()
    provider.initialize(user_id=user_id, agent_id=agent_id)
    target = "user" if agent_id is None else "memory"
    return {"entries": provider.read(target)}


def _read_v1_memory_bucket(user_id: str, agent_id: str | None) -> dict:
    """Read one V1 memory.json into its raw dict (facts / user / history)."""
    provider = _build_v1_provider(user_id, agent_id)
    return provider.read_v1_data()


@router.get("/{user_id}/memory", response_model=AdminMemoryResponse)
async def get_user_memory(user_id: str, request: Request) -> AdminMemoryResponse:
    """Return every memory bucket for one user (V1 or V2 depending on config)."""
    await _require_admin(request)

    # Source agents from the database: only agents owned by this specific user.
    agent_store = get_agent_store(request)
    all_agents = await agent_store.list_visible(user_id)
    owned_agents = sorted(
        [a for a in all_agents if a.get("user_id") == user_id],
        key=lambda a: (a.get("name") or a["id"]).lower(),
    )

    if _is_v1(user_id):
        # V1 模式：每个桶读 memory.json，内容为 {facts, user, history}
        buckets: list[AdminMemoryBucket] = []
        try:
            buckets.append(AdminMemoryBucket(name="default", display_name="默认助手", data=_read_v1_memory_bucket(user_id, None)))
        except Exception:
            logger.exception("Failed to read V1 global memory for user %s", user_id)
            buckets.append(AdminMemoryBucket(name="default", display_name="默认助手", data={}))

        for agent in owned_agents:
            agent_id: str = agent["id"]
            display_name: str = agent.get("name") or agent_id
            try:
                data = _read_v1_memory_bucket(user_id, agent_id)
            except Exception:
                data = {}
            buckets.append(AdminMemoryBucket(name=agent_id, display_name=display_name, data=data))

        return AdminMemoryResponse(user_id=user_id, version="v1", buckets=buckets)

    # V2 模式：读 USER.md / MEMORY.md，内容为 {entries: [...]}
    buckets = [
        AdminMemoryBucket(name="default", display_name="默认助手", data=_read_memory_bucket(user_id, None))
    ]
    for agent in owned_agents:
        agent_id = agent["id"]
        display_name = agent.get("name") or agent_id
        try:
            data = _read_memory_bucket(user_id, agent_id)
        except Exception:
            data = {"entries": []}
        buckets.append(AdminMemoryBucket(name=agent_id, display_name=display_name, data=data))

    return AdminMemoryResponse(user_id=user_id, version="v2", buckets=buckets)


@router.post("/{user_id}/memory/entries", response_model=AdminMemoryBucket)
async def admin_add_memory_entry(
    user_id: str,
    body: AdminMemoryEntryAddRequest,
    request: Request,
) -> AdminMemoryBucket:
    """Add a memory entry for a user (admin only)."""
    await _require_admin(request)
    target = "user" if body.agent is None else "memory"
    provider = BuiltinFileProvider()
    provider.initialize(user_id=user_id, agent_id=body.agent)
    try:
        result = provider.add(target, body.content)
        if result is False:
            raise HTTPException(status_code=400, detail="Failed to add entry (quota exceeded or write error)")
    except HTTPException:
        raise
    except Exception as exc:
        logger.error("admin_add_memory_entry failed for user %s: %s", user_id, exc, exc_info=True)
        raise HTTPException(status_code=500, detail="Failed to write memory entry") from exc
    bucket_name = "global" if body.agent is None else body.agent
    return AdminMemoryBucket(name=bucket_name, data=_read_memory_bucket(user_id, body.agent))


@router.put("/{user_id}/memory/entries", response_model=AdminMemoryBucket)
async def admin_replace_memory_entry(
    user_id: str,
    body: AdminMemoryEntryReplaceRequest,
    request: Request,
) -> AdminMemoryBucket:
    """Replace a memory entry for a user (admin only)."""
    await _require_admin(request)
    target = "user" if body.agent is None else "memory"
    provider = BuiltinFileProvider()
    provider.initialize(user_id=user_id, agent_id=body.agent)
    try:
        result = provider.replace(target, body.old_text, body.content)
        if result is False:
            raise HTTPException(status_code=404, detail="Entry not found or write failed")
    except HTTPException:
        raise
    except Exception as exc:
        logger.error("admin_replace_memory_entry failed for user %s: %s", user_id, exc, exc_info=True)
        raise HTTPException(status_code=500, detail="Failed to replace memory entry") from exc
    bucket_name = "global" if body.agent is None else body.agent
    return AdminMemoryBucket(name=bucket_name, data=_read_memory_bucket(user_id, body.agent))


@router.delete("/{user_id}/memory/entries", response_model=AdminMemoryBucket)
async def admin_delete_memory_entry(
    user_id: str,
    request: Request,
    old_text: str = Query(..., min_length=1, description="Short unique substring to locate the entry"),
    agent: str | None = Query(None, description="Agent name; None = global USER.md bucket"),
) -> AdminMemoryBucket:
    """Delete a memory entry for a user (admin only)."""
    await _require_admin(request)
    target = "user" if agent is None else "memory"
    provider = BuiltinFileProvider()
    provider.initialize(user_id=user_id, agent_id=agent)
    try:
        result = provider.remove(target, old_text)
        if result is False:
            raise HTTPException(status_code=404, detail="Entry not found")
    except HTTPException:
        raise
    except Exception as exc:
        logger.error("admin_delete_memory_entry failed for user %s: %s", user_id, exc, exc_info=True)
        raise HTTPException(status_code=500, detail="Failed to delete memory entry") from exc
    bucket_name = "global" if agent is None else agent
    return AdminMemoryBucket(name=bucket_name, data=_read_memory_bucket(user_id, agent))


# ── V1 Memory CRUD (admin) ─────────────────────────────────────────────────

_V1_USER_FIELDS = {"workContext", "personalContext", "topOfMind"}
_V1_HISTORY_FIELDS = {"recentMonths", "earlierContext", "longTermBackground"}


class AdminV1SectionUpdateRequest(BaseModel):
    section_type: str = Field(..., description="'user' 或 'history'")
    field: str = Field(..., description="字段名，如 workContext / recentMonths")
    summary: str = Field(..., description="新的摘要文本")
    agent: str | None = Field(None, description="Agent id；None 表示用户级全局记忆")


class AdminV1FactAddRequest(BaseModel):
    content: str = Field(..., min_length=1)
    category: str = Field(default="context")
    confidence: float = Field(default=0.8, ge=0.0, le=1.0)
    agent: str | None = Field(None)


class AdminV1FactUpdateRequest(BaseModel):
    content: str | None = Field(None)
    category: str | None = Field(None)
    confidence: float | None = Field(None, ge=0.0, le=1.0)
    agent: str | None = Field(None)


@router.put("/{user_id}/memory/v1/sections", response_model=AdminMemoryBucket)
async def admin_update_v1_section(
    user_id: str,
    body: AdminV1SectionUpdateRequest,
    request: Request,
) -> AdminMemoryBucket:
    """更新 V1 记忆的某个摘要字段（用户画像 / 历史）。"""
    await _require_admin(request)
    if body.section_type == "user" and body.field not in _V1_USER_FIELDS:
        raise HTTPException(status_code=400, detail=f"无效 user 字段：{body.field}")
    if body.section_type == "history" and body.field not in _V1_HISTORY_FIELDS:
        raise HTTPException(status_code=400, detail=f"无效 history 字段：{body.field}")
    if body.section_type not in ("user", "history"):
        raise HTTPException(status_code=400, detail="section_type 须为 'user' 或 'history'")

    from deerflow.agents.memory.v1.storage import get_memory_storage, utc_now_iso_z
    storage = get_memory_storage()
    data = storage.load(body.agent, user_id=user_id)
    data.setdefault(body.section_type, {})[body.field] = {
        "summary": body.summary,
        "updatedAt": utc_now_iso_z(),
    }
    if not storage.save(data, body.agent, user_id=user_id):
        raise HTTPException(status_code=500, detail="写入失败")

    bucket_name = "global" if body.agent is None else body.agent
    return AdminMemoryBucket(name=bucket_name, data=_read_v1_memory_bucket(user_id, body.agent))


@router.post("/{user_id}/memory/v1/facts", response_model=AdminMemoryBucket)
async def admin_add_v1_fact(
    user_id: str,
    body: AdminV1FactAddRequest,
    request: Request,
) -> AdminMemoryBucket:
    """新增一条 V1 事实记录。"""
    await _require_admin(request)
    try:
        from deerflow.agents.memory.v1.updater import create_memory_fact
        create_memory_fact(
            body.content,
            category=body.category,
            confidence=body.confidence,
            agent_name=body.agent,
            user_id=user_id,
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(status_code=500, detail="新增失败") from exc
    bucket_name = "global" if body.agent is None else body.agent
    return AdminMemoryBucket(name=bucket_name, data=_read_v1_memory_bucket(user_id, body.agent))


@router.put("/{user_id}/memory/v1/facts/{fact_id}", response_model=AdminMemoryBucket)
async def admin_update_v1_fact(
    user_id: str,
    fact_id: str,
    body: AdminV1FactUpdateRequest,
    request: Request,
) -> AdminMemoryBucket:
    """更新一条 V1 事实记录（内容 / 类别 / 置信度）。"""
    await _require_admin(request)
    try:
        from deerflow.agents.memory.v1.updater import update_memory_fact
        update_memory_fact(
            fact_id,
            content=body.content,
            category=body.category,
            confidence=body.confidence,
            agent_name=body.agent,
            user_id=user_id,
        )
    except KeyError:
        raise HTTPException(status_code=404, detail=f"事实 '{fact_id}' 不存在")
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(status_code=500, detail="更新失败") from exc
    bucket_name = "global" if body.agent is None else body.agent
    return AdminMemoryBucket(name=bucket_name, data=_read_v1_memory_bucket(user_id, body.agent))


@router.delete("/{user_id}/memory/v1/facts/{fact_id}", response_model=AdminMemoryBucket)
async def admin_delete_v1_fact(
    user_id: str,
    fact_id: str,
    request: Request,
    agent: str | None = Query(None, description="Agent id；None 表示用户级全局记忆"),
) -> AdminMemoryBucket:
    """删除一条 V1 事实记录。"""
    await _require_admin(request)
    try:
        from deerflow.agents.memory.v1.updater import delete_memory_fact
        delete_memory_fact(fact_id, agent_name=agent, user_id=user_id)
    except KeyError:
        raise HTTPException(status_code=404, detail=f"事实 '{fact_id}' 不存在")
    except Exception as exc:
        raise HTTPException(status_code=500, detail="删除失败") from exc
    bucket_name = "global" if agent is None else agent
    return AdminMemoryBucket(name=bucket_name, data=_read_v1_memory_bucket(user_id, agent))