+ {!isEditing && activeTabMeta && (
+ <>
+
+
+
+ )}
{isFileTab && fileIndex != null && (
);
}
+
+export function PublicScheduledTaskFullscreenPage() {
+ const { taskId, runId } = useParams<{ taskId: string; runId: string }>();
+ const [searchParams] = useSearchParams();
+ const requestedTab = searchParams.get("tab");
+ const tabKind: ContentTab["kind"] =
+ requestedTab === "body" || requestedTab === "file" || requestedTab === "html"
+ ? requestedTab
+ : "html";
+ const fileIndexValue = Number(searchParams.get("file"));
+ const fileIndexValid = Number.isInteger(fileIndexValue) && fileIndexValue >= 0;
+
+ const { data: run, isLoading, isError, error } = useQuery({
+ queryKey: ["public-scheduled-task", "run", runId],
+ queryFn: () => fetchPublicRun(runId!),
+ enabled: !!runId,
+ refetchOnWindowFocus: false,
+ retry: false,
+ });
+
+ const isHtmlRun =
+ run?.result?.format === "html" ||
+ (run?.result?.files ?? []).some(
+ (file) => file.kind === "html" || (file.name ?? "").toLowerCase().endsWith(".html"),
+ );
+
+ const { data: htmlData, isLoading: htmlLoading } = useQuery({
+ queryKey: ["public-scheduled-task", "html", runId],
+ queryFn: () => fetchPublicRunHtml(runId!),
+ enabled: !!runId && tabKind === "html" && isHtmlRun && !run?.result?.html,
+ refetchOnWindowFocus: false,
+ retry: false,
+ });
+
+ const {
+ data: fileData,
+ isLoading: fileLoading,
+ isError: fileIsError,
+ error: fileError,
+ } = useQuery({
+ queryKey: ["public-scheduled-task", "file", runId, fileIndexValue],
+ queryFn: () => fetchPublicRunFile(runId!, fileIndexValue),
+ enabled: !!runId && tabKind === "file" && fileIndexValid,
+ refetchOnWindowFocus: false,
+ retry: false,
+ });
+
+ const htmlContent = run?.result?.html ?? htmlData?.html ?? null;
+ const backHref = taskId ? `#/public/scheduled-tasks/${encodeURIComponent(taskId)}` : "#/public/scheduled-tasks";
+ const pageTitle =
+ tabKind === "file"
+ ? fileData?.name || "文件预览"
+ : tabKind === "body"
+ ? run?.result?.title || "正文预览"
+ : htmlData?.title || run?.result?.title || "HTML 预览";
+
+ return (
+
+
+
+
+
+ {pageTitle}
+
+ {tabKind === "file" && runId && fileIndexValid && (
+
+ )}
+
+
+
+ {!runId || !taskId ? (
+
+ 缺少任务或执行记录 ID。
+
+ ) : isLoading ? (
+
+
+ 正在加载预览...
+
+ ) : isError ? (
+
+ 加载失败：{error instanceof Error ? error.message : "未知错误"}
+
+ ) : tabKind === "html" ? (
+ htmlContent ? (
+
+ ) : (
+
+ {htmlLoading && }
+ HTML 页面尚未生成或正在生成中。
+
+ )
+ ) : tabKind === "file" ? (
+ fileLoading ? (
+
+
+ 正在加载文件...
+
+ ) : !fileIndexValid || fileIsError || !fileData ? (
+
+ 文件加载失败：{fileError instanceof Error ? fileError.message : "文件不存在或不可预览"}
+
+ ) : (
+
+
+
+ )
+ ) : (
+
+
+
+ )}
+
+
+ );
+}
diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py
index 80f1c2ce..8da8c6c3 100644
--- a/offline-backend-20260512/backend/app/gateway/app.py
+++ b/offline-backend-20260512/backend/app/gateway/app.py
@@ -2,10 +2,12 @@ import asyncio
import logging
import os
import queue
-import sys
-import threading
-from collections.abc import AsyncGenerator
-from contextlib import asynccontextmanager
+import sys
+import threading
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+from datetime import datetime
+from time import perf_counter
from dotenv import load_dotenv
from fastapi import FastAPI
@@ -468,6 +470,84 @@ _LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS = _env_float(
)
+def _leaderboard_scheduler_sql_for_log(stmt) -> str:
+ try:
+ return str(stmt.compile(compile_kwargs={"literal_binds": True}))
+ except Exception:
+ return str(stmt)
+
+
+async def _execute_leaderboard_scheduler_query(session, step: str, stmt, **context):
+ sql = _leaderboard_scheduler_sql_for_log(stmt)
+ logger.info(
+ "Leaderboard snapshot DB query start: step=%s stat_date=%s job=%s settings_hash=%s started_at=%s sql=%s",
+ step,
+ context.get("stat_date") or context.get("stat_dates") or "",
+ context.get("job_key") or "",
+ context.get("settings_hash") or "",
+ datetime.now().isoformat(),
+ sql,
+ )
+ start = perf_counter()
+ try:
+ result = await session.execute(stmt)
+ except Exception:
+ logger.exception(
+ "Leaderboard snapshot DB query failed: step=%s stat_date=%s job=%s settings_hash=%s elapsed_ms=%.2f",
+ step,
+ context.get("stat_date") or context.get("stat_dates") or "",
+ context.get("job_key") or "",
+ context.get("settings_hash") or "",
+ (perf_counter() - start) * 1000,
+ )
+ raise
+ logger.info(
+ "Leaderboard snapshot DB query done: step=%s stat_date=%s job=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f",
+ step,
+ context.get("stat_date") or context.get("stat_dates") or "",
+ context.get("job_key") or "",
+ context.get("settings_hash") or "",
+ datetime.now().isoformat(),
+ (perf_counter() - start) * 1000,
+ )
+ return result
+
+
+async def _time_leaderboard_scheduler_db_operation(step: str, operation, *, sql: str = "", **context):
+ logger.info(
+ "Leaderboard snapshot DB query start: step=%s stat_date=%s job=%s settings_hash=%s started_at=%s sql=%s",
+ step,
+ context.get("stat_date") or context.get("stat_dates") or "",
+ context.get("job_key") or "",
+ context.get("settings_hash") or "",
+ datetime.now().isoformat(),
+ sql,
+ )
+ start = perf_counter()
+ try:
+ result = await operation()
+ except Exception:
+ logger.exception(
+ "Leaderboard snapshot DB query failed: step=%s stat_date=%s job=%s settings_hash=%s elapsed_ms=%.2f",
+ step,
+ context.get("stat_date") or context.get("stat_dates") or "",
+ context.get("job_key") or "",
+ context.get("settings_hash") or "",
+ (perf_counter() - start) * 1000,
+ )
+ raise
+ logger.info(
+ "Leaderboard snapshot DB query done: step=%s stat_date=%s job=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f",
+ step,
+ context.get("stat_date") or context.get("stat_dates") or "",
+ context.get("job_key") or "",
+ context.get("settings_hash") or "",
+ datetime.now().isoformat(),
+ (perf_counter() - start) * 1000,
+ )
+ return result
+
+
def _recent_leaderboard_stat_dates(days: int) -> list[str]:
from datetime import datetime, timedelta
@@ -493,21 +573,27 @@ async def _leaderboard_snapshot_due_dates(
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,
- ),
+ stmt = (
+ 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)
+ ),
+ )
+ .order_by(AdminLeaderboardDailyStatRow.stat_date.desc())
+ .limit(limit)
+ )
+ rows = (
+ await _execute_leaderboard_scheduler_query(
+ session,
+ "scheduler.due_dates",
+ stmt,
+ settings_hash=settings_hash,
)
).all()
return [str(row[0]) for row in rows]
@@ -527,9 +613,15 @@ async def _claim_leaderboard_snapshot(
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},
+ row = await _time_leaderboard_scheduler_db_operation(
+ "scheduler.claim.get",
+ lambda: session.get(
+ AdminLeaderboardDailyStatRow,
+ {"stat_date": stat_date, "settings_hash": settings_hash},
+ ),
+ sql="SELECT admin_leaderboard_daily_stats by primary key",
+ stat_date=stat_date,
+ settings_hash=settings_hash,
)
if row is None or _snapshot_is_fresh(row, stat_date):
return False
@@ -545,7 +637,13 @@ async def _claim_leaderboard_snapshot(
row.message = "排行榜后台统计任务已认领。"
row.error = None
row.updated_at = datetime.now(UTC)
- await session.commit()
+ await _time_leaderboard_scheduler_db_operation(
+ "scheduler.claim.commit",
+ lambda: session.commit(),
+ sql="COMMIT",
+ stat_date=stat_date,
+ settings_hash=settings_hash,
+ )
return True
@@ -1152,10 +1250,11 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
# Admin LLM metrics audit + export
app.include_router(llm_metrics.router)
- # Admin window into other users' threads / messages / memory
- app.include_router(admin_users.router)
-
- # Admin tool / skill call metrics audit + export
+ # Admin window into other users' threads / messages / memory
+ app.include_router(admin_users.router)
+ app.include_router(admin_users.self_router)
+
+ # Admin tool / skill call metrics audit + export
app.include_router(tool_metrics.router)
# Channels API is mounted at /api/channels
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 753471ea..0865fcdc 100644
--- a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py
+++ b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py
@@ -19,8 +19,11 @@ import logging
import os
import re
from collections.abc import AsyncIterator
+from contextvars import ContextVar
from difflib import SequenceMatcher
from hashlib import sha256
+from time import perf_counter
+from typing import Any
from datetime import UTC, datetime, timedelta, time as datetime_time
@@ -56,6 +59,7 @@ from deerflow.persistence.user.model import UserRow
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/admin/users", tags=["admin-users"])
+self_router = APIRouter(prefix="/api/users/me", tags=["user-qa"])
_LEADERBOARD_MAX_CONCURRENT = max(
1,
@@ -99,6 +103,119 @@ _LEADERBOARD_QUEUE_ON_REQUEST_ENABLED = os.getenv(
"off",
}
_leaderboard_semaphore = asyncio.Semaphore(_LEADERBOARD_MAX_CONCURRENT)
+_LEADERBOARD_TRACE_CONTEXT: ContextVar[dict[str, object] | None] = ContextVar(
+ "leaderboard_trace_context",
+ default=None,
+)
+
+
+def _leaderboard_log_context(**context: object) -> dict[str, object] | None:
+ current = _LEADERBOARD_TRACE_CONTEXT.get()
+ merged: dict[str, object] = dict(current or {})
+ for key, value in context.items():
+ if value is None or value == "":
+ continue
+ merged[key] = value
+ return merged or None
+
+
+def _leaderboard_sql_for_log(stmt) -> str:
+ try:
+ return str(stmt.compile(compile_kwargs={"literal_binds": True}))
+ except Exception:
+ return str(stmt)
+
+
+async def _execute_leaderboard_query(session, step: str, stmt, **context: object):
+ log_context = _leaderboard_log_context(**context)
+ if not log_context:
+ return await session.execute(stmt)
+
+ started_at = datetime.now(BEIJING_TZ).isoformat()
+ sql = _leaderboard_sql_for_log(stmt)
+ logger.info(
+ "Leaderboard snapshot DB query start: step=%s stat_date=%s job=%s settings_hash=%s started_at=%s sql=%s",
+ step,
+ log_context.get("stat_date") or log_context.get("stat_dates") or "",
+ log_context.get("job_key") or "",
+ log_context.get("settings_hash") or "",
+ started_at,
+ sql,
+ )
+ start = perf_counter()
+ try:
+ result = await session.execute(stmt)
+ except Exception:
+ elapsed_ms = (perf_counter() - start) * 1000
+ logger.exception(
+ "Leaderboard snapshot DB query failed: step=%s stat_date=%s job=%s settings_hash=%s elapsed_ms=%.2f",
+ step,
+ log_context.get("stat_date") or log_context.get("stat_dates") or "",
+ log_context.get("job_key") or "",
+ log_context.get("settings_hash") or "",
+ elapsed_ms,
+ )
+ raise
+
+ elapsed_ms = (perf_counter() - start) * 1000
+ logger.info(
+ "Leaderboard snapshot DB query done: step=%s stat_date=%s job=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f",
+ step,
+ log_context.get("stat_date") or log_context.get("stat_dates") or "",
+ log_context.get("job_key") or "",
+ log_context.get("settings_hash") or "",
+ datetime.now(BEIJING_TZ).isoformat(),
+ elapsed_ms,
+ )
+ return result
+
+
+async def _time_leaderboard_db_operation(
+ step: str,
+ operation,
+ *,
+ sql: str = "",
+ **context: object,
+):
+ log_context = _leaderboard_log_context(**context)
+ if not log_context:
+ return await operation()
+
+ logger.info(
+ "Leaderboard snapshot DB query start: step=%s stat_date=%s job=%s settings_hash=%s started_at=%s sql=%s",
+ step,
+ log_context.get("stat_date") or log_context.get("stat_dates") or "",
+ log_context.get("job_key") or "",
+ log_context.get("settings_hash") or "",
+ datetime.now(BEIJING_TZ).isoformat(),
+ sql,
+ )
+ start = perf_counter()
+ try:
+ result = await operation()
+ except Exception:
+ elapsed_ms = (perf_counter() - start) * 1000
+ logger.exception(
+ "Leaderboard snapshot DB query failed: step=%s stat_date=%s job=%s settings_hash=%s elapsed_ms=%.2f",
+ step,
+ log_context.get("stat_date") or log_context.get("stat_dates") or "",
+ log_context.get("job_key") or "",
+ log_context.get("settings_hash") or "",
+ elapsed_ms,
+ )
+ raise
+
+ elapsed_ms = (perf_counter() - start) * 1000
+ logger.info(
+ "Leaderboard snapshot DB query done: step=%s stat_date=%s job=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f",
+ step,
+ log_context.get("stat_date") or log_context.get("stat_dates") or "",
+ log_context.get("job_key") or "",
+ log_context.get("settings_hash") or "",
+ datetime.now(BEIJING_TZ).isoformat(),
+ elapsed_ms,
+ )
+ return result
async def _require_admin(request: Request) -> None:
@@ -285,6 +402,13 @@ def _parse_datetime(value: str | None, fallback: datetime) -> datetime:
raise HTTPException(status_code=400, detail=f"Invalid datetime: {value}") from exc
+def _parse_optional_datetime(value: str | None) -> datetime | None:
+ if not value:
+ return None
+ parsed = _parse_datetime(value, _now())
+ return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=BEIJING_TZ)
+
+
def _resolve_range(days: int, since: str | None, until: str | None) -> tuple[datetime, datetime, int]:
end = _parse_datetime(until, _now())
start = _parse_datetime(since, end - timedelta(days=max(1, min(days, 365))))
@@ -325,7 +449,7 @@ def _bounds_for_stat_date(stat_date: str) -> tuple[datetime, datetime]:
async def _try_set_session_value(session, sql: str) -> None:
try:
- await session.execute(text(sql))
+ await _execute_leaderboard_query(session, "session_setting", text(sql))
except Exception:
logger.debug("Failed to apply analytics session setting: %s", sql, exc_info=True)
try:
@@ -665,7 +789,7 @@ async def _query_user_leaderboard(
desc(func.coalesce(run_agg.c.run_count, 0) + func.coalesce(tool_agg.c.tool_call_count, 0)),
desc(func.coalesce(run_agg.c.total_tokens, 0)),
)
- rows = (await session.execute(stmt)).all()
+ rows = (await _execute_leaderboard_query(session, "users", stmt)).all()
return [
AdminLeaderboardUser(
user_id=r[0],
@@ -725,7 +849,7 @@ async def _query_call_rankings(
include_admins=include_admins,
include_scheduled=include_scheduled,
)
- rows = (await session.execute(stmt)).all()
+ rows = (await _execute_leaderboard_query(session, f"{field}_rankings", stmt)).all()
return [
AdminLeaderboardSkill(
name=r[0] or "",
@@ -791,7 +915,7 @@ async def _query_run_dimensions(
include_failed=include_failed,
include_scheduled=include_scheduled,
)
- rows = (await session.execute(stmt)).all()
+ rows = (await _execute_leaderboard_query(session, f"{field}_dimensions", stmt)).all()
return [
AdminLeaderboardDimension(
name=r[0] or "",
@@ -848,7 +972,7 @@ async def _query_model_rankings(
stmt = stmt.where(LlmCallMetricRow.status != "error")
if not include_scheduled:
stmt = stmt.where(or_(LlmCallMetricRow.run_id.is_(None), LlmCallMetricRow.run_id.not_in(_scheduled_run_ids_stmt())))
- rows = (await session.execute(stmt)).all()
+ rows = (await _execute_leaderboard_query(session, "model_rankings", stmt)).all()
return [
AdminLeaderboardDimension(
name=r[0] or "",
@@ -895,7 +1019,7 @@ async def _query_recent_questions(
include_failed=include_failed,
include_scheduled=include_scheduled,
)
- rows = (await session.execute(stmt)).all()
+ rows = (await _execute_leaderboard_query(session, "recent_questions", stmt)).all()
questions: list[AdminLeaderboardQuestion] = []
for r in rows:
text = _clean_question_text(r[0]) if cleanup_system_questions else str(r[0] or "").strip()
@@ -947,7 +1071,7 @@ async def _query_frequent_questions(
include_failed=include_failed,
include_scheduled=include_scheduled,
)
- rows = (await session.execute(stmt)).all()
+ rows = (await _execute_leaderboard_query(session, "frequent_questions", stmt)).all()
groups: list[dict] = []
for r in rows:
text = _clean_question_text(r[0]) if cleanup_system_questions else str(r[0] or "").strip()
@@ -1022,7 +1146,7 @@ async def _query_trends(
include_failed=include_failed,
include_scheduled=include_scheduled,
)
- rows = (await session.execute(stmt)).all()
+ rows = (await _execute_leaderboard_query(session, "trends", stmt)).all()
return [
AdminLeaderboardTrendPoint(
date=str(r[0]),
@@ -1071,8 +1195,8 @@ async def _query_overview(
total_users_stmt = select(func.count(UserRow.id))
total_users_stmt = _apply_user_scope(total_users_stmt, UserRow.id, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
- total_users = 1 if user_id else (await session.execute(total_users_stmt)).scalar_one()
- run_row = (await session.execute(
+ total_users = 1 if user_id else (await _execute_leaderboard_query(session, "overview.total_users", total_users_stmt)).scalar_one()
+ run_summary_stmt = (
select(
func.count(RunRow.run_id),
_question_count_expr(),
@@ -1080,16 +1204,19 @@ async def _query_overview(
func.count(func.distinct(RunRow.user_id)),
func.sum(case((RunRow.status == "error", 1), else_=0)),
).where(*run_where)
- )).one()
- thread_count = (await session.execute(select(func.count(ThreadMetaRow.thread_id)).where(*thread_where))).scalar_one()
- tool_row = (await session.execute(
+ )
+ run_row = (await _execute_leaderboard_query(session, "overview.run_summary", run_summary_stmt)).one()
+ thread_count_stmt = select(func.count(ThreadMetaRow.thread_id)).where(*thread_where)
+ thread_count = (await _execute_leaderboard_query(session, "overview.thread_count", thread_count_stmt)).scalar_one()
+ tool_summary_stmt = (
select(
func.count(ToolCallMetricRow.id),
func.sum(case((func.nullif(ToolCallMetricRow.skill_name, "").is_not(None), 1), else_=0)),
)
.outerjoin(RunRow, RunRow.run_id == ToolCallMetricRow.run_id)
.where(*tool_where)
- )).one()
+ )
+ tool_row = (await _execute_leaderboard_query(session, "overview.tool_summary", tool_summary_stmt)).one()
return AdminLeaderboardOverview(
total_users=int(total_users or 0),
@@ -1579,13 +1706,12 @@ async def _load_daily_snapshot_rows(
if not stat_dates:
return {}
async with session_factory() as session:
+ stmt = select(AdminLeaderboardDailyStatRow).where(
+ AdminLeaderboardDailyStatRow.stat_date.in_(stat_dates),
+ AdminLeaderboardDailyStatRow.settings_hash == settings_hash,
+ )
rows = (
- await session.execute(
- select(AdminLeaderboardDailyStatRow).where(
- AdminLeaderboardDailyStatRow.stat_date.in_(stat_dates),
- AdminLeaderboardDailyStatRow.settings_hash == settings_hash,
- )
- )
+ await _execute_leaderboard_query(session, "snapshot_rows.load", stmt)
).scalars().all()
return {row.stat_date: row for row in rows}
@@ -1603,9 +1729,15 @@ async def _store_daily_snapshot(
) -> None:
async with session_factory() as session:
now = datetime.now(UTC)
- row = await session.get(
- AdminLeaderboardDailyStatRow,
- {"stat_date": stat_date, "settings_hash": settings_hash},
+ row = await _time_leaderboard_db_operation(
+ "snapshot.store.get",
+ lambda: session.get(
+ AdminLeaderboardDailyStatRow,
+ {"stat_date": stat_date, "settings_hash": settings_hash},
+ ),
+ sql="SELECT admin_leaderboard_daily_stats by primary key",
+ stat_date=stat_date,
+ settings_hash=settings_hash,
)
if row is None:
row = AdminLeaderboardDailyStatRow(
@@ -1630,7 +1762,13 @@ async def _store_daily_snapshot(
if generated_at is not None:
row.generated_at = generated_at
row.updated_at = now
- await session.commit()
+ await _time_leaderboard_db_operation(
+ "snapshot.store.commit",
+ lambda: session.commit(),
+ sql="COMMIT",
+ stat_date=stat_date,
+ settings_hash=settings_hash,
+ )
async def _queue_daily_leaderboard_snapshots(
@@ -1646,12 +1784,17 @@ async def _queue_daily_leaderboard_snapshots(
return queued
async with session_factory() as session:
now = datetime.now(UTC)
+ stmt = select(AdminLeaderboardDailyStatRow).where(
+ AdminLeaderboardDailyStatRow.stat_date.in_(stat_dates),
+ AdminLeaderboardDailyStatRow.settings_hash == settings_hash,
+ )
rows = (
- await session.execute(
- select(AdminLeaderboardDailyStatRow).where(
- AdminLeaderboardDailyStatRow.stat_date.in_(stat_dates),
- AdminLeaderboardDailyStatRow.settings_hash == settings_hash,
- )
+ await _execute_leaderboard_query(
+ session,
+ "snapshot_queue.load_existing",
+ stmt,
+ stat_dates=stat_dates,
+ settings_hash=settings_hash,
)
).scalars().all()
rows_by_date = {row.stat_date: row for row in rows}
@@ -1680,7 +1823,13 @@ async def _queue_daily_leaderboard_snapshots(
row.updated_at = now
queued.append(stat_date)
if queued:
- await session.commit()
+ await _time_leaderboard_db_operation(
+ "snapshot_queue.commit",
+ lambda: session.commit(),
+ sql="COMMIT",
+ stat_dates=queued,
+ settings_hash=settings_hash,
+ )
return queued
@@ -1722,7 +1871,22 @@ async def _refresh_daily_leaderboard_snapshot(
include_failed: bool,
) -> None:
acquired = False
+ trace_token = _LEADERBOARD_TRACE_CONTEXT.set(
+ {
+ "job_key": job_key,
+ "stat_date": stat_date,
+ "settings_hash": settings_hash,
+ }
+ )
+ snapshot_start = perf_counter()
try:
+ logger.info(
+ "Leaderboard snapshot build start: job=%s stat_date=%s settings_hash=%s started_at=%s",
+ job_key,
+ stat_date,
+ settings_hash,
+ datetime.now(BEIJING_TZ).isoformat(),
+ )
if _LEADERBOARD_START_DELAY_SECONDS > 0:
await asyncio.sleep(_LEADERBOARD_START_DELAY_SECONDS)
await _store_daily_snapshot(
@@ -1758,6 +1922,14 @@ async def _refresh_daily_leaderboard_snapshot(
error=None,
generated_at=generated_at,
)
+ logger.info(
+ "Leaderboard snapshot build done: job=%s stat_date=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f",
+ job_key,
+ stat_date,
+ settings_hash,
+ datetime.now(BEIJING_TZ).isoformat(),
+ (perf_counter() - snapshot_start) * 1000,
+ )
except asyncio.TimeoutError:
logger.warning("Daily leaderboard snapshot timed out: %s", job_key)
await _store_daily_snapshot(
@@ -1781,6 +1953,7 @@ async def _refresh_daily_leaderboard_snapshot(
finally:
if acquired:
_leaderboard_semaphore.release()
+ _LEADERBOARD_TRACE_CONTEXT.reset(trace_token)
async def _load_or_schedule_daily_leaderboard(
@@ -2248,22 +2421,182 @@ class AdminUserQaPageResponse(BaseModel):
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")
+def _qa_sort_datetime(value: str | None) -> datetime:
+ if not value:
+ return datetime.min.replace(tzinfo=BEIJING_TZ)
+ text_value = str(value).strip()
+ if text_value.endswith("Z"):
+ text_value = text_value[:-1] + "+00:00"
+ try:
+ parsed = datetime.fromisoformat(text_value)
+ except ValueError:
+ return datetime.min.replace(tzinfo=BEIJING_TZ)
+ if parsed.tzinfo is None:
+ return parsed.replace(tzinfo=BEIJING_TZ)
+ return parsed.astimezone(BEIJING_TZ)
+
+def _qa_dedupe_key(item: AdminUserQaItem) -> tuple[str, str, str]:
+ question_key = _question_similarity_key(item.question)[:160]
+ answer_key = _question_similarity_key(item.answer)[:160]
+ return (item.thread_id, question_key, answer_key)
+
+
+def _message_value(message: Any, key: str, default: Any = None) -> Any:
+ if isinstance(message, dict):
+ return message.get(key, default)
+ return getattr(message, key, default)
+
+
+def _qa_message_type(message: Any) -> str:
+ value = _message_value(message, "type") or _message_value(message, "role") or ""
+ if value == "user":
+ return "human"
+ if value == "assistant":
+ return "ai"
+ return str(value)
+
+
+def _qa_message_id(message: Any, fallback: str) -> str:
+ return str(_message_value(message, "id") or fallback)
+
+
+def _qa_message_text(message: Any) -> str:
+ return _message_content_to_text(_message_value(message, "content")).strip()
+
+
+async def _read_thread_checkpoint_messages(checkpointer, thread_id: str) -> list[dict]:
+ config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
+ try:
+ checkpoint_tuple = await checkpointer.aget_tuple(config)
+ except Exception:
+ logger.warning("Failed to read checkpoint while listing Q&A: thread=%s", thread_id, exc_info=True)
+ return []
+ if checkpoint_tuple is None:
+ return []
+ checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {}
+ channel_values = checkpoint.get("channel_values", {}) if isinstance(checkpoint, dict) else {}
+ values = serialize_channel_values(channel_values)
+ messages = values.get("messages", [])
+ return messages if isinstance(messages, list) else []
+
+
+def _qa_items_from_checkpoint_messages(
+ *,
+ user_id: str,
+ thread_id: str,
+ thread_title: str,
+ assistant_id: str,
+ thread_time: str,
+ messages: list[dict],
+) -> list[AdminUserQaItem]:
+ items: list[AdminUserQaItem] = []
+ current_question = ""
+ current_answer = ""
+ current_id = ""
+ current_index = 0
+
+ def flush() -> None:
+ nonlocal current_question, current_answer, current_id, current_index
+ question = _clean_question_text(current_question)
+ if not _is_displayable_question(question):
+ current_question = ""
+ current_answer = ""
+ current_id = ""
+ return
+ items.append(
+ AdminUserQaItem(
+ run_id=current_id or f"{thread_id}:checkpoint:{current_index}",
+ thread_id=thread_id,
+ thread_title=thread_title,
+ assistant_id=assistant_id,
+ status="checkpoint",
+ question=question[:2000],
+ answer=current_answer.strip()[:6000],
+ created_at=thread_time,
+ updated_at=thread_time,
+ )
+ )
+ current_question = ""
+ current_answer = ""
+ current_id = ""
+
+ for index, message in enumerate(messages):
+ msg_type = _qa_message_type(message)
+ if msg_type == "human":
+ if current_question:
+ flush()
+ current_question = _qa_message_text(message)
+ current_answer = ""
+ current_id = _qa_message_id(message, f"{thread_id}:checkpoint:{index}")
+ current_index = index
+ continue
+ if msg_type == "ai" and current_question:
+ text = _qa_message_text(message)
+ if text:
+ current_answer = text
+
+ if current_question:
+ flush()
+
+ return items
+
+
+async def _list_thread_checkpoint_qa_items(
+ *,
+ request: Request,
+ user_id: str,
+) -> list[AdminUserQaItem]:
+ repo = get_thread_store(request)
+ checkpointer = get_checkpointer(request)
+ items: list[AdminUserQaItem] = []
+ offset = 0
+ page_size = 200
+ while True:
+ rows = await repo.search(limit=page_size, offset=offset, user_id=user_id)
+ if not rows:
+ break
+ for row in rows:
+ thread_id = str(row.get("thread_id") or "")
+ if not thread_id:
+ continue
+ messages = await _read_thread_checkpoint_messages(checkpointer, thread_id)
+ if not messages:
+ continue
+ metadata = row.get("metadata") if isinstance(row.get("metadata"), dict) else {}
+ context = row.get("context") if isinstance(row.get("context"), dict) else {}
+ assistant_id = str(
+ row.get("assistant_id")
+ or (context or {}).get("agent_id")
+ or (metadata or {}).get("agent_id")
+ or (context or {}).get("agent_name")
+ or (metadata or {}).get("agent_name")
+ or ""
+ )
+ thread_time = str(row.get("updated_at") or row.get("created_at") or "")
+ items.extend(
+ _qa_items_from_checkpoint_messages(
+ user_id=user_id,
+ thread_id=thread_id,
+ thread_title=str(row.get("display_name") or ""),
+ assistant_id=assistant_id,
+ thread_time=thread_time,
+ messages=messages,
+ )
+ )
+ if len(rows) < page_size:
+ break
+ offset += len(rows)
+ return items
+
+
+async def _list_run_summary_qa_items(
+ *,
+ session_factory,
+ user_id: str,
+ since: str | None,
+ until: str | None,
+) -> list[AdminUserQaItem]:
filters = [
RunRow.user_id == user_id,
RunRow.first_human_message.is_not(None),
@@ -2272,26 +2605,9 @@ async def list_user_qa(
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(
@@ -2316,32 +2632,152 @@ async def list_user_qa(
)
.where(*filters)
.order_by(RunRow.created_at.desc(), RunRow.run_id.desc())
- .limit(page_size)
- .offset(offset)
)
).all()
+ return [
+ AdminUserQaItem(
+ run_id=r[0] or "",
+ thread_id=r[1] or "",
+ thread_title=r[2] or "",
+ assistant_id=r[3] or "",
+ status=r[4] or "",
+ model_name=r[5] or "",
+ question=_clean_question_text(r[6])[:2000],
+ answer=str(r[7] or "").strip()[:6000],
+ total_tokens=int(r[8] or 0),
+ created_at=_iso(r[9]),
+ updated_at=_iso(r[10]),
+ )
+ for r in rows
+ if _is_displayable_question(r[6])
+ ]
+
+
+def _filter_qa_items(
+ items: list[AdminUserQaItem],
+ *,
+ q: str | None,
+ since: str | None,
+ until: str | None,
+) -> list[AdminUserQaItem]:
+ needle = (q or "").strip().lower()
+ since_dt = _parse_optional_datetime(since) if since else None
+ until_dt = _parse_optional_datetime(until) if until else None
+ filtered: list[AdminUserQaItem] = []
+ for item in items:
+ item_dt = _qa_sort_datetime(item.created_at or item.updated_at)
+ if since_dt and item_dt < since_dt:
+ continue
+ if until_dt and item_dt > until_dt:
+ continue
+ if needle:
+ haystack = "\n".join(
+ [
+ item.question,
+ item.answer,
+ item.run_id,
+ item.thread_id,
+ item.thread_title,
+ item.assistant_id,
+ ]
+ ).lower()
+ if needle not in haystack:
+ continue
+ filtered.append(item)
+ return filtered
+
+
+async def _list_user_qa_page(
+ *,
+ request: Request,
+ user_id: str,
+ page: int,
+ page_size: int,
+ q: str | None,
+ since: str | None,
+ until: str | None,
+ include_checkpoints: bool = True,
+) -> AdminUserQaPageResponse:
+ session_factory = get_session_factory()
+ if session_factory is None:
+ raise HTTPException(status_code=503, detail="Q&A audit requires database persistence")
+
+ items = await _list_run_summary_qa_items(
+ session_factory=session_factory,
+ user_id=user_id,
+ since=since,
+ until=until,
+ )
+ if include_checkpoints:
+ items.extend(await _list_thread_checkpoint_qa_items(request=request, user_id=user_id))
+
+ by_key: dict[tuple[str, str, str], AdminUserQaItem] = {}
+ for item in items:
+ key = _qa_dedupe_key(item)
+ existing = by_key.get(key)
+ if existing is None or (existing.status == "checkpoint" and item.status != "checkpoint"):
+ by_key[key] = item
+
+ filtered = _filter_qa_items(list(by_key.values()), q=q, since=since, until=until)
+ filtered.sort(
+ key=lambda item: (_qa_sort_datetime(item.created_at or item.updated_at), item.thread_id, item.run_id),
+ reverse=True,
+ )
+ offset = (page - 1) * page_size
return AdminUserQaPageResponse(
user_id=user_id,
page=page,
page_size=page_size,
- total=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
- ],
+ total=len(filtered),
+ items=filtered[offset : offset + page_size],
+ )
+
+
+@router.get("/{user_id}/qa", response_model=AdminUserQaPageResponse)
+async def list_user_qa(
+ user_id: str,
+ request: Request,
+ page: int = Query(default=1, ge=1),
+ page_size: int = Query(default=20, ge=1, le=100),
+ q: str | None = Query(default=None, description="Search question, answer, run id, or thread id"),
+ since: str | None = Query(default=None, description="Inclusive range start, ISO datetime"),
+ until: str | None = Query(default=None, description="Inclusive range end, ISO datetime"),
+) -> AdminUserQaPageResponse:
+ """Return one user's Q&A runs, including older checkpoint-only threads."""
+ await _require_admin(request)
+ return await _list_user_qa_page(
+ request=request,
+ user_id=user_id,
+ page=page,
+ page_size=page_size,
+ q=q,
+ since=since,
+ until=until,
+ )
+
+
+@self_router.get("/qa", response_model=AdminUserQaPageResponse)
+async def list_my_qa(
+ request: Request,
+ page: int = Query(default=1, ge=1),
+ page_size: int = Query(default=20, ge=1, le=100),
+ q: str | None = Query(default=None, description="Search question, answer, run id, or thread id"),
+ since: str | None = Query(default=None, description="Inclusive range start, ISO datetime"),
+ until: str | None = Query(default=None, description="Inclusive range end, ISO datetime"),
+) -> AdminUserQaPageResponse:
+ """Return the current user's own Q&A history."""
+ user = getattr(request.state, "user", None)
+ if user is None:
+ user = await get_current_user_from_request(request)
+ return await _list_user_qa_page(
+ request=request,
+ user_id=str(user.id),
+ page=page,
+ page_size=page_size,
+ q=q,
+ since=since,
+ until=until,
)