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 (
+
+ );
+}
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: