From 0001f86826a28eb6a71fe47c1e0e78ca9a370457 Mon Sep 17 00:00:00 2001 From: Administrator Date: Sun, 7 Jun 2026 21:18:16 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E5=96=84=E5=85=AC=E5=BC=80?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E9=A2=84=E8=A7=88=E5=92=8C=E7=94=A8=E6=88=B7?= =?UTF-8?q?=E9=97=AE=E7=AD=94=E5=8E=86=E5=8F=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- frontend-web/src/App.tsx | 8 +- frontend-web/src/core/user-qa/api.ts | 47 ++ .../src/pages/PublicScheduledTaskPage.tsx | 289 ++++++++- .../backend/app/gateway/app.py | 151 ++++- .../app/gateway/routers/admin_users.py | 598 +++++++++++++++--- 5 files changed, 984 insertions(+), 109 deletions(-) create mode 100644 frontend-web/src/core/user-qa/api.ts diff --git a/frontend-web/src/App.tsx b/frontend-web/src/App.tsx index a24f9b53..4ca83b8e 100644 --- a/frontend-web/src/App.tsx +++ b/frontend-web/src/App.tsx @@ -15,7 +15,9 @@ import { HashRouter, useLocation } from "react-router-dom"; import LandingPage from "./pages/LandingPage"; import LoginPage from "./pages/LoginPage"; import PageRoutes from "./pages/PageRoutes"; -import PublicScheduledTaskPage from "./pages/PublicScheduledTaskPage"; +import PublicScheduledTaskPage, { + PublicScheduledTaskFullscreenPage, +} from "./pages/PublicScheduledTaskPage"; import PublicSharePage from "./pages/PublicSharePage"; import PublicRoundtableSharePage from "./pages/PublicRoundtableSharePage"; import EmbedChatPage from "./pages/EmbedChatPage"; @@ -78,6 +80,10 @@ export function App() { path="/public/scheduled-tasks/:taskId" element={} /> + } + /> {/* Embedded workspace Q&A — no page/workspace sidebars; parent passes message/username/theme */} (res: Response, what: string): Promise { + if (!res.ok) { + const err = (await res.json().catch(() => ({}))) as { detail?: string }; + throw new Error(err.detail ?? `${what} (HTTP ${res.status})`); + } + return res.json() as Promise; +} + +export async function listMyQa( + params: { page?: number; pageSize?: number; q?: string } = {}, +): Promise { + const search = new URLSearchParams({ + page: String(params.page ?? 1), + page_size: String(params.pageSize ?? 20), + }); + const query = (params.q ?? "").trim(); + if (query) search.set("q", query); + return parse( + await apiFetch(`${getBackendBaseURL()}/api/users/me/qa?${search}`), + "Failed to load my Q&A", + ); +} diff --git a/frontend-web/src/pages/PublicScheduledTaskPage.tsx b/frontend-web/src/pages/PublicScheduledTaskPage.tsx index a57f653c..e3e251f3 100644 --- a/frontend-web/src/pages/PublicScheduledTaskPage.tsx +++ b/frontend-web/src/pages/PublicScheduledTaskPage.tsx @@ -6,16 +6,18 @@ import { ChevronLeft, ChevronRight, Download, + ExternalLink, FileText, List, Loader2, + Maximize2, Pencil, Trash2, X, } from "lucide-react"; import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import type { PointerEvent as ReactPointerEvent } from "react"; -import { useParams } from "react-router-dom"; +import { useParams, useSearchParams } from "react-router-dom"; import { toast } from "sonner"; import { Badge } from "@/components/ui/badge"; @@ -127,6 +129,10 @@ async function fetchPublicTaskRuns(taskId: string): Promise return fetchJson(`${publicBaseUrl()}/${encodeURIComponent(taskId)}/runs?limit=200`); } +async function fetchPublicRun(runId: string): Promise { + return fetchJson(`${publicBaseUrl()}/runs/${encodeURIComponent(runId)}`); +} + async function fetchPublicRunFile(runId: string, index: number): Promise { return fetchJson( `${publicBaseUrl()}/runs/${encodeURIComponent(runId)}/files/${index}`, @@ -667,12 +673,61 @@ type ContentTab = { fileName?: string; }; +type PublicFullscreenPreview = + | { mode: "html"; title: string; html: string } + | { mode: "markdown"; title: string; content: string } + | { mode: "text"; title: string; content: string } + | { mode: "file"; title: string; file: PublicFileContent }; + +function PublicFullscreenPreviewContent({ preview }: { preview: PublicFullscreenPreview }) { + if (preview.mode === "html") { + return ; + } + if (preview.mode === "markdown") { + return ( + + ); + } + if (preview.mode === "file") { + const content = preview.file.content || "无文件内容"; + if (isMarkdownFile(preview.file)) { + return ( + + ); + } + return 
{content}
; + } + return 
{preview.content || "无正文内容"}
; +} + +function publicFullscreenHref(taskId: string, runId: string, tab: ContentTab): string { + const params = new URLSearchParams(); + params.set("tab", tab.kind); + if (tab.kind === "file" && tab.fileIndex != null) { + params.set("file", String(tab.fileIndex)); + } + return `#/public/scheduled-tasks/${encodeURIComponent(taskId)}/runs/${encodeURIComponent(runId)}/fullscreen?${params.toString()}`; +} + /** * Right-pane controller for a single run: builds the tab list (body + MDs), * fetches each tab's source text, and hosts the view/edit toggle. */ function RunDetailPane({ run, taskId }: { run: PublicRun; taskId: string }) { const queryClient = useQueryClient(); + const [fullscreenPreview, setFullscreenPreview] = useState(null); // HTML page run: render a sandboxed iframe preview instead of the MD tabs. const isHtmlRun = @@ -837,9 +892,67 @@ function RunDetailPane({ run, taskId }: { run: PublicRun; taskId: string }) { const fileIndex = activeTabMeta?.fileIndex; const fileIsLoading = isFileTab && fileIndex != null ? !!fileLoading[fileIndex] : false; const showEmptyHint = !activeSource && !fileIsLoading; + const fullscreenHref = activeTabMeta ? publicFullscreenHref(taskId, run.id, activeTabMeta) : ""; + const fullscreenDisabled = + !activeTabMeta || + (activeTabMeta.kind === "html" ? !htmlContent : activeTabMeta.kind === "file" ? fileIsLoading : false); + + const handleOpenFullscreen = useCallback(() => { + if (!activeTabMeta) return; + if (activeTabMeta.kind === "html") { + if (!htmlContent) return; + setFullscreenPreview({ + mode: "html", + title: run.result?.title || "HTML 预览", + html: htmlContent, + }); + return; + } + if (activeTabMeta.kind === "body") { + setFullscreenPreview({ + mode: run.result?.format === "markdown" || run.result?.format === "mixed" ? "markdown" : "text", + title: run.result?.title || "正文预览", + content: activeSource || "无正文内容", + }); + return; + } + const sourceFile = markdownFiles.find((file) => file.index === activeTabMeta.fileIndex); + setFullscreenPreview({ + mode: "file", + title: activeTabMeta.fileName || sourceFile?.name || "文件预览", + file: { + index: activeTabMeta.fileIndex ?? 0, + name: activeTabMeta.fileName || sourceFile?.name || "result.md", + mime_type: sourceFile?.mime_type ?? "text/markdown", + kind: sourceFile?.kind ?? "markdown", + size: sourceFile?.size ?? activeSource.length, + previewable: true, + content: activeSource, + }, + }); + }, [activeTabMeta, activeSource, htmlContent, markdownFiles, run.result?.format, run.result?.title]); return ( <> + !open && setFullscreenPreview(null)}> + + + + {fullscreenPreview?.title ?? "预览"} + + + {fullscreenPreview && ( +
+ +
+ )} + +
@@ -859,6 +972,30 @@ function RunDetailPane({ run, taskId }: { run: PublicRun; taskId: string }) {

+ {!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, )