feat: 知识库一键后台保存 + 对话置顶仅首页 + 排行榜页面增强

- 知识库「保存」改为一键触发，无需填表；后端经 BackgroundTasks 异步沉淀到 wiki，点击立即返回 {queued:true}
- 对话列表置顶分区仅在第一页展示，翻页后不再重复顶置
- 公开/管理员排行榜页面重构与维度增强；排行榜页改为整页统一滚动

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

offline-backend-20260512/backend/app/gateway/routers/admin_users.py

@@ -808,6 +808,10 @@ async def _query_user_leaderboard(
     if user_id:
         stmt = stmt.where(UserRow.id == user_id)
     stmt = _apply_user_scope(stmt, UserRow.id, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
+    # 排行榜只统计真正产生过问答的用户；从未问答的账号（仅注册、零活跃）不计入榜单与人数。
+    # 个人分析（指定 user_id）不受此限制，仍可查看任意单个用户。
+    if user_id is None:
+        stmt = stmt.where(func.coalesce(run_agg.c.qa_count, 0) > 0)
     stmt = stmt.order_by(
         desc(func.coalesce(run_agg.c.run_count, 0) + func.coalesce(tool_agg.c.tool_call_count, 0)),
         desc(func.coalesce(run_agg.c.total_tokens, 0)),

offline-backend-20260512/backend/app/gateway/routers/knowledge.py

@@ -15,7 +15,7 @@ import logging
 import mimetypes
 from typing import Any
 
-from fastapi import APIRouter, HTTPException, Request
+from fastapi import APIRouter, BackgroundTasks, HTTPException, Request
 from fastapi.responses import FileResponse
 from pydantic import BaseModel, Field
 
@@ -40,6 +40,10 @@ class IngestThreadRequest(BaseModel):
     status: str = Field(default="approved", description="draft | approved")
     tags: list[str] = Field(default_factory=list)
     include_sources: bool = True
+    background: bool = Field(
+        default=False,
+        description="When true, sediment the thread asynchronously in a background task and return immediately.",
+    )
 
 
 class BatchIngestRequest(BaseModel):
@@ -117,16 +121,64 @@ def _messages_and_meta(values: dict[str, Any]) -> tuple[list[dict[str, Any]], st
     return messages, title, batches
 
 
+async def _ingest_thread_background(service, checkpointer, thread_id: str, body: IngestThreadRequest, created_by: str | None) -> None:
+    """Sediment a thread into a note without an HTTP context.
+
+    Runs after the response is sent (FastAPI ``BackgroundTasks``); ``created_by``
+    must be captured by the request handler since the user contextvar is gone
+    by the time this executes. All failures are swallowed (logged) — the user
+    has already been told the save was queued.
+    """
+    try:
+        values = await _read_thread_values(checkpointer, thread_id)
+        messages, title, batches = _messages_and_meta(values)
+        if not messages:
+            logger.warning("Background knowledge ingest skipped: thread %s has no messages", sanitize_log_param(thread_id))
+            return
+        await service.capture_thread(
+            thread_id=thread_id,
+            messages=messages,
+            thread_title=title,
+            mode=body.mode,
+            title=body.title,
+            status=body.status,
+            tags=body.tags,
+            include_sources=body.include_sources,
+            reference_batches=batches,
+            created_by=created_by,
+        )
+    except EmptyThreadError:
+        logger.warning("Background knowledge ingest skipped: thread %s has no sediment-able content", sanitize_log_param(thread_id))
+    except Exception:
+        logger.exception("Background knowledge ingest failed for thread %s", sanitize_log_param(thread_id))
+
+
 # --------------------------------------------------------------------------- #
 # Ingest
 # --------------------------------------------------------------------------- #
 @router.post("/ingest/thread/{thread_id}")
 @require_auth
 @require_permission("threads", "read", owner_check=True)
-async def ingest_thread(thread_id: str, body: IngestThreadRequest, request: Request) -> dict[str, Any]:
-    """Sediment a single thread into a knowledge note."""
+async def ingest_thread(thread_id: str, body: IngestThreadRequest, request: Request, background_tasks: BackgroundTasks) -> dict[str, Any]:
+    """Sediment a single thread into a knowledge note.
+
+    With ``background=true`` the heavy capture (read checkpoint → extract →
+    write vault + DB) is deferred to a background task and the endpoint returns
+    ``{"queued": true}`` immediately, so the user's "保存" click never blocks.
+    """
     service = get_knowledge_service(request)
     checkpointer = get_checkpointer(request)
+    if body.background:
+        # Capture the user id now — the contextvar is gone once the task runs.
+        background_tasks.add_task(
+            _ingest_thread_background,
+            service,
+            checkpointer,
+            thread_id,
+            body,
+            get_effective_user_id(),
+        )
+        return {"queued": True}
     values = await _read_thread_values(checkpointer, thread_id)
     messages, title, batches = _messages_and_meta(values)
     if not messages: