This commit is contained in:
fengxiangdebaozi committed 2026-08-25 17:07:09 +08:00
commit dd1064b3cb
28 files changed
+616 -68

No files matched your search

+15 -4
View File
@@ -27,6 +27,16 @@ export type KnowledgeSourceLike = {
slug?: string | null; slug?: string | null;
}; };
/** Build a HashRouter URL without dropping a reverse-proxy path such as /web/. */
export function buildAppHashUrl(path: string): string {
const route = path.startsWith("/") ? path : `/${path}`;
if (typeof window !== "undefined") {
return `${window.location.pathname || "/"}#${route}`;
}
const base = import.meta.env.BASE_URL || "/";
return `${base}#${route}`;
}
export function parseWeKnoraPreviewUrl(url?: string | null): WeKnoraSourceTarget | null { export function parseWeKnoraPreviewUrl(url?: string | null): WeKnoraSourceTarget | null {
if (!url?.startsWith("/weknora-source-preview?")) { if (!url?.startsWith("/weknora-source-preview?")) {
return null; return null;
@@ -92,7 +102,7 @@ export function buildWeKnoraSourceIframeUrl(target: WeKnoraSourceTarget): string
if (target.documentId) params.set("documentId", target.documentId); if (target.documentId) params.set("documentId", target.documentId);
if (target.wikiSlug) params.set("wikiSlug", target.wikiSlug); if (target.wikiSlug) params.set("wikiSlug", target.wikiSlug);
const query = params.toString(); const query = params.toString();
return `/#/page/workspace/knowledge/${encodeURIComponent(target.knowledgeBaseId)}${query ? `?${query}` : ""}`; return buildAppHashUrl(`/page/workspace/knowledge/${encodeURIComponent(target.knowledgeBaseId)}${query ? `?${query}` : ""}`);
} }
/** /**
@@ -103,7 +113,7 @@ export function buildWeKnoraWikiPageUrl(
knowledgeBaseId: string, knowledgeBaseId: string,
wikiSlug: string, wikiSlug: string,
): string { ): string {
return `/#${buildWeKnoraWikiPagePath(knowledgeBaseId, wikiSlug)}`; return buildAppHashUrl(buildWeKnoraWikiPagePath(knowledgeBaseId, wikiSlug));
} }
export function buildWeKnoraWikiPagePath( export function buildWeKnoraWikiPagePath(
@@ -129,7 +139,7 @@ export function buildPublicKnowledgeWikiUrl(
knowledgeBaseId: string, knowledgeBaseId: string,
wikiSlug?: string, wikiSlug?: string,
): string { ): string {
return `/#${buildPublicKnowledgeWikiPath(knowledgeBaseId, wikiSlug)}`; return buildAppHashUrl(buildPublicKnowledgeWikiPath(knowledgeBaseId, wikiSlug));
} }
export function buildLegacyKnowledgeSourceUrl(source: KnowledgeSourceLike): string { export function buildLegacyKnowledgeSourceUrl(source: KnowledgeSourceLike): string {
@@ -160,6 +170,7 @@ export function normalizeMalformedWeKnoraHashRoute(): boolean {
const { pathname, search, hash } = window.location; const { pathname, search, hash } = window.location;
if (!pathname.startsWith("/page/workspace/knowledge/")) return false; if (!pathname.startsWith("/page/workspace/knowledge/")) return false;
if (hash && !hash.startsWith("#/")) return false; if (hash && !hash.startsWith("#/")) return false;
window.location.replace(`/#${pathname}${search}`); const base = import.meta.env.BASE_URL || "/";
window.location.replace(`${base}#${pathname}${search}`);
return true; return true;
} }
@@ -179,6 +179,28 @@ export interface LlmWikiWikiIndexStatus {
last_error?: string | null; last_error?: string | null;
last_started_at?: string | null; last_started_at?: string | null;
last_completed_at?: string | null; last_completed_at?: string | null;
progress?: {
total_pages: number;
processed_pages: number;
vectorized_pages: number;
failed_pages: number;
vector_count: number;
completion_percent: number;
};
pages?: Array<{
id: string;
slug: string;
title: string;
page_type: string;
status: string;
index_status: string;
vector_count: number;
vectorized: boolean;
processed_in_current_sync: boolean;
indexed_at?: string | null;
updated_at?: string | null;
last_error?: string | null;
}>;
} }
export interface LlmWikiSourceContext { export interface LlmWikiSourceContext {
@@ -444,21 +466,22 @@ export async function getLlmWikiDocumentPreview(
return response.blob(); return response.blob();
} }
export async function getLlmWikiProtectedFile( export async function getLlmWikiProtectedFileUrl(
knowledgeBaseId: string, knowledgeBaseId: string,
filePath: string, filePath: string,
publicAccess = false, publicAccess = false,
): Promise<Blob> { ): Promise<string> {
const base = publicAccess ? PUBLIC_API() : API();
const params = new URLSearchParams({ file_path: filePath }); const params = new URLSearchParams({ file_path: filePath });
const response = await apiFetch( if (publicAccess) {
`${base}/knowledge-bases/${encodeURIComponent(knowledgeBaseId)}/files?${params.toString()}`, return `${PUBLIC_API()}/knowledge-bases/${encodeURIComponent(knowledgeBaseId)}/files?${params.toString()}`;
);
if (!response.ok) {
await readJson<void>(response, "读取 Wiki 图片");
throw new Error("读取 Wiki 图片失败");
} }
return response.blob(); const access = await readJson<{ token: string; expires_in: number }>(
await apiFetch(
`${API()}/knowledge-bases/${encodeURIComponent(knowledgeBaseId)}/files/access-url?${params.toString()}`,
),
"读取 Wiki 图片",
);
return `${PUBLIC_API()}/files/${encodeURIComponent(access.token)}`;
} }
export async function getLlmWikiGraph(knowledgeBaseId: string): Promise<LlmWikiGraph> { export async function getLlmWikiGraph(knowledgeBaseId: string): Promise<LlmWikiGraph> {
@@ -6,6 +6,7 @@ import { toast } from "sonner";
import { Button } from "@/components/ui/button"; import { Button } from "@/components/ui/button";
import { Badge } from "@/components/ui/badge"; import { Badge } from "@/components/ui/badge";
import { Input } from "@/components/ui/input"; import { Input } from "@/components/ui/input";
import { Progress } from "@/components/ui/progress";
import { Textarea } from "@/components/ui/textarea"; import { Textarea } from "@/components/ui/textarea";
import { MarkdownCodeEditor } from "@/components/wiki/MarkdownCodeEditor"; import { MarkdownCodeEditor } from "@/components/wiki/MarkdownCodeEditor";
import { import {
@@ -331,6 +332,10 @@ export function WeKnoraWikiManagementPage() {
() => moveTargetOptions(pagesList), () => moveTargetOptions(pagesList),
[pagesList], [pagesList],
); );
const pageIndexBySlug = useMemo(
() => new Map((indexStatus.data?.pages ?? []).map((item) => [item.slug, item])),
[indexStatus.data?.pages],
);
const currentMoveTarget = moveTargetValue(draft.parentSlug, draft.categoryPathText, draft.folderId); const currentMoveTarget = moveTargetValue(draft.parentSlug, draft.categoryPathText, draft.folderId);
useEffect(() => { useEffect(() => {
@@ -387,6 +392,17 @@ export function WeKnoraWikiManagementPage() {
}); });
}; };
const renderVectorBadge = (page: WikiPageListItem) => {
const item = pageIndexBySlug.get(page.slug);
if (item?.last_error || item?.index_status === "failed" || item?.index_status === "stale") {
return <Badge variant="destructive" className="ml-auto shrink-0 text-[10px]" title={item?.last_error || "向量化失败"}>失败</Badge>;
}
if (item?.vectorized) {
return <Badge variant="secondary" className="ml-auto shrink-0 text-[10px]" title={`最近向量化:${item.indexed_at || "未知"}`}>已向量化 · {item.vector_count} 段</Badge>;
}
return <Badge variant="outline" className="ml-auto shrink-0 text-[10px]">未向量化</Badge>;
};
const renderWikiNode = (node: WikiTreeNode, depth = 0) => { const renderWikiNode = (node: WikiTreeNode, depth = 0) => {
const hasChildren = node.children.length > 0; const hasChildren = node.children.length > 0;
const isExpanded = query.trim() ? true : expandedDirectories.has(node.key); const isExpanded = query.trim() ? true : expandedDirectories.has(node.key);
@@ -432,7 +448,8 @@ export function WeKnoraWikiManagementPage() {
> >
<div className="flex min-w-0 items-center gap-2"> <div className="flex min-w-0 items-center gap-2">
<FileText className="size-3.5 shrink-0 text-muted-foreground" /> <FileText className="size-3.5 shrink-0 text-muted-foreground" />
<span className="truncate font-medium">{page.title || page.slug}</span> <span className="truncate font-medium">{page.title || page.slug}</span>
{renderVectorBadge(page)}
</div> </div>
<div className="mt-1 truncate pl-5 text-xs text-muted-foreground">{page.slug}</div> <div className="mt-1 truncate pl-5 text-xs text-muted-foreground">{page.slug}</div>
</button> </button>
@@ -460,6 +477,7 @@ export function WeKnoraWikiManagementPage() {
<div className="flex min-w-0 items-center gap-2"> <div className="flex min-w-0 items-center gap-2">
<FileText className="size-3.5 shrink-0 text-muted-foreground" /> <FileText className="size-3.5 shrink-0 text-muted-foreground" />
<span className="truncate font-medium">{page.title || page.slug}</span> <span className="truncate font-medium">{page.title || page.slug}</span>
{renderVectorBadge(page)}
</div> </div>
<div className="mt-1 truncate pl-5 text-xs text-muted-foreground">{page.slug}</div> <div className="mt-1 truncate pl-5 text-xs text-muted-foreground">{page.slug}</div>
{page.aliases?.length ? ( {page.aliases?.length ? (
@@ -634,6 +652,26 @@ export function WeKnoraWikiManagementPage() {
<main className="grid min-h-0 flex-1 grid-cols-[320px_1fr] overflow-hidden"> <main className="grid min-h-0 flex-1 grid-cols-[320px_1fr] overflow-hidden">
<aside className="flex min-h-0 flex-col border-r bg-muted/20"> <aside className="flex min-h-0 flex-col border-r bg-muted/20">
<div className="border-b p-3"> <div className="border-b p-3">
{indexStatus.data && (
<div className="mb-3 rounded-lg border bg-background p-3">
<div className="flex items-center justify-between gap-2 text-xs">
<span className="flex items-center gap-2 font-medium">
向量化进度
{!indexStatus.data.local_index_enabled && <Badge variant="outline" className="text-[10px]">当前未启用</Badge>}
</span>
<span className="tabular-nums text-muted-foreground">
{indexStatus.data.progress?.processed_pages ?? 0}/{indexStatus.data.progress?.total_pages ?? 0} 篇 · {indexStatus.data.progress?.completion_percent ?? 0}%
</span>
</div>
<Progress value={indexStatus.data.progress?.completion_percent ?? 0} className="mt-2 h-2" />
<div className="mt-2 flex flex-wrap gap-x-3 gap-y-1 text-[11px] text-muted-foreground">
<span>已向量化 {indexStatus.data.progress?.vectorized_pages ?? 0} 篇</span>
<span>向量章节 {indexStatus.data.progress?.vector_count ?? 0}</span>
<span className={(indexStatus.data.progress?.failed_pages ?? 0) > 0 ? "text-destructive" : ""}>失败 {indexStatus.data.progress?.failed_pages ?? 0} 篇</span>
</div>
{indexStatus.data.last_error && <div className="mt-2 break-words text-[11px] text-destructive">{indexStatus.data.last_error}</div>}
</div>
)}
<div className="relative"> <div className="relative">
<Search className="absolute left-3 top-1/2 size-4 -translate-y-1/2 text-muted-foreground" /> <Search className="absolute left-3 top-1/2 size-4 -translate-y-1/2 text-muted-foreground" />
<Input <Input
@@ -19,6 +19,7 @@ import { Button } from "@/components/ui/button";
import { Input } from "@/components/ui/input"; import { Input } from "@/components/ui/input";
import { MarkdownContent } from "@/components/workspace/messages/markdown-content"; import { MarkdownContent } from "@/components/workspace/messages/markdown-content";
import { import {
buildAppHashUrl,
buildPublicKnowledgeWikiPath, buildPublicKnowledgeWikiPath,
buildWeKnoraWikiPagePath, buildWeKnoraWikiPagePath,
} from "@/core/llmwiki/source-links"; } from "@/core/llmwiki/source-links";
@@ -98,9 +99,11 @@ function downloadBlob(blob: Blob, fileName: string) {
} }
function publicWikiTargetFromHref(href: string, knowledgeBaseId: string) { function publicWikiTargetFromHref(href: string, knowledgeBaseId: string) {
const routePrefix = `/#${buildPublicKnowledgeWikiPath(knowledgeBaseId)}`; const hashIndex = href.indexOf("#");
if (!href.startsWith(`${routePrefix}?`)) return null; if (hashIndex < 0) return null;
const route = href.slice(2); const route = href.slice(hashIndex + 1);
const routePrefix = buildPublicKnowledgeWikiPath(knowledgeBaseId);
if (!route.startsWith(`${routePrefix}?`)) return null;
const queryIndex = route.indexOf("?"); const queryIndex = route.indexOf("?");
const slug = new URLSearchParams(route.slice(queryIndex + 1)).get("wikiSlug")?.trim(); const slug = new URLSearchParams(route.slice(queryIndex + 1)).get("wikiSlug")?.trim();
return slug ? { route, slug } : null; return slug ? { route, slug } : null;
@@ -219,7 +222,7 @@ export function WeKnoraWikiPageView({
? (knowledgeBaseId, slug) => { ? (knowledgeBaseId, slug) => {
const next = new URLSearchParams(searchParams); const next = new URLSearchParams(searchParams);
next.set("wikiSlug", slug); next.set("wikiSlug", slug);
return `/#${buildPublicKnowledgeWikiPath(knowledgeBaseId)}?${next.toString()}`; return buildAppHashUrl(`${buildPublicKnowledgeWikiPath(knowledgeBaseId)}?${next.toString()}`);
} }
: undefined, : undefined,
publicEmbed ? pagesList : [], publicEmbed ? pagesList : [],
@@ -2,7 +2,7 @@ import { ImageOff, Loader2 } from "lucide-react";
import { type ImgHTMLAttributes, useEffect, useState } from "react"; import { type ImgHTMLAttributes, useEffect, useState } from "react";
import { parseProtectedWikiImageSource } from "@/core/llmwiki/wiki-page-markdown"; import { parseProtectedWikiImageSource } from "@/core/llmwiki/wiki-page-markdown";
import { getLlmWikiProtectedFile } from "@/strategy-components/api/llmwiki"; import { getLlmWikiProtectedFileUrl } from "@/strategy-components/api/llmwiki";
type WikiProtectedImageProps = ImgHTMLAttributes<HTMLImageElement> & { type WikiProtectedImageProps = ImgHTMLAttributes<HTMLImageElement> & {
knowledgeBaseId: string; knowledgeBaseId: string;
@@ -18,24 +18,22 @@ export function WikiProtectedImage({
...props ...props
}: WikiProtectedImageProps) { }: WikiProtectedImageProps) {
const filePath = parseProtectedWikiImageSource(src); const filePath = parseProtectedWikiImageSource(src);
const [objectUrl, setObjectUrl] = useState(""); const [httpUrl, setHttpUrl] = useState("");
const [error, setError] = useState(""); const [error, setError] = useState("");
useEffect(() => { useEffect(() => {
if (!filePath || !knowledgeBaseId) { if (!filePath || !knowledgeBaseId) {
setObjectUrl(""); setHttpUrl("");
setError(""); setError("");
return; return;
} }
let cancelled = false; let cancelled = false;
let nextObjectUrl = ""; setHttpUrl("");
setObjectUrl("");
setError(""); setError("");
void getLlmWikiProtectedFile(knowledgeBaseId, filePath, publicAccess) void getLlmWikiProtectedFileUrl(knowledgeBaseId, filePath, publicAccess)
.then((blob) => { .then((url) => {
if (cancelled) return; if (cancelled) return;
nextObjectUrl = URL.createObjectURL(blob); setHttpUrl(url);
setObjectUrl(nextObjectUrl);
}) })
.catch((reason: unknown) => { .catch((reason: unknown) => {
if (!cancelled) { if (!cancelled) {
@@ -44,7 +42,6 @@ export function WikiProtectedImage({
}); });
return () => { return () => {
cancelled = true; cancelled = true;
if (nextObjectUrl) URL.revokeObjectURL(nextObjectUrl);
}; };
}, [filePath, knowledgeBaseId, publicAccess]); }, [filePath, knowledgeBaseId, publicAccess]);
@@ -64,12 +61,12 @@ export function WikiProtectedImage({
</span> </span>
); );
} }
if (!objectUrl) { if (!httpUrl) {
return ( return (
<span className="my-3 flex min-h-24 items-center justify-center rounded-lg bg-muted/30 text-muted-foreground"> <span className="my-3 flex min-h-24 items-center justify-center rounded-lg bg-muted/30 text-muted-foreground">
<Loader2 className="size-5 animate-spin" aria-label="正在加载 Wiki 图片" /> <Loader2 className="size-5 animate-spin" aria-label="正在加载 Wiki 图片" />
</span> </span>
); );
} }
return <img {...props} src={objectUrl} alt={alt} className={className} />; return <img {...props} src={httpUrl} alt={alt} className={className} />;
} }
+24 -3
View File
@@ -638,13 +638,20 @@ Two `TypeDecorator`s keep ORM columns correct across the `sqlite` / `postgres` /
### Schema Migrations / Manual Column-Add SQL (`scripts/sql/`) ### Schema Migrations / Manual Column-Add SQL (`scripts/sql/`)
Alembic auto-migration is **disabled** ? `_run_alembic_upgrade(url)` is commented out in `persistence/engine.py`. On startup only `Base.metadata.create_all` runs, which creates **new tables** but never patches **existing** ones (no new columns, no new indexes). So a column added to an ORM model appears automatically on a brand-new deployment, but is **missing on any existing MySQL database** ? and code reading/writing that column will error there until the column is added by hand. Gateway startup runs Alembic through `app.gateway.app._run_db_migrations` after ORM table creation. The migration graph must always have unique revision ids and exactly one head; `tests/test_migration_graph.py` enforces both. Historical duplicate revision `20260821_01` is repaired by the `20260821_03` / `20260821_04` fixed-question chain and merge revision `20260825_01`. Startup DDL has a bounded timeout and failures are logged without preventing the app from serving.
**Convention (required):** whenever code adds a field to an existing table (a new `mapped_column` on an ORM model), you must, in the same change: **Convention (required):** whenever code adds a field to an existing table (a new `mapped_column` on an ORM model), you must, in the same change:
1. Keep the matching alembic version file under `persistence/migrations/versions/` (guarded with `_has_table`/`_has_column`), and 1. Keep the matching alembic version file under `persistence/migrations/versions/` (guarded with `_has_table`/`_has_column`), and
2. Append the idempotent `ADD COLUMN` statement to **`scripts/sql/column_additions_mysql.sql`**, with a definition matching the ORM column exactly (type, length, nullability). 2. Append the idempotent `ADD COLUMN` statement to **`scripts/sql/column_additions_mysql.sql`**, with a definition matching the ORM column exactly (type, length, nullability).
Operators upgrading an existing DB run the whole file (`USE <db>;` then execute) ? it is idempotent (existing columns are skipped via `information_schema.columns`), so it is safe to re-run. Index-only changes go in the sibling `scripts/sql/leaderboard_indexes_mysql.sql` (same idempotent pattern). Operators upgrading an existing DB may run the whole file (`USE <db>;` then execute) as a recovery path; it is idempotent (existing columns are skipped via `information_schema.columns`). Index-only changes go in the sibling `scripts/sql/leaderboard_indexes_mysql.sql` (same idempotent pattern) and stay out of startup migrations because online index creation on large metric tables can delay service boot.
Leaderboard scheduled-run exclusion uses correlated `NOT EXISTS` against the
indexed `scheduled_task_runs.agent_run_id`; do not restore `NOT IN`. Legacy
scheduler detection for tool metrics is also a correlated existence check and
must not outer-join the complete `runs` table into metric aggregations. Keep the
created-at + dimension compound indexes in ORM metadata and the idempotent
MySQL index script synchronized.
### Sandbox System (`packages/harness/deerflow/sandbox/`) ### Sandbox System (`packages/harness/deerflow/sandbox/`)
@@ -1496,6 +1503,16 @@ Wiki directory, reconstruct hierarchy from `parent_slug`, `wiki_path`,
directory navigation on the same lightweight route rather than mounting the directory navigation on the same lightweight route rather than mounting the
WeKnora iframe. WeKnora iframe.
Generated HashRouter URLs must retain `window.location.pathname` (for example
the `/web/` reverse-proxy prefix); never emit root-absolute `/#/...` citation,
reader, or public-embed links. Protected Wiki image rendering must use real HTTP
URLs, never `URL.createObjectURL`. Public readers use the published file proxy;
authenticated readers obtain a short-lived signed capability from
`/files/access-url` and load `/api/public/llmwiki/files/{token}`. The signed
payload is restricted to `kind=wiki_file`, mapping id, remote id, validated
protected path, and expiry, and responses provide an inline UTF-8 filename for
new-tab viewing and browser Save As.
Third-party pages embed published Wiki content through the chrome-free route Third-party pages embed published Wiki content through the chrome-free route
`/#/embed/knowledge/{mapping_id}` (optional exact-page `wikiSlug`). It defaults `/#/embed/knowledge/{mapping_id}` (optional exact-page `wikiSlug`). It defaults
to a recognized Wiki index page (title `索引`/`index`, page type `index`, or an to a recognized Wiki index page (title `索引`/`index`, page type `index`, or an
@@ -1528,7 +1545,11 @@ mapping id is hard to guess.
Owners manually force a full sync/vectorization through `POST Owners manually force a full sync/vectorization through `POST
/api/llmwiki/knowledge-bases/{id}/vectorize`; status is available through `GET /api/llmwiki/knowledge-bases/{id}/vectorize`; status is available through `GET
/api/llmwiki/knowledge-bases/{id}/wiki-index-status`. The Wiki management page /api/llmwiki/knowledge-bases/{id}/wiki-index-status`. That response includes
aggregate `progress` and per-page `pages` details; current-scan progress is
derived from `active_sync_id` / `last_seen_sync_id`, while each page exposes its
index state, vector count, timestamps, and bounded error. The Wiki management
page must show the progress bar and a status badge beside every article. It
also exports all remote processed Wiki pages via `GET also exports all remote processed Wiki pages via `GET
/api/llmwiki/knowledge-bases/{id}/wiki/export.xlsx`. Directory export must keep /api/llmwiki/knowledge-bases/{id}/wiki/export.xlsx`. Directory export must keep
the full path, current directory, parent path, and dynamic per-level columns; the full path, current directory, parent path, and dynamic per-level columns;
+29 -2
View File
@@ -804,8 +804,12 @@ database leases; searches use revision-checked per-knowledge-base LRU snapshots
after the full-library readiness gate succeeds. Knowledge-base owners can click after the full-library readiness gate succeeds. Knowledge-base owners can click
**手动向量化** in Wiki management (`POST **手动向量化** in Wiki management (`POST
/api/llmwiki/knowledge-bases/{id}/vectorize`) and follow its status at /api/llmwiki/knowledge-bases/{id}/vectorize`) and follow its status at
`GET /api/llmwiki/knowledge-bases/{id}/wiki-index-status`. Existing chunk `GET /api/llmwiki/knowledge-bases/{id}/wiki-index-status`. The response includes
citations remain readable while all new retrieval citations open Wiki pages. aggregate `progress` plus a `pages` list with every Wiki page's index state,
vector-section count, timestamps, and last error. The management page renders
the percentage and labels each article as 已向量化 / 待向量化 / 失败, so it is
clear exactly what is searchable. Existing chunk citations remain readable
while all new retrieval citations open Wiki pages.
The frontend labels these results simply as **Wiki** references; provider-brand The frontend labels these results simply as **Wiki** references; provider-brand
wording is intentionally hidden from user-facing labels and status text. Clicking any wording is intentionally hidden from user-facing labels and status text. Clicking any
@@ -828,6 +832,29 @@ left. It reconstructs hierarchy from `parent_slug`, `wiki_path`,
page, and switches directory entries through the same lightweight reader route page, and switches directory entries through the same lightweight reader route
without mounting the WeKnora iframe. without mounting the WeKnora iframe.
All generated SPA deep links preserve the deployment document path (for
example `/web/`) before the `#` route, including Q&A references and Wiki index
links. Protected Wiki images no longer render from temporary browser `blob:`
URLs. Public images use the published file proxy directly; private images first
obtain a 24-hour signed capability and then load through
`GET /api/public/llmwiki/files/{token}` with a real HTTP URL and inline
filename, allowing zoom, new-tab viewing, and browser Save As without exposing
the user's bearer token.
## Leaderboard query and schema startup safety
The leaderboard excludes scheduled runs with indexed correlated `NOT EXISTS`
checks. Tool metrics no longer outer-join the full `runs` table merely to
recognize legacy scheduler messages. New databases receive dimension-aware
analytics indexes from ORM metadata; existing large MySQL deployments should
run `scripts/sql/leaderboard_indexes_mysql.sql` during a low-traffic window so
index creation never delays application startup.
Startup Alembic migrations now have one unambiguous head. The historical
duplicate `20260821_01` identifier is repaired by unique fixed-question
revisions and the `20260825_01` merge revision, which also fills the
report-structure column for databases stamped by either old variant.
Published knowledge bases also have a login-free, chrome-free reader at Published knowledge bases also have a login-free, chrome-free reader at
`/#/embed/knowledge/{mapping_id}` for direct third-party iframe use. With no `/#/embed/knowledge/{mapping_id}` for direct third-party iframe use. With no
`wikiSlug`, it selects the Wiki **索引** page first (recognized by title, page `wikiSlug`, it selects the Wiki **索引** page first (recognized by title, page
@@ -637,10 +637,16 @@ def _non_scheduled_run_filter():
def _non_scheduled_tool_filter(): def _non_scheduled_tool_filter():
return or_( """Exclude legacy scheduled runs without joining runs into the metric scan."""
ToolCallMetricRow.run_id.is_(None),
RunRow.run_id.is_(None), return ~(
_non_scheduled_run_filter(), select(1)
.select_from(RunRow)
.where(
RunRow.run_id == ToolCallMetricRow.run_id,
RunRow.first_human_message.like("【定时任务触发】%"),
)
.exists()
) )
@@ -667,8 +673,15 @@ def _admin_user_ids_stmt():
return select(UserRow.id).where(UserRow.system_role == "admin") return select(UserRow.id).where(UserRow.system_role == "admin")
def _scheduled_run_ids_stmt(): def _not_scheduled_run_id_filter(run_id_column):
return select(ScheduledTaskRunRow.agent_run_id).where(ScheduledTaskRunRow.agent_run_id.is_not(None)) """Use a NULL-safe, index-friendly anti-semi join instead of NOT IN."""
return ~(
select(1)
.select_from(ScheduledTaskRunRow)
.where(ScheduledTaskRunRow.agent_run_id == run_id_column)
.exists()
)
def _apply_user_scope(stmt, column, *, exclude_user_ids: set[str], include_admins: bool): def _apply_user_scope(stmt, column, *, exclude_user_ids: set[str], include_admins: bool):
@@ -692,7 +705,7 @@ def _apply_run_scope(
if not include_failed: if not include_failed:
stmt = stmt.where(RunRow.status != "error") stmt = stmt.where(RunRow.status != "error")
if not include_scheduled: if not include_scheduled:
stmt = stmt.where(or_(run_id_column.is_(None), run_id_column.not_in(_scheduled_run_ids_stmt()))) stmt = stmt.where(_not_scheduled_run_id_filter(run_id_column))
stmt = stmt.where(_non_scheduled_run_filter()) stmt = stmt.where(_non_scheduled_run_filter())
return stmt return stmt
@@ -706,8 +719,8 @@ def _apply_tool_scope(
): ):
stmt = _apply_user_scope(stmt, ToolCallMetricRow.user_id, exclude_user_ids=exclude_user_ids, include_admins=include_admins) stmt = _apply_user_scope(stmt, ToolCallMetricRow.user_id, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
if not include_scheduled: if not include_scheduled:
stmt = stmt.where(or_(ToolCallMetricRow.run_id.is_(None), ToolCallMetricRow.run_id.not_in(_scheduled_run_ids_stmt()))) stmt = stmt.where(_not_scheduled_run_id_filter(ToolCallMetricRow.run_id))
stmt = stmt.outerjoin(RunRow, RunRow.run_id == ToolCallMetricRow.run_id).where(_non_scheduled_tool_filter()) stmt = stmt.where(_non_scheduled_tool_filter())
return stmt return stmt
@@ -985,7 +998,7 @@ async def _query_model_rankings(
if not include_failed: if not include_failed:
stmt = stmt.where(LlmCallMetricRow.status != "error") stmt = stmt.where(LlmCallMetricRow.status != "error")
if not include_scheduled: if not include_scheduled:
stmt = stmt.where(or_(LlmCallMetricRow.run_id.is_(None), LlmCallMetricRow.run_id.not_in(_scheduled_run_ids_stmt()))) stmt = stmt.where(_not_scheduled_run_id_filter(LlmCallMetricRow.run_id))
rows = (await _execute_leaderboard_query(session, "model_rankings", stmt)).all() rows = (await _execute_leaderboard_query(session, "model_rankings", stmt)).all()
return [ return [
AdminLeaderboardDimension( AdminLeaderboardDimension(
@@ -1080,8 +1093,8 @@ async def _query_overview(
if not include_failed: if not include_failed:
run_where.append(RunRow.status != "error") run_where.append(RunRow.status != "error")
if not include_scheduled: if not include_scheduled:
run_where.append(or_(RunRow.run_id.is_(None), RunRow.run_id.not_in(_scheduled_run_ids_stmt()))) run_where.append(_not_scheduled_run_id_filter(RunRow.run_id))
tool_where.append(or_(ToolCallMetricRow.run_id.is_(None), ToolCallMetricRow.run_id.not_in(_scheduled_run_ids_stmt()))) tool_where.append(_not_scheduled_run_id_filter(ToolCallMetricRow.run_id))
tool_where.append(_non_scheduled_tool_filter()) tool_where.append(_non_scheduled_tool_filter())
total_users_stmt = select(func.count(UserRow.id)) total_users_stmt = select(func.count(UserRow.id))
@@ -1104,7 +1117,6 @@ async def _query_overview(
func.count(ToolCallMetricRow.id), func.count(ToolCallMetricRow.id),
func.sum(case((func.nullif(ToolCallMetricRow.skill_name, "").is_not(None), 1), else_=0)), 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) .where(*tool_where)
) )
tool_row = (await _execute_leaderboard_query(session, "overview.tool_summary", tool_summary_stmt)).one() tool_row = (await _execute_leaderboard_query(session, "overview.tool_summary", tool_summary_stmt)).one()
@@ -6,6 +6,7 @@ import asyncio
import hashlib import hashlib
import json import json
import logging import logging
import mimetypes
import re import re
import time import time
from dataclasses import replace from dataclasses import replace
@@ -47,6 +48,9 @@ from app.gateway.weknora_embed import (
from app.gateway.weknora_embed import ( from app.gateway.weknora_embed import (
verify_weknora_embed_cookie as _verify_embed_cookie, verify_weknora_embed_cookie as _verify_embed_cookie,
) )
from app.gateway.weknora_embed import (
verify_weknora_signed_payload as _verify_signed_payload,
)
from deerflow.config.agents_config import load_agent_config from deerflow.config.agents_config import load_agent_config
from deerflow.config.llmwiki_config import LlmWikiRuntimeOverride, normalize_service_url from deerflow.config.llmwiki_config import LlmWikiRuntimeOverride, normalize_service_url
from deerflow.config.system_settings import load_system_settings, save_system_settings from deerflow.config.system_settings import load_system_settings, save_system_settings
@@ -61,6 +65,7 @@ router = APIRouter(tags=["llmwiki"])
proxy_router = APIRouter(tags=["llmwiki-proxy"]) proxy_router = APIRouter(tags=["llmwiki-proxy"])
_MAX_UPLOAD_BYTES = 100 * 1024 * 1024 _MAX_UPLOAD_BYTES = 100 * 1024 * 1024
_WIKI_FILE_ACCESS_TTL_SECONDS = 24 * 60 * 60
_PROTECTED_WIKI_FILE_PATH = re.compile( _PROTECTED_WIKI_FILE_PATH = re.compile(
r"^(?:storage://[0-9A-Za-z_-]+/)?(?:resource|local|minio|cos|tos|s3|oss|ks3|obs)://", r"^(?:storage://[0-9A-Za-z_-]+/)?(?:resource|local|minio|cos|tos|s3|oss|ks3|obs)://",
re.IGNORECASE, re.IGNORECASE,
@@ -1909,19 +1914,23 @@ def _validate_protected_wiki_file_path(file_path: str) -> str:
async def _wiki_file_response(row: dict[str, Any], file_path: str, *, public: bool) -> Response: async def _wiki_file_response(row: dict[str, Any], file_path: str, *, public: bool) -> Response:
safe_file_path = _validate_protected_wiki_file_path(file_path)
try: try:
content, media_type = await _client_or_503().get_knowledge_base_file( content, media_type = await _client_or_503().get_knowledge_base_file(
str(row["weknora_id"]), str(row["weknora_id"]),
_validate_protected_wiki_file_path(file_path), safe_file_path,
) )
except WeKnoraError as exc: except WeKnoraError as exc:
_raise_upstream(exc) _raise_upstream(exc)
filename = unquote(safe_file_path.rstrip("/").rsplit("/", 1)[-1]).replace("\r", "").replace("\n", "") or "wiki-file"
if "." not in filename:
filename += mimetypes.guess_extension(str(media_type).split(";", 1)[0].strip()) or ""
return Response( return Response(
content=content, content=content,
media_type=media_type, media_type=media_type,
headers={ headers={
"Cache-Control": "public, max-age=300" if public else "private, max-age=60", "Cache-Control": "public, max-age=300" if public else "private, max-age=3600",
"Content-Disposition": "inline", "Content-Disposition": f"inline; filename*=UTF-8''{quote(filename, safe='')}",
"X-Content-Type-Options": "nosniff", "X-Content-Type-Options": "nosniff",
}, },
) )
@@ -2629,6 +2638,25 @@ async def get_public_wiki_file(
return await _wiki_file_response(row, file_path, public=True) return await _wiki_file_response(row, file_path, public=True)
@router.get("/api/public/llmwiki/files/{token}")
async def get_signed_wiki_file(token: str) -> Response:
"""Serve a private Wiki image through a short-lived, signed HTTP URL."""
payload = _verify_signed_payload(token)
if payload.get("kind") != "wiki_file":
raise HTTPException(status_code=401, detail="Signed Wiki file token is invalid")
mapping_id = str(payload.get("mapping_id") or "")
weknora_id = str(payload.get("weknora_id") or "")
file_path = str(payload.get("file_path") or "")
if not mapping_id or not weknora_id or not file_path:
raise HTTPException(status_code=401, detail="Signed Wiki file token is invalid")
return await _wiki_file_response(
{"id": mapping_id, "weknora_id": weknora_id},
file_path,
public=False,
)
_PUBLIC_WIKI_INDEX_TYPES = ("summary", "entity", "concept", "synthesis", "comparison") _PUBLIC_WIKI_INDEX_TYPES = ("summary", "entity", "concept", "synthesis", "comparison")
@@ -2951,6 +2979,28 @@ async def get_wiki_file(
return await _wiki_file_response(row, file_path, public=False) return await _wiki_file_response(row, file_path, public=False)
@router.get("/api/llmwiki/knowledge-bases/{mapping_id}/files/access-url")
async def create_wiki_file_access(
request: Request,
mapping_id: str,
file_path: str = Query(..., min_length=1, max_length=8192),
) -> dict[str, Any]:
"""Issue a browser-usable HTTP capability after normal KB authorization."""
row, _, _ = await _authorized_mapping(request, mapping_id, write=False)
safe_file_path = _validate_protected_wiki_file_path(file_path)
token = _sign_embed_payload(
{
"kind": "wiki_file",
"mapping_id": mapping_id,
"weknora_id": str(row["weknora_id"]),
"file_path": safe_file_path,
"exp": int(time.time()) + _WIKI_FILE_ACCESS_TTL_SECONDS,
}
)
return {"token": token, "expires_in": _WIKI_FILE_ACCESS_TTL_SECONDS}
@router.get("/api/llmwiki/knowledge-bases/{mapping_id}/graph") @router.get("/api/llmwiki/knowledge-bases/{mapping_id}/graph")
async def get_knowledge_base_graph( async def get_knowledge_base_graph(
request: Request, request: Request,
@@ -245,6 +245,33 @@ async def knowledge_base_wiki_index_status(request: Request, mapping_id: str) ->
raise HTTPException(status_code=503, detail={"code": "WIKI_INDEX_UNAVAILABLE"}) raise HTTPException(status_code=503, detail={"code": "WIKI_INDEX_UNAVAILABLE"})
states = await index_store.list_index_status([mapping_id]) states = await index_store.list_index_status([mapping_id])
state = states[0] if states else {"knowledge_base_mapping_id": mapping_id, "state": "not_synced", "index_revision": 0} state = states[0] if states else {"knowledge_base_mapping_id": mapping_id, "state": "not_synced", "index_revision": 0}
if (
state.get("state") == "idle"
and int(state.get("index_revision") or 0) == 0
and not state.get("last_completed_at")
and not state.get("last_success_at")
):
state = {**state, "state": "not_synced"}
pages = await index_store.list_page_index_status(mapping_id)
active_sync_id = str(state.get("active_sync_id") or "")
is_running = str(state.get("state") or "") in {"running", "rebuilding"}
for page in pages:
page["vectorized"] = page.get("index_status") == "ready" and int(page.get("vector_count") or 0) > 0
page["processed_in_current_sync"] = bool(active_sync_id) and page.get("last_seen_sync_id") == active_sync_id
total_pages = max(int(state.get("remote_page_count") or 0), len(pages))
processed_pages = (
sum(bool(page["processed_in_current_sync"]) for page in pages)
if is_running
else total_pages if str(state.get("state") or "") in {"idle", "partial"} else 0
)
completion_percent = (
min(100, round((processed_pages / total_pages) * 100))
if total_pages
else 100
if str(state.get("state") or "") in {"idle", "partial"}
and bool(state.get("last_completed_at") or state.get("last_success_at"))
else 0
)
embedding = getattr(request.app.state, "llmwiki_embedding", None) embedding = getattr(request.app.state, "llmwiki_embedding", None)
fingerprint = embedding.fingerprint if embedding is not None else None fingerprint = embedding.fingerprint if embedding is not None else None
fully_vectorized = bool(mapping.get("wiki_index_enabled", True)) and is_fully_vectorized(state, fingerprint) fully_vectorized = bool(mapping.get("wiki_index_enabled", True)) and is_fully_vectorized(state, fingerprint)
@@ -255,6 +282,15 @@ async def knowledge_base_wiki_index_status(request: Request, mapping_id: str) ->
"wiki_index_enabled": mapping.get("wiki_index_enabled", True), "wiki_index_enabled": mapping.get("wiki_index_enabled", True),
"fully_vectorized": fully_vectorized, "fully_vectorized": fully_vectorized,
"retrieval_mode": "vector" if fully_vectorized else "wiki_api", "retrieval_mode": "vector" if fully_vectorized else "wiki_api",
"progress": {
"total_pages": total_pages,
"processed_pages": processed_pages,
"vectorized_pages": sum(bool(page["vectorized"]) for page in pages),
"failed_pages": sum(bool(page.get("last_error")) or page.get("index_status") in {"failed", "stale"} for page in pages),
"vector_count": sum(int(page.get("vector_count") or 0) for page in pages),
"completion_percent": completion_percent,
},
"pages": pages,
} }
@@ -106,6 +106,26 @@ def sign_weknora_embed_payload(payload: dict[str, Any]) -> str:
return f"{body}.{_b64url_encode(sig)}" return f"{body}.{_b64url_encode(sig)}"
def verify_weknora_signed_payload(raw: str | None) -> dict[str, Any]:
"""Verify a short-lived capability signed with the iframe proxy secret."""
if not raw or "." not in raw:
raise HTTPException(status_code=401, detail="Signed resource token is missing")
body, sig = raw.rsplit(".", 1)
expected = _b64url_encode(hmac.new(WEKNORA_PROXY_SECRET.encode("utf-8"), body.encode("ascii"), hashlib.sha256).digest())
if not hmac.compare_digest(sig, expected):
raise HTTPException(status_code=401, detail="Signed resource token is invalid")
try:
payload = json.loads(_b64url_decode(body))
except (ValueError, TypeError, json.JSONDecodeError):
raise HTTPException(status_code=401, detail="Signed resource token is invalid") from None
if not isinstance(payload, dict):
raise HTTPException(status_code=401, detail="Signed resource token is invalid")
if int(payload.get("exp") or 0) < int(time.time()):
raise HTTPException(status_code=401, detail="Signed resource token has expired")
return payload
def verify_weknora_embed_cookie(raw: str | None) -> WeKnoraEmbedContext: def verify_weknora_embed_cookie(raw: str | None) -> WeKnoraEmbedContext:
if not raw or "." not in raw: if not raw or "." not in raw:
raise HTTPException(status_code=401, detail="WeKnora iframe session is missing") raise HTTPException(status_code=401, detail="WeKnora iframe session is missing")
@@ -90,6 +90,13 @@ class WikiSyncService:
batch = await self.client.list_wiki_pages(remote_id, page=page_number, page_size=self.config.sync_page_size) batch = await self.client.list_wiki_pages(remote_id, page=page_number, page_size=self.config.sync_page_size)
pages = [page for page in batch.get("pages", []) if isinstance(page, dict)] pages = [page for page in batch.get("pages", []) if isinstance(page, dict)]
remote_count += len(pages) remote_count += len(pages)
reported_total = int(batch.get("total") or batch.get("total_count") or remote_count)
await self.store.update_sync_progress(
mapping_id,
owner=self.owner,
sync_id=sync_id,
remote_page_count=max(remote_count, reported_total),
)
for summary in pages: for summary in pages:
try: try:
_ensure_lease() _ensure_lease()
@@ -18,6 +18,7 @@ class LlmCallMetricRow(Base):
__table_args__ = ( __table_args__ = (
Index("ix_llm_call_metrics_created_user", "created_at", "user_id"), Index("ix_llm_call_metrics_created_user", "created_at", "user_id"),
Index("ix_llm_call_metrics_run_created", "run_id", "created_at"), Index("ix_llm_call_metrics_run_created", "run_id", "created_at"),
Index("ix_llm_call_metrics_created_model_user", "created_at", "model_name", "user_id"),
) )
id: Mapped[str] = mapped_column(String(64), primary_key=True) id: Mapped[str] = mapped_column(String(64), primary_key=True)
@@ -44,11 +44,24 @@ class LlmWikiIndexStore(ABC):
@abstractmethod @abstractmethod
async def renew_sync_lease(self, mapping_id: str, *, owner: str, lease_seconds: int) -> bool: ... async def renew_sync_lease(self, mapping_id: str, *, owner: str, lease_seconds: int) -> bool: ...
@abstractmethod
async def update_sync_progress(
self,
mapping_id: str,
*,
owner: str,
sync_id: str,
remote_page_count: int,
) -> bool: ...
@abstractmethod @abstractmethod
async def finish_sync(self, mapping_id: str, *, owner: str, sync_id: str, success: bool, remote_page_count: int, error: str | None = None) -> None: ... async def finish_sync(self, mapping_id: str, *, owner: str, sync_id: str, success: bool, remote_page_count: int, error: str | None = None) -> None: ...
@abstractmethod @abstractmethod
async def list_index_status(self, mapping_ids: list[str] | None = None) -> list[dict[str, Any]]: ... async def list_index_status(self, mapping_ids: list[str] | None = None) -> list[dict[str, Any]]: ...
@abstractmethod
async def list_page_index_status(self, mapping_id: str) -> list[dict[str, Any]]: ...
@abstractmethod @abstractmethod
async def mark_rebuild_required(self, mapping_ids: list[str], fingerprint: str) -> int: ... async def mark_rebuild_required(self, mapping_ids: list[str], fingerprint: str) -> int: ...
@@ -123,6 +123,21 @@ class MemoryLlmWikiIndexStore(LlmWikiIndexStore):
state["lease_expires_at"] = _now() + timedelta(seconds=lease_seconds) state["lease_expires_at"] = _now() + timedelta(seconds=lease_seconds)
return True return True
async def update_sync_progress(
self,
mapping_id: str,
*,
owner: str,
sync_id: str,
remote_page_count: int,
) -> bool:
state = self._state(mapping_id)
if state.get("lease_owner") != owner or state.get("active_sync_id") != sync_id:
return False
state["remote_page_count"] = max(0, remote_page_count)
state["updated_at"] = _now()
return True
async def finish_sync(self, mapping_id: str, *, owner: str, sync_id: str, success: bool, remote_page_count: int, error: str | None = None) -> None: async def finish_sync(self, mapping_id: str, *, owner: str, sync_id: str, success: bool, remote_page_count: int, error: str | None = None) -> None:
state = self._state(mapping_id) state = self._state(mapping_id)
if state.get("lease_owner") != owner or state.get("active_sync_id") != sync_id: if state.get("lease_owner") != owner or state.get("active_sync_id") != sync_id:
@@ -150,6 +165,33 @@ class MemoryLlmWikiIndexStore(LlmWikiIndexStore):
wanted = set(mapping_ids or []) wanted = set(mapping_ids or [])
return [dict(value) for key, value in self.states.items() if not wanted or key in wanted] return [dict(value) for key, value in self.states.items() if not wanted or key in wanted]
async def list_page_index_status(self, mapping_id: str) -> list[dict[str, Any]]:
result = []
for page in sorted(
(
page
for page in self.pages.values()
if page["knowledge_base_mapping_id"] == mapping_id and not page.get("is_remote_deleted")
),
key=lambda page: (str(page.get("wiki_path") or ""), str(page.get("slug") or "")),
):
result.append(
{
"id": page["id"],
"slug": page["slug"],
"title": page["title"],
"page_type": page["page_type"],
"status": page["status"],
"index_status": page["index_status"],
"vector_count": len(self.vectors.get(page["id"], [])),
"indexed_at": page.get("indexed_at"),
"last_error": page.get("last_error"),
"last_seen_sync_id": page.get("last_seen_sync_id"),
"updated_at": page.get("updated_at"),
}
)
return result
async def mark_rebuild_required(self, mapping_ids: list[str], fingerprint: str) -> int: async def mark_rebuild_required(self, mapping_ids: list[str], fingerprint: str) -> int:
for mapping_id in mapping_ids: for mapping_id in mapping_ids:
state = self._state(mapping_id) state = self._state(mapping_id)
@@ -235,6 +235,27 @@ class SqlLlmWikiIndexStore(LlmWikiIndexStore):
await session.commit() await session.commit()
return bool(result.rowcount) return bool(result.rowcount)
async def update_sync_progress(
self,
mapping_id: str,
*,
owner: str,
sync_id: str,
remote_page_count: int,
) -> bool:
async with self._sf() as session:
result = await session.execute(
update(LlmWikiWikiSyncStateRow)
.where(
LlmWikiWikiSyncStateRow.knowledge_base_mapping_id == mapping_id,
LlmWikiWikiSyncStateRow.lease_owner == owner,
LlmWikiWikiSyncStateRow.active_sync_id == sync_id,
)
.values(remote_page_count=max(0, remote_page_count), updated_at=_now())
)
await session.commit()
return bool(result.rowcount)
async def finish_sync(self, mapping_id: str, *, owner: str, sync_id: str, success: bool, remote_page_count: int, error: str | None = None) -> None: async def finish_sync(self, mapping_id: str, *, owner: str, sync_id: str, success: bool, remote_page_count: int, error: str | None = None) -> None:
async with self._sf() as session: async with self._sf() as session:
state = await session.get(LlmWikiWikiSyncStateRow, mapping_id) state = await session.get(LlmWikiWikiSyncStateRow, mapping_id)
@@ -286,6 +307,47 @@ class SqlLlmWikiIndexStore(LlmWikiIndexStore):
async with self._sf() as session: async with self._sf() as session:
return [_dict(row) for row in (await session.execute(stmt)).scalars()] return [_dict(row) for row in (await session.execute(stmt)).scalars()]
async def list_page_index_status(self, mapping_id: str) -> list[dict[str, Any]]:
vector_counts = (
select(
LlmWikiWikiVectorRow.page_id.label("page_id"),
func.count(LlmWikiWikiVectorRow.id).label("vector_count"),
)
.where(LlmWikiWikiVectorRow.knowledge_base_mapping_id == mapping_id)
.group_by(LlmWikiWikiVectorRow.page_id)
.subquery()
)
stmt = (
select(LlmWikiWikiPageRow, func.coalesce(vector_counts.c.vector_count, 0))
.outerjoin(vector_counts, vector_counts.c.page_id == LlmWikiWikiPageRow.id)
.where(
LlmWikiWikiPageRow.knowledge_base_mapping_id == mapping_id,
LlmWikiWikiPageRow.is_remote_deleted.is_(False),
)
.order_by(LlmWikiWikiPageRow.wiki_path, LlmWikiWikiPageRow.slug)
)
async with self._sf() as session:
rows = (await session.execute(stmt)).all()
result: list[dict[str, Any]] = []
for row, vector_count in rows:
page = _dict(row)
result.append(
{
"id": page["id"],
"slug": page["slug"],
"title": page["title"],
"page_type": page["page_type"],
"status": page["status"],
"index_status": page["index_status"],
"vector_count": int(vector_count or 0),
"indexed_at": page.get("indexed_at"),
"last_error": page.get("last_error"),
"last_seen_sync_id": page.get("last_seen_sync_id"),
"updated_at": page.get("updated_at"),
}
)
return result
async def mark_rebuild_required(self, mapping_ids: list[str], fingerprint: str) -> int: async def mark_rebuild_required(self, mapping_ids: list[str], fingerprint: str) -> int:
if not mapping_ids: if not mapping_ids:
return 0 return 0
@@ -1,7 +1,7 @@
"""Create administrator-managed exact-match fixed questions. """Create administrator-managed exact-match fixed questions.
Revision ID: 20260821_01 Revision ID: 20260821_03
Revises: 20260816_01 Revises: 20260821_01
Create Date: 2026-08-21 Create Date: 2026-08-21
""" """
@@ -11,8 +11,8 @@ import sqlalchemy as sa
from alembic import op from alembic import op
from sqlalchemy.dialects import mysql from sqlalchemy.dialects import mysql
revision = "20260821_01" revision = "20260821_03"
down_revision = "20260816_01" down_revision = "20260821_01"
branch_labels = None branch_labels = None
depends_on = None depends_on = None
@@ -1,7 +1,7 @@
"""Add configurable fixed-answer streaming speed. """Add configurable fixed-answer streaming speed.
Revision ID: 20260821_02 Revision ID: 20260821_04
Revises: 20260821_01 Revises: 20260821_03
Create Date: 2026-08-21 Create Date: 2026-08-21
""" """
@@ -10,8 +10,8 @@ from __future__ import annotations
import sqlalchemy as sa import sqlalchemy as sa
from alembic import op from alembic import op
revision = "20260821_02" revision = "20260821_04"
down_revision = "20260821_01" down_revision = "20260821_03"
branch_labels = None branch_labels = None
depends_on = None depends_on = None
@@ -1,7 +1,7 @@
"""Add DeerFlow-local WeKnora Wiki mirror and vector index. """Add DeerFlow-local WeKnora Wiki mirror and vector index.
Revision ID: 20260823_01 Revision ID: 20260823_01
Revises: 20260821_02 Revises: 20260821_04
Create Date: 2026-08-23 Create Date: 2026-08-23
""" """
@@ -12,7 +12,7 @@ from alembic import op
from sqlalchemy.dialects import mysql from sqlalchemy.dialects import mysql
revision = "20260823_01" revision = "20260823_01"
down_revision = "20260821_02" down_revision = "20260821_04"
branch_labels = None branch_labels = None
depends_on = None depends_on = None
@@ -0,0 +1,34 @@
"""Merge report-structure and local-Wiki heads, repairing the old duplicate stamp.
Revision ID: 20260825_01
Revises: 20260822_02, 20260823_01
Create Date: 2026-08-25
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
revision = "20260825_01"
down_revision = ("20260822_02", "20260823_01")
branch_labels = None
depends_on = None
def upgrade() -> None:
# An older build accidentally used 20260821_01 for two migrations. A DB
# stamped by the fixed-question variant may therefore still lack this
# report column even though Alembic considers the surviving 01 applied.
inspector = sa.inspect(op.get_bind())
if inspector.has_table("report_structures"):
columns = {column["name"] for column in inspector.get_columns("report_structures")}
if "structure_mode" not in columns:
op.add_column(
"report_structures",
sa.Column("structure_mode", sa.String(length=16), nullable=False, server_default="adaptive"),
)
def downgrade() -> None:
pass
@@ -52,4 +52,6 @@ class RunRow(Base):
Index("ix_runs_created_user_status", "created_at", "user_id", "status"), Index("ix_runs_created_user_status", "created_at", "user_id", "status"),
Index("ix_runs_user_created", "user_id", "created_at"), Index("ix_runs_user_created", "user_id", "created_at"),
Index("ix_runs_thread_created", "thread_id", "created_at"), Index("ix_runs_thread_created", "thread_id", "created_at"),
Index("ix_runs_created_assistant_user", "created_at", "assistant_id", "user_id"),
Index("ix_runs_created_model_user", "created_at", "model_name", "user_id"),
) )
@@ -24,6 +24,8 @@ class ToolCallMetricRow(Base):
__table_args__ = ( __table_args__ = (
Index("ix_tool_call_metrics_created_user", "created_at", "user_id"), Index("ix_tool_call_metrics_created_user", "created_at", "user_id"),
Index("ix_tool_call_metrics_run_created", "run_id", "created_at"), Index("ix_tool_call_metrics_run_created", "run_id", "created_at"),
Index("ix_tool_call_metrics_created_skill_user", "created_at", "skill_name", "user_id"),
Index("ix_tool_call_metrics_created_tool_user", "created_at", "tool_name", "user_id"),
) )
id: Mapped[str] = mapped_column(String(64), primary_key=True) id: Mapped[str] = mapped_column(String(64), primary_key=True)
@@ -4,9 +4,9 @@
-- --
-- 背景: -- 背景:
-- 这些索引声明在各 ORM model 的 __table_args__ 里,但 Base.metadata.create_all -- 这些索引声明在各 ORM model 的 __table_args__ 里,但 Base.metadata.create_all
-- 对【已存在的表】不会补建索引,而 alembic 自动迁移在本项目里是关闭的 -- 对【已存在的表】不会补建索引。本项目虽会在启动时运行 Alembic,但大表
-- (engine.py 中 `# await _run_alembic_upgrade(url)` 被注释)。 -- 补索引刻意不放进启动迁移,避免在线建索引拖长或阻塞服务启动。
-- 因此在【已有数据的 MySQL 部署】上,这些索引很可能并未真正创建,排行榜的 -- 因此在【已有数据的 MySQL 部署】上需要低峰期执行本脚本,否则排行榜的
-- 单日聚合查询会退化为全表扫描,拖慢并占用连接。 -- 单日聚合查询会退化为全表扫描,拖慢并占用连接。
-- --
-- 用法: -- 用法:
@@ -55,19 +55,28 @@ DELIMITER ;
CALL _create_index_if_missing('runs', 'ix_runs_created_user_status', '`created_at`, `user_id`, `status`'); CALL _create_index_if_missing('runs', 'ix_runs_created_user_status', '`created_at`, `user_id`, `status`');
CALL _create_index_if_missing('runs', 'ix_runs_user_created', '`user_id`, `created_at`'); CALL _create_index_if_missing('runs', 'ix_runs_user_created', '`user_id`, `created_at`');
CALL _create_index_if_missing('runs', 'ix_runs_thread_created', '`thread_id`, `created_at`'); CALL _create_index_if_missing('runs', 'ix_runs_thread_created', '`thread_id`, `created_at`');
CALL _create_index_if_missing('runs', 'ix_runs_created_assistant_user', '`created_at`, `assistant_id`, `user_id`');
CALL _create_index_if_missing('runs', 'ix_runs_created_model_user', '`created_at`, `model_name`, `user_id`');
-- ---- tool_call_metrics ------------------------------------------------------- -- ---- tool_call_metrics -------------------------------------------------------
CALL _create_index_if_missing('tool_call_metrics', 'ix_tool_call_metrics_created_user', '`created_at`, `user_id`'); CALL _create_index_if_missing('tool_call_metrics', 'ix_tool_call_metrics_created_user', '`created_at`, `user_id`');
CALL _create_index_if_missing('tool_call_metrics', 'ix_tool_call_metrics_run_created', '`run_id`, `created_at`'); CALL _create_index_if_missing('tool_call_metrics', 'ix_tool_call_metrics_run_created', '`run_id`, `created_at`');
CALL _create_index_if_missing('tool_call_metrics', 'ix_tool_call_metrics_created_skill_user', '`created_at`, `skill_name`, `user_id`');
CALL _create_index_if_missing('tool_call_metrics', 'ix_tool_call_metrics_created_tool_user', '`created_at`, `tool_name`, `user_id`');
-- ---- llm_call_metrics -------------------------------------------------------- -- ---- llm_call_metrics --------------------------------------------------------
CALL _create_index_if_missing('llm_call_metrics', 'ix_llm_call_metrics_created_user', '`created_at`, `user_id`'); CALL _create_index_if_missing('llm_call_metrics', 'ix_llm_call_metrics_created_user', '`created_at`, `user_id`');
CALL _create_index_if_missing('llm_call_metrics', 'ix_llm_call_metrics_run_created', '`run_id`, `created_at`'); CALL _create_index_if_missing('llm_call_metrics', 'ix_llm_call_metrics_run_created', '`run_id`, `created_at`');
CALL _create_index_if_missing('llm_call_metrics', 'ix_llm_call_metrics_created_model_user', '`created_at`, `model_name`, `user_id`');
-- ---- threads_meta ------------------------------------------------------------ -- ---- threads_meta ------------------------------------------------------------
CALL _create_index_if_missing('threads_meta', 'ix_threads_meta_created_user', '`created_at`, `user_id`'); CALL _create_index_if_missing('threads_meta', 'ix_threads_meta_created_user', '`created_at`, `user_id`');
CALL _create_index_if_missing('threads_meta', 'ix_threads_meta_user_created', '`user_id`, `created_at`'); CALL _create_index_if_missing('threads_meta', 'ix_threads_meta_user_created', '`user_id`, `created_at`');
-- ---- scheduled_task_runs -----------------------------------------------------
-- NOT EXISTS anti-join used when the leaderboard excludes scheduler runs.
CALL _create_index_if_missing('scheduled_task_runs', 'ix_scheduled_task_runs_agent_run_id', '`agent_run_id`');
-- ---- position_roundtable_sessions -------------------------------------------- -- ---- position_roundtable_sessions --------------------------------------------
-- Explicit taskId history is loaded by taskId and newest update time. -- Explicit taskId history is loaded by taskId and newest update time.
CALL _create_index_if_missing('position_roundtable_sessions', 'ix_position_roundtable_sessions_task_updated', '`external_task_id`, `updated_at`'); CALL _create_index_if_missing('position_roundtable_sessions', 'ix_position_roundtable_sessions_task_updated', '`external_task_id`, `updated_at`');
@@ -75,15 +84,19 @@ CALL _create_index_if_missing('position_roundtable_sessions', 'ix_position_round
-- ---- 清理临时存储过程 -------------------------------------------------------- -- ---- 清理临时存储过程 --------------------------------------------------------
DROP PROCEDURE IF EXISTS _create_index_if_missing; DROP PROCEDURE IF EXISTS _create_index_if_missing;
-- ---- 执行后自检:确认 9 个索引都已存在 --------------------------------------- -- ---- 执行后自检:确认排行榜索引都已存在 ----------------------------------------
SELECT table_name, index_name SELECT table_name, index_name
FROM information_schema.statistics FROM information_schema.statistics
WHERE table_schema = DATABASE() WHERE table_schema = DATABASE()
AND index_name IN ( AND index_name IN (
'ix_runs_created_user_status', 'ix_runs_user_created', 'ix_runs_thread_created', 'ix_runs_created_user_status', 'ix_runs_user_created', 'ix_runs_thread_created',
'ix_runs_created_assistant_user', 'ix_runs_created_model_user',
'ix_tool_call_metrics_created_user', 'ix_tool_call_metrics_run_created', 'ix_tool_call_metrics_created_user', 'ix_tool_call_metrics_run_created',
'ix_tool_call_metrics_created_skill_user', 'ix_tool_call_metrics_created_tool_user',
'ix_llm_call_metrics_created_user', 'ix_llm_call_metrics_run_created', 'ix_llm_call_metrics_created_user', 'ix_llm_call_metrics_run_created',
'ix_threads_meta_created_user', 'ix_threads_meta_user_created' 'ix_llm_call_metrics_created_model_user',
'ix_threads_meta_created_user', 'ix_threads_meta_user_created',
'ix_scheduled_task_runs_agent_run_id'
) )
GROUP BY table_name, index_name GROUP BY table_name, index_name
ORDER BY table_name, index_name; ORDER BY table_name, index_name;
@@ -0,0 +1,43 @@
from __future__ import annotations
from sqlalchemy import select
from sqlalchemy.dialects import mysql
from app.gateway.routers.admin_users import _apply_run_scope, _apply_tool_scope
from deerflow.persistence.run.model import RunRow
from deerflow.persistence.tool_metrics.model import ToolCallMetricRow
def _mysql_sql(stmt) -> str:
return str(stmt.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True})).upper()
def test_excluding_scheduled_runs_uses_index_friendly_not_exists() -> None:
stmt = _apply_run_scope(
select(RunRow.run_id),
user_column=RunRow.user_id,
run_id_column=RunRow.run_id,
exclude_user_ids=set(),
include_admins=True,
include_failed=True,
include_scheduled=False,
)
sql = _mysql_sql(stmt)
assert "NOT (EXISTS" in sql
assert "SCHEDULED_TASK_RUNS" in sql
assert " NOT IN " not in sql
def test_tool_aggregation_does_not_join_the_full_runs_table() -> None:
stmt = _apply_tool_scope(
select(ToolCallMetricRow.tool_name),
exclude_user_ids=set(),
include_admins=True,
include_scheduled=False,
)
sql = _mysql_sql(stmt)
assert sql.count("NOT (EXISTS") == 2
assert "LEFT OUTER JOIN" not in sql
assert " NOT IN " not in sql
@@ -66,6 +66,22 @@ async def test_sql_store_atomically_replaces_vectors_and_marks_missing_deleted(t
assert store.write_batch_size == 1 assert store.write_batch_size == 1
assert len(snapshot["rows"]) == 3 assert len(snapshot["rows"]) == 3
assert decode_vector(snapshot["rows"][0]["vector_blob"], 2).tolist() == pytest.approx([0.6, 0.8]) assert decode_vector(snapshot["rows"][0]["vector_blob"], 2).tolist() == pytest.approx([0.6, 0.8])
page_statuses = await store.list_page_index_status("mapping-1")
assert len(page_statuses) == 1
assert page_statuses[0] == {
**page_statuses[0],
"id": saved["id"],
"slug": "people/alice",
"title": "Alice",
"page_type": "article",
"status": "published",
"index_status": "ready",
"vector_count": 3,
"last_error": None,
"last_seen_sync_id": "sync-1",
}
assert page_statuses[0]["indexed_at"]
assert page_statuses[0]["updated_at"]
assert await store.mark_missing_pages_deleted("mapping-1", "sync-2") == 1 assert await store.mark_missing_pages_deleted("mapping-1", "sync-2") == 1
assert (await store.load_vector_snapshot("mapping-1", "fp"))["rows"] == [] assert (await store.load_vector_snapshot("mapping-1", "fp"))["rows"] == []
@@ -54,7 +54,11 @@ class FakeWeKnora:
return {"id": knowledge_base_id, "indexing_strategy": {"wiki_enabled": True}} return {"id": knowledge_base_id, "indexing_strategy": {"wiki_enabled": True}}
async def list_wiki_pages(self, knowledge_base_id: str, *, page: int, page_size: int): async def list_wiki_pages(self, knowledge_base_id: str, *, page: int, page_size: int):
return {"pages": [{"slug": item["slug"], "title": item["title"]} for item in self.pages.values()], "total_pages": 1} return {
"pages": [{"slug": item["slug"], "title": item["title"]} for item in self.pages.values()],
"total": len(self.pages),
"total_pages": 1,
}
async def get_wiki_page(self, knowledge_base_id: str, slug: str): async def get_wiki_page(self, knowledge_base_id: str, slug: str):
return dict(self.pages[slug]) return dict(self.pages[slug])
@@ -139,6 +143,10 @@ async def test_sync_is_incremental_and_full_scan_deletes_missing_pages() -> None
first = await service.sync_mapping(mapping) first = await service.sync_mapping(mapping)
assert first["changed_pages"] == 1 assert first["changed_pages"] == 1
assert embedding.calls == 1 assert embedding.calls == 1
statuses = await store.list_page_index_status("kb")
assert statuses[0]["title"] == "One"
assert statuses[0]["vector_count"] > 0
assert store.states["kb"]["remote_page_count"] == 1
second = await service.sync_mapping(mapping) second = await service.sync_mapping(mapping)
assert second["changed_pages"] == 0 assert second["changed_pages"] == 0
assert embedding.calls == 1 assert embedding.calls == 1
@@ -0,0 +1,17 @@
from __future__ import annotations
from pathlib import Path
from alembic.config import Config
from alembic.script import ScriptDirectory
def test_migration_graph_has_unique_revisions_and_one_head() -> None:
migrations = Path(__file__).parents[1] / "packages" / "harness" / "deerflow" / "persistence" / "migrations"
config = Config()
config.set_main_option("script_location", str(migrations))
scripts = ScriptDirectory.from_config(config)
revisions = list(scripts.walk_revisions())
assert len({revision.revision for revision in revisions}) == len(revisions)
assert len(scripts.get_heads()) == 1
@@ -7,6 +7,7 @@ from fastapi import HTTPException
from starlette.requests import Request from starlette.requests import Request
from app.gateway.routers import llmwiki from app.gateway.routers import llmwiki
from app.gateway.weknora_embed import verify_weknora_signed_payload
def _request(store: object) -> Request: def _request(store: object) -> Request:
@@ -199,6 +200,38 @@ async def test_public_wiki_file_is_authorized_and_streamed(monkeypatch) -> None:
assert response.body == b"png-bytes" assert response.body == b"png-bytes"
assert response.media_type == "image/png" assert response.media_type == "image/png"
assert response.headers["cache-control"] == "public, max-age=300" assert response.headers["cache-control"] == "public, max-age=300"
assert response.headers["content-disposition"] == "inline; filename*=UTF-8''figure.png"
@pytest.mark.asyncio
async def test_private_wiki_file_uses_signed_http_access_instead_of_blob_url(monkeypatch) -> None:
async def authorized_mapping(*_args, **_kwargs):
return _published_row(), "user-one", False
class Client:
async def get_knowledge_base_file(self, knowledge_base_id: str, file_path: str):
assert knowledge_base_id == "remote-one"
assert file_path == "storage://tenant/resource://wiki/figure.png"
return b"png-bytes", "image/png"
monkeypatch.setattr(llmwiki, "_authorized_mapping", authorized_mapping)
monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())
access = await llmwiki.create_wiki_file_access(
_request(PublicStore(_published_row())),
"kb-one",
"storage://tenant/resource://wiki/figure.png",
)
payload = verify_weknora_signed_payload(access["token"])
assert payload["kind"] == "wiki_file"
assert payload["mapping_id"] == "kb-one"
assert payload["weknora_id"] == "remote-one"
assert payload["file_path"] == "storage://tenant/resource://wiki/figure.png"
response = await llmwiki.get_signed_wiki_file(access["token"])
assert response.body == b"png-bytes"
assert response.headers["cache-control"] == "private, max-age=3600"
assert response.headers["content-disposition"] == "inline; filename*=UTF-8''figure.png"
def test_wiki_file_proxy_rejects_non_weknora_urls() -> None: def test_wiki_file_proxy_rejects_non_weknora_urls() -> None:
@@ -208,6 +241,23 @@ def test_wiki_file_proxy_rejects_non_weknora_urls() -> None:
assert exc_info.value.status_code == 422 assert exc_info.value.status_code == 422
@pytest.mark.asyncio
async def test_wiki_file_proxy_adds_image_extension_for_opaque_resource_ids(monkeypatch) -> None:
class Client:
async def get_knowledge_base_file(self, knowledge_base_id: str, file_path: str):
return b"png-bytes", "image/png"
monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())
response = await llmwiki._wiki_file_response(
{"weknora_id": "remote-one"},
"resource://opaque-image-id",
public=True,
)
assert response.headers["content-disposition"] == "inline; filename*=UTF-8''opaque-image-id.png"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_public_wiki_index_preserves_weknora_group_and_item_order(monkeypatch) -> None: async def test_public_wiki_index_preserves_weknora_group_and_item_order(monkeypatch) -> None:
store = PublicStore(_published_row()) store = PublicStore(_published_row())