-
- 请从中间列选择一个会话
+
+
+
+
+
+ {
+ setQaQuery(event.target.value);
+ setThreadId(null);
+ setQaPage(1);
+ }}
+ placeholder="搜索问题、回答、运行 ID 或会话 ID"
+ className="pl-9"
+ />
+
+ {qaQuery && (
+
+ )}
- )}
- {threadId && messagesLoading && (
+
+ 共 {fmtNumber(qaData?.total ?? 0)} 条问答
+ {threadId && 已按会话过滤}
+ {qaFetching && !qaLoading && (
+
+ 更新中
+
+ )}
+
+
+
+ {qaLoading && (
加载中
)}
- {threadId && !messagesLoading && (
+ {!qaLoading && (qaData?.items.length ?? 0) === 0 && (
+
+
+ 暂无匹配的问答记录
+
+ )}
+ {!qaLoading && (qaData?.items.length ?? 0) > 0 && (
- {(messages ?? []).map((m, i) => (
-
+ {(qaData?.items ?? []).map((item) => (
+
))}
- {(messages ?? []).every((m) => !roleOf(m)) && (
-
- 该会话暂无问答记录
-
- )}
+
)}
diff --git a/frontend-web/src/pages/ScheduledTaskRunDetailPage.tsx b/frontend-web/src/pages/ScheduledTaskRunDetailPage.tsx
index 2f9cad01..9002fbcb 100644
--- a/frontend-web/src/pages/ScheduledTaskRunDetailPage.tsx
+++ b/frontend-web/src/pages/ScheduledTaskRunDetailPage.tsx
@@ -1,7 +1,7 @@
"use client";
import { useCallback, useEffect, useMemo, useState } from "react";
-import { Activity, ChevronDown, ChevronLeft, Download, Edit3, FileText, Heart, Loader2, Save, Code2, Trash2, X } from "lucide-react";
+import { Activity, ChevronDown, ChevronLeft, Download, Edit3, FileText, Heart, Loader2, Maximize2, Save, Code2, Trash2, X } from "lucide-react";
import { useParams, useSearchParams } from "react-router-dom";
import { useRouteNavigate as useNavigate } from "@/core/route-base";
import { toast } from "sonner";
@@ -9,6 +9,7 @@ import { toast } from "sonner";
import { Badge } from "@/components/ui/badge";
import { Button } from "@/components/ui/button";
import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
+import { Dialog, DialogContent, DialogHeader, DialogTitle } from "@/components/ui/dialog";
import { ScrollArea } from "@/components/ui/scroll-area";
import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs";
import { Textarea } from "@/components/ui/textarea";
@@ -101,6 +102,32 @@ function RenderPreview({ file }: { file: ScheduledTaskResultFileContent }) {
return
{file.content};
}
+type FullscreenPreview =
+ | { mode: "html"; title: string; html: string }
+ | { mode: "markdown"; title: string; content: string }
+ | { mode: "text"; title: string; content: string }
+ | { mode: "file"; title: string; file: ScheduledTaskResultFileContent };
+
+function FullscreenPreviewContent({ preview }: { preview: FullscreenPreview }) {
+ if (preview.mode === "html") {
+ return
;
+ }
+ if (preview.mode === "markdown") {
+ return (
+
+ );
+ }
+ if (preview.mode === "file") {
+ return
;
+ }
+ return
{preview.content || "无正文内容"};
+}
+
type EditState = { draft: string; saving: boolean };
/** Flatten a LangChain-style message content (string | parts | nested) to text. */
@@ -218,6 +245,7 @@ export default function ScheduledTaskRunDetailPage() {
const [filesLoading, setFilesLoading] = useState(false);
const [showMeta, setShowMeta] = useState(false);
const [edits, setEdits] = useState
>({});
+ const [fullscreenPreview, setFullscreenPreview] = useState(null);
const canEditRun = run?.can_edit === true;
const canDeleteRun = canEditRun;
@@ -428,7 +456,27 @@ export default function ScheduledTaskRunDetailPage() {
};
return (
-
+ <>
+
+
@@ -544,6 +592,23 @@ export default function ScheduledTaskRunDetailPage() {
{isHtmlRun && (
+
+ {file.previewable && !isEditing && (
+
+ )}
{isMarkdown && (
-
+
+
);
}
diff --git a/frontend-web/src/pages/ScheduledTasksPage.tsx b/frontend-web/src/pages/ScheduledTasksPage.tsx
index 418f807e..6a527443 100644
--- a/frontend-web/src/pages/ScheduledTasksPage.tsx
+++ b/frontend-web/src/pages/ScheduledTasksPage.tsx
@@ -45,7 +45,6 @@ import { TagChipList } from "@/components/workspace/tags/tag-chip";
import { TagPickerDialog } from "@/components/workspace/tags/tag-picker-dialog";
import { TagSearchBar } from "@/components/workspace/tags/tag-search-bar";
import { useAgents } from "@/core/agents";
-import { getStoredAuth } from "@/core/auth";
import { useModels } from "@/core/models/hooks";
import {
downloadScheduledTaskRun,
@@ -188,7 +187,6 @@ export default function ScheduledTasksPage() {
const [subscribeTarget, setSubscribeTarget] = useState(null);
const [subscribeNotify, setSubscribeNotify] = useState(false);
const [tagPickerTask, setTagPickerTask] = useState(null);
- const isAdmin = useMemo(() => getStoredAuth()?.system_role === "admin", []);
const sortedTasks = useMemo(
() =>
@@ -552,19 +550,17 @@ export default function ScheduledTasksPage() {
)}
- {isAdmin && (
-
- )}
+
{isOwner ? (
<>
diff --git a/frontend-web/src/pages/WorkspaceRoutes.tsx b/frontend-web/src/pages/WorkspaceRoutes.tsx
index 923eb137..2f2a40c0 100644
--- a/frontend-web/src/pages/WorkspaceRoutes.tsx
+++ b/frontend-web/src/pages/WorkspaceRoutes.tsx
@@ -420,13 +420,9 @@ export default function WorkspaceRoutes({
-
-
- ) : (
-
- )
+
+
+
}
/>
None:
pass
-async def _curator_scheduler_loop() -> None:
+async def _curator_scheduler_loop() -> None:
"""常驻后台任务:每隔 _CURATOR_CHECK_INTERVAL_SECONDS 检查一次需不需要进化。
多 worker 部署下用文件锁做 leader 选举 —— 只有抢到锁的那个 worker 进程真正
@@ -404,11 +404,248 @@ async def _curator_scheduler_loop() -> None:
except Exception:
logger.exception("Curator scheduler tick failed (non-fatal)")
await asyncio.sleep(_CURATOR_CHECK_INTERVAL_SECONDS)
- finally:
- _release_singleton_lock(lock_handle)
-
-
-def _warn_if_multi_worker() -> None:
+ finally:
+ _release_singleton_lock(lock_handle)
+
+
+def _env_bool(name: str, default: bool) -> bool:
+ raw = os.getenv(name)
+ if raw is None:
+ return default
+ return raw.strip().lower() not in {"0", "false", "no", "off"}
+
+
+def _env_int(name: str, default: int, *, alias: str | None = None, minimum: int = 1) -> int:
+ raw = os.getenv(name)
+ if raw is None and alias:
+ raw = os.getenv(alias)
+ if raw is None:
+ raw = str(default)
+ try:
+ value = int(raw)
+ except ValueError:
+ return default
+ return max(minimum, value)
+
+
+def _env_float(name: str, default: float, *, alias: str | None = None, minimum: float = 0.0) -> float:
+ raw = os.getenv(name)
+ if raw is None and alias:
+ raw = os.getenv(alias)
+ if raw is None:
+ raw = str(default)
+ try:
+ value = float(raw)
+ except ValueError:
+ return default
+ return max(minimum, value)
+
+
+_LEADERBOARD_SNAPSHOT_BACKGROUND_ENABLED = _env_bool("ADMIN_LEADERBOARD_BACKGROUND_ENABLED", True)
+_LEADERBOARD_SNAPSHOT_PREWARM_DAYS = _env_int(
+ "ADMIN_LEADERBOARD_BACKGROUND_PREWARM_DAYS",
+ 30,
+)
+_LEADERBOARD_SNAPSHOT_BATCH_SIZE = _env_int(
+ "ADMIN_LEADERBOARD_BACKGROUND_BATCH_SIZE",
+ 1,
+)
+_LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS = _env_float(
+ "ADMIN_LEADERBOARD_BACKGROUND_INTERVAL_SECONDS",
+ 300.0,
+)
+_LEADERBOARD_SNAPSHOT_SLEEP_SECONDS = _env_float(
+ "ADMIN_LEADERBOARD_BACKGROUND_SLEEP_SECONDS",
+ 2.0,
+)
+_LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS = _env_int(
+ "ADMIN_LEADERBOARD_BACKGROUND_RUNNING_TTL_SECONDS",
+ 1800,
+)
+_LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS = _env_float(
+ "ADMIN_LEADERBOARD_BACKGROUND_STARTUP_DELAY_SECONDS",
+ 60.0,
+)
+
+
+def _recent_leaderboard_stat_dates(days: int) -> list[str]:
+ from datetime import datetime, timedelta
+
+ from deerflow.persistence.types import BEIJING_TZ
+
+ today = datetime.now(BEIJING_TZ).date()
+ start = today - timedelta(days=max(1, days) - 1)
+ return [(start + timedelta(days=i)).isoformat() for i in range((today - start).days + 1)]
+
+
+async def _leaderboard_snapshot_due_dates(
+ session_factory,
+ *,
+ settings_hash: str,
+ limit: int,
+ running_ttl_seconds: int,
+) -> list[str]:
+ from datetime import UTC, datetime, timedelta
+
+ from sqlalchemy import and_, or_, select
+
+ from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
+
+ cutoff = datetime.now(UTC) - timedelta(seconds=running_ttl_seconds)
+ async with session_factory() as session:
+ rows = (
+ await session.execute(
+ select(AdminLeaderboardDailyStatRow.stat_date)
+ .where(
+ AdminLeaderboardDailyStatRow.settings_hash == settings_hash,
+ or_(
+ AdminLeaderboardDailyStatRow.status.in_(["queued", "error"]),
+ and_(
+ AdminLeaderboardDailyStatRow.status == "running",
+ AdminLeaderboardDailyStatRow.updated_at < cutoff,
+ ),
+ ),
+ )
+ .order_by(AdminLeaderboardDailyStatRow.stat_date.desc())
+ .limit(limit)
+ )
+ ).all()
+ return [str(row[0]) for row in rows]
+
+
+async def _claim_leaderboard_snapshot(
+ session_factory,
+ *,
+ stat_date: str,
+ settings_hash: str,
+ running_ttl_seconds: int,
+) -> bool:
+ from datetime import UTC, datetime
+
+ from app.gateway.routers.admin_users import _snapshot_is_fresh
+ from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
+ from deerflow.persistence.types import BEIJING_TZ
+
+ async with session_factory() as session:
+ row = await session.get(
+ AdminLeaderboardDailyStatRow,
+ {"stat_date": stat_date, "settings_hash": settings_hash},
+ )
+ if row is None or _snapshot_is_fresh(row, stat_date):
+ return False
+ if row.status == "running":
+ updated_at = row.updated_at
+ if isinstance(updated_at, datetime):
+ if updated_at.tzinfo is None:
+ updated_at = updated_at.replace(tzinfo=BEIJING_TZ)
+ age = (datetime.now(BEIJING_TZ) - updated_at.astimezone(BEIJING_TZ)).total_seconds()
+ if age < running_ttl_seconds:
+ return False
+ row.status = "running"
+ row.message = "排行榜后台统计任务已认领。"
+ row.error = None
+ row.updated_at = datetime.now(UTC)
+ await session.commit()
+ return True
+
+
+async def _run_leaderboard_snapshot_tick() -> tuple[int, int]:
+ from app.gateway.routers.admin_users import (
+ _daily_job_key,
+ _queue_daily_leaderboard_snapshots,
+ _refresh_daily_leaderboard_snapshot,
+ _resolve_leaderboard_snapshot_scope,
+ )
+ from deerflow.config.system_settings import load_system_settings
+ from deerflow.persistence.engine import get_session_factory
+
+ session_factory = get_session_factory()
+ if session_factory is None:
+ return 0, 0
+
+ settings = load_system_settings().leaderboard
+ excluded, include_scheduled, include_admins, include_failed, settings_hash = _resolve_leaderboard_snapshot_scope(settings)
+ stat_dates = _recent_leaderboard_stat_dates(_LEADERBOARD_SNAPSHOT_PREWARM_DAYS)
+ queued_dates = await _queue_daily_leaderboard_snapshots(
+ session_factory,
+ stat_dates,
+ settings_hash,
+ message="排行榜后台统计任务定期预热/补算队列。",
+ )
+
+ due_dates = await _leaderboard_snapshot_due_dates(
+ session_factory,
+ settings_hash=settings_hash,
+ limit=_LEADERBOARD_SNAPSHOT_BATCH_SIZE,
+ running_ttl_seconds=_LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS,
+ )
+ processed = 0
+ for stat_date in due_dates:
+ claimed = await _claim_leaderboard_snapshot(
+ session_factory,
+ stat_date=stat_date,
+ settings_hash=settings_hash,
+ running_ttl_seconds=_LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS,
+ )
+ if not claimed:
+ continue
+ logger.info("Leaderboard snapshot scheduler: building date=%s hash=%s", stat_date, settings_hash[:10])
+ await _refresh_daily_leaderboard_snapshot(
+ _daily_job_key(stat_date, settings_hash),
+ session_factory=session_factory,
+ settings=settings,
+ stat_date=stat_date,
+ settings_hash=settings_hash,
+ excluded=excluded,
+ include_scheduled=include_scheduled,
+ include_admins=include_admins,
+ include_failed=include_failed,
+ )
+ processed += 1
+ if _LEADERBOARD_SNAPSHOT_SLEEP_SECONDS > 0 and processed < len(due_dates):
+ await asyncio.sleep(_LEADERBOARD_SNAPSHOT_SLEEP_SECONDS)
+ return processed, len(queued_dates)
+
+
+async def _leaderboard_snapshot_scheduler_loop() -> None:
+ """Background task inside the existing gateway process for daily snapshots."""
+ if not _LEADERBOARD_SNAPSHOT_BACKGROUND_ENABLED:
+ logger.info("Leaderboard snapshot scheduler disabled by ADMIN_LEADERBOARD_BACKGROUND_ENABLED")
+ return
+
+ from deerflow.config.runtime_paths import runtime_home
+
+ lock_handle = _acquire_singleton_lock(runtime_home() / ".leaderboard_snapshot_scheduler.lock")
+ if lock_handle is None:
+ logger.info("Leaderboard snapshot scheduler: another gateway worker holds the lock; this worker skips")
+ return
+
+ logger.info(
+ "Leaderboard snapshot scheduler started: prewarm_days=%d batch_size=%d interval=%.1fs",
+ _LEADERBOARD_SNAPSHOT_PREWARM_DAYS,
+ _LEADERBOARD_SNAPSHOT_BATCH_SIZE,
+ _LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS,
+ )
+ try:
+ if _LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS > 0:
+ await asyncio.sleep(_LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS)
+ while True:
+ try:
+ processed, queued = await _run_leaderboard_snapshot_tick()
+ if processed or queued:
+ logger.info(
+ "Leaderboard snapshot scheduler tick: queued=%d processed=%d",
+ queued,
+ processed,
+ )
+ except Exception:
+ logger.exception("Leaderboard snapshot scheduler tick failed (non-fatal)")
+ await asyncio.sleep(_LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS)
+ finally:
+ _release_singleton_lock(lock_handle)
+
+
+def _warn_if_multi_worker() -> None:
"""检测 uvicorn / gunicorn 多 worker 部署并打 ERROR 日志。
主 agent / canvas / AI 写作流式响应在 Gateway 进程内通过 ``StreamBridge``
@@ -562,16 +799,21 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
# 默认每天凌晨 3 点扫一次,删 retention_days(默认 7 天)以前的会话,联动
# 清理 MySQL 业务表 + SQLite checkpointer 里的 thread 状态。多 worker
# 下用文件锁选 leader,只有抢到锁的 worker 真正跑清理。
- try:
- from app.gateway.ai_writing_cleanup import cleanup_scheduler_loop
-
- app.state.ai_writing_cleanup_task = asyncio.create_task(cleanup_scheduler_loop(app))
- except Exception:
- logger.exception("AI writing cleanup scheduler failed to start (non-fatal)")
-
- # Start IM channel service if any channels are configured
- try:
- from app.channels.service import start_channel_service
+ try:
+ from app.gateway.ai_writing_cleanup import cleanup_scheduler_loop
+
+ app.state.ai_writing_cleanup_task = asyncio.create_task(cleanup_scheduler_loop(app))
+ except Exception:
+ logger.exception("AI writing cleanup scheduler failed to start (non-fatal)")
+
+ try:
+ app.state.leaderboard_snapshot_scheduler_task = asyncio.create_task(_leaderboard_snapshot_scheduler_loop())
+ except Exception:
+ logger.exception("Leaderboard snapshot scheduler failed to start (non-fatal)")
+
+ # Start IM channel service if any channels are configured
+ try:
+ from app.channels.service import start_channel_service
channel_service = await start_channel_service(app.state.config)
logger.info("Channel service started: %s", channel_service.get_status())
@@ -615,7 +857,17 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
except asyncio.CancelledError:
pass
except Exception:
- logger.exception("AI writing cleanup scheduler shutdown error")
+ logger.exception("AI writing cleanup scheduler shutdown error")
+
+ leaderboard_snapshot_task = getattr(app.state, "leaderboard_snapshot_scheduler_task", None)
+ if leaderboard_snapshot_task is not None:
+ leaderboard_snapshot_task.cancel()
+ try:
+ await leaderboard_snapshot_task
+ except asyncio.CancelledError:
+ pass
+ except Exception:
+ logger.exception("Leaderboard snapshot scheduler shutdown error")
# 阶段 7 删冗余：AI 写作 graph 不再自管 checkpointer / in-flight task；
# thread 状态由 LangGraph Server 主 checkpointer 接管，lifespan 出口由
diff --git a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py
index 516b5ab8..753471ea 100644
--- a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py
+++ b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py
@@ -18,17 +18,16 @@ import json
import logging
import os
import re
-import time
from collections.abc import AsyncIterator
from difflib import SequenceMatcher
from hashlib import sha256
-from datetime import datetime, timedelta
+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
+from sqlalchemy import and_, case, desc, func, not_, or_, select, text
from langchain_core.messages import HumanMessage, SystemMessage
from app.gateway.deps import (
@@ -46,11 +45,13 @@ 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__)
@@ -64,14 +65,6 @@ _LEADERBOARD_TIMEOUT_SECONDS = max(
5.0,
float(os.getenv("ADMIN_LEADERBOARD_TIMEOUT_SECONDS", "45")),
)
-_LEADERBOARD_ACQUIRE_TIMEOUT_SECONDS = max(
- 0.1,
- float(os.getenv("ADMIN_LEADERBOARD_ACQUIRE_TIMEOUT_SECONDS", "0.5")),
-)
-_LEADERBOARD_CACHE_TTL_SECONDS = max(
- 5.0,
- float(os.getenv("ADMIN_LEADERBOARD_CACHE_TTL_SECONDS", "300")),
-)
_LEADERBOARD_QUERY_PAUSE_SECONDS = max(
0.0,
float(os.getenv("ADMIN_LEADERBOARD_QUERY_PAUSE_SECONDS", "0.05")),
@@ -80,10 +73,32 @@ _LEADERBOARD_START_DELAY_SECONDS = max(
0.0,
float(os.getenv("ADMIN_LEADERBOARD_START_DELAY_SECONDS", "0.2")),
)
+_LEADERBOARD_DAILY_SNAPSHOT_LIMIT = max(
+ 100,
+ int(os.getenv("ADMIN_LEADERBOARD_DAILY_SNAPSHOT_LIMIT", "1000")),
+)
+_LEADERBOARD_TODAY_TTL_SECONDS = max(
+ 30.0,
+ float(os.getenv("ADMIN_LEADERBOARD_TODAY_TTL_SECONDS", "300")),
+)
+_LEADERBOARD_QUERY_TIMEOUT_MS = max(
+ 500,
+ int(os.getenv("ADMIN_LEADERBOARD_QUERY_TIMEOUT_MS", "8000")),
+)
+_LEADERBOARD_LOCK_WAIT_SECONDS = max(
+ 1,
+ int(os.getenv("ADMIN_LEADERBOARD_LOCK_WAIT_SECONDS", "5")),
+)
+_LEADERBOARD_QUEUE_ON_REQUEST_ENABLED = os.getenv(
+ "ADMIN_LEADERBOARD_QUEUE_ON_REQUEST_ENABLED",
+ os.getenv("ADMIN_LEADERBOARD_WEB_QUEUE_ENABLED", "true"),
+).strip().lower() not in {
+ "0",
+ "false",
+ "no",
+ "off",
+}
_leaderboard_semaphore = asyncio.Semaphore(_LEADERBOARD_MAX_CONCURRENT)
-_leaderboard_jobs: dict[str, asyncio.Task] = {}
-_leaderboard_cache: dict[str, dict] = {}
-_leaderboard_jobs_lock = asyncio.Lock()
async def _require_admin(request: Request) -> None:
@@ -209,6 +224,45 @@ class AdminLeaderboardReportResponse(BaseModel):
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()
@@ -240,6 +294,56 @@ def _resolve_range(days: int, since: str | None, until: str | None) -> tuple[dat
return start, end, span_days
+def _as_beijing(value: datetime) -> datetime:
+ if value.tzinfo is None:
+ value = value.replace(tzinfo=UTC)
+ return value.astimezone(BEIJING_TZ)
+
+
+def _today_stat_date() -> str:
+ return datetime.now(BEIJING_TZ).date().isoformat()
+
+
+def _stat_dates_for_range(since: datetime, until: datetime) -> list[str]:
+ start_date = _as_beijing(since).date()
+ end_date = _as_beijing(until).date()
+ if start_date > end_date:
+ raise HTTPException(status_code=400, detail="since must be before until")
+ count = min(366, (end_date - start_date).days + 1)
+ return [(start_date + timedelta(days=i)).isoformat() for i in range(count)]
+
+
+def _bounds_for_stat_date(stat_date: str) -> tuple[datetime, datetime]:
+ try:
+ day = datetime.strptime(stat_date, "%Y-%m-%d").date()
+ except ValueError as exc:
+ raise HTTPException(status_code=400, detail=f"Invalid stat date: {stat_date}") from exc
+ start = datetime.combine(day, datetime_time.min).replace(tzinfo=BEIJING_TZ)
+ end = datetime.combine(day, datetime_time.max).replace(tzinfo=BEIJING_TZ)
+ return start, end
+
+
+async def _try_set_session_value(session, sql: str) -> None:
+ try:
+ await session.execute(text(sql))
+ except Exception:
+ logger.debug("Failed to apply analytics session setting: %s", sql, exc_info=True)
+ try:
+ await session.rollback()
+ except Exception:
+ logger.debug("Failed to rollback after analytics session setting failure", exc_info=True)
+
+
+async def _set_analytics_read_timeout(session) -> None:
+ conn = await session.connection()
+ dialect = conn.dialect.name
+ if dialect == "mysql":
+ await _try_set_session_value(session, f"SET SESSION max_execution_time = {_LEADERBOARD_QUERY_TIMEOUT_MS}")
+ await _try_set_session_value(session, f"SET SESSION innodb_lock_wait_timeout = {_LEADERBOARD_LOCK_WAIT_SECONDS}")
+ elif dialect == "postgresql":
+ await _try_set_session_value(session, f"SET LOCAL statement_timeout = {_LEADERBOARD_QUERY_TIMEOUT_MS}")
+
+
def _question_count_expr():
return func.sum(case((RunRow.first_human_message.is_not(None), 1), else_=0))
@@ -1018,31 +1122,6 @@ def _empty_leaderboard_response(
)
-def _leaderboard_cache_key(
- *,
- since_dt: datetime,
- until_dt: datetime,
- limit: int,
- excluded: set[str],
- include_scheduled: bool,
- include_admins: bool,
- include_failed: bool,
- cleanup_system_questions: bool,
-) -> str:
- payload = {
- "version": 2,
- "since": _iso(since_dt),
- "until": _iso(until_dt),
- "limit": limit,
- "excluded": 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()
-
-
async def _leaderboard_pause() -> None:
if _LEADERBOARD_QUERY_PAUSE_SECONDS > 0:
await asyncio.sleep(_LEADERBOARD_QUERY_PAUSE_SECONDS)
@@ -1061,6 +1140,7 @@ async def _build_user_leaderboard_response(
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,
@@ -1119,15 +1199,523 @@ async def _build_user_leaderboard_response(
)
-async def _refresh_leaderboard_cache(
- cache_key: str,
+def _leaderboard_daily_settings_hash(
*,
- session_factory,
+ excluded: set[str],
+ include_scheduled: bool,
+ include_admins: bool,
+ include_failed: bool,
+ cleanup_system_questions: bool,
+) -> str:
+ payload = {
+ "version": 2,
+ "snapshot_limit": _LEADERBOARD_DAILY_SNAPSHOT_LIMIT,
+ "exclude_user_ids": sorted(excluded),
+ "include_scheduled": include_scheduled,
+ "include_admins": include_admins,
+ "include_failed": include_failed,
+ "cleanup_system_questions": cleanup_system_questions,
+ }
+ return sha256(json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest()
+
+
+def _resolve_leaderboard_snapshot_scope(
settings,
+ *,
+ exclude_user_ids: list[str] | None = None,
+ include_scheduled: bool | None = None,
+ include_admins: bool | None = None,
+ include_failed: bool | None = None,
+) -> tuple[set[str], bool, bool, bool, str]:
+ excluded = {uid for uid in settings.exclude_user_ids if uid}
+ excluded.update({uid for uid in (exclude_user_ids or []) if uid})
+ resolved_include_scheduled = settings.include_scheduled if include_scheduled is None else include_scheduled
+ resolved_include_admins = settings.include_admins if include_admins is None else include_admins
+ resolved_include_failed = settings.include_failed if include_failed is None else include_failed
+ settings_hash = _leaderboard_daily_settings_hash(
+ excluded=excluded,
+ include_scheduled=resolved_include_scheduled,
+ include_admins=resolved_include_admins,
+ include_failed=resolved_include_failed,
+ cleanup_system_questions=settings.cleanup_system_questions,
+ )
+ return excluded, resolved_include_scheduled, resolved_include_admins, resolved_include_failed, settings_hash
+
+
+def _daily_job_key(stat_date: str, settings_hash: str) -> str:
+ return f"{stat_date}:{settings_hash}"
+
+
+def _sort_datetime(value: str | None) -> datetime:
+ if not value:
+ return datetime.min
+ text_value = str(value).strip()
+ if text_value.endswith("Z"):
+ text_value = text_value[:-1] + "+00:00"
+ try:
+ parsed = datetime.fromisoformat(text_value)
+ except ValueError:
+ return datetime.min
+ if parsed.tzinfo is not None:
+ return parsed.astimezone(UTC).replace(tzinfo=None)
+ return parsed
+
+
+def _latest_iso(left: str | None, right: str | None) -> str:
+ if not left:
+ return str(right or "")
+ if not right:
+ return str(left or "")
+ return str(right if _sort_datetime(right) > _sort_datetime(left) else left)
+
+
+def _snapshot_is_fresh(row: AdminLeaderboardDailyStatRow | None, stat_date: str) -> bool:
+ if row is None or not row.payload_json or row.status != "ready":
+ return False
+ if stat_date != _today_stat_date():
+ return True
+ generated_at = row.generated_at
+ if not isinstance(generated_at, datetime):
+ return False
+ age = (datetime.now(BEIJING_TZ) - _as_beijing(generated_at)).total_seconds()
+ return age <= _LEADERBOARD_TODAY_TTL_SECONDS
+
+
+def _snapshot_response_from_row(row: AdminLeaderboardDailyStatRow | None) -> AdminLeaderboardResponse | None:
+ if row is None or not row.payload_json:
+ return None
+ payload = row.payload_json
+ if isinstance(payload, str):
+ try:
+ payload = json.loads(payload)
+ except json.JSONDecodeError:
+ return None
+ try:
+ return AdminLeaderboardResponse.model_validate(payload)
+ except Exception:
+ logger.warning("Invalid leaderboard snapshot payload: %s/%s", row.stat_date, row.settings_hash, exc_info=True)
+ return None
+
+
+def _merge_users_from_daily(responses: list[AdminLeaderboardResponse], limit: int) -> list[AdminLeaderboardUser]:
+ users_by_id: dict[str, dict] = {}
+ for response in responses:
+ day_users: dict[str, AdminLeaderboardUser] = {}
+ for ranking in (response.users_by_activity, response.users_by_questions, response.users_by_tokens):
+ for user in ranking:
+ if user.user_id:
+ day_users[user.user_id] = user
+ for user in day_users.values():
+ item = users_by_id.setdefault(
+ user.user_id,
+ {
+ "user_id": user.user_id,
+ "email": user.email,
+ "system_role": user.system_role or "user",
+ "thread_count": 0,
+ "run_count": 0,
+ "qa_count": 0,
+ "llm_call_count": 0,
+ "tool_call_count": 0,
+ "skill_call_count": 0,
+ "total_tokens": 0,
+ "input_tokens": 0,
+ "output_tokens": 0,
+ "last_active_at": "",
+ },
+ )
+ if user.email:
+ item["email"] = user.email
+ if user.system_role:
+ item["system_role"] = user.system_role
+ for field in (
+ "thread_count",
+ "run_count",
+ "qa_count",
+ "llm_call_count",
+ "tool_call_count",
+ "skill_call_count",
+ "total_tokens",
+ "input_tokens",
+ "output_tokens",
+ ):
+ item[field] += int(getattr(user, field) or 0)
+ item["last_active_at"] = _latest_iso(item["last_active_at"], user.last_active_at)
+ users = [AdminLeaderboardUser(**item) for item in users_by_id.values()]
+ return sorted(users, key=lambda u: (u.run_count + u.tool_call_count, u.total_tokens), reverse=True)[:limit]
+
+
+def _merge_call_rankings_from_daily(
+ responses: list[AdminLeaderboardResponse],
+ field_name: str,
+ limit: int,
+) -> list[AdminLeaderboardSkill]:
+ items_by_name: dict[str, dict] = {}
+ for response in responses:
+ for item in getattr(response, field_name):
+ if not item.name:
+ continue
+ merged = items_by_name.setdefault(
+ item.name,
+ {
+ "name": item.name,
+ "call_count": 0,
+ "user_count": 0,
+ "success_count": 0,
+ "error_count": 0,
+ "duration_weight": 0,
+ "last_used_at": "",
+ },
+ )
+ calls = int(item.call_count or 0)
+ merged["call_count"] += calls
+ merged["user_count"] += int(item.user_count or 0)
+ merged["success_count"] += int(item.success_count or 0)
+ merged["error_count"] += int(item.error_count or 0)
+ merged["duration_weight"] += int(item.avg_duration_ms or 0) * calls
+ merged["last_used_at"] = _latest_iso(merged["last_used_at"], item.last_used_at)
+ merged_items = [
+ AdminLeaderboardSkill(
+ name=item["name"],
+ call_count=item["call_count"],
+ user_count=item["user_count"],
+ success_count=item["success_count"],
+ error_count=item["error_count"],
+ avg_duration_ms=int(item["duration_weight"] / item["call_count"]) if item["call_count"] else 0,
+ last_used_at=item["last_used_at"],
+ )
+ for item in items_by_name.values()
+ ]
+ return sorted(merged_items, key=lambda item: (item.call_count, _sort_datetime(item.last_used_at)), reverse=True)[:limit]
+
+
+def _merge_dimensions_from_daily(
+ responses: list[AdminLeaderboardResponse],
+ field_name: str,
+ limit: int,
+) -> list[AdminLeaderboardDimension]:
+ items_by_name: dict[str, dict] = {}
+ for response in responses:
+ for item in getattr(response, field_name):
+ if not item.name:
+ continue
+ merged = items_by_name.setdefault(
+ item.name,
+ {
+ "name": item.name,
+ "count": 0,
+ "user_count": 0,
+ "success_count": 0,
+ "error_count": 0,
+ "total_tokens": 0,
+ "last_seen_at": "",
+ },
+ )
+ for field in ("count", "user_count", "success_count", "error_count", "total_tokens"):
+ merged[field] += int(getattr(item, field) or 0)
+ merged["last_seen_at"] = _latest_iso(merged["last_seen_at"], item.last_seen_at)
+ merged_items = [AdminLeaderboardDimension(**item) for item in items_by_name.values()]
+ return sorted(merged_items, key=lambda item: (item.count, _sort_datetime(item.last_seen_at)), reverse=True)[:limit]
+
+
+def _merge_recent_questions_from_daily(
+ responses: list[AdminLeaderboardResponse],
+ limit: int,
+) -> list[AdminLeaderboardQuestion]:
+ questions: list[AdminLeaderboardQuestion] = []
+ seen: set[str] = set()
+ for response in responses:
+ for question in response.recent_questions:
+ key = question.run_id or f"{question.user_id}:{question.asked_at}:{question.text}"
+ if key in seen:
+ continue
+ seen.add(key)
+ questions.append(question)
+ questions.sort(key=lambda item: _sort_datetime(item.asked_at), reverse=True)
+ return questions[:limit]
+
+
+_FREQUENT_QUESTION_COUNT_RE = re.compile(r"（相似问题\s+(\d+)\s+次）$")
+
+
+def _split_frequent_question_count(text: str) -> tuple[str, int]:
+ match = _FREQUENT_QUESTION_COUNT_RE.search(text)
+ if not match:
+ return text, 1
+ return _FREQUENT_QUESTION_COUNT_RE.sub("", text).strip(), max(1, int(match.group(1)))
+
+
+def _merge_frequent_questions_from_daily(
+ responses: list[AdminLeaderboardResponse],
+ limit: int,
+) -> list[AdminLeaderboardQuestion]:
+ groups: list[dict] = []
+ for response in responses:
+ for question in response.frequent_questions:
+ base_text, count = _split_frequent_question_count(question.text)
+ key = _question_similarity_key(base_text)
+ if not key:
+ continue
+ target = None
+ for group in groups:
+ score = SequenceMatcher(None, key, group["key"]).ratio()
+ if score >= 0.72 or key in group["key"] or group["key"] in key:
+ target = group
+ break
+ if target is None:
+ groups.append(
+ {
+ "key": key,
+ "text": base_text,
+ "count": count,
+ "user_id": question.user_id,
+ "email": question.email,
+ "thread_id": question.thread_id,
+ "run_id": question.run_id,
+ "asked_at": question.asked_at,
+ }
+ )
+ else:
+ target["count"] += count
+ if _sort_datetime(question.asked_at) > _sort_datetime(target.get("asked_at")):
+ target.update(
+ {
+ "text": base_text,
+ "user_id": question.user_id,
+ "email": question.email,
+ "thread_id": question.thread_id,
+ "run_id": question.run_id,
+ "asked_at": question.asked_at,
+ }
+ )
+ groups.sort(key=lambda item: (item["count"], _sort_datetime(item.get("asked_at"))), reverse=True)
+ questions: list[AdminLeaderboardQuestion] = []
+ for group in groups[:limit]:
+ text_value = group["text"][:460]
+ if group["count"] > 1:
+ text_value = f"{text_value}（相似问题 {group['count']} 次）"
+ questions.append(
+ AdminLeaderboardQuestion(
+ text=text_value,
+ user_id=group["user_id"],
+ email=group["email"],
+ thread_id=group["thread_id"],
+ run_id=group["run_id"],
+ asked_at=group["asked_at"],
+ )
+ )
+ return questions
+
+
+def _merge_trends_from_daily(responses: list[AdminLeaderboardResponse]) -> list[AdminLeaderboardTrendPoint]:
+ trends_by_date: dict[str, AdminLeaderboardTrendPoint] = {}
+ for response in responses:
+ for point in response.trends:
+ item = trends_by_date.setdefault(point.date, AdminLeaderboardTrendPoint(date=point.date))
+ item.runs += int(point.runs or 0)
+ item.questions += int(point.questions or 0)
+ item.tokens += int(point.tokens or 0)
+ item.users += int(point.users or 0)
+ return sorted(trends_by_date.values(), key=lambda item: item.date)
+
+
+def _merge_daily_leaderboard_payloads(
+ responses: list[AdminLeaderboardResponse],
+ *,
since_dt: datetime,
until_dt: datetime,
range_days: int,
limit: int,
+ status: str,
+ message: str,
+) -> AdminLeaderboardResponse:
+ users = _merge_users_from_daily(responses, _LEADERBOARD_DAILY_SNAPSHOT_LIMIT)
+ active_user_ids = {
+ user.user_id
+ for user in users
+ if user.user_id and (user.run_count or user.qa_count or user.tool_call_count or user.thread_count)
+ }
+ overview = AdminLeaderboardOverview(
+ total_users=max((item.overview.total_users for item in responses), default=0),
+ active_users=len(active_user_ids) if active_user_ids else sum(item.overview.active_users for item in responses),
+ total_threads=sum(item.overview.total_threads for item in responses),
+ total_runs=sum(item.overview.total_runs for item in responses),
+ total_questions=sum(item.overview.total_questions for item in responses),
+ total_tokens=sum(item.overview.total_tokens for item in responses),
+ total_skill_calls=sum(item.overview.total_skill_calls for item in responses),
+ total_tool_calls=sum(item.overview.total_tool_calls for item in responses),
+ error_runs=sum(item.overview.error_runs for item in responses),
+ )
+ generated_at = ""
+ for response in responses:
+ generated_at = _latest_iso(generated_at, response.generated_at)
+ return AdminLeaderboardResponse(
+ range_days=range_days,
+ since=_iso(since_dt),
+ until=_iso(until_dt),
+ status=status,
+ generated_at=generated_at or _iso(datetime.now(BEIJING_TZ)),
+ message=message,
+ overview=overview,
+ users_by_activity=sorted(users, key=lambda u: (u.run_count + u.tool_call_count, u.total_tokens), reverse=True)[:limit],
+ users_by_questions=sorted(users, key=lambda u: (u.qa_count, u.run_count), reverse=True)[:limit],
+ users_by_tokens=sorted(users, key=lambda u: u.total_tokens, reverse=True)[:limit],
+ skills=_merge_call_rankings_from_daily(responses, "skills", limit),
+ tools=_merge_call_rankings_from_daily(responses, "tools", limit),
+ models=_merge_dimensions_from_daily(responses, "models", limit),
+ agents=_merge_dimensions_from_daily(responses, "agents", limit),
+ run_statuses=_merge_dimensions_from_daily(responses, "run_statuses", limit),
+ recent_questions=_merge_recent_questions_from_daily(responses, limit),
+ frequent_questions=_merge_frequent_questions_from_daily(responses, limit),
+ trends=_merge_trends_from_daily(responses),
+ )
+
+
+async def _load_daily_snapshot_rows(
+ session_factory,
+ stat_dates: list[str],
+ settings_hash: str,
+) -> dict[str, AdminLeaderboardDailyStatRow]:
+ if not stat_dates:
+ return {}
+ async with session_factory() as session:
+ rows = (
+ await session.execute(
+ select(AdminLeaderboardDailyStatRow).where(
+ AdminLeaderboardDailyStatRow.stat_date.in_(stat_dates),
+ AdminLeaderboardDailyStatRow.settings_hash == settings_hash,
+ )
+ )
+ ).scalars().all()
+ return {row.stat_date: row for row in rows}
+
+
+async def _store_daily_snapshot(
+ session_factory,
+ stat_date: str,
+ settings_hash: str,
+ *,
+ status: str,
+ payload: dict | None = None,
+ message: str | None = None,
+ error: str | None = None,
+ generated_at: datetime | None = None,
+) -> None:
+ async with session_factory() as session:
+ now = datetime.now(UTC)
+ row = await session.get(
+ AdminLeaderboardDailyStatRow,
+ {"stat_date": stat_date, "settings_hash": settings_hash},
+ )
+ if row is None:
+ row = AdminLeaderboardDailyStatRow(
+ stat_date=stat_date,
+ settings_hash=settings_hash,
+ status=status,
+ payload_json=payload or {},
+ message=message,
+ error=error,
+ generated_at=generated_at,
+ created_at=now,
+ updated_at=now,
+ )
+ session.add(row)
+ else:
+ row.status = status
+ if payload is not None:
+ row.payload_json = payload
+ if message is not None:
+ row.message = message
+ row.error = error
+ if generated_at is not None:
+ row.generated_at = generated_at
+ row.updated_at = now
+ await session.commit()
+
+
+async def _queue_daily_leaderboard_snapshots(
+ session_factory,
+ stat_dates: list[str],
+ settings_hash: str,
+ *,
+ force: bool = False,
+ message: str = "等待排行榜后台统计任务处理。",
+) -> list[str]:
+ queued: list[str] = []
+ if not stat_dates:
+ return queued
+ async with session_factory() as session:
+ now = datetime.now(UTC)
+ rows = (
+ await session.execute(
+ select(AdminLeaderboardDailyStatRow).where(
+ AdminLeaderboardDailyStatRow.stat_date.in_(stat_dates),
+ AdminLeaderboardDailyStatRow.settings_hash == settings_hash,
+ )
+ )
+ ).scalars().all()
+ rows_by_date = {row.stat_date: row for row in rows}
+ for stat_date in stat_dates:
+ row = rows_by_date.get(stat_date)
+ if row is None:
+ row = AdminLeaderboardDailyStatRow(
+ stat_date=stat_date,
+ settings_hash=settings_hash,
+ status="queued",
+ payload_json={},
+ message=message,
+ created_at=now,
+ updated_at=now,
+ )
+ session.add(row)
+ queued.append(stat_date)
+ continue
+ if not force and _snapshot_is_fresh(row, stat_date):
+ continue
+ if row.status == "running":
+ continue
+ row.status = "queued"
+ row.message = message
+ row.error = None
+ row.updated_at = now
+ queued.append(stat_date)
+ if queued:
+ await session.commit()
+ return queued
+
+
+async def _build_daily_leaderboard_snapshot(
+ session_factory,
+ settings,
+ stat_date: str,
+ *,
+ excluded: set[str],
+ include_scheduled: bool,
+ include_admins: bool,
+ include_failed: bool,
+) -> AdminLeaderboardResponse:
+ since_dt, until_dt = _bounds_for_stat_date(stat_date)
+ return await _build_user_leaderboard_response(
+ session_factory,
+ settings,
+ since_dt,
+ until_dt,
+ 1,
+ _LEADERBOARD_DAILY_SNAPSHOT_LIMIT,
+ excluded,
+ include_scheduled,
+ include_admins,
+ include_failed,
+ )
+
+
+async def _refresh_daily_leaderboard_snapshot(
+ job_key: str,
+ *,
+ session_factory,
+ settings,
+ stat_date: str,
+ settings_hash: str,
excluded: set[str],
include_scheduled: bool,
include_admins: bool,
@@ -1137,47 +1725,140 @@ async def _refresh_leaderboard_cache(
try:
if _LEADERBOARD_START_DELAY_SECONDS > 0:
await asyncio.sleep(_LEADERBOARD_START_DELAY_SECONDS)
+ await _store_daily_snapshot(
+ session_factory,
+ stat_date,
+ settings_hash,
+ status="running",
+ message="正在生成当天排行榜快照。",
+ )
await _leaderboard_semaphore.acquire()
acquired = True
response = await asyncio.wait_for(
- _build_user_leaderboard_response(
+ _build_daily_leaderboard_snapshot(
session_factory,
settings,
- since_dt,
- until_dt,
- range_days,
- limit,
- excluded,
- include_scheduled,
- include_admins,
- include_failed,
+ stat_date,
+ excluded=excluded,
+ include_scheduled=include_scheduled,
+ include_admins=include_admins,
+ include_failed=include_failed,
),
timeout=_LEADERBOARD_TIMEOUT_SECONDS,
)
- _leaderboard_cache[cache_key] = {
- "created_at": time.monotonic(),
- "response": response,
- }
+ generated_at = datetime.now(BEIJING_TZ)
+ response.generated_at = _iso(generated_at)
+ await _store_daily_snapshot(
+ session_factory,
+ stat_date,
+ settings_hash,
+ status="ready",
+ payload=response.model_dump(mode="json"),
+ message="",
+ error=None,
+ generated_at=generated_at,
+ )
except asyncio.TimeoutError:
- logger.warning("Leaderboard refresh timed out for key=%s", cache_key)
- except Exception:
- logger.exception("Leaderboard refresh failed for key=%s", cache_key)
+ logger.warning("Daily leaderboard snapshot timed out: %s", job_key)
+ await _store_daily_snapshot(
+ session_factory,
+ stat_date,
+ settings_hash,
+ status="error",
+ message="当天排行榜快照统计超时，后台会在下次访问时重试。",
+ error="timeout",
+ )
+ except Exception as exc:
+ logger.exception("Daily leaderboard snapshot failed: %s", job_key)
+ await _store_daily_snapshot(
+ session_factory,
+ stat_date,
+ settings_hash,
+ status="error",
+ message="当天排行榜快照统计失败，后台会在下次访问时重试。",
+ error=str(exc),
+ )
finally:
if acquired:
_leaderboard_semaphore.release()
- async with _leaderboard_jobs_lock:
- _leaderboard_jobs.pop(cache_key, None)
-async def _ensure_leaderboard_refresh(cache_key: str, **kwargs) -> None:
- async with _leaderboard_jobs_lock:
- task = _leaderboard_jobs.get(cache_key)
- if task and not task.done():
- return
- _leaderboard_jobs[cache_key] = asyncio.create_task(
- _refresh_leaderboard_cache(cache_key, **kwargs)
+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(
@@ -1198,146 +1879,103 @@ async def get_user_leaderboard(
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)
- 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
- cache_key = _leaderboard_cache_key(
- since_dt=since_dt,
- until_dt=until_dt,
- limit=limit,
- excluded=excluded,
- include_scheduled=resolved_include_scheduled,
- include_admins=resolved_include_admins,
- include_failed=resolved_include_failed,
- cleanup_system_questions=settings.cleanup_system_questions,
- )
-
- cached = _leaderboard_cache.get(cache_key)
- if cached:
- response = cached["response"].model_copy()
- age = time.monotonic() - float(cached["created_at"])
- if age <= _LEADERBOARD_CACHE_TTL_SECONDS:
- response.status = "ready"
- return response
- response.status = "refreshing"
- response.message = "排行榜缓存已返回，后台正在刷新最新统计。"
- await _ensure_leaderboard_refresh(
- cache_key,
- session_factory=session_factory,
- settings=settings,
- since_dt=since_dt,
- until_dt=until_dt,
- range_days=range_days,
- limit=limit,
- excluded=excluded,
- include_scheduled=resolved_include_scheduled,
- include_admins=resolved_include_admins,
- include_failed=resolved_include_failed,
- )
- return response
-
- await _ensure_leaderboard_refresh(
- cache_key,
- session_factory=session_factory,
- settings=settings,
- since_dt=since_dt,
- until_dt=until_dt,
- range_days=range_days,
- limit=limit,
- excluded=excluded,
- include_scheduled=resolved_include_scheduled,
- include_admins=resolved_include_admins,
- include_failed=resolved_include_failed,
- )
- return _empty_leaderboard_response(
+ 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,
- range_days,
- status="running",
- message="排行榜统计已在后台运行。线上用户较多时会慢一些，页面会自动刷新结果。",
+ limit=limit,
+ exclude_user_ids=exclude_user_ids,
+ include_scheduled=include_scheduled,
+ include_admins=include_admins,
+ include_failed=include_failed,
)
- try:
- await asyncio.wait_for(
- _leaderboard_semaphore.acquire(),
- timeout=_LEADERBOARD_ACQUIRE_TIMEOUT_SECONDS,
- )
- except TimeoutError as exc:
- raise HTTPException(
- status_code=429,
- detail="排行榜正在统计中，请稍后再试。线上用户较多时统计时间可能较长，请耐心等待。",
- ) from exc
- async def build_response() -> AdminLeaderboardResponse:
- settings = load_system_settings().leaderboard
- since_dt, until_dt, range_days = _resolve_range(days, since, until)
- 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
- async with session_factory() as session:
- common = {
- "exclude_user_ids": excluded,
- "include_scheduled": resolved_include_scheduled,
- "include_admins": resolved_include_admins,
- "include_failed": resolved_include_failed,
- }
- tool_common = {
- "exclude_user_ids": excluded,
- "include_scheduled": resolved_include_scheduled,
- "include_admins": resolved_include_admins,
- }
- question_common = {
- **common,
- "cleanup_system_questions": settings.cleanup_system_questions,
- }
- overview = await _query_overview(session, since_dt, until_dt, **common)
- users = await _query_user_leaderboard(session, since_dt, until_dt, **common)
- skills = await _query_call_rankings(session, since_dt, until_dt, user_id=None, field="skill", limit=limit, **tool_common)
- tools = await _query_call_rankings(session, since_dt, until_dt, user_id=None, field="tool", limit=limit, **tool_common)
- 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)
- agents = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="agent", limit=limit, **common)
- run_statuses = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="status", limit=limit, **common)
- recent_questions = await _query_recent_questions(session, since_dt, until_dt, user_id=None, limit=limit, **question_common)
- frequent_questions = await _query_frequent_questions(session, since_dt, until_dt, limit=limit, **question_common)
- trends = await _query_trends(session, since_dt, until_dt, **common)
+@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")
- return AdminLeaderboardResponse(
- range_days=range_days,
- since=_iso(since_dt),
- until=_iso(until_dt),
- overview=overview,
- users_by_activity=sorted(users, key=lambda u: (u.run_count + u.tool_call_count, u.total_tokens), reverse=True)[:limit],
- users_by_questions=sorted(users, key=lambda u: (u.qa_count, u.run_count), reverse=True)[:limit],
- users_by_tokens=sorted(users, key=lambda u: u.total_tokens, reverse=True)[:limit],
- skills=skills,
- tools=tools,
- models=models,
- agents=agents,
- run_statuses=run_statuses,
- recent_questions=recent_questions,
- frequent_questions=frequent_questions,
- trends=trends,
+ 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,
+ )
- try:
- return await asyncio.wait_for(
- build_response(),
- timeout=_LEADERBOARD_TIMEOUT_SECONDS,
- )
- except TimeoutError as exc:
- raise HTTPException(
- status_code=504,
- detail="排行榜统计时间过长，请缩短时间范围后重试，或稍后再查看。",
- ) from exc
- finally:
- _leaderboard_semaphore.release()
+
+@router.post("/leaderboard/snapshots/backfill", response_model=AdminLeaderboardSnapshotBackfillResponse)
+async def queue_leaderboard_snapshot_backfill(
+ body: AdminLeaderboardSnapshotBackfillRequest,
+ request: Request,
+) -> AdminLeaderboardSnapshotBackfillResponse:
+ """Queue a date range for the leaderboard background snapshot scheduler."""
+ await _require_admin(request)
+ session_factory = get_session_factory()
+ if session_factory is None:
+ raise HTTPException(status_code=503, detail="Analytics requires database persistence")
+
+ settings = load_system_settings().leaderboard
+ since_dt, until_dt, _range_days = _resolve_range(body.days, body.since, body.until)
+ stat_dates = _stat_dates_for_range(since_dt, until_dt)
+ *_scope, settings_hash = _resolve_leaderboard_snapshot_scope(settings)
+ queued_dates = await _queue_daily_leaderboard_snapshots(
+ session_factory,
+ stat_dates,
+ settings_hash,
+ force=body.force,
+ message="管理员手动补算请求已加入排行榜后台统计队列。",
+ )
+ return AdminLeaderboardSnapshotBackfillResponse(
+ settings_hash=settings_hash,
+ since=_iso(since_dt),
+ until=_iso(until_dt),
+ total_days=len(stat_dates),
+ queued_dates=queued_dates,
+ message=f"已加入 {len(queued_dates)} 天快照队列，后台统计任务会继续补齐。",
+ )
async def _build_leaderboard_report_prompt(
@@ -1347,38 +1985,26 @@ async def _build_leaderboard_report_prompt(
until_dt: datetime,
limit: int,
) -> str:
- excluded = {uid for uid in settings.exclude_user_ids if uid}
- common = {
- "exclude_user_ids": excluded,
- "include_scheduled": settings.include_scheduled,
- "include_admins": settings.include_admins,
- "include_failed": settings.include_failed,
- }
- tool_common = {
- "exclude_user_ids": excluded,
- "include_scheduled": settings.include_scheduled,
- "include_admins": settings.include_admins,
- }
- question_common = {
- **common,
- "cleanup_system_questions": settings.cleanup_system_questions,
- }
+ 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="排行榜快照正在后台生成，请稍后再生成报告。")
- async with session_factory() as session:
- overview = await _query_overview(session, since_dt, until_dt, **common)
- users = await _query_user_leaderboard(session, since_dt, until_dt, **common)
- skills = await _query_call_rankings(session, since_dt, until_dt, user_id=None, field="skill", limit=limit, **tool_common)
- tools = await _query_call_rankings(session, since_dt, until_dt, user_id=None, field="tool", limit=limit, **tool_common)
- 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)
- agents = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="agent", limit=limit, **common)
- run_statuses = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="status", limit=limit, **common)
- recent_questions = await _query_recent_questions(session, since_dt, until_dt, user_id=None, limit=limit, **question_common)
- frequent_questions = await _query_frequent_questions(session, since_dt, until_dt, limit=limit, **question_common)
- trends = await _query_trends(session, since_dt, until_dt, **common)
-
- top_users = sorted(users, key=lambda u: (u.run_count + u.tool_call_count, u.total_tokens), reverse=True)[:12]
+ overview = snapshot.overview
+ skills = snapshot.skills
+ tools = snapshot.tools
+ models = snapshot.models
+ agents = snapshot.agents
+ run_statuses = snapshot.run_statuses
+ recent_questions = snapshot.recent_questions
+ frequent_questions = snapshot.frequent_questions
+ trends = snapshot.trends
+ top_users = snapshot.users_by_activity[:12]
context = {
"range": {"since": _iso(since_dt), "until": _iso(until_dt)},
"overview": overview.model_dump(),
@@ -1566,6 +2192,7 @@ async def get_user_analytics(
}
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")
@@ -1596,6 +2223,128 @@ async def get_user_analytics(
)
+# ── Q&A audit list ─────────────────────────────────────────────────────────
+
+
+class AdminUserQaItem(BaseModel):
+ run_id: str
+ thread_id: str = ""
+ thread_title: str = ""
+ assistant_id: str = ""
+ status: str = ""
+ model_name: str = ""
+ question: str = ""
+ answer: str = ""
+ total_tokens: int = 0
+ created_at: str = ""
+ updated_at: str = ""
+
+
+class AdminUserQaPageResponse(BaseModel):
+ user_id: str
+ page: int
+ page_size: int
+ total: int
+ items: list[AdminUserQaItem] = Field(default_factory=list)
+
+
+@router.get("/{user_id}/qa", response_model=AdminUserQaPageResponse)
+async def list_user_qa(
+ user_id: str,
+ request: Request,
+ page: int = Query(default=1, ge=1),
+ page_size: int = Query(default=20, ge=1, le=100),
+ q: str | None = Query(default=None, description="Search question, answer, run id, or thread id"),
+ since: str | None = Query(default=None, description="Inclusive range start, ISO datetime"),
+ until: str | None = Query(default=None, description="Inclusive range end, ISO datetime"),
+) -> AdminUserQaPageResponse:
+ """Return one user's Q&A runs from lightweight run summary fields."""
+ await _require_admin(request)
+ session_factory = get_session_factory()
+ if session_factory is None:
+ raise HTTPException(status_code=503, detail="Q&A audit requires database persistence")
+
+ filters = [
+ RunRow.user_id == user_id,
+ RunRow.first_human_message.is_not(None),
+ RunRow.first_human_message != "",
+ ]
+ if since or until:
+ since_dt, until_dt, _range_days = _resolve_range(365, since, until)
+ filters.append(_run_time_filter(since_dt, until_dt))
+ needle = (q or "").strip()
+ if needle:
+ like = f"%{needle}%"
+ filters.append(
+ or_(
+ RunRow.first_human_message.like(like),
+ RunRow.last_ai_message.like(like),
+ RunRow.run_id.like(like),
+ RunRow.thread_id.like(like),
+ )
+ )
+
+ offset = (page - 1) * page_size
+ async with session_factory() as session:
+ await _set_analytics_read_timeout(session)
+ total = (
+ await session.execute(
+ select(func.count(RunRow.run_id)).where(*filters)
+ )
+ ).scalar_one()
+ rows = (
+ await session.execute(
+ select(
+ RunRow.run_id,
+ RunRow.thread_id,
+ ThreadMetaRow.display_name,
+ RunRow.assistant_id,
+ RunRow.status,
+ RunRow.model_name,
+ RunRow.first_human_message,
+ RunRow.last_ai_message,
+ RunRow.total_tokens,
+ RunRow.created_at,
+ RunRow.updated_at,
+ )
+ .outerjoin(
+ ThreadMetaRow,
+ and_(
+ ThreadMetaRow.thread_id == RunRow.thread_id,
+ ThreadMetaRow.user_id == RunRow.user_id,
+ ),
+ )
+ .where(*filters)
+ .order_by(RunRow.created_at.desc(), RunRow.run_id.desc())
+ .limit(page_size)
+ .offset(offset)
+ )
+ ).all()
+
+ return AdminUserQaPageResponse(
+ user_id=user_id,
+ page=page,
+ page_size=page_size,
+ total=int(total or 0),
+ items=[
+ AdminUserQaItem(
+ run_id=r[0] or "",
+ thread_id=r[1] or "",
+ thread_title=r[2] or "",
+ assistant_id=r[3] or "",
+ status=r[4] or "",
+ model_name=r[5] or "",
+ question=_clean_question_text(r[6])[:2000],
+ answer=str(r[7] or "").strip()[:6000],
+ total_tokens=int(r[8] or 0),
+ created_at=_iso(r[9]),
+ updated_at=_iso(r[10]),
+ )
+ for r in rows
+ ],
+ )
+
+
# ── Threads ────────────────────────────────────────────────────────────────
diff --git a/offline-backend-20260512/backend/app/gateway/routers/tags.py b/offline-backend-20260512/backend/app/gateway/routers/tags.py
index a8960602..e85a5464 100644
--- a/offline-backend-20260512/backend/app/gateway/routers/tags.py
+++ b/offline-backend-20260512/backend/app/gateway/routers/tags.py
@@ -1,9 +1,9 @@
"""Tag management API.
Tags are split into three types (``agent`` / ``skill`` / ``scheduled_task``).
-A tag may only be applied to a resource of its own type. Tag CRUD and tag
-assignment are admin-only; listing tags and reading a resource's tags is open
-to everyone (the frontend needs them to render chips and search filters).
+A tag may only be applied to a resource of its own type. Tag CRUD, listing,
+and assignment are available to ordinary users so teams can organize resources
+without asking an administrator to tag every item.
"""
from __future__ import annotations
@@ -11,10 +11,10 @@ from __future__ import annotations
import logging
import uuid
-from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
-from app.gateway.deps import get_optional_user_from_request, get_tag_store
+from app.gateway.deps import get_tag_store
from deerflow.persistence.tags import TAG_TYPES, TagStore
logger = logging.getLogger(__name__)
@@ -22,18 +22,6 @@ logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/tags", tags=["tags"])
-# --------------------------------------------------------------------------- auth
-
-
-async def _require_admin(request: Request) -> None:
- """Reject non-admins. Lets the call through when auth is disabled."""
- user = await get_optional_user_from_request(request)
- if user is None:
- return
- if getattr(user, "system_role", None) != "admin":
- raise HTTPException(status_code=403, detail="Tag management is admin-only")
-
-
def _validate_tag_type(tag_type: str) -> str:
if tag_type not in TAG_TYPES:
raise HTTPException(status_code=422, detail=f"Invalid tag type '{tag_type}'. Allowed: {', '.join(TAG_TYPES)}")
@@ -96,10 +84,8 @@ async def list_tags(
@router.post("", response_model=TagResponse, status_code=201, summary="Create Tag")
async def create_tag(
body: TagCreateRequest,
- request: Request,
store: TagStore = Depends(get_tag_store),
) -> TagResponse:
- await _require_admin(request)
_validate_tag_type(body.tag_type)
try:
record = await store.create_tag(
@@ -125,10 +111,8 @@ async def create_tag(
async def update_tag(
tag_id: str,
body: TagUpdateRequest,
- request: Request,
store: TagStore = Depends(get_tag_store),
) -> TagResponse:
- await _require_admin(request)
update_data = body.model_dump(exclude_unset=True)
if "name" in update_data and update_data["name"] is not None:
update_data["name"] = update_data["name"].strip()
@@ -149,10 +133,8 @@ async def update_tag(
@router.delete("/{tag_id}", status_code=204, summary="Delete Tag")
async def delete_tag(
tag_id: str,
- request: Request,
store: TagStore = Depends(get_tag_store),
) -> None:
- await _require_admin(request)
try:
deleted = await store.delete_tag(tag_id)
if not deleted:
@@ -195,10 +177,8 @@ async def set_resource_tags(
target_type: str,
target_id: str,
body: TagAssignmentRequest,
- request: Request,
store: TagStore = Depends(get_tag_store),
) -> TagListResponse:
- await _require_admin(request)
_validate_tag_type(target_type)
try:
records = await store.set_target_tags(target_type, target_id, body.tag_ids)
diff --git a/offline-backend-20260512/backend/config.yaml b/offline-backend-20260512/backend/config.yaml
index fae7f00d..8f727d9c 100644
--- a/offline-backend-20260512/backend/config.yaml
+++ b/offline-backend-20260512/backend/config.yaml
@@ -224,7 +224,9 @@ auth_login:
database:
backend: sqlite
sqlite_dir: /app/backend/.deer-flow/data
- pool_size: 10
+ pool_size: 20 # 基础连接池(postgres/mysql 生效;sqlite 忽略)
+ max_overflow: 20 # 高峰期可临时超出 pool_size 的额外连接数,真实上限 = pool_size + max_overflow
+ pool_timeout: 30 # 池满时等待空闲连接的秒数,超时报错
echo_sql: false
checkpointer:
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/database_config.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/database_config.py
index bd78d38d..2ed2100c 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/config/database_config.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/config/database_config.py
@@ -11,13 +11,13 @@ blocking, making a unified file safe for both workloads. Writers
that contend for the lock wait via the default 5-second sqlite3
busy timeout rather than failing immediately.
-Postgres mode: both use the same database URL but maintain independent
-connection pools with different lifecycles.
-
-MySQL mode: application ORM data uses MySQL. LangGraph's checkpointer does
-not have a built-in MySQL backend in this project, so deployments that choose
-MySQL for business tables should configure the legacy ``checkpointer:``
-section separately (usually sqlite) to preserve conversation checkpoints.
+Postgres mode: both use the same database URL but maintain independent
+connection pools with different lifecycles.
+
+MySQL mode: application ORM data uses MySQL. LangGraph's checkpointer does
+not have a built-in MySQL backend in this project, so deployments that choose
+MySQL for business tables should configure the legacy ``checkpointer:``
+section separately (usually sqlite) to preserve conversation checkpoints.
Memory mode: checkpointer uses MemorySaver, app uses in-memory stores.
No database is initialized.
@@ -26,12 +26,12 @@ Sensitive values (postgres_url) should use $VAR syntax in config.yaml
to reference environment variables from .env:
database:
- backend: postgres
- postgres_url: $DATABASE_URL
-
-The $VAR resolution is handled by AppConfig.resolve_env_variables()
-before this config is instantiated -- DatabaseConfig itself does not
-need to do any environment variable processing.
+ backend: postgres
+ postgres_url: $DATABASE_URL
+
+The $VAR resolution is handled by AppConfig.resolve_env_variables()
+before this config is instantiated -- DatabaseConfig itself does not
+need to do any environment variable processing.
"""
from __future__ import annotations
@@ -42,40 +42,54 @@ from typing import Literal
from pydantic import BaseModel, Field
-class DatabaseConfig(BaseModel):
- backend: Literal["memory", "sqlite", "postgres", "mysql"] = Field(
- default="memory",
- description=("Storage backend for application data. 'memory' for development (no persistence across restarts), 'sqlite' for single-node deployment, 'postgres' for production multi-node deployment, 'mysql' for MySQL-backed business tables."),
- )
+class DatabaseConfig(BaseModel):
+ backend: Literal["memory", "sqlite", "postgres", "mysql"] = Field(
+ default="memory",
+ description=("Storage backend for application data. 'memory' for development (no persistence across restarts), 'sqlite' for single-node deployment, 'postgres' for production multi-node deployment, 'mysql' for MySQL-backed business tables."),
+ )
sqlite_dir: str = Field(
default=".deer-flow/data",
description=("Directory for the SQLite database file. Both checkpointer and application data share {sqlite_dir}/deerflow.db."),
)
- postgres_url: str = Field(
- default="",
- description=(
- "PostgreSQL connection URL, shared by checkpointer and app. "
+ postgres_url: str = Field(
+ default="",
+ description=(
+ "PostgreSQL connection URL, shared by checkpointer and app. "
"Use $DATABASE_URL in config.yaml to reference .env. "
"Example: postgresql://user:pass@host:5432/deerflow "
"(the +asyncpg driver suffix is added automatically where needed)."
- ),
- )
- mysql_url: str = Field(
- default="",
- description=(
- "MySQL connection URL for the application ORM. "
- "Use $MYSQL_DATABASE_URL in config.yaml to reference .env. "
- "Example: mysql+asyncmy://user:pass@host:3306/deerflow?charset=utf8mb4. "
- "If mysql:// is provided, the +asyncmy driver suffix is added automatically."
- ),
- )
+ ),
+ )
+ mysql_url: str = Field(
+ default="",
+ description=(
+ "MySQL connection URL for the application ORM. "
+ "Use $MYSQL_DATABASE_URL in config.yaml to reference .env. "
+ "Example: mysql+asyncmy://user:pass@host:3306/deerflow?charset=utf8mb4. "
+ "If mysql:// is provided, the +asyncmy driver suffix is added automatically."
+ ),
+ )
echo_sql: bool = Field(
default=False,
description="Echo all SQL statements to log (debug only).",
)
pool_size: int = Field(
default=5,
- description="Connection pool size for the app ORM engine (postgres/mysql only).",
+ description="Connection pool size for the app ORM engine (postgres/mysql only).",
+ )
+ max_overflow: int = Field(
+ default=10,
+ description=(
+ "Extra connections allowed beyond pool_size under load (postgres/mysql only). "
+ "The real connection ceiling is pool_size + max_overflow. SQLAlchemy default is 10."
+ ),
+ )
+ pool_timeout: int = Field(
+ default=30,
+ description=(
+ "Seconds a caller waits for a free pooled connection before erroring (postgres/mysql only). "
+ "SQLAlchemy default is 30."
+ ),
)
# -- Derived helpers (not user-configured) --
@@ -108,16 +122,16 @@ class DatabaseConfig(BaseModel):
"""SQLAlchemy async URL for the application ORM engine."""
if self.backend == "sqlite":
return f"sqlite+aiosqlite:///{self.sqlite_path}"
- if self.backend == "postgres":
- url = self.postgres_url
- if url.startswith("postgresql://"):
- url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
- return url
- if self.backend == "mysql":
- url = self.mysql_url
- if not url:
- raise ValueError("database.mysql_url is required for backend='mysql'")
- if url.startswith("mysql://"):
- url = url.replace("mysql://", "mysql+asyncmy://", 1)
- return url
- raise ValueError(f"No SQLAlchemy URL for backend={self.backend!r}")
+ if self.backend == "postgres":
+ url = self.postgres_url
+ if url.startswith("postgresql://"):
+ url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
+ return url
+ if self.backend == "mysql":
+ url = self.mysql_url
+ if not url:
+ raise ValueError("database.mysql_url is required for backend='mysql'")
+ if url.startswith("mysql://"):
+ url = url.replace("mysql://", "mysql+asyncmy://", 1)
+ return url
+ raise ValueError(f"No SQLAlchemy URL for backend={self.backend!r}")
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/__init__.py
new file mode 100644
index 00000000..e5dec995
--- /dev/null
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/__init__.py
@@ -0,0 +1,5 @@
+"""Persistence models for admin analytics snapshots."""
+
+from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
+
+__all__ = ["AdminLeaderboardDailyStatRow"]
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/model.py
new file mode 100644
index 00000000..434e02f9
--- /dev/null
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/model.py
@@ -0,0 +1,31 @@
+"""ORM models for admin analytics snapshots."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import Index, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+from deerflow.persistence.types import BeijingDateTime, PortableJSON
+
+
+class AdminLeaderboardDailyStatRow(Base):
+ """One precomputed leaderboard snapshot for a single local calendar day."""
+
+ __tablename__ = "admin_leaderboard_daily_stats"
+ __table_args__ = (
+ Index("ix_admin_leaderboard_daily_stats_status", "status"),
+ Index("ix_admin_leaderboard_daily_stats_updated_at", "updated_at"),
+ )
+
+ stat_date: Mapped[str] = mapped_column(String(10), primary_key=True)
+ settings_hash: Mapped[str] = mapped_column(String(64), primary_key=True)
+ status: Mapped[str] = mapped_column(String(16), default="ready")
+ payload_json: Mapped[dict] = mapped_column(PortableJSON(), default=dict)
+ message: Mapped[str | None] = mapped_column(Text)
+ error: Mapped[str | None] = mapped_column(Text)
+ generated_at: Mapped[datetime | None] = mapped_column(BeijingDateTime())
+ created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
+ updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py
index 384f86af..a65f6136 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py
@@ -11,16 +11,16 @@ None and fall back to in-memory implementations.
from __future__ import annotations
import asyncio
-import json
-import logging
-import os
-import sys
-from importlib.util import find_spec
-from pathlib import Path
-
-from sqlalchemy import text
-from sqlalchemy.engine import make_url
-from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
+import json
+import logging
+import os
+import sys
+from importlib.util import find_spec
+from pathlib import Path
+
+from sqlalchemy import text
+from sqlalchemy.engine import make_url
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
def _json_serializer(obj: object) -> str:
@@ -133,7 +133,7 @@ async def _run_alembic_upgrade(url: str) -> None:
logger.exception("Auto-applying Alembic migrations failed; continuing with create_all schema")
-async def _auto_create_postgres_db(url: str) -> None:
+async def _auto_create_postgres_db(url: str) -> None:
"""Connect to the ``postgres`` maintenance DB and CREATE DATABASE.
The target database name is extracted from *url*. The connection is
@@ -156,55 +156,57 @@ async def _auto_create_postgres_db(url: str) -> None:
async with maint_engine.connect() as conn:
await conn.execute(text(f'CREATE DATABASE "{db_name}"'))
logger.info("Auto-created PostgreSQL database: %s", db_name)
- finally:
- await maint_engine.dispose()
-
-
-async def _auto_create_mysql_db(url: str) -> None:
- """Create the target MySQL database when it is missing.
-
- ``create_all`` can create tables, but MySQL requires the database/schema to
- exist before the application URL can connect. We connect to the built-in
- ``mysql`` maintenance database on the same server and issue a conservative
- ``CREATE DATABASE IF NOT EXISTS``.
- """
-
- parsed = make_url(url)
- db_name = parsed.database
- if not db_name:
- raise ValueError("Cannot auto-create MySQL database: no database name in URL")
-
- maint_url = parsed.set(database="mysql")
- maint_engine = create_async_engine(
- maint_url.render_as_string(hide_password=False),
- isolation_level="AUTOCOMMIT",
- pool_pre_ping=True,
- )
- quoted = "`" + db_name.replace("`", "``") + "`"
- try:
- async with maint_engine.connect() as conn:
- await conn.execute(
- text(
- f"CREATE DATABASE IF NOT EXISTS {quoted} "
- "CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
- )
- )
- logger.info("Auto-created MySQL database if missing: %s", db_name)
- finally:
- await maint_engine.dispose()
-
-
-def _is_mysql_unknown_database_error(exc: Exception) -> bool:
- message = str(exc).lower()
- return "unknown database" in message or "1049" in message
-
-
-async def init_engine(
+ finally:
+ await maint_engine.dispose()
+
+
+async def _auto_create_mysql_db(url: str) -> None:
+ """Create the target MySQL database when it is missing.
+
+ ``create_all`` can create tables, but MySQL requires the database/schema to
+ exist before the application URL can connect. We connect to the built-in
+ ``mysql`` maintenance database on the same server and issue a conservative
+ ``CREATE DATABASE IF NOT EXISTS``.
+ """
+
+ parsed = make_url(url)
+ db_name = parsed.database
+ if not db_name:
+ raise ValueError("Cannot auto-create MySQL database: no database name in URL")
+
+ maint_url = parsed.set(database="mysql")
+ maint_engine = create_async_engine(
+ maint_url.render_as_string(hide_password=False),
+ isolation_level="AUTOCOMMIT",
+ pool_pre_ping=True,
+ )
+ quoted = "`" + db_name.replace("`", "``") + "`"
+ try:
+ async with maint_engine.connect() as conn:
+ await conn.execute(
+ text(
+ f"CREATE DATABASE IF NOT EXISTS {quoted} "
+ "CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
+ )
+ )
+ logger.info("Auto-created MySQL database if missing: %s", db_name)
+ finally:
+ await maint_engine.dispose()
+
+
+def _is_mysql_unknown_database_error(exc: Exception) -> bool:
+ message = str(exc).lower()
+ return "unknown database" in message or "1049" in message
+
+
+async def init_engine(
backend: str,
*,
url: str = "",
echo: bool = False,
pool_size: int = 5,
+ max_overflow: int = 10,
+ pool_timeout: int = 30,
sqlite_dir: str = "",
) -> None:
"""Create the async engine and session factory, then auto-create tables.
@@ -213,7 +215,9 @@ async def init_engine(
backend: "memory", "sqlite", "postgres", or "mysql".
url: SQLAlchemy async URL (for sqlite/postgres/mysql).
echo: Echo SQL to log.
- pool_size: Postgres connection pool size.
+ pool_size: Base connection pool size (postgres/mysql).
+ max_overflow: Extra connections allowed beyond pool_size under load (postgres/mysql).
+ pool_timeout: Seconds to wait for a free pooled connection before erroring (postgres/mysql).
sqlite_dir: Directory to create for SQLite (ensured to exist).
"""
global _engine, _session_factory
@@ -271,6 +275,8 @@ async def init_engine(
url,
echo=echo,
pool_size=pool_size,
+ max_overflow=max_overflow,
+ pool_timeout=pool_timeout,
pool_pre_ping=True,
json_serializer=_json_serializer,
)
@@ -279,6 +285,8 @@ async def init_engine(
url,
echo=echo,
pool_size=pool_size,
+ max_overflow=max_overflow,
+ pool_timeout=pool_timeout,
pool_pre_ping=True,
pool_recycle=1800,
json_serializer=_json_serializer,
@@ -302,37 +310,39 @@ async def init_engine(
# This is expected during initial scaffolding or minimal installs.
logger.debug("deerflow.persistence.models not found; skipping auto-create tables")
- try:
- async with _engine.begin() as conn:
- await conn.run_sync(Base.metadata.create_all)
- except Exception as exc:
- if backend == "postgres" and "does not exist" in str(exc):
+ try:
+ async with _engine.begin() as conn:
+ await conn.run_sync(Base.metadata.create_all)
+ except Exception as exc:
+ if backend == "postgres" and "does not exist" in str(exc):
# Database not yet created — attempt to auto-create it, then retry.
await _auto_create_postgres_db(url)
# Rebuild engine against the now-existing database
await _engine.dispose()
- _engine = create_async_engine(url, echo=echo, pool_size=pool_size, pool_pre_ping=True, json_serializer=_json_serializer)
- _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
- async with _engine.begin() as conn:
- await conn.run_sync(Base.metadata.create_all)
- elif backend == "mysql" and _is_mysql_unknown_database_error(exc):
- # Database not yet created. Create it, rebuild the app engine
- # against the target DB, then let SQLAlchemy create all tables.
- await _auto_create_mysql_db(url)
- await _engine.dispose()
- _engine = create_async_engine(
- url,
- echo=echo,
- pool_size=pool_size,
- pool_pre_ping=True,
- pool_recycle=1800,
- json_serializer=_json_serializer,
- )
- _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
- async with _engine.begin() as conn:
- await conn.run_sync(Base.metadata.create_all)
- else:
- raise
+ _engine = create_async_engine(url, echo=echo, pool_size=pool_size, max_overflow=max_overflow, pool_timeout=pool_timeout, pool_pre_ping=True, json_serializer=_json_serializer)
+ _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
+ async with _engine.begin() as conn:
+ await conn.run_sync(Base.metadata.create_all)
+ elif backend == "mysql" and _is_mysql_unknown_database_error(exc):
+ # Database not yet created. Create it, rebuild the app engine
+ # against the target DB, then let SQLAlchemy create all tables.
+ await _auto_create_mysql_db(url)
+ await _engine.dispose()
+ _engine = create_async_engine(
+ url,
+ echo=echo,
+ pool_size=pool_size,
+ max_overflow=max_overflow,
+ pool_timeout=pool_timeout,
+ pool_pre_ping=True,
+ pool_recycle=1800,
+ json_serializer=_json_serializer,
+ )
+ _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
+ async with _engine.begin() as conn:
+ await conn.run_sync(Base.metadata.create_all)
+ else:
+ raise
# Apply in-place schema fixes (column widenings, etc.) that ``create_all``
# cannot do on its own. Failures don't abort startup — see the helper.
@@ -351,6 +361,8 @@ async def init_engine_from_config(config) -> None:
url=config.app_sqlalchemy_url,
echo=config.echo_sql,
pool_size=config.pool_size,
+ max_overflow=getattr(config, "max_overflow", 10),
+ pool_timeout=getattr(config, "pool_timeout", 30),
sqlite_dir=config.sqlite_dir if config.backend == "sqlite" else "",
)
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/llm_metrics/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/llm_metrics/model.py
index 9d16e1ba..e6882fa9 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/llm_metrics/model.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/llm_metrics/model.py
@@ -4,7 +4,7 @@ from __future__ import annotations
from datetime import UTC, datetime
-from sqlalchemy import Float, Integer, String, Text
+from sqlalchemy import Float, Index, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
@@ -15,6 +15,10 @@ class LlmCallMetricRow(Base):
"""One row per completed (or failed) LLM provider invocation."""
__tablename__ = "llm_call_metrics"
+ __table_args__ = (
+ Index("ix_llm_call_metrics_created_user", "created_at", "user_id"),
+ Index("ix_llm_call_metrics_run_created", "run_id", "created_at"),
+ )
id: Mapped[str] = mapped_column(String(64), primary_key=True)
# Wall-clock time the LLM call ended. Indexed because admin views always
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260608_01_admin_leaderboard_daily_stats.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260608_01_admin_leaderboard_daily_stats.py
new file mode 100644
index 00000000..50a14e64
--- /dev/null
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260608_01_admin_leaderboard_daily_stats.py
@@ -0,0 +1,85 @@
+"""Create admin leaderboard daily snapshots and analytics indexes.
+
+Revision ID: 20260608_01
+Revises: 20260607_02
+Create Date: 2026-06-08
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+
+from deerflow.persistence.types import PortableJSON
+
+
+revision = "20260608_01"
+down_revision = "20260607_02"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+ inspector = inspect(op.get_bind())
+ return table_name in inspector.get_table_names()
+
+
+def _has_index(table_name: str, index_name: str) -> bool:
+ if not _has_table(table_name):
+ return False
+ inspector = inspect(op.get_bind())
+ return any(idx["name"] == index_name for idx in inspector.get_indexes(table_name))
+
+
+def _create_index(table_name: str, index_name: str, columns: list[str]) -> None:
+ if _has_table(table_name) and not _has_index(table_name, index_name):
+ op.create_index(index_name, table_name, columns)
+
+
+def _drop_index(table_name: str, index_name: str) -> None:
+ if _has_table(table_name) and _has_index(table_name, index_name):
+ op.drop_index(index_name, table_name=table_name)
+
+
+def upgrade() -> None:
+ if not _has_table("admin_leaderboard_daily_stats"):
+ op.create_table(
+ "admin_leaderboard_daily_stats",
+ sa.Column("stat_date", sa.String(10), nullable=False),
+ sa.Column("settings_hash", sa.String(64), nullable=False),
+ sa.Column("status", sa.String(16), nullable=False, server_default="ready"),
+ sa.Column("payload_json", PortableJSON(), nullable=True),
+ sa.Column("message", sa.Text(), nullable=True),
+ sa.Column("error", sa.Text(), nullable=True),
+ sa.Column("generated_at", sa.DateTime(timezone=True), nullable=True),
+ sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+ sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+ sa.PrimaryKeyConstraint("stat_date", "settings_hash", name="pk_admin_leaderboard_daily_stats"),
+ )
+ op.create_index("ix_admin_leaderboard_daily_stats_status", "admin_leaderboard_daily_stats", ["status"])
+ op.create_index("ix_admin_leaderboard_daily_stats_updated_at", "admin_leaderboard_daily_stats", ["updated_at"])
+
+ _create_index("runs", "ix_runs_created_user_status", ["created_at", "user_id", "status"])
+ _create_index("runs", "ix_runs_user_created", ["user_id", "created_at"])
+ _create_index("runs", "ix_runs_thread_created", ["thread_id", "created_at"])
+ _create_index("tool_call_metrics", "ix_tool_call_metrics_created_user", ["created_at", "user_id"])
+ _create_index("tool_call_metrics", "ix_tool_call_metrics_run_created", ["run_id", "created_at"])
+ _create_index("llm_call_metrics", "ix_llm_call_metrics_created_user", ["created_at", "user_id"])
+ _create_index("llm_call_metrics", "ix_llm_call_metrics_run_created", ["run_id", "created_at"])
+ _create_index("threads_meta", "ix_threads_meta_created_user", ["created_at", "user_id"])
+ _create_index("threads_meta", "ix_threads_meta_user_created", ["user_id", "created_at"])
+
+
+def downgrade() -> None:
+ _drop_index("threads_meta", "ix_threads_meta_user_created")
+ _drop_index("threads_meta", "ix_threads_meta_created_user")
+ _drop_index("llm_call_metrics", "ix_llm_call_metrics_run_created")
+ _drop_index("llm_call_metrics", "ix_llm_call_metrics_created_user")
+ _drop_index("tool_call_metrics", "ix_tool_call_metrics_run_created")
+ _drop_index("tool_call_metrics", "ix_tool_call_metrics_created_user")
+ _drop_index("runs", "ix_runs_thread_created")
+ _drop_index("runs", "ix_runs_user_created")
+ _drop_index("runs", "ix_runs_created_user_status")
+ if _has_table("admin_leaderboard_daily_stats"):
+ op.drop_table("admin_leaderboard_daily_stats")
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py
index 4ac8baa7..45122546 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py
@@ -21,6 +21,7 @@ from deerflow.persistence.agents.model import (
AgentRow,
AgentSkillRow,
)
+from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
from deerflow.persistence.memory_config.model import UserMemoryConfigRow
from deerflow.persistence.feedback.model import FeedbackRow
from deerflow.persistence.llm_metrics.model import LlmCallMetricRow
@@ -58,6 +59,7 @@ __all__ = [
"AgentFavoriteRow",
"AgentRow",
"AgentSkillRow",
+ "AdminLeaderboardDailyStatRow",
"LlmCallMetricRow",
"NotificationRow",
"RecommendedQuestionRow",
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/run/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/run/model.py
index 175e7ed7..8f75f10a 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/run/model.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/run/model.py
@@ -4,7 +4,7 @@ from __future__ import annotations
from datetime import UTC, datetime
-from sqlalchemy import Index, String, Text
+from sqlalchemy import Index, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
@@ -47,4 +47,9 @@ class RunRow(Base):
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
- __table_args__ = (Index("ix_runs_thread_status", "thread_id", "status"),)
+ __table_args__ = (
+ Index("ix_runs_thread_status", "thread_id", "status"),
+ Index("ix_runs_created_user_status", "created_at", "user_id", "status"),
+ Index("ix_runs_user_created", "user_id", "created_at"),
+ Index("ix_runs_thread_created", "thread_id", "created_at"),
+ )
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/model.py
index 6b0331cb..594b38f6 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/model.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/model.py
@@ -4,15 +4,19 @@ from __future__ import annotations
from datetime import UTC, datetime
-from sqlalchemy import String
+from sqlalchemy import Index, String
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime, PortableJSON
-class ThreadMetaRow(Base):
- __tablename__ = "threads_meta"
+class ThreadMetaRow(Base):
+ __tablename__ = "threads_meta"
+ __table_args__ = (
+ Index("ix_threads_meta_created_user", "created_at", "user_id"),
+ Index("ix_threads_meta_user_created", "user_id", "created_at"),
+ )
thread_id: Mapped[str] = mapped_column(String(64), primary_key=True)
assistant_id: Mapped[str | None] = mapped_column(String(128), index=True)
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/tool_metrics/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/tool_metrics/model.py
index bd70c038..b55d407f 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/tool_metrics/model.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/tool_metrics/model.py
@@ -4,7 +4,7 @@ from __future__ import annotations
from datetime import UTC, datetime
-from sqlalchemy import Integer, String, Text
+from sqlalchemy import Index, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
@@ -21,6 +21,10 @@ class ToolCallMetricRow(Base):
"""
__tablename__ = "tool_call_metrics"
+ __table_args__ = (
+ Index("ix_tool_call_metrics_created_user", "created_at", "user_id"),
+ Index("ix_tool_call_metrics_run_created", "run_id", "created_at"),
+ )
id: Mapped[str] = mapped_column(String(64), primary_key=True)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), index=True)
diff --git a/offline-backend-20260512/backend/scripts/sql/leaderboard_indexes_mysql.sql b/offline-backend-20260512/backend/scripts/sql/leaderboard_indexes_mysql.sql
new file mode 100644
index 00000000..8c4151c3
--- /dev/null
+++ b/offline-backend-20260512/backend/scripts/sql/leaderboard_indexes_mysql.sql
@@ -0,0 +1,85 @@
+-- ============================================================================
+-- 排行榜(admin leaderboard)统计所需的复合索引 —— MySQL 手动执行版
+-- ============================================================================
+--
+-- 背景:
+-- 这些索引声明在各 ORM model 的 __table_args__ 里,但 Base.metadata.create_all
+-- 对【已存在的表】不会补建索引,而 alembic 自动迁移在本项目里是关闭的
+-- (engine.py 中 `# await _run_alembic_upgrade(url)` 被注释)。
+-- 因此在【已有数据的 MySQL 部署】上,这些索引很可能并未真正创建,排行榜的
+-- 单日聚合查询会退化为全表扫描,拖慢并占用连接。
+--
+-- 用法:
+-- 1) 先选中目标库(与 database.mysql_url 里的库名一致):
+-- USE your_deerflow_db;
+-- 2) 直接整文件执行即可。本脚本是【幂等】的:已存在的索引会跳过,不会报错,
+-- 可重复执行。
+--
+-- 注意:
+-- * 大表上建索引可能耗时;MySQL 8.0 在线 DDL 默认 ALGORITHM=INPLACE 一般不阻塞
+-- 写入,但仍建议在低峰期执行,或先在备库/预发验证。
+-- * 涉及列均为 String(64) / String(20) 等短列,utf8mb4 下复合索引长度远低于
+-- 3072 字节上限,无需前缀索引。
+-- * 索引名与 migration 20260608_01 保持一致,日后若启用 alembic 也不会重复创建。
+-- ============================================================================
+
+-- ---- 通用幂等封装:仅当索引不存在时才创建 --------------------------------------
+DELIMITER //
+DROP PROCEDURE IF EXISTS _create_index_if_missing //
+CREATE PROCEDURE _create_index_if_missing(
+ IN p_table VARCHAR(64),
+ IN p_index VARCHAR(64),
+ IN p_columns VARCHAR(255)
+)
+BEGIN
+ DECLARE v_exists INT DEFAULT 0;
+ SELECT COUNT(*) INTO v_exists
+ FROM information_schema.statistics
+ WHERE table_schema = DATABASE()
+ AND table_name = p_table
+ AND index_name = p_index;
+
+ IF v_exists = 0 THEN
+ SET @ddl = CONCAT('CREATE INDEX `', p_index, '` ON `', p_table, '` (', p_columns, ')');
+ PREPARE stmt FROM @ddl;
+ EXECUTE stmt;
+ DEALLOCATE PREPARE stmt;
+ SELECT CONCAT('CREATED ', p_table, '.', p_index) AS result;
+ ELSE
+ SELECT CONCAT('SKIPPED ', p_table, '.', p_index, ' (already exists)') AS result;
+ END IF;
+END //
+DELIMITER ;
+
+-- ---- runs ---------------------------------------------------------------------
+CALL _create_index_if_missing('runs', 'ix_runs_created_user_status', '`created_at`, `user_id`, `status`');
+CALL _create_index_if_missing('runs', 'ix_runs_user_created', '`user_id`, `created_at`');
+CALL _create_index_if_missing('runs', 'ix_runs_thread_created', '`thread_id`, `created_at`');
+
+-- ---- tool_call_metrics -------------------------------------------------------
+CALL _create_index_if_missing('tool_call_metrics', 'ix_tool_call_metrics_created_user', '`created_at`, `user_id`');
+CALL _create_index_if_missing('tool_call_metrics', 'ix_tool_call_metrics_run_created', '`run_id`, `created_at`');
+
+-- ---- llm_call_metrics --------------------------------------------------------
+CALL _create_index_if_missing('llm_call_metrics', 'ix_llm_call_metrics_created_user', '`created_at`, `user_id`');
+CALL _create_index_if_missing('llm_call_metrics', 'ix_llm_call_metrics_run_created', '`run_id`, `created_at`');
+
+-- ---- threads_meta ------------------------------------------------------------
+CALL _create_index_if_missing('threads_meta', 'ix_threads_meta_created_user', '`created_at`, `user_id`');
+CALL _create_index_if_missing('threads_meta', 'ix_threads_meta_user_created', '`user_id`, `created_at`');
+
+-- ---- 清理临时存储过程 --------------------------------------------------------
+DROP PROCEDURE IF EXISTS _create_index_if_missing;
+
+-- ---- 执行后自检:确认 9 个索引都已存在 ---------------------------------------
+SELECT table_name, index_name
+ FROM information_schema.statistics
+ WHERE table_schema = DATABASE()
+ AND index_name IN (
+ 'ix_runs_created_user_status', 'ix_runs_user_created', 'ix_runs_thread_created',
+ 'ix_tool_call_metrics_created_user', 'ix_tool_call_metrics_run_created',
+ 'ix_llm_call_metrics_created_user', 'ix_llm_call_metrics_run_created',
+ 'ix_threads_meta_created_user', 'ix_threads_meta_user_created'
+ )
+ GROUP BY table_name, index_name
+ ORDER BY table_name, index_name;