"""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 contextvars import ContextVar
from hashlib import sha256
from time import perf_counter
from typing import Any

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"])
self_router = APIRouter(prefix="/api/users/me", tags=["user-qa"])

_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", "43200")),
)
_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)
_LEADERBOARD_TRACE_CONTEXT: ContextVar[dict[str, object] | None] = ContextVar(
    "leaderboard_trace_context",
    default=None,
)


def _leaderboard_log_context(**context: object) -> dict[str, object] | None:
    current = _LEADERBOARD_TRACE_CONTEXT.get()
    merged: dict[str, object] = dict(current or {})
    for key, value in context.items():
        if value is None or value == "":
            continue
        merged[key] = value
    return merged or None


def _leaderboard_sql_for_log(stmt) -> str:
    try:
        return str(stmt.compile(compile_kwargs={"literal_binds": True}))
    except Exception:
        return str(stmt)


async def _execute_leaderboard_query(session, step: str, stmt, **context: object):
    log_context = _leaderboard_log_context(**context)
    if not log_context:
        return await session.execute(stmt)

    started_at = datetime.now(BEIJING_TZ).isoformat()
    sql = _leaderboard_sql_for_log(stmt)
    logger.info(
        "Leaderboard snapshot DB query start: step=%s stat_date=%s job=%s settings_hash=%s started_at=%s sql=%s",
        step,
        log_context.get("stat_date") or log_context.get("stat_dates") or "",
        log_context.get("job_key") or "",
        log_context.get("settings_hash") or "",
        started_at,
        sql,
    )
    start = perf_counter()
    try:
        result = await session.execute(stmt)
    except Exception:
        elapsed_ms = (perf_counter() - start) * 1000
        logger.exception(
            "Leaderboard snapshot DB query failed: step=%s stat_date=%s job=%s settings_hash=%s elapsed_ms=%.2f",
            step,
            log_context.get("stat_date") or log_context.get("stat_dates") or "",
            log_context.get("job_key") or "",
            log_context.get("settings_hash") or "",
            elapsed_ms,
        )
        raise

    elapsed_ms = (perf_counter() - start) * 1000
    logger.info(
        "Leaderboard snapshot DB query done: step=%s stat_date=%s job=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f",
        step,
        log_context.get("stat_date") or log_context.get("stat_dates") or "",
        log_context.get("job_key") or "",
        log_context.get("settings_hash") or "",
        datetime.now(BEIJING_TZ).isoformat(),
        elapsed_ms,
    )
    return result


async def _time_leaderboard_db_operation(
    step: str,
    operation,
    *,
    sql: str = "",
    **context: object,
):
    log_context = _leaderboard_log_context(**context)
    if not log_context:
        return await operation()

    logger.info(
        "Leaderboard snapshot DB query start: step=%s stat_date=%s job=%s settings_hash=%s started_at=%s sql=%s",
        step,
        log_context.get("stat_date") or log_context.get("stat_dates") or "",
        log_context.get("job_key") or "",
        log_context.get("settings_hash") or "",
        datetime.now(BEIJING_TZ).isoformat(),
        sql,
    )
    start = perf_counter()
    try:
        result = await operation()
    except Exception:
        elapsed_ms = (perf_counter() - start) * 1000
        logger.exception(
            "Leaderboard snapshot DB query failed: step=%s stat_date=%s job=%s settings_hash=%s elapsed_ms=%.2f",
            step,
            log_context.get("stat_date") or log_context.get("stat_dates") or "",
            log_context.get("job_key") or "",
            log_context.get("settings_hash") or "",
            elapsed_ms,
        )
        raise

    elapsed_ms = (perf_counter() - start) * 1000
    logger.info(
        "Leaderboard snapshot DB query done: step=%s stat_date=%s job=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f",
        step,
        log_context.get("stat_date") or log_context.get("stat_dates") or "",
        log_context.get("job_key") or "",
        log_context.get("settings_hash") or "",
        datetime.now(BEIJING_TZ).isoformat(),
        elapsed_ms,
    )
    return result


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 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)
    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)
    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 _parse_optional_datetime(value: str | None) -> datetime | None:
    if not value:
        return None
    parsed = _parse_datetime(value, _now())
    return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=BEIJING_TZ)


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 _execute_leaderboard_query(session, "session_setting", 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)


def _is_mariadb(dialect) -> bool:
    """SQLAlchemy reports ``dialect.name == 'mysql'`` for MariaDB too."""
    if getattr(dialect, "_is_mariadb", False) or getattr(dialect, "is_mariadb", False):
        return True
    version_info = getattr(dialect, "server_version_info", None) or ()
    return any("mariadb" in str(part).lower() for part in version_info)


async def _set_analytics_read_timeout(session) -> None:
    """Best-effort per-statement read timeout for the analytics queries.

    The right knob differs by server, and some servers have none at all:
      - MySQL >= 5.7.4: ``max_execution_time`` (milliseconds)
      - MariaDB:        ``max_statement_time`` (seconds)
      - MySQL <= 5.6:   no per-statement timeout variable — skip it entirely
                        (5.6 raises "Unknown system variable" otherwise)
      - PostgreSQL/openGauss: ``statement_timeout`` (milliseconds)
    """
    conn = await session.connection()
    dialect = conn.dialect
    name = dialect.name
    if name == "mysql":
        if _is_mariadb(dialect):
            # MariaDB: max_statement_time is in SECONDS (fractional ok).
            await _try_set_session_value(session, f"SET SESSION max_statement_time = {_LEADERBOARD_QUERY_TIMEOUT_MS / 1000}")
        elif (getattr(dialect, "server_version_info", None) or ()) >= (5, 7, 4):
            await _try_set_session_value(session, f"SET SESSION max_execution_time = {_LEADERBOARD_QUERY_TIMEOUT_MS}")
        # else: MySQL <= 5.6 has no statement-timeout variable — skip silently.
        await _try_set_session_value(session, f"SET SESSION innodb_lock_wait_timeout = {_LEADERBOARD_LOCK_WAIT_SECONDS}")
    elif name in {"postgresql", "opengauss"}:
        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():
    """Exclude legacy scheduled runs without joining runs into the metric scan."""

    return ~(
        select(1)
        .select_from(RunRow)
        .where(
            RunRow.run_id == ToolCallMetricRow.run_id,
            RunRow.first_human_message.like("【定时任务触发】%"),
        )
        .exists()
    )


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 _not_scheduled_run_id_filter(run_id_column):
    """Use a NULL-safe, index-friendly anti-semi join instead of NOT IN."""

    return ~(
        select(1)
        .select_from(ScheduledTaskRunRow)
        .where(ScheduledTaskRunRow.agent_run_id == run_id_column)
        .exists()
    )


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(_not_scheduled_run_id_filter(run_id_column))
        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(_not_scheduled_run_id_filter(ToolCallMetricRow.run_id))
        stmt = stmt.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)
    # 排行榜只统计真正产生过问答的用户；从未问答的账号（仅注册、零活跃）不计入榜单与人数。
    # 个人分析（指定 user_id）不受此限制，仍可查看任意单个用户。
    if user_id is None:
        stmt = stmt.where(func.coalesce(run_agg.c.qa_count, 0) > 0)
    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 _execute_leaderboard_query(session, "users", 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 _execute_leaderboard_query(session, f"{field}_rankings", 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 _execute_leaderboard_query(session, f"{field}_dimensions", 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(_not_scheduled_run_id_filter(LlmCallMetricRow.run_id))
    rows = (await _execute_leaderboard_query(session, "model_rankings", 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_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 _execute_leaderboard_query(session, "trends", 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(_not_scheduled_run_id_filter(RunRow.run_id))
        tool_where.append(_not_scheduled_run_id_filter(ToolCallMetricRow.run_id))
        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 _execute_leaderboard_query(session, "overview.total_users", total_users_stmt)).scalar_one()
    run_summary_stmt = (
        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)
    )
    run_row = (await _execute_leaderboard_query(session, "overview.run_summary", run_summary_stmt)).one()
    thread_count_stmt = select(func.count(ThreadMetaRow.thread_id)).where(*thread_where)
    thread_count = (await _execute_leaderboard_query(session, "overview.thread_count", thread_count_stmt)).scalar_one()
    tool_summary_stmt = (
        select(
            func.count(ToolCallMetricRow.id),
            func.sum(case((func.nullif(ToolCallMetricRow.skill_name, "").is_not(None), 1), else_=0)),
        )
        .where(*tool_where)
    )
    tool_row = (await _execute_leaderboard_query(session, "overview.tool_summary", tool_summary_stmt)).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,
        }
        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()
        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,
        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_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),
        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:
        stmt = select(AdminLeaderboardDailyStatRow).where(
            AdminLeaderboardDailyStatRow.stat_date.in_(stat_dates),
            AdminLeaderboardDailyStatRow.settings_hash == settings_hash,
        )
        rows = (
            await _execute_leaderboard_query(session, "snapshot_rows.load", stmt)
        ).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 _time_leaderboard_db_operation(
            "snapshot.store.get",
            lambda: session.get(
                AdminLeaderboardDailyStatRow,
                {"stat_date": stat_date, "settings_hash": settings_hash},
            ),
            sql="SELECT admin_leaderboard_daily_stats by primary key",
            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 _time_leaderboard_db_operation(
            "snapshot.store.commit",
            lambda: session.commit(),
            sql="COMMIT",
            stat_date=stat_date,
            settings_hash=settings_hash,
        )


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)
        stmt = select(AdminLeaderboardDailyStatRow).where(
            AdminLeaderboardDailyStatRow.stat_date.in_(stat_dates),
            AdminLeaderboardDailyStatRow.settings_hash == settings_hash,
        )
        rows = (
            await _execute_leaderboard_query(
                session,
                "snapshot_queue.load_existing",
                stmt,
                stat_dates=stat_dates,
                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 _time_leaderboard_db_operation(
                "snapshot_queue.commit",
                lambda: session.commit(),
                sql="COMMIT",
                stat_dates=queued,
                settings_hash=settings_hash,
            )
    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
    trace_token = _LEADERBOARD_TRACE_CONTEXT.set(
        {
            "job_key": job_key,
            "stat_date": stat_date,
            "settings_hash": settings_hash,
        }
    )
    snapshot_start = perf_counter()
    try:
        logger.info(
            "Leaderboard snapshot build start: job=%s stat_date=%s settings_hash=%s started_at=%s",
            job_key,
            stat_date,
            settings_hash,
            datetime.now(BEIJING_TZ).isoformat(),
        )
        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,
        )
        logger.info(
            "Leaderboard snapshot build done: job=%s stat_date=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f",
            job_key,
            stat_date,
            settings_hash,
            datetime.now(BEIJING_TZ).isoformat(),
            (perf_counter() - snapshot_start) * 1000,
        )
    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()
        _LEADERBOARD_TRACE_CONTEXT.reset(trace_token)


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)} 天快照队列，后台统计任务会继续补齐。",
    )


@router.post("/leaderboard/refresh-today", response_model=AdminLeaderboardResponse)
async def refresh_today_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:
    """Force a synchronous rebuild of *today's* snapshot, then return the range.

    Today's snapshot otherwise refreshes only every ``_LEADERBOARD_TODAY_TTL_SECONDS``
    (~5 min) via the background scheduler. This endpoint lets an admin pull the
    latest real-time records into the stats on demand: it rebuilds today's daily
    snapshot immediately (bounded by the same semaphore/timeout the scheduler uses)
    and then returns the merged leaderboard for the requested range.
    """
    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
    (
        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,
    )
    today = _today_stat_date()
    await _refresh_daily_leaderboard_snapshot(
        _daily_job_key(today, settings_hash),
        session_factory=session_factory,
        settings=settings,
        stat_date=today,
        settings_hash=settings_hash,
        excluded=excluded,
        include_scheduled=resolved_include_scheduled,
        include_admins=resolved_include_admins,
        include_failed=resolved_include_failed,
    )

    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,
    )


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
    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]],
        "trends": [t.model_dump() for t in trends[-21:]],
    }
    return (
        "请基于以下管理员排行榜统计数据，生成一份中文用户行为分析报告。"
        "报告需要有结论，不要只罗列数字；请使用 Markdown 标题、列表和表格，让前端可以直接渲染。"
        "重点要求："
        "1) 你是专门负责分析用户行为的专家，不要自称企业 AI 应用运营分析师；"
        "2) 结合 top_users、top_skills、top_tools、top_models 和 trends，说明使用频率、使用增长、常用能力和重点用户行为画像；"
        "3) 可以提及使用强度、关注变化和行为机会，但不要把报告重点放在系统本身的问题、故障、失败率或回答质量缺陷上；"
        "4) 给管理员 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,
    }
    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)
        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,
        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)


def _qa_sort_datetime(value: str | None) -> datetime:
    if not value:
        return datetime.min.replace(tzinfo=BEIJING_TZ)
    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.replace(tzinfo=BEIJING_TZ)
    if parsed.tzinfo is None:
        return parsed.replace(tzinfo=BEIJING_TZ)
    return parsed.astimezone(BEIJING_TZ)


def _qa_dedupe_key(item: AdminUserQaItem) -> tuple[str, str, str]:
    question_key = _question_similarity_key(item.question)[:160]
    answer_key = _question_similarity_key(item.answer)[:160]
    return (item.thread_id, question_key, answer_key)


def _message_value(message: Any, key: str, default: Any = None) -> Any:
    if isinstance(message, dict):
        return message.get(key, default)
    return getattr(message, key, default)


def _qa_message_type(message: Any) -> str:
    value = _message_value(message, "type") or _message_value(message, "role") or ""
    if value == "user":
        return "human"
    if value == "assistant":
        return "ai"
    return str(value)


def _qa_message_id(message: Any, fallback: str) -> str:
    return str(_message_value(message, "id") or fallback)


def _qa_message_text(message: Any) -> str:
    return _message_content_to_text(_message_value(message, "content")).strip()


async def _read_thread_checkpoint_messages(checkpointer, thread_id: str) -> list[dict]:
    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
    try:
        checkpoint_tuple = await checkpointer.aget_tuple(config)
    except Exception:
        logger.warning("Failed to read checkpoint while listing Q&A: thread=%s", thread_id, exc_info=True)
        return []
    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", [])
    return messages if isinstance(messages, list) else []


def _qa_items_from_checkpoint_messages(
    *,
    user_id: str,
    thread_id: str,
    thread_title: str,
    assistant_id: str,
    thread_time: str,
    messages: list[dict],
) -> list[AdminUserQaItem]:
    items: list[AdminUserQaItem] = []
    current_question = ""
    current_answer = ""
    current_id = ""
    current_index = 0

    def flush() -> None:
        nonlocal current_question, current_answer, current_id, current_index
        question = _clean_question_text(current_question)
        if not _is_displayable_question(question):
            current_question = ""
            current_answer = ""
            current_id = ""
            return
        items.append(
            AdminUserQaItem(
                run_id=current_id or f"{thread_id}:checkpoint:{current_index}",
                thread_id=thread_id,
                thread_title=thread_title,
                assistant_id=assistant_id,
                status="checkpoint",
                question=question[:2000],
                answer=current_answer.strip()[:6000],
                created_at=thread_time,
                updated_at=thread_time,
            )
        )
        current_question = ""
        current_answer = ""
        current_id = ""

    for index, message in enumerate(messages):
        msg_type = _qa_message_type(message)
        if msg_type == "human":
            if current_question:
                flush()
            current_question = _qa_message_text(message)
            current_answer = ""
            current_id = _qa_message_id(message, f"{thread_id}:checkpoint:{index}")
            current_index = index
            continue
        if msg_type == "ai" and current_question:
            text = _qa_message_text(message)
            if text:
                current_answer = text

    if current_question:
        flush()

    return items


async def _list_thread_checkpoint_qa_items(
    *,
    request: Request,
    user_id: str,
) -> list[AdminUserQaItem]:
    repo = get_thread_store(request)
    checkpointer = get_checkpointer(request)
    items: list[AdminUserQaItem] = []
    offset = 0
    page_size = 200
    while True:
        rows = await repo.search(limit=page_size, offset=offset, user_id=user_id)
        if not rows:
            break
        for row in rows:
            thread_id = str(row.get("thread_id") or "")
            if not thread_id:
                continue
            messages = await _read_thread_checkpoint_messages(checkpointer, thread_id)
            if not messages:
                continue
            metadata = row.get("metadata") if isinstance(row.get("metadata"), dict) else {}
            context = row.get("context") if isinstance(row.get("context"), dict) else {}
            assistant_id = str(
                row.get("assistant_id")
                or (context or {}).get("agent_id")
                or (metadata or {}).get("agent_id")
                or (context or {}).get("agent_name")
                or (metadata or {}).get("agent_name")
                or ""
            )
            thread_time = str(row.get("updated_at") or row.get("created_at") or "")
            items.extend(
                _qa_items_from_checkpoint_messages(
                    user_id=user_id,
                    thread_id=thread_id,
                    thread_title=str(row.get("display_name") or ""),
                    assistant_id=assistant_id,
                    thread_time=thread_time,
                    messages=messages,
                )
            )
        if len(rows) < page_size:
            break
        offset += len(rows)
    return items


async def _list_run_summary_qa_items(
    *,
    session_factory,
    user_id: str,
    since: str | None,
    until: str | None,
) -> list[AdminUserQaItem]:
    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))

    async with session_factory() as session:
        await _set_analytics_read_timeout(session)
        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())
            )
        ).all()

    return [
        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
        if _is_displayable_question(r[6])
    ]


def _filter_qa_items(
    items: list[AdminUserQaItem],
    *,
    q: str | None,
    since: str | None,
    until: str | None,
) -> list[AdminUserQaItem]:
    needle = (q or "").strip().lower()
    since_dt = _parse_optional_datetime(since) if since else None
    until_dt = _parse_optional_datetime(until) if until else None
    filtered: list[AdminUserQaItem] = []
    for item in items:
        item_dt = _qa_sort_datetime(item.created_at or item.updated_at)
        if since_dt and item_dt < since_dt:
            continue
        if until_dt and item_dt > until_dt:
            continue
        if needle:
            haystack = "\n".join(
                [
                    item.question,
                    item.answer,
                    item.run_id,
                    item.thread_id,
                    item.thread_title,
                    item.assistant_id,
                ]
            ).lower()
            if needle not in haystack:
                continue
        filtered.append(item)
    return filtered


async def _list_user_qa_page(
    *,
    request: Request,
    user_id: str,
    page: int,
    page_size: int,
    q: str | None,
    since: str | None,
    until: str | None,
    include_checkpoints: bool = True,
) -> AdminUserQaPageResponse:
    session_factory = get_session_factory()
    if session_factory is None:
        raise HTTPException(status_code=503, detail="Q&A audit requires database persistence")

    items = await _list_run_summary_qa_items(
        session_factory=session_factory,
        user_id=user_id,
        since=since,
        until=until,
    )
    if include_checkpoints:
        items.extend(await _list_thread_checkpoint_qa_items(request=request, user_id=user_id))

    by_key: dict[tuple[str, str, str], AdminUserQaItem] = {}
    for item in items:
        key = _qa_dedupe_key(item)
        existing = by_key.get(key)
        if existing is None or (existing.status == "checkpoint" and item.status != "checkpoint"):
            by_key[key] = item

    filtered = _filter_qa_items(list(by_key.values()), q=q, since=since, until=until)
    filtered.sort(
        key=lambda item: (_qa_sort_datetime(item.created_at or item.updated_at), item.thread_id, item.run_id),
        reverse=True,
    )
    offset = (page - 1) * page_size
    return AdminUserQaPageResponse(
        user_id=user_id,
        page=page,
        page_size=page_size,
        total=len(filtered),
        items=filtered[offset : offset + page_size],
    )


@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, including older checkpoint-only threads."""
    await _require_admin(request)
    return await _list_user_qa_page(
        request=request,
        user_id=user_id,
        page=page,
        page_size=page_size,
        q=q,
        since=since,
        until=until,
    )


@self_router.get("/qa", response_model=AdminUserQaPageResponse)
async def list_my_qa(
    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 the current user's own Q&A history."""
    user = getattr(request.state, "user", None)
    if user is None:
        user = await get_current_user_from_request(request)
    return await _list_user_qa_page(
        request=request,
        user_id=str(user.id),
        page=page,
        page_size=page_size,
        q=q,
        since=since,
        until=until,
    )


# ── 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))