feat: 完善公开任务预览和用户问答历史

This commit is contained in:
Administrator committed 2026-06-07 21:18:16 +08:00
1 parent e7c745a512
commit 0001f86826
5 files changed
+984 -109

No files matched your search

+7 -1
View File
@@ -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={<PublicScheduledTaskPage />}
/>
<Route
path="/public/scheduled-tasks/:taskId/runs/:runId/fullscreen"
element={<PublicScheduledTaskFullscreenPage />}
/>
{/* Embedded workspace Q&A — no page/workspace sidebars; parent passes message/username/theme */}
<Route
path="/embed/chats/:thread_id"
+47
View File
@@ -0,0 +1,47 @@
import { apiFetch } from "@/core/api";
import { getBackendBaseURL } from "@/core/config";
export interface UserQaItem {
run_id: string;
thread_id: string;
thread_title: string;
assistant_id: string;
status: string;
model_name: string;
question: string;
answer: string;
total_tokens: number;
created_at: string;
updated_at: string;
}
export interface UserQaPageResponse {
user_id: string;
page: number;
page_size: number;
total: number;
items: UserQaItem[];
}
async function parse<T>(res: Response, what: string): Promise<T> {
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<T>;
}
export async function listMyQa(
params: { page?: number; pageSize?: number; q?: string } = {},
): Promise<UserQaPageResponse> {
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",
);
}
@@ -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<PublicRunsResponse>
return fetchJson<PublicRunsResponse>(`${publicBaseUrl()}/${encodeURIComponent(taskId)}/runs?limit=200`);
}
async function fetchPublicRun(runId: string): Promise<PublicRun> {
return fetchJson<PublicRun>(`${publicBaseUrl()}/runs/${encodeURIComponent(runId)}`);
}
async function fetchPublicRunFile(runId: string, index: number): Promise<PublicFileContent> {
return fetchJson<PublicFileContent>(
`${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 <HtmlPreviewFrame html={preview.html} className="h-full w-full border-0 bg-white" />;
}
if (preview.mode === "markdown") {
return (
<MarkdownContent
className="max-w-none"
content={preview.content || "无正文内容"}
isLoading={false}
remarkPlugins={reasoningPlugins.remarkPlugins}
rehypePlugins={reasoningPlugins.rehypePlugins}
/>
);
}
if (preview.mode === "file") {
const content = preview.file.content || "无文件内容";
if (isMarkdownFile(preview.file)) {
return (
<MarkdownContent
className="max-w-none"
content={content}
isLoading={false}
remarkPlugins={reasoningPlugins.remarkPlugins}
rehypePlugins={reasoningPlugins.rehypePlugins}
/>
);
}
return <pre className="whitespace-pre-wrap break-words text-sm">{content}</pre>;
}
return <pre className="whitespace-pre-wrap break-words text-sm">{preview.content || "无正文内容"}</pre>;
}
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<PublicFullscreenPreview | null>(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 (
<>
<Dialog open={fullscreenPreview !== null} onOpenChange={(open) => !open && setFullscreenPreview(null)}>
<DialogContent className="flex h-[96vh] w-[96vw] max-w-none flex-col gap-0 p-0 sm:max-w-none">
<DialogHeader className="shrink-0 border-b px-4 py-3 pr-12">
<DialogTitle className="truncate text-base">
{fullscreenPreview?.title ?? "预览"}
</DialogTitle>
</DialogHeader>
{fullscreenPreview && (
<div
className={cn(
"min-h-0 flex-1",
fullscreenPreview.mode === "html" ? "" : "overflow-auto p-4",
)}
>
<PublicFullscreenPreviewContent preview={fullscreenPreview} />
</div>
)}
</DialogContent>
</Dialog>
<CardHeader className="flex flex-wrap items-start justify-between gap-2 space-y-0">
<div className="min-w-0 flex-1">
<Tooltip>
@@ -859,6 +972,30 @@ function RunDetailPane({ run, taskId }: { run: PublicRun; taskId: string }) {
</p>
</div>
<div className="flex flex-wrap items-center gap-2">
{!isEditing && activeTabMeta && (
<>
<Button
size="sm"
variant="outline"
onClick={handleOpenFullscreen}
disabled={fullscreenDisabled}
title="全屏查看"
>
<Maximize2 className="size-4" />
全屏
</Button>
<Button
size="sm"
variant="outline"
onClick={() => window.open(fullscreenHref, "_blank", "noopener,noreferrer")}
disabled={fullscreenDisabled}
title="新页面打开全屏预览"
>
<ExternalLink className="size-4" />
新页面
</Button>
</>
)}
{isFileTab && fileIndex != null && (
<a
className="bg-background hover:bg-accent inline-flex h-9 items-center gap-1 rounded-md border px-3 text-xs"
@@ -999,3 +1136,153 @@ function RunDetailPane({ run, taskId }: { run: PublicRun; taskId: string }) {
</>
);
}
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 (
<TooltipProvider delayDuration={150}>
<div className="bg-background flex h-screen min-h-0 flex-col">
<div className="flex h-12 shrink-0 items-center gap-2 border-b px-3">
<Button asChild variant="ghost" size="sm" className="h-8 px-2 text-xs">
<a href={backHref}>
<ChevronLeft className="size-3" />
返回列表
</a>
</Button>
<div className="min-w-0 flex-1 truncate text-sm font-medium" title={pageTitle}>
{pageTitle}
</div>
{tabKind === "file" && runId && fileIndexValid && (
<Button asChild variant="outline" size="sm" className="h-8 px-2 text-xs">
<a
href={publicRunFileDownloadUrl(runId, fileIndexValue)}
target="_blank"
rel="noopener noreferrer"
>
<Download className="size-3" />
下载
</a>
</Button>
)}
</div>
<div className="min-h-0 flex-1">
{!runId || !taskId ? (
<div className="text-muted-foreground flex h-full items-center justify-center text-sm">
缺少任务或执行记录 ID。
</div>
) : isLoading ? (
<div className="text-muted-foreground flex h-full items-center justify-center gap-2 text-sm">
<Loader2 className="size-4 animate-spin" />
正在加载预览...
</div>
) : isError ? (
<div className="text-destructive flex h-full items-center justify-center text-sm">
加载失败:{error instanceof Error ? error.message : "未知错误"}
</div>
) : tabKind === "html" ? (
htmlContent ? (
<HtmlPreviewFrame html={htmlContent} className="h-full w-full border-0 bg-white" />
) : (
<div className="text-muted-foreground flex h-full items-center justify-center gap-2 text-sm">
{htmlLoading && <Loader2 className="size-4 animate-spin" />}
HTML 页面尚未生成或正在生成中。
</div>
)
) : tabKind === "file" ? (
fileLoading ? (
<div className="text-muted-foreground flex h-full items-center justify-center gap-2 text-sm">
<Loader2 className="size-4 animate-spin" />
正在加载文件...
</div>
) : !fileIndexValid || fileIsError || !fileData ? (
<div className="text-destructive flex h-full items-center justify-center text-sm">
文件加载失败:{fileError instanceof Error ? fileError.message : "文件不存在或不可预览"}
</div>
) : (
<ScrollArea className="h-full">
<div className="mx-auto max-w-5xl p-6">
<PublicFullscreenPreviewContent
preview={{
mode: "file",
title: fileData.name,
file: fileData,
}}
/>
</div>
</ScrollArea>
)
) : (
<ScrollArea className="h-full">
<div className="mx-auto max-w-5xl p-6">
<PublicFullscreenPreviewContent
preview={{
mode:
run?.result?.format === "markdown" || run?.result?.format === "mixed"
? "markdown"
: "text",
title: run?.result?.title || "正文预览",
content: run?.result?.content || "无正文内容",
}}
/>
</div>
</ScrollArea>
)}
</div>
</div>
</TooltipProvider>
);
}
@@ -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
@@ -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,
)