From 1b86373a5490d78c63231762b4b07cc246157769 Mon Sep 17 00:00:00 2001 From: Administrator Date: Thu, 11 Jun 2026 23:18:58 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=9F=A5=E8=AF=86=E5=BA=93=E6=A0=87?= =?UTF-8?q?=E7=AD=BE/=E5=AE=9E=E4=BD=93=E5=85=B3=E8=81=94=E8=B7=B3?= =?UTF-8?q?=E8=BD=AC=20+=20=E4=BC=9A=E8=AF=9D=E9=87=8D=E8=BF=9E=E5=AE=B9?= =?UTF-8?q?=E9=94=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 前端: - Wiki 标签与实体改为可点击，弹出 RelatedNotesDialog 查看关联知识 - 知识库首次进入自动打开最近笔记，避免空白阅读区 - 流式重连识别 "Run not found"，清理 sessionStorage 残留 key 并静默回退历史 后端: - 线程鉴权区分 not found / not accessible，并支持 DB 切换产生的孤儿线程认领 - run 查询/join 在内存未命中时回退持久化存储，重启后返回干净的终止 SSE 而非 404 --- .../components/wiki/RelatedNotesDialog.tsx | 110 ++++++++++++++++++ .../src/components/wiki/WikiContent.tsx | 41 +++++-- frontend-web/src/core/threads/hooks.ts | 33 +++++- frontend-web/src/pages/KnowledgeBasePage.tsx | 11 ++ .../backend/app/gateway/authz.py | 74 ++++++++++-- .../app/gateway/routers/thread_runs.py | 51 ++++++-- .../backend/app/gateway/services.py | 14 +++ .../harness/deerflow/runtime/runs/manager.py | 46 ++++++++ 8 files changed, 357 insertions(+), 23 deletions(-) create mode 100644 frontend-web/src/components/wiki/RelatedNotesDialog.tsx diff --git a/frontend-web/src/components/wiki/RelatedNotesDialog.tsx b/frontend-web/src/components/wiki/RelatedNotesDialog.tsx new file mode 100644 index 00000000..1cc5e048 --- /dev/null +++ b/frontend-web/src/components/wiki/RelatedNotesDialog.tsx @@ -0,0 +1,110 @@ +import { FileText, Hash, Loader2, Sparkles } from "lucide-react"; + +import { + Dialog, + DialogContent, + DialogDescription, + DialogHeader, + DialogTitle, +} from "@/components/ui/dialog"; +import { ScrollArea } from "@/components/ui/scroll-area"; +import { useKnowledgeNotes } from "@/core/knowledge/hooks"; +import type { KnowledgeNote } from "@/core/knowledge/types"; + +import { SOURCE_TYPE_LABELS, formatDateTime } from "./wikiUtils"; + +/** What was clicked: a tag (exact filter) or an entity (keyword match in body). */ +export type RelatedNotesFilter = + | { kind: "tag"; value: string } + | { kind: "entity"; value: string }; + +/** + * Click a tag / entity chip → list every other note that shares it. + * Tags use the backend's exact `tag` filter; entities fall back to a keyword + * match (entity names appear verbatim in titles/summaries/bodies). + */ +export function RelatedNotesDialog({ + filter, + currentNoteId, + onOpenChange, + onSelect, +}: { + filter: RelatedNotesFilter | null; + currentNoteId?: string; + onOpenChange: (open: boolean) => void; + onSelect: (noteId: string) => void; +}) { + const params = + filter?.kind === "tag" + ? { tag: filter.value, limit: 50 } + : filter?.kind === "entity" + ? { keyword: filter.value, limit: 50 } + : { limit: 0 }; + + const { data, isLoading, isFetching } = useKnowledgeNotes(params); + const items = (data?.items ?? []).filter( + (n: KnowledgeNote) => n.id !== currentNoteId, + ); + + return ( + + + + + {filter?.kind === "entity" ? ( + + ) : ( + + )} + {filter?.value} + + + {filter?.kind === "entity" + ? "提到该实体的关联知识" + : "使用该标签的关联知识"} + + + + + {isLoading || (isFetching && items.length === 0) ? ( +
+ 加载中... +
+ ) : items.length === 0 ? ( +
+ 暂无其他关联知识。 +
+ ) : ( +
    + {items.map((note) => ( +
  • + +
    • + ))} +
+ )} + + + + ); +} diff --git a/frontend-web/src/components/wiki/WikiContent.tsx b/frontend-web/src/components/wiki/WikiContent.tsx index 38d4705f..354f190b 100644 --- a/frontend-web/src/components/wiki/WikiContent.tsx +++ b/frontend-web/src/components/wiki/WikiContent.tsx @@ -25,6 +25,10 @@ import { DocumentToolbar } from "./DocumentToolbar"; import { FloatingToc } from "./FloatingToc"; import { MarkdownViewer } from "./MarkdownViewer"; import { RelatedDocs } from "./RelatedDocs"; +import { + RelatedNotesDialog, + type RelatedNotesFilter, +} from "./RelatedNotesDialog"; import { SOURCE_TYPE_LABELS, STATUS_LABELS, @@ -91,6 +95,10 @@ export function WikiContent({ const [draftContent, setDraftContent] = useState(note.content_md); const [draftStatus, setDraftStatus] = useState(note.status); const [activeHeadingId, setActiveHeadingId] = useState(null); + // Tag / entity chip clicked → show notes that share it. + const [relatedFilter, setRelatedFilter] = useState( + null, + ); // Pull the backend-injected `## Related` wikilinks out of the body so they // render as a structured module; editing still works on the full source. @@ -270,12 +278,15 @@ export function WikiContent({ 创建：{formatDateTime(note.created_at).slice(0, 10)} ) : null} {note.tags.map((tag) => ( - setRelatedFilter({ kind: "tag", value: tag })} + className="inline-flex h-6 cursor-pointer items-center rounded-md bg-rose-50 px-2 text-xs font-medium text-rose-500 transition-colors hover:bg-rose-100 hover:text-rose-600 dark:bg-rose-500/10 dark:text-rose-300 dark:hover:bg-rose-500/20" > #{tag} - + ))} @@ -287,11 +298,15 @@ export function WikiContent({ {note.entities.map((ent) => { const meta = entityMeta(ent.entity_type); return ( - + setRelatedFilter({ kind: "entity", value: ent.name }) + } className={cn( - "inline-flex items-center gap-1.5 rounded-full border py-1 pl-2 pr-1 text-xs font-medium", + "inline-flex cursor-pointer items-center gap-1.5 rounded-full border py-1 pl-2 pr-1 text-xs font-medium transition-opacity hover:opacity-80", meta.className, )} > @@ -300,7 +315,7 @@ export function WikiContent({ {meta.label} - + ); })} @@ -373,6 +388,18 @@ export function WikiContent({ + + { + if (!open) setRelatedFilter(null); + }} + onSelect={(id) => { + setRelatedFilter(null); + navigate(`/page/workspace/knowledge/${id}`); + }} + /> ); } diff --git a/frontend-web/src/core/threads/hooks.ts b/frontend-web/src/core/threads/hooks.ts index caea58c4..79a1273b 100644 --- a/frontend-web/src/core/threads/hooks.ts +++ b/frontend-web/src/core/threads/hooks.ts @@ -152,7 +152,10 @@ function getStreamErrorMessage(error: unknown): string { */ function isThreadNotAccessibleError(message: string): boolean { const lowered = message.toLowerCase(); - if (/thread\s+[\w-]{8,}\s+not\s+found/i.test(message)) { + // Backend now returns "not found" (no row + no checkpoint) and + // "not accessible" (row owned by another user / stale login session) as + // distinct details — both mean the current user can't use this thread. + if (/thread\s+[\w-]{8,}\s+not\s+(found|accessible)/i.test(message)) { return true; } // Fallback: a bare 404/403 from the threads endpoint when the SDK has @@ -163,6 +166,21 @@ function isThreadNotAccessibleError(message: string): boolean { ); } +/** + * Detect a "Run not found" 404 from the LangGraph SDK. + * + * The SDK persists the in-flight run id under ``lg:stream:{threadId}`` in + * sessionStorage and, on mount, auto-``joinStream``s it to resume a dropped + * SSE. After a backend restart the in-memory run registry is gone, so the join + * 404s — and the SDK only clears that key on reconnect *success*, never on + * failure. So every switch back into the conversation re-fires the same 404. + * We detect it here to clear the stale key ourselves and stay silent (the + * conversation renders from history; there is nothing live to reconnect to). + */ +function isRunNotFoundError(message: string): boolean { + return /run\s+[\w-]{8,}\s+not\s+found/i.test(message); +} + function buildCurrentUserLabel(): string { const auth = getStoredAuth(); if (!auth) { @@ -335,6 +353,19 @@ export function useThreadStream({ onError(error) { setOptimisticMessages([]); const message = getStreamErrorMessage(error); + // A stale auto-reconnect ("Run not found") after a backend restart: + // drop the dirty ``lg:stream:{threadId}`` key so switching back into this + // conversation stops re-firing the same 404, and render from history + // silently — there is no live run to rejoin. + if (isRunNotFoundError(message)) { + const staleThreadId = threadIdRef.current ?? onStreamThreadId ?? null; + if (staleThreadId && runMetadataStorageRef.current) { + runMetadataStorageRef.current.removeItem( + `lg:stream:${staleThreadId}`, + ); + } + return; + } // A "thread not found" 404 almost always means the URL points at a // thread from a previous login session — either this tab outlived a // logout/re-login or the thread genuinely belongs to a different user. diff --git a/frontend-web/src/pages/KnowledgeBasePage.tsx b/frontend-web/src/pages/KnowledgeBasePage.tsx index 7ce56d67..b12cfb5f 100644 --- a/frontend-web/src/pages/KnowledgeBasePage.tsx +++ b/frontend-web/src/pages/KnowledgeBasePage.tsx @@ -468,6 +468,8 @@ export default function KnowledgeBasePage() { const [status, setStatus] = useState("all"); const [expandedIds, setExpandedIds] = useState>(new Set()); const treeInitializedRef = useRef(false); + // Auto-open a default note once on entry so the reading pane is never blank. + const defaultNoteShownRef = useRef(false); const [mobileSidebarOpen, setMobileSidebarOpen] = useState(false); const [sidebarCollapsed, setSidebarCollapsed] = useState(readSidebarCollapsed); @@ -495,6 +497,15 @@ export default function KnowledgeBasePage() { const { data: note, isLoading: noteLoading } = useKnowledgeNote(noteId); const { data: wikiLinks } = useNoteWikilinks(noteId); + // On first entry without an explicit note, open the most recent one so the + // reading pane shows real content instead of the empty placeholder. + useEffect(() => { + if (defaultNoteShownRef.current || noteId || isLoading) return; + if (notes.length === 0) return; + defaultNoteShownRef.current = true; + navigate(`/page/workspace/knowledge/${notes[0].id}`, { replace: true }); + }, [noteId, isLoading, notes, navigate]); + // First load: expand the root-level folders so the tree reads as a directory. useEffect(() => { if (treeInitializedRef.current || tree.length === 0) return; diff --git a/offline-backend-20260512/backend/app/gateway/authz.py b/offline-backend-20260512/backend/app/gateway/authz.py index 89fd943c..0d8d2899 100644 --- a/offline-backend-20260512/backend/app/gateway/authz.py +++ b/offline-backend-20260512/backend/app/gateway/authz.py @@ -31,21 +31,43 @@ from __future__ import annotations import functools import inspect - -from app.gateway.auth.disabled_mode import get_assumed_user_for_disabled_auth, is_auth_disabled +import logging from collections.abc import Callable from types import SimpleNamespace from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar from fastapi import HTTPException, Request +from app.gateway.auth.disabled_mode import get_assumed_user_for_disabled_auth, is_auth_disabled + if TYPE_CHECKING: from app.gateway.auth.models import User +logger = logging.getLogger(__name__) + P = ParamSpec("P") T = TypeVar("T") +async def _thread_checkpoint_exists(request: Request, thread_id: str) -> bool: + """Return True if a LangGraph checkpoint exists for ``thread_id``. + + A checkpoint with no matching ``threads_meta`` row is the signature of a + DB-switch orphan (sqlite↔mysql/postgres juggling): the conversation state + survives in the checkpointer while the business row is gone. Normal thread + deletion removes the checkpoint too, so this can never resurrect a + deliberately-deleted thread. + """ + try: + from app.gateway.deps import get_checkpointer + + checkpointer = get_checkpointer(request) + tup = await checkpointer.aget_tuple({"configurable": {"thread_id": thread_id}}) + return tup is not None + except Exception: + return False + + # Permission constants class Permissions: """Permission constants for resource:action format.""" @@ -205,6 +227,7 @@ def require_permission( action: str, owner_check: bool = False, require_existing: bool = False, + claim_orphan: bool = False, ) -> Callable[[Callable[P, T]], Callable[P, T]]: """Decorator that checks permission for resource:action. @@ -221,6 +244,15 @@ def require_permission( **destructive / mutating** routes (DELETE, PATCH, state-update) so a deleted thread can't be re-targeted by another user via the missing-row code path. + claim_orphan: Only meaningful with ``owner_check=True`` and + ``require_existing=True``. When the ``threads_meta`` + row is missing **and** a LangGraph checkpoint exists + for the thread (a DB-switch orphan), adopt it for the + current user instead of 404ing. Use on **run-create** + routes so a conversation whose business row was lost + (sqlite↔mysql swap) stays answerable. Safe: normal + deletion removes the checkpoint, so this never revives + a deliberately-deleted thread. Usage: # Read-style: legacy untracked threads are allowed @@ -289,16 +321,44 @@ def require_permission( from app.gateway.deps import get_thread_store thread_store = get_thread_store(request) + user_id = str(auth.user.id) allowed = await thread_store.check_access( thread_id, - str(auth.user.id), + user_id, require_existing=require_existing, ) if not allowed: - raise HTTPException( - status_code=404, - detail=f"Thread {thread_id} not found", - ) + # Tell apart three denial causes so the client gets an + # accurate message and so DB-switch orphans can be adopted + # rather than hard-404'd: + # 1. row exists but owned by another user -> not accessible + # 2. no row, no checkpoint -> truly missing + # 3. no row + checkpoint + claim_orphan -> orphan, claim + existing = await thread_store.get(thread_id, user_id=None) + if existing is not None: + # Distinct detail string lets the frontend show "无权访问" + # instead of "不存在" for a cross-user / stale-session hit. + raise HTTPException( + status_code=404, + detail=f"Thread {thread_id} not accessible", + ) + claimed = False + if claim_orphan and await _thread_checkpoint_exists(request, thread_id): + try: + await thread_store.create(thread_id, user_id=user_id) + logger.info( + "Claimed orphan thread %s for user %s (checkpoint present, no meta row)", + thread_id, + user_id, + ) + claimed = True + except Exception: + logger.warning("Failed to claim orphan thread %s", thread_id, exc_info=True) + if not claimed: + raise HTTPException( + status_code=404, + detail=f"Thread {thread_id} not found", + ) return await func(*args, **kwargs) diff --git a/offline-backend-20260512/backend/app/gateway/routers/thread_runs.py b/offline-backend-20260512/backend/app/gateway/routers/thread_runs.py index 38129fa5..1f108dcc 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/thread_runs.py +++ b/offline-backend-20260512/backend/app/gateway/routers/thread_runs.py @@ -21,7 +21,7 @@ from pydantic import BaseModel, Field from app.gateway.authz import require_permission from app.gateway.deps import get_checkpointer, get_current_user, get_feedback_repo, get_run_event_store, get_run_manager, get_run_store, get_stream_bridge -from app.gateway.services import sse_consumer, start_run +from app.gateway.services import immediate_end_sse, sse_consumer, start_run from deerflow.runtime import RunRecord, serialize_channel_values logger = logging.getLogger(__name__) @@ -96,7 +96,7 @@ def _record_to_response(record: RunRecord) -> RunResponse: @router.post("/{thread_id}/runs", response_model=RunResponse) -@require_permission("runs", "create", owner_check=True, require_existing=True) +@require_permission("runs", "create", owner_check=True, require_existing=True, claim_orphan=True) async def create_run(thread_id: str, body: RunCreateRequest, request: Request) -> RunResponse: """Create a background run (returns immediately).""" record = await start_run(body, thread_id, request) @@ -104,7 +104,7 @@ async def create_run(thread_id: str, body: RunCreateRequest, request: Request) - @router.post("/{thread_id}/runs/stream") -@require_permission("runs", "create", owner_check=True, require_existing=True) +@require_permission("runs", "create", owner_check=True, require_existing=True, claim_orphan=True) async def stream_run(thread_id: str, body: RunCreateRequest, request: Request) -> StreamingResponse: """Create a run and stream events via SSE. @@ -132,7 +132,7 @@ async def stream_run(thread_id: str, body: RunCreateRequest, request: Request) - @router.post("/{thread_id}/runs/wait", response_model=dict) -@require_permission("runs", "create", owner_check=True, require_existing=True) +@require_permission("runs", "create", owner_check=True, require_existing=True, claim_orphan=True) async def wait_run(thread_id: str, body: RunCreateRequest, request: Request) -> dict: """Create a run and block until it completes, returning the final state.""" record = await start_run(body, thread_id, request) @@ -173,7 +173,11 @@ async def get_run(thread_id: str, run_id: str, request: Request) -> RunResponse: run_mgr = get_run_manager(request) record = run_mgr.get(run_id) if record is None or record.thread_id != thread_id: - raise HTTPException(status_code=404, detail=f"Run {run_id} not found") + # In-memory miss: fall back to the persistent store so a run still + # resolves after a Gateway restart instead of 404ing. + record = await run_mgr.get_persisted(run_id) + if record is None or record.thread_id != thread_id: + raise HTTPException(status_code=404, detail=f"Run {run_id} not found") return _record_to_response(record) @@ -222,11 +226,27 @@ async def join_run(thread_id: str, run_id: str, request: Request) -> StreamingRe bridge = get_stream_bridge(request) run_mgr = get_run_manager(request) record = run_mgr.get(run_id) - if record is None or record.thread_id != thread_id: + if record is not None and record.thread_id == thread_id: + return StreamingResponse( + sse_consumer(bridge, record, request, run_mgr), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + }, + ) + + # Not in memory: a run that still exists in the store but has no live task + # means the Gateway restarted (or the finished run was cleaned up). There + # is nothing to reconnect to, so stream a clean terminal end instead of a + # 404 — the SDK then clears its reconnect key and renders history. + persisted = await run_mgr.get_persisted(run_id) + if persisted is None or persisted.thread_id != thread_id: raise HTTPException(status_code=404, detail=f"Run {run_id} not found") return StreamingResponse( - sse_consumer(bridge, record, request, run_mgr), + immediate_end_sse(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", @@ -255,7 +275,22 @@ async def stream_existing_run( run_mgr = get_run_manager(request) record = run_mgr.get(run_id) if record is None or record.thread_id != thread_id: - raise HTTPException(status_code=404, detail=f"Run {run_id} not found") + # In-memory miss. If the run survives in the store it's a dead record + # from before a restart (or already finished) — nothing to join or + # cancel, so emit a clean terminal end so the SDK stops reconnecting + # instead of 404ing. A truly unknown run still 404s. + persisted = await run_mgr.get_persisted(run_id) + if persisted is None or persisted.thread_id != thread_id: + raise HTTPException(status_code=404, detail=f"Run {run_id} not found") + return StreamingResponse( + immediate_end_sse(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + }, + ) # Cancel if an action was requested (stop-button / interrupt flow) if action is not None: diff --git a/offline-backend-20260512/backend/app/gateway/services.py b/offline-backend-20260512/backend/app/gateway/services.py index 46625e28..3f0e8c37 100644 --- a/offline-backend-20260512/backend/app/gateway/services.py +++ b/offline-backend-20260512/backend/app/gateway/services.py @@ -42,6 +42,20 @@ logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- +async def immediate_end_sse(): + """Yield a single terminal ``end`` SSE frame and stop. + + Used when a run exists in the persistent store but not in the in-memory + registry — typically after a Gateway restart, or once a finished run's + record has been cleaned up. There is no live stream to join, so instead of + returning 404 (which the SDK's ``joinStream`` surfaces as a hard error and + never clears its reconnect key for) we stream a clean end. The SDK treats + this as success, clears ``lg:stream:{threadId}``, and the conversation + falls back to history rendering — killing the repeating switch-back 404. + """ + yield format_sse("end", None) + + def format_sse(event: str, data: Any, *, event_id: str | None = None) -> str: """Format a single SSE frame. diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py index 9bd39c06..64deed00 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py @@ -112,6 +112,52 @@ class RunManager: """Return a run record by ID, or ``None``.""" return self._runs.get(run_id) + def _record_from_store(self, data: dict) -> RunRecord: + """Rebuild a :class:`RunRecord` from a persisted store row. + + The reconstructed record has no live ``task``/``abort_event`` wired to a + running coroutine — it represents *history*, not an in-flight run. + """ + try: + status = RunStatus(data.get("status") or RunStatus.success.value) + except ValueError: + status = RunStatus.success + return RunRecord( + run_id=data["run_id"], + thread_id=data["thread_id"], + assistant_id=data.get("assistant_id"), + status=status, + # Disconnect mode is irrelevant for a store-only record (no task to + # cancel); ``continue_`` keeps sse_consumer's finally block a no-op. + on_disconnect=DisconnectMode.continue_, + multitask_strategy=data.get("multitask_strategy", "reject"), + metadata=data.get("metadata") or {}, + kwargs=data.get("kwargs") or {}, + created_at=data.get("created_at") or "", + updated_at=data.get("updated_at") or "", + error=data.get("error"), + ) + + async def get_persisted(self, run_id: str) -> RunRecord | None: + """Load a run from the backing store (process-restart fallback). + + The in-memory registry is wiped on every Gateway restart, but the + ``RunStore`` row survives. When :meth:`get` misses, callers fall back + here so a stale ``joinStream``/``GET run`` can report the run's final + status instead of 404ing. Returns ``None`` when there is no store, no + matching row, or the lookup fails (e.g. no user context). + """ + if self._store is None: + return None + try: + data = await self._store.get(run_id) + except Exception: + logger.warning("Failed to load run %s from store", run_id, exc_info=True) + return None + if not data: + return None + return self._record_from_store(data) + async def list_by_thread(self, thread_id: str) -> list[RunRecord]: """Return all runs for a given thread, newest first.""" async with self._lock: