"""cmzs LLMWiki BFF backed by an independently deployed WeKnora."""

from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import mimetypes
import re
import time
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import replace
from datetime import UTC, datetime
from http.cookiejar import CookieJar, DefaultCookiePolicy
from io import BytesIO
from types import SimpleNamespace
from typing import Any, Literal
from urllib.parse import quote, unquote, urlencode, urlsplit
from uuid import uuid4

import httpx
from fastapi import APIRouter, BackgroundTasks, File, HTTPException, Query, Request, UploadFile
from fastapi.responses import JSONResponse, RedirectResponse, Response, StreamingResponse
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field

from app.gateway.csrf_middleware import is_secure_request
from app.gateway.deps import get_agent_store, get_checkpointer, get_optional_user_from_request, get_thread_store
from app.gateway.llmwiki_deposit import (
    CONVERSATION_DEPOSIT_KB_DESCRIPTION,
    CONVERSATION_DEPOSIT_KB_NAME,
    CONVERSATION_DEPOSIT_OWNER_USER_ID,
    ConversationTurnDeposit,
    deposit_conversation_turn,
    ensure_conversation_deposit_mapping,
)
from app.gateway.llmwiki_rag import build_llmwiki_rag_context
from app.gateway.perf_logging import finish_trace, trace_request, trace_stage
from app.gateway.weknora_embed import (
    WEKNORA_EMBED_API_PREFIX as _WEKNORA_EMBED_API_PREFIX,
)
from app.gateway.weknora_embed import (
    WEKNORA_EMBED_COOKIE as _WEKNORA_EMBED_COOKIE,
)
from app.gateway.weknora_embed import (
    WEKNORA_EMBED_TTL_SECONDS as _WEKNORA_EMBED_TTL_SECONDS,
)
from app.gateway.weknora_embed import (
    WeKnoraEmbedContext,
)
from app.gateway.weknora_embed import (
    sign_weknora_embed_payload as _sign_embed_payload,
)
from app.gateway.weknora_embed import (
    verify_weknora_embed_cookie as _verify_embed_cookie,
)
from app.gateway.weknora_embed import (
    verify_weknora_signed_payload as _verify_signed_payload,
)
from app.gateway.weknora_static import StaticAsset, StaticAssetCache, prepare_static_asset
from app.gateway.wiki_directory_constraints import (
    directory_roots,
    ensure_directory_paths,
    normalize_directory_paths,
    paths_from_parent_directories,
    resolve_directory_paths,
)
from app.gateway.wiki_generation_config import (
    WEKNORA_CUSTOM_INSTRUCTION_MAX,
    WIKI_CONFIG_SCHEMA_VERSION,
    WIKI_FOCUS_INSTRUCTION_MAX,
    compile_wiki_instructions,
    normalize_focus_instructions,
)
from deerflow.assistant_knowledge.package_io import looks_like_knowledge_package
from deerflow.config.agents_config import load_agent_config
from deerflow.config.llmwiki_config import LlmWikiRuntimeOverride, normalize_service_url
from deerflow.config.system_settings import load_system_settings, save_system_settings
from deerflow.integrations.weknora.client import WeKnoraClient, WeKnoraError
from deerflow.integrations.weknora.knowledge_profile import (
    DEFAULT_RETRIEVAL_SETTINGS,
    are_test_questions_customized,
    resolved_retrieval_settings,
    resolved_test_questions,
)
from deerflow.integrations.weknora.local_index.retrieval import is_fully_vectorized
from deerflow.integrations.weknora.local_index.search import WikiIndexError
from deerflow.integrations.weknora.runtime import build_weknora_client, get_resolved_llmwiki_runtime
from deerflow.persistence.llmwiki import LlmWikiStore

logger = logging.getLogger(__name__)
KNOWLEDGE_RESEARCH_DIRECTORY = "深度研究"


@asynccontextmanager
async def _proxy_lifespan(app: Any) -> AsyncIterator[None]:
    # One pool per Gateway/event loop: the embedded app loads many assets and
    # API requests together, which must not each establish a new connection.
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(60.0),
        follow_redirects=False,
        trust_env=False,
        # Only explicit per-request service headers authenticate upstream.
        # Retain connections, never upstream cookies across DeerFlow users.
        cookies=CookieJar(policy=DefaultCookiePolicy(allowed_domains=[])),
    ) as client:
        app.state.llmwiki_proxy_http_client = client
        app.state.llmwiki_static_assets = StaticAssetCache()
        try:
            yield
        finally:
            app.state.llmwiki_proxy_http_client = None
            app.state.llmwiki_static_assets = None


@asynccontextmanager
async def _proxy_http_client(request: Request) -> AsyncIterator[httpx.AsyncClient]:
    app = getattr(request, "scope", {}).get("app")
    client = getattr(getattr(app, "state", None), "llmwiki_proxy_http_client", None)
    if client is not None:
        yield client
    else:
        # Standalone router callers without the Gateway lifespan still work.
        async with httpx.AsyncClient(timeout=httpx.Timeout(60.0), follow_redirects=False, trust_env=False) as client:
            yield client


router = APIRouter(tags=["llmwiki"], lifespan=_proxy_lifespan)
proxy_router = APIRouter(tags=["llmwiki-proxy"])

_MAX_UPLOAD_BYTES = 100 * 1024 * 1024
_KNOWLEDGE_PACKAGE_UPLOAD_DETAIL = (
    "检测到知识库 JSONL 数据包。为避免大模型把数据包当普通文档重新识别，"
    "请使用“导入 / 导出数据包”中的“上传并导入”；本次文件未送入模型。"
)
_WIKI_FILE_ACCESS_TTL_SECONDS = 24 * 60 * 60
_PROTECTED_WIKI_FILE_PATH = re.compile(
    r"^(?:storage://[0-9A-Za-z_-]+/)?(?:resource|local|minio|cos|tos|s3|oss|ks3|obs)://",
    re.IGNORECASE,
)
_WEKNORA_ADMIN_SESSION_CACHE: dict[str, dict[str, Any]] = {}
_WEKNORA_ADMIN_SESSION_TASKS: dict[str, asyncio.Task[dict[str, Any]]] = {}
_WEKNORA_PUBLIC_PREFIX = "/deerflow"
_WEKNORA_PUBLIC_ROOT_PATHS = (
    "/platform",
    "/assets",
    "/css",
    "/fonts",
    "/img",
    "/images",
    "/js",
    "/locales",
    "/media",
    "/static",
    "/tdesign-icons",
    "/api/v1",
    "/favicon.ico",
    "/favicon.svg",
    "/config.js",
    "/runtime-config.js",
    "/logo.svg",
    "/manifest.webmanifest",
)
_WEKNORA_PUBLIC_RELATIVE_ROOT_PATHS = (
    "assets",
    "css",
    "fonts",
    "img",
    "images",
    "js",
    "locales",
    "media",
    "static",
    "tdesign-icons",
)
_WEKNORA_PUBLIC_RELATIVE_FILE_PATHS = (
    "config.js",
    "runtime-config.js",
    "favicon.ico",
    "favicon.svg",
    "logo.svg",
    "manifest.webmanifest",
)
_WEKNORA_STATIC_FILE_SUFFIXES = (
    ".avif",
    ".css",
    ".eot",
    ".gif",
    ".ico",
    ".jpeg",
    ".jpg",
    ".js",
    ".json",
    ".map",
    ".mjs",
    ".otf",
    ".png",
    ".svg",
    ".ttf",
    ".txt",
    ".wasm",
    ".webmanifest",
    ".webp",
    ".woff",
    ".woff2",
)
_WEKNORA_STATIC_ROOT_PREFIXES = (
    "/assets/",
    "/css/",
    "/fonts/",
    "/img/",
    "/images/",
    "/js/",
    "/locales/",
    "/media/",
    "/static/",
    "/tdesign-icons/",
)
_WEKNORA_ROOT_ASSET_FILES = (
    "/config.js",
    "/runtime-config.js",
    "/favicon.ico",
    "/favicon.svg",
    "/logo.svg",
    "/manifest.webmanifest",
)
_HOP_BY_HOP_RESPONSE_HEADERS = {
    "connection",
    "content-encoding",
    "content-length",
    "keep-alive",
    "proxy-authenticate",
    "proxy-authorization",
    "te",
    "trailer",
    "transfer-encoding",
    "upgrade",
    "x-frame-options",
    "content-security-policy",
}


class LlmWikiSettingsUpdate(BaseModel):
    override_enabled: bool = True
    api_base_url: str = ""
    web_base_url: str = ""


class LlmWikiConnectionTest(BaseModel):
    api_base_url: str | None = None


class WikiGenerationInstructions(BaseModel):
    """User-owned guidance for WeKnora's stable Wiki ingest prompts."""

    content_instructions: str | None = Field(
        default=None,
        max_length=WEKNORA_CUSTOM_INSTRUCTION_MAX,
    )
    extraction_instructions: str | None = Field(
        default=None,
        max_length=WEKNORA_CUSTOM_INSTRUCTION_MAX,
    )
    # ``directory_paths`` is generation guidance only. WeKnora may adapt it to
    # the source material; DeerFlow must not pre-create, move, or prune folders
    # to make the generated Wiki match this reference tree. The older
    # ``parent_directories`` input remains for API compatibility.
    parent_directories: list[str] | None = Field(default=None, max_length=100)
    directory_paths: list[list[str]] | None = Field(default=None, max_length=100)
    focus_instructions: str | None = Field(
        default=None,
        max_length=WIKI_FOCUS_INSTRUCTION_MAX,
    )


class KnowledgeBaseCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=255)
    description: str = Field(default="", max_length=2048)
    type: Literal["document", "faq"] = "document"
    wiki_enabled: bool = True
    embedding_model_id: str | None = Field(default=None, min_length=1, max_length=255)
    wiki_config: WikiGenerationInstructions | None = None


class KnowledgeRetrievalSettingsUpdate(BaseModel):
    entity_expansion_enabled: bool = True
    max_related_terms: int = Field(default=6, ge=0, le=12)
    statistics_enabled: bool = True


class KnowledgeBaseUpdate(BaseModel):
    name: str | None = Field(default=None, min_length=1, max_length=255)
    description: str | None = Field(default=None, max_length=2048)
    test_questions: list[str] | None = Field(default=None, max_length=12)
    retrieval_settings: KnowledgeRetrievalSettingsUpdate | None = None
    wiki_config: WikiGenerationInstructions | None = None


class SearchRequest(BaseModel):
    query: str = Field(..., min_length=1, max_length=4000)
    scope: Literal["personal", "public", "all"] = "all"
    knowledge_base_ids: list[str] = Field(default_factory=list, max_length=100)
    top_k: int = Field(default=8, ge=1, le=50)
    include_drafts: bool = False


class RetrievalDebugRequest(BaseModel):
    """Run the exact pre-answer knowledge retrieval path without calling an LLM."""

    query: str = Field(..., min_length=1, max_length=4000)
    knowledge_base_ids: list[str] = Field(default_factory=list, max_length=100)
    assistant_knowledge_base_ids: list[str] = Field(default_factory=list, max_length=100)


class KnowledgeUrlImport(BaseModel):
    url: str = Field(..., min_length=1, max_length=4096)


class ManualKnowledgeCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=255)
    content: str = Field(..., min_length=1, max_length=2_000_000)


class WikiPageCreate(BaseModel):
    slug: str = Field(..., min_length=1, max_length=512)
    title: str = Field(..., min_length=1, max_length=255)
    content: str = Field(..., min_length=1, max_length=2_000_000)
    summary: str | None = Field(default=None, max_length=200_000)
    page_type: str | None = Field(default="article", min_length=1, max_length=64)
    status: str | None = Field(default="published", min_length=1, max_length=64)
    aliases: list[str] | None = Field(default=None, max_length=100)
    parent_slug: str | None = Field(default=None, max_length=512)
    category_path: list[str] | None = Field(default=None, max_length=50)
    folder_id: str | None = Field(default=None, max_length=128)
    wiki_path: str | None = Field(default=None, max_length=1024)


class WikiPageUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=255)
    content: str | None = Field(default=None, min_length=1, max_length=2_000_000)
    summary: str | None = Field(default=None, max_length=200_000)
    page_type: str | None = Field(default=None, min_length=1, max_length=64)
    status: str | None = Field(default=None, min_length=1, max_length=64)
    aliases: list[str] | None = Field(default=None, max_length=100)
    parent_slug: str | None = Field(default=None, max_length=512)
    category_path: list[str] | None = Field(default=None, max_length=50)
    folder_id: str | None = Field(default=None, max_length=128)
    wiki_path: str | None = Field(default=None, max_length=1024)


class KnowledgeResearchSessionCreate(BaseModel):
    question: str = Field(..., min_length=1, max_length=20_000)


class KnowledgeResearchSessionUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=255)
    draft_markdown: str | None = Field(default=None, max_length=2_000_000)


class KnowledgeResearchPublicationSave(BaseModel):
    title: str = Field(..., min_length=1, max_length=255)
    expected_saved_hash: str | None = Field(default=None, min_length=64, max_length=64)


class ConversationDepositCreate(BaseModel):
    thread_id: str = Field(..., min_length=1, max_length=128)
    question: str = Field(default="", max_length=200_000)
    answer: str = Field(default="", max_length=2_000_000)
    human_message_id: str | None = Field(default=None, max_length=128)
    assistant_message_id: str = Field(..., min_length=1, max_length=128)
    assistant_id: str | None = Field(default=None, max_length=128)
    knowledge_base_ids: list[str] = Field(default_factory=list, max_length=100)


def _store(request: Request) -> LlmWikiStore:
    value = getattr(request.app.state, "llmwiki_store", None)
    if value is None:
        raise HTTPException(status_code=503, detail="LLMWiki metadata store is unavailable")
    return value


def _schedule_local_wiki_sync(
    request: Request,
    mapping: dict[str, Any],
    *,
    constraints_changed: bool = False,
) -> None:
    service = getattr(request.app.state, "llmwiki_sync_service", None)
    config = getattr(request.app.state, "llmwiki_effective_config", None)
    if config is None:
        config = request.app.state.config.llmwiki.local_wiki_index
    local_sync_enabled = bool(
        service is not None
        and config.enabled
        and config.auto_sync
        and mapping.get("wiki_index_enabled", True)
    )
    # Directory settings are prompt guidance only. Post-generation work is
    # reserved for the independently configured strict focus filter.
    constrained = bool(
        str(mapping.get("wiki_focus_instructions") or "").strip()
    )
    if local_sync_enabled or constrained or constraints_changed:
        from app.gateway.llmwiki_index_scheduler import schedule_local_wiki_mutation_sync

        # WeKnora parses documents and regenerates Wiki pages asynchronously.
        # The settle coordinator applies strict focus filtering and refreshes
        # the optional local vector mirror; directory references need neither.
        schedule_local_wiki_mutation_sync(
            request.app,
            mapping,
            client=_client_or_503() if constrained else None,
            sync_local=local_sync_enabled,
        )


async def _actor(request: Request) -> tuple[str, bool]:
    user = await get_optional_user_from_request(request)
    if user is None:
        return "default", True
    return str(user.id), getattr(user, "system_role", None) == "admin"


async def _require_admin(request: Request) -> tuple[str, bool]:
    actor = await _actor(request)
    if not actor[1]:
        raise HTTPException(status_code=403, detail="Only administrators can configure WeKnora")
    return actor


def _runtime_or_legacy() -> dict[str, Any]:
    runtime = get_resolved_llmwiki_runtime()
    return {
        "provider": runtime.provider,
        "api_base_url": runtime.api_base_url,
        "web_base_url": runtime.web_base_url,
        "override_enabled": runtime.override_enabled,
    }


def _client_or_503() -> WeKnoraClient:
    runtime = get_resolved_llmwiki_runtime()
    if not runtime.weknora_enabled:
        raise HTTPException(status_code=409, detail="WeKnora mode is not enabled")
    try:
        return build_weknora_client(runtime)
    except RuntimeError as exc:
        raise HTTPException(status_code=503, detail=str(exc)) from None


def _raise_upstream(exc: WeKnoraError) -> None:
    status = 422 if exc.status_code == 400 else exc.status_code if exc.status_code in {404, 502, 504} else 502
    raise HTTPException(status_code=status, detail=str(exc)) from None


def _embedding_model_view(source: dict[str, Any]) -> dict[str, Any] | None:
    """Return the small, credential-free model shape needed by the create dialog."""

    if str(source.get("type") or "").strip().lower() != "embedding":
        return None
    model_id = str(source.get("id") or "").strip()
    if not model_id:
        return None
    status = str(source.get("status") or "active").strip().lower() or "active"
    explicitly_available = source.get("available")
    available = (
        bool(explicitly_available)
        if isinstance(explicitly_available, bool)
        else status in {"active", "available", "enabled", "ready"}
    )
    parameters = source.get("parameters") if isinstance(source.get("parameters"), dict) else {}

    def safe_text(*keys: str) -> str:
        for key in keys:
            value = source.get(key)
            if isinstance(value, (str, int, float)) and str(value).strip():
                return str(value).strip()
        return ""

    nested_provider = parameters.get("provider")
    provider = safe_text("provider", "provider_type")
    if not provider and isinstance(nested_provider, str):
        provider = nested_provider.strip()
    if not provider:
        provider = safe_text("source", "platform")
    nested_model = parameters.get("model") or parameters.get("model_name")
    model_name = safe_text("model_name", "model")
    if not model_name and isinstance(nested_model, str):
        model_name = nested_model.strip()

    return {
        "id": model_id,
        "name": safe_text("display_name", "name", "model_name", "model") or model_name or model_id,
        "model_name": model_name,
        "provider": provider,
        "status": status,
        "is_default": source.get("is_default") is True,
        "available": available,
    }


async def _embedding_models(client: WeKnoraClient) -> list[dict[str, Any]]:
    return [view for item in await client.list_models() if (view := _embedding_model_view(item)) is not None]


def _select_embedding_model(models: list[dict[str, Any]], requested_id: str | None) -> str:
    available = [item for item in models if item["available"]]
    if requested_id:
        selected = next((item for item in models if item["id"] == requested_id), None)
        if selected is None:
            raise WeKnoraError("The selected WeKnora embedding model does not exist", status_code=400)
        if not selected["available"]:
            raise WeKnoraError("The selected WeKnora embedding model is unavailable", status_code=400)
        return requested_id
    selected = next((item for item in available if item["is_default"]), None) or next(iter(available), None)
    if selected is None:
        raise WeKnoraError(
            "WeKnora has no available embedding model; configure one in WeKnora before creating a knowledge base",
            status_code=400,
        )
    return str(selected["id"])


def _select_knowledge_qa_model(models: list[dict[str, Any]]) -> str:
    available = [
        item
        for item in models
        if str(item.get("type") or "").strip().lower() == "knowledgeqa"
        and str(item.get("id") or "").strip()
        and str(item.get("status") or "active").strip().lower() == "active"
    ]
    available.sort(
        key=lambda item: (
            0 if item.get("is_default") is True else 1,
            str(item.get("id") or "").strip(),
        )
    )
    if not available:
        raise WeKnoraError(
            "WeKnora has no active KnowledgeQA model; configure one before creating a Wiki",
            status_code=400,
        )
    return str(available[0]["id"]).strip()


def _json_response(payload: Any, *, status_code: int = 200) -> Response:
    return Response(
        content=json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
        status_code=status_code,
        media_type="application/json",
    )


def _runtime_for_iframe_proxy(request: Request | None = None):
    source_config = None
    if request is not None:
        source_config = getattr(request.app.state, "llmwiki_source_config", None)
    runtime = get_resolved_llmwiki_runtime(source_config)
    if not runtime.weknora_enabled:
        raise HTTPException(status_code=409, detail="WeKnora mode is not enabled")
    if not runtime.web_base_url:
        raise HTTPException(status_code=422, detail="WeKnora web_base_url is required for iframe embedding")
    if not runtime.admin_email or not runtime.admin_password:
        raise HTTPException(
            status_code=422,
            detail="Configure llmwiki.weknora.admin_email and admin_password for WeKnora iframe SSO",
        )
    return runtime


def _admin_session_cache_key(runtime: Any) -> str:
    digest = hashlib.sha256(runtime.admin_password.encode("utf-8")).hexdigest()[:12]
    return f"{runtime.api_base_url}|{runtime.admin_email}|{digest}"


def _extract_weknora_token(payload: dict[str, Any]) -> tuple[str, str, dict[str, Any] | None, dict[str, Any] | None]:
    source = payload.get("data") if isinstance(payload.get("data"), dict) else payload
    token = str(source.get("token") or source.get("access_token") or payload.get("token") or payload.get("access_token") or "")
    refresh_token = str(source.get("refresh_token") or source.get("refreshToken") or payload.get("refresh_token") or payload.get("refreshToken") or "")
    user = source.get("user") if isinstance(source.get("user"), dict) else payload.get("user")
    tenant = source.get("tenant") if isinstance(source.get("tenant"), dict) else source.get("active_tenant") if isinstance(source.get("active_tenant"), dict) else payload.get("tenant")
    return token, refresh_token, user if isinstance(user, dict) else None, tenant if isinstance(tenant, dict) else None


async def _get_weknora_admin_session(runtime: Any, *, force_refresh: bool = False) -> dict[str, Any]:
    key = _admin_session_cache_key(runtime)
    cached = _WEKNORA_ADMIN_SESSION_CACHE.get(key)
    now = time.time()
    if cached and not force_refresh and float(cached.get("expires_at") or 0) > now + 60:
        return cached

    pending = _WEKNORA_ADMIN_SESSION_TASKS.get(key)
    if pending is None:
        pending = asyncio.create_task(_login_weknora_admin_session(runtime))
        _WEKNORA_ADMIN_SESSION_TASKS[key] = pending

        def clear_pending(task: asyncio.Task[dict[str, Any]]) -> None:
            if _WEKNORA_ADMIN_SESSION_TASKS.get(key) is task:
                _WEKNORA_ADMIN_SESSION_TASKS.pop(key, None)
            if not task.cancelled():
                task.exception()  # Also retrieve failures if every caller disconnected.

        pending.add_done_callback(clear_pending)
    # A navigation cancelling one request must not cancel everybody's login.
    return await asyncio.shield(pending)


async def _login_weknora_admin_session(runtime: Any) -> dict[str, Any]:
    key = _admin_session_cache_key(runtime)
    now = time.time()
    try:
        async with httpx.AsyncClient(timeout=httpx.Timeout(20.0), follow_redirects=False, trust_env=False) as client:
            response = await client.post(
                f"{runtime.api_base_url}/api/v1/auth/login",
                json={"email": runtime.admin_email, "password": runtime.admin_password},
                headers={"Accept": "application/json"},
            )
    except httpx.TimeoutException as exc:
        logger.warning("WeKnora admin login timed out: base_url=%s", runtime.api_base_url)
        raise HTTPException(status_code=504, detail="WeKnora admin login timed out") from exc
    except httpx.HTTPError as exc:
        logger.warning("WeKnora admin login unavailable: base_url=%s error=%r", runtime.api_base_url, exc)
        raise HTTPException(status_code=502, detail="WeKnora admin login is unavailable") from exc

    if response.status_code >= 400:
        raise HTTPException(status_code=502, detail=f"WeKnora admin login failed ({response.status_code})")
    try:
        payload = response.json()
    except ValueError as exc:
        raise HTTPException(status_code=502, detail="WeKnora admin login returned invalid JSON") from exc
    if isinstance(payload, dict) and payload.get("success") is False:
        raise HTTPException(status_code=502, detail=str(payload.get("message") or "WeKnora admin login failed"))
    token, refresh_token, user, tenant = _extract_weknora_token(payload if isinstance(payload, dict) else {})
    if not token:
        raise HTTPException(status_code=502, detail="WeKnora admin login did not return a token")

    session = {
        "token": token,
        "refresh_token": refresh_token,
        "user": user or {},
        "tenant": tenant or {},
        # WeKnora tokens do not consistently expose expiry in the login body.
        # Relogin periodically and retry once on upstream 401.
        "expires_at": now + 25 * 60,
    }
    _WEKNORA_ADMIN_SESSION_CACHE[key] = session
    return session


def _tenant_id_from_session(session: dict[str, Any]) -> str:
    tenant = session.get("tenant") if isinstance(session.get("tenant"), dict) else {}
    user = session.get("user") if isinstance(session.get("user"), dict) else {}
    return str(tenant.get("id") or user.get("tenant_id") or "")


async def _read_embed_context(request: Request) -> WeKnoraEmbedContext:
    ctx = _verify_embed_cookie(request.cookies.get(_WEKNORA_EMBED_COOKIE))
    if not ctx.mapping_id or not ctx.weknora_id or not ctx.user_id:
        raise HTTPException(status_code=401, detail="WeKnora iframe session is invalid")
    store = _store(request)
    readable = await store.get_authorized(ctx.mapping_id, ctx.user_id, write=False, is_admin=ctx.is_admin)
    if readable is None or str(readable.get("weknora_id") or "") != ctx.weknora_id:
        raise HTTPException(status_code=403, detail="WeKnora iframe session is no longer authorized")
    writable = None
    if not _is_conversation_deposit_mapping(readable):
        writable = await store.get_authorized(ctx.mapping_id, ctx.user_id, write=True, is_admin=ctx.is_admin)
    return replace(
        ctx,
        can_write=writable is not None,
        is_conversation_deposit=_is_conversation_deposit_mapping(readable),
        allowed_weknora_ids=_embed_allowed_weknora_ids(ctx),
    )


def _embed_allowed_weknora_ids(ctx: WeKnoraEmbedContext) -> tuple[str, ...]:
    """Return the signed knowledge-base allowlist, always retaining the active base."""

    return tuple(dict.fromkeys(value for value in (ctx.weknora_id, *ctx.allowed_weknora_ids) if value))


def _embed_context_payload(ctx: WeKnoraEmbedContext) -> dict[str, Any]:
    # Keep this payload bounded: it is signed directly into the browser cookie.
    # The visible knowledge-base list is hydrated only for the embedded page or
    # its list API call, after the cookie has been verified.
    return {
        "mapping_id": ctx.mapping_id,
        "weknora_id": ctx.weknora_id,
        "user_id": ctx.user_id,
        "is_admin": ctx.is_admin,
        "can_write": ctx.can_write,
        "is_conversation_deposit": ctx.is_conversation_deposit,
        "exp": ctx.exp,
    }


async def _with_visible_embed_knowledge_bases(
    request: Request,
    ctx: WeKnoraEmbedContext,
) -> WeKnoraEmbedContext:
    """Hydrate the iframe-only picker allowlist without putting it in a cookie."""

    visible_rows = await _store(request).list_visible(
        ctx.user_id,
        scope="all",
        # Mirror the regular picker. An administrator must not receive another
        # user's private bases merely because the iframe uses an admin token.
        is_admin=False,
    )
    return replace(
        ctx,
        allowed_weknora_ids=tuple(
            dict.fromkeys(
                value
                for value in (
                    ctx.weknora_id,
                    *(str(item.get("weknora_id") or "") for item in visible_rows),
                )
                if value
            )
        ),
    )


def _embed_directory_time(value: Any) -> str | None:
    if isinstance(value, datetime):
        return value.isoformat()
    if isinstance(value, str) and value.strip():
        return value
    return None


async def _local_embed_knowledge_base_directory(
    request: Request,
    ctx: WeKnoraEmbedContext,
) -> Response:
    """Serve the iframe picker from DeerFlow mappings, never WeKnora's full directory."""

    rows = await _store(request).list_visible(
        ctx.user_id,
        scope="all",
        # Match the normal picker: the WeKnora service-admin token must not
        # make an administrator's private libraries discoverable to others.
        is_admin=False,
    )
    items = [
        {
            "id": str(row.get("weknora_id") or ""),
            "name": str(row.get("name") or ""),
            "description": str(row.get("description") or ""),
            "type": str(row.get("kb_type") or "document"),
            "status": "active",
            "created_at": _embed_directory_time(row.get("created_at")),
            "updated_at": _embed_directory_time(row.get("updated_at")),
        }
        for row in rows
        if str(row.get("weknora_id") or "")
    ]
    response = _json_response(
        {"success": True, "data": items, "total": len(items)},
    )
    response.headers["Cache-Control"] = "no-store"
    return response


def _weknora_id_from_platform_path(path: str) -> str | None:
    prefix = "/platform/knowledge-bases/"
    if not path.startswith(prefix):
        return None
    raw_id = path.removeprefix(prefix)
    if not raw_id or "/" in raw_id:
        return None
    return unquote(raw_id)


async def _switch_embed_context(
    request: Request,
    ctx: WeKnoraEmbedContext,
    weknora_id: str,
) -> WeKnoraEmbedContext:
    """Resolve a native-UI switch to the exact cmzs mapping.

    The embedded WeKnora UI uses a service-admin token, so the target must be
    checked against the current cmzs visibility rules on every switch. This is
    deliberately dynamic rather than a signed ID list in the browser cookie.
    """

    store = _store(request)
    mapped = await store.get_by_weknora_id(weknora_id)
    if mapped is None:
        raise HTTPException(status_code=403, detail="Knowledge base is not managed by cmzs")
    readable = await store.get_authorized(
        str(mapped["id"]),
        ctx.user_id,
        write=False,
        is_admin=ctx.is_admin,
    )
    if readable is None or str(readable.get("weknora_id") or "") != weknora_id:
        raise HTTPException(status_code=403, detail="Knowledge base is no longer authorized")
    writable = None
    if not _is_conversation_deposit_mapping(readable):
        writable = await store.get_authorized(
            str(readable["id"]),
            ctx.user_id,
            write=True,
            is_admin=ctx.is_admin,
        )
    return replace(
        ctx,
        mapping_id=str(readable["id"]),
        weknora_id=weknora_id,
        can_write=writable is not None,
        is_conversation_deposit=_is_conversation_deposit_mapping(readable),
    )


def _encode_upstream_path(path: str) -> str:
    if not path.startswith("/"):
        path = f"/{path}"
    return "/".join(quote(part, safe="") for part in path.split("/"))


def _safe_response_headers(headers: httpx.Headers) -> dict[str, str]:
    result: dict[str, str] = {}
    for key, value in headers.items():
        if key.lower() in _HOP_BY_HOP_RESPONSE_HEADERS:
            continue
        if key.lower() == "set-cookie":
            continue
        result[key] = value
    return result


def _should_follow_weknora_binary_redirect(path: str) -> bool:
    """Keep protected preview/image bytes inside the DeerFlow proxy."""

    segments = _path_segments(path)
    return (len(segments) == 5 and segments[:3] == ["api", "v1", "knowledge"] and segments[4] == "preview") or (len(segments) == 5 and segments[:3] == ["api", "v1", "knowledge-bases"] and segments[4] == "files")


def _is_weknora_content_mutation(method: str, path: str) -> bool:
    if method.upper() not in {"POST", "PUT", "PATCH", "DELETE"}:
        return False
    segments = _path_segments(path)
    if len(segments) < 3 or segments[:2] != ["api", "v1"]:
        return False
    return segments[2] in {"knowledge-bases", "knowledgebase", "knowledge", "chunks"}


async def _schedule_proxy_local_wiki_sync(
    request: Request,
    ctx: WeKnoraEmbedContext,
) -> None:
    """Bridge successful native WeKnora UI writes back to the local mirror."""

    try:
        mapping = await _store(request).get_authorized(
            ctx.mapping_id,
            ctx.user_id,
            write=False,
            is_admin=ctx.is_admin,
        )
        if mapping is not None and str(mapping.get("weknora_id") or "") == ctx.weknora_id:
            _schedule_local_wiki_sync(request, mapping)
    except Exception:
        # The upstream mutation already succeeded. A scheduling failure must
        # not turn that durable write into a false client-side failure; the
        # periodic scanner remains the fallback.
        logger.warning(
            "Could not schedule local Wiki sync after proxied mutation mapping_id=%s",
            ctx.mapping_id,
            exc_info=True,
        )


def _forward_request_headers(request: Request) -> dict[str, str]:
    blocked = {
        "host",
        "connection",
        "content-length",
        "cookie",
        "authorization",
        "x-forwarded-host",
        "x-forwarded-for",
        "x-forwarded-proto",
    }
    headers = {key: value for key, value in request.headers.items() if key.lower() not in blocked}
    headers.setdefault("Accept", "*/*")
    return headers


def _path_segments(path: str) -> list[str]:
    return [segment for segment in path.strip("/").split("/") if segment]


def _json_from_body(body: bytes, content_type: str | None) -> Any:
    if not body or "application/json" not in (content_type or "").lower():
        return None
    try:
        return json.loads(body)
    except ValueError:
        return None


def _body_kb_id(body: Any) -> str:
    return str(body.get("kb_id") or body.get("knowledge_base_id") or body.get("source_kb_id") or "") if isinstance(body, dict) else ""


def _request_kb_id(request: Request, body: Any) -> str:
    return _body_kb_id(body) or str(request.query_params.get("kb_id") or request.query_params.get("knowledge_base_id") or request.query_params.get("source_kb_id") or "")


async def _admin_json(runtime: Any, path: str, *, token: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
    try:
        async with httpx.AsyncClient(timeout=httpx.Timeout(20.0), follow_redirects=False, trust_env=False) as client:
            response = await client.get(
                f"{runtime.api_base_url}{_encode_upstream_path(path)}",
                params=params,
                headers={"Accept": "application/json", "Authorization": f"Bearer {token}"},
            )
    except httpx.HTTPError as exc:
        logger.warning("WeKnora verification request failed: base_url=%s path=%s error=%r", runtime.api_base_url, path, exc)
        raise HTTPException(status_code=502, detail="WeKnora verification request failed") from exc
    if response.status_code == 404:
        raise HTTPException(status_code=404, detail="WeKnora resource was not found")
    if response.status_code >= 400:
        raise HTTPException(status_code=502, detail=f"WeKnora verification failed ({response.status_code})")
    try:
        payload = response.json()
    except ValueError as exc:
        raise HTTPException(status_code=502, detail="WeKnora verification returned invalid JSON") from exc
    return payload if isinstance(payload, dict) else {}


def _payload_data(payload: dict[str, Any]) -> Any:
    return payload.get("data", payload)


async def _verify_proxy_document_belongs_to_kb(runtime: Any, ctx: WeKnoraEmbedContext, document_id: str, token: str) -> None:
    payload = await _admin_json(runtime, f"/api/v1/knowledge/{document_id}", token=token)
    data = _payload_data(payload)
    if not isinstance(data, dict) or str(data.get("knowledge_base_id") or "") != ctx.weknora_id:
        raise HTTPException(status_code=403, detail="Document is outside the authorized WeKnora knowledge base")


async def _verify_proxy_chunk_belongs_to_kb(runtime: Any, ctx: WeKnoraEmbedContext, chunk_id: str, token: str) -> None:
    payload = await _admin_json(runtime, f"/api/v1/chunks/by-id/{chunk_id}", token=token)
    data = _payload_data(payload)
    if not isinstance(data, dict) or str(data.get("knowledge_base_id") or "") != ctx.weknora_id:
        raise HTTPException(status_code=403, detail="Chunk is outside the authorized WeKnora knowledge base")


async def _authorize_weknora_api_proxy(
    request: Request,
    ctx: WeKnoraEmbedContext,
    runtime: Any,
    path: str,
    body: bytes,
    token: str,
) -> None:
    method = request.method.upper()
    if method == "OPTIONS":
        return
    read_method = method in {"GET", "HEAD"}
    write_method = method in {"POST", "PUT", "PATCH", "DELETE"}
    if write_method and not ctx.can_write:
        raise HTTPException(status_code=403, detail="Current cmzs user cannot write this knowledge base")

    segments = _path_segments(path)
    if len(segments) < 2 or segments[0] != "api" or segments[1] != "v1":
        raise HTTPException(status_code=403, detail="Unsupported WeKnora API path")
    tail = segments[2:]
    json_body = _json_from_body(body, request.headers.get("content-type"))

    if tail[:1] == ["auth"]:
        if len(tail) >= 2 and tail[1] in {"me", "tenant", "validate", "refresh"}:
            return
        raise HTTPException(status_code=403, detail="Only WeKnora session introspection is allowed")

    if read_method and tail == ["me", "invitations", "pending-count"]:
        return

    if read_method and tail == ["tenants", "kv", "retrieval-config"]:
        return

    if tail[:1] == ["knowledge-bases"]:
        if len(tail) == 1:
            if method == "GET":
                return
            raise HTTPException(status_code=403, detail="Creating or deleting other WeKnora knowledge bases is not allowed")
        if tail[1] != ctx.weknora_id:
            raise HTTPException(status_code=403, detail="Only the current WeKnora knowledge base is allowed")
        if (
            method == "POST"
            and tail[2:] == ["knowledge", "file"]
            and looks_like_knowledge_package(body)
        ):
            raise HTTPException(status_code=422, detail=_KNOWLEDGE_PACKAGE_UPLOAD_DETAIL)
        if read_method or (write_method and ctx.can_write):
            return

    if tail[:1] == ["knowledgebase"]:
        if len(tail) < 2 or tail[1] != ctx.weknora_id:
            raise HTTPException(status_code=403, detail="Only the current WeKnora wiki is allowed")
        if read_method or (write_method and ctx.can_write):
            return

    if tail[:1] == ["knowledge"]:
        if len(tail) == 1:
            raise HTTPException(status_code=403, detail="Global WeKnora knowledge collection is not allowed")
        action = tail[1]
        if read_method and action == "batch":
            ids = [item.strip() for raw in request.query_params.getlist("ids") for item in raw.split(",") if item.strip()]
            if not ids:
                raise HTTPException(status_code=403, detail="Batch operation must include document ids")
            for document_id in ids:
                await _verify_proxy_document_belongs_to_kb(runtime, ctx, document_id, token)
            return
        if action in {"batch", "batch-delete", "batch-reparse", "folder", "move", "tags"}:
            if _request_kb_id(request, json_body) != ctx.weknora_id:
                raise HTTPException(status_code=403, detail="Batch operation must be scoped to the current knowledge base")
            return
        if action in {"search"}:
            raise HTTPException(status_code=403, detail="Global WeKnora search is disabled in embedded detail mode")
        document_id = tail[2] if action == "manual" and len(tail) >= 3 else action
        await _verify_proxy_document_belongs_to_kb(runtime, ctx, document_id, token)
        return

    if tail[:1] == ["chunks"]:
        if len(tail) < 2:
            raise HTTPException(status_code=403, detail="Chunk path is incomplete")
        if tail[1] == "by-id":
            if len(tail) < 3:
                raise HTTPException(status_code=403, detail="Chunk path is incomplete")
            await _verify_proxy_chunk_belongs_to_kb(runtime, ctx, tail[2], token)
            return
        await _verify_proxy_document_belongs_to_kb(runtime, ctx, tail[1], token)
        return

    if tail[:1] == ["knowledge-search"]:
        ids = json_body.get("knowledge_base_ids") if isinstance(json_body, dict) else None
        if ids == [ctx.weknora_id] or ids == (ctx.weknora_id,):
            return
        raise HTTPException(status_code=403, detail="Search must be scoped to the current WeKnora knowledge base")

    # A few read-only metadata endpoints are required by the original detail
    # page to render upload controls and current-tenant state. They do not
    # expose other knowledge-base contents.
    if (
        read_method
        and tail
        and tail[0]
        in {
            "models",
            "system",
            "storage-backends",
            "vector-stores",
            "web-search-providers",
            "mcp-services",
        }
    ):
        return

    raise HTTPException(status_code=403, detail="This WeKnora endpoint is blocked in embedded detail mode")


def _filter_knowledge_base_list_response(response: httpx.Response, ctx: WeKnoraEmbedContext) -> bytes | None:
    content_type = response.headers.get("content-type", "")
    if "application/json" not in content_type.lower():
        return None
    try:
        payload = response.json()
    except ValueError:
        return None
    if not isinstance(payload, dict):
        return None

    allowed_weknora_ids = set(_embed_allowed_weknora_ids(ctx))

    def keep_allowed(value: Any) -> Any:
        if isinstance(value, list):
            return [item for item in value if isinstance(item, dict) and str(item.get("id") or "") in allowed_weknora_ids]
        if isinstance(value, dict):
            result = dict(value)
            for key in ("items", "list", "knowledge_bases", "knowledge", "records", "data"):
                if isinstance(result.get(key), list):
                    result[key] = keep_allowed(result[key])
                    result["total"] = len(result[key])
            return result
        return value

    if isinstance(payload.get("data"), list):
        payload["data"] = keep_allowed(payload["data"])
        payload["total"] = len(payload["data"])
    elif isinstance(payload.get("data"), dict):
        payload["data"] = keep_allowed(payload["data"])
    else:
        for key in ("items", "list", "knowledge_bases", "records"):
            if isinstance(payload.get(key), list):
                payload[key] = keep_allowed(payload[key])
                payload["total"] = len(payload[key])
    return json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")


def _escape_script_json(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, separators=(",", ":")).replace("</", "<\\/")


def _normalize_public_prefix(value: str | None) -> str:
    raw = (value or "").strip()
    if not raw:
        return ""
    if "://" in raw:
        try:
            raw = urlsplit(raw).path
        except ValueError:
            return ""
    if not raw.startswith("/"):
        raw = f"/{raw}"
    if (
        raw.startswith("//")
        or any(marker in raw for marker in ("\\", "?", "#"))
        or any(part in {".", ".."} for part in raw.split("/"))
        or any(ord(char) <= 32 for char in raw)
    ):
        return ""
    raw = raw.rstrip("/")
    return "" if raw == "/" else raw


def _request_public_prefix(request: Request) -> str:
    # The public backend prefix belongs to the reverse proxy, not to the
    # Gateway route itself.  Offline deployments use /deerflow-api by default
    # and pass it explicitly, while direct/local deployments retain the
    # historical /deerflow fallback.
    forwarded = _normalize_public_prefix(request.headers.get("x-forwarded-prefix"))
    scope = getattr(request, "scope", None)
    root_path = _normalize_public_prefix(
        str(scope.get("root_path") or "") if isinstance(scope, dict) else ""
    )
    return forwarded or root_path or _WEKNORA_PUBLIC_PREFIX


def _with_public_prefix(prefix: str, path: str) -> str:
    if not path.startswith("/"):
        path = f"/{path}"
    return f"{prefix}{path}" if prefix else path


def _without_public_prefix(prefix: str, path: str) -> str:
    if prefix and path.startswith(f"{prefix}/"):
        return path[len(prefix) :]
    if prefix and path == prefix:
        return "/"
    return path


def _rewrite_weknora_html_public_paths(
    html: str,
    public_prefix: str,
    *,
    rewrite_relative: bool = True,
) -> str:
    if not public_prefix:
        return html
    for path in _WEKNORA_PUBLIC_ROOT_PATHS:
        if path not in html and path.replace("/", "\\/") not in html:
            continue
        prefixed = f"{public_prefix}{path}"
        html = re.sub(
            rf"(?P<prefix>[\"'`(=:\s]){re.escape(path)}(?=[/?#\"'`),;\]\}}\s>]|$)",
            rf"\g<prefix>{prefixed}",
            html,
        )
        escaped_path = rf"\\/{re.escape(path.lstrip('/'))}"
        escaped_prefixed = f"\\/{prefixed.lstrip('/')}"
        html = re.sub(
            rf"(?<!\\/{re.escape(public_prefix.lstrip('/'))}){escaped_path}(?=[\\/?#\"'`),;\]\}}\s>]|$)",
            escaped_prefixed,
            html,
        )
    if rewrite_relative:
        for path in _WEKNORA_PUBLIC_RELATIVE_ROOT_PATHS:
            prefixed = f"{public_prefix}/{path}"
            html = re.sub(
                rf"(?P<prefix>[\"'`(=:\s])(?:\./)?{re.escape(path)}(?=/)",
                rf"\g<prefix>{prefixed}",
                html,
            )
        for path in _WEKNORA_PUBLIC_RELATIVE_FILE_PATHS:
            prefixed = f"{public_prefix}/{path}"
            html = re.sub(
                rf"(?P<prefix>[\"'`(=:\s])(?:\./)?{re.escape(path)}(?=[?#\"'`),;\]\}}\s>]|$)",
                rf"\g<prefix>{prefixed}",
                html,
            )
    return html


def _rewrite_weknora_script_public_paths(text: str, public_prefix: str) -> str:
    if not public_prefix:
        return text
    for path in _WEKNORA_PUBLIC_ROOT_PATHS:
        if path not in text:
            continue
        prefixed = f"{public_prefix}{path}"
        text = re.sub(
            rf"(?P<quote>[\"'`]){re.escape(path)}(?=[/?#\"'`),;\]\}}\s]|$)",
            rf"\g<quote>{prefixed}",
            text,
        )
    return _rewrite_weknora_vite_dependency_map(text, public_prefix)


def _should_rewrite_weknora_static_response(content_type: str, upstream_path: str) -> bool:
    lower_type = content_type.lower()
    if any(
        marker in lower_type
        for marker in (
            "javascript",
            "text/css",
            "application/json",
            "manifest",
            "image/svg+xml",
        )
    ):
        return True
    lower_path = upstream_path.lower()
    return lower_path.endswith((".js", ".mjs", ".css", ".json", ".webmanifest", ".svg"))


def _rewrite_weknora_vite_dependency_map(text: str, public_prefix: str) -> str:
    if not public_prefix or "__vite__mapDeps" not in text:
        return text
    marker = "m.f||(m.f=["
    start = text.find(marker)
    if start < 0:
        return text
    end = text.find("]))", start + len(marker))
    if end < 0:
        return text
    prefix = public_prefix.strip("/")
    dependency_map = text[start:end]
    dependency_map = re.sub(
        r'(?P<quote>["\'])assets/',
        rf"\g<quote>{prefix}/assets/",
        dependency_map,
    )
    return f"{text[:start]}{dependency_map}{text[end:]}"


def _weknora_static_fallback_upstream_path(request: Request) -> str | None:
    path = _without_public_prefix(
        _request_public_prefix(request),
        str(request.scope.get("path") or request.url.path),
    )
    if not path.startswith("/"):
        path = f"/{path}"
    lower_path = path.lower()
    if lower_path.startswith("/api/"):
        return None
    if lower_path.startswith(_WEKNORA_STATIC_ROOT_PREFIXES):
        return path
    for root_prefix in _WEKNORA_STATIC_ROOT_PREFIXES:
        index = lower_path.find(root_prefix)
        if index > 0:
            return path[index:]
    for asset_file in _WEKNORA_ROOT_ASSET_FILES:
        if lower_path.endswith(asset_file):
            return asset_file
    if lower_path.endswith(_WEKNORA_STATIC_FILE_SUFFIXES):
        return path
    return None


def _is_weknora_static_asset_path(path: str) -> bool:
    if any(part in {".", ".."} for part in path.split("/")) or "\\" in path or "\x00" in path:
        return False
    lower = path.lower()
    return lower in _WEKNORA_ROOT_ASSET_FILES or (lower.startswith(_WEKNORA_STATIC_ROOT_PREFIXES) and lower.endswith(_WEKNORA_STATIC_FILE_SUFFIXES))


def _prepare_weknora_static_asset(content: bytes, content_type: str, path: str, public_prefix: str, *, cacheable: bool) -> StaticAsset:
    # All full-bundle regex passes, hashing and gzip work run off the event loop.
    if _should_rewrite_weknora_static_response(content_type, path):
        text = content.decode("utf-8", errors="replace")
        if "javascript" in content_type.lower() or path.lower().endswith((".js", ".mjs")):
            text = _rewrite_weknora_script_public_paths(text, public_prefix)
        else:
            text = _rewrite_weknora_html_public_paths(text, public_prefix, rewrite_relative=False)
        content = text.encode("utf-8")
    fingerprint = re.search(r"[-.]([A-Za-z0-9_-]{8,32})\.[A-Za-z0-9]+$", path)
    versioned = path.startswith("/assets/") and fingerprint is not None and re.search(r"[0-9A-Z]", fingerprint[1]) is not None
    if not cacheable:
        control, ttl = "private, no-store", 0.0
    elif path.lower() in {"/config.js", "/runtime-config.js"}:
        control, ttl = "private, no-cache", 0.0
    elif versioned:
        control, ttl = "private, max-age=31536000, immutable", 3600.0
    else:
        control, ttl = "private, max-age=300", 300.0
    return prepare_static_asset(content, content_type=content_type, cache_control=control, ttl=ttl)


async def _proxy_weknora_static_asset(request: Request, upstream_path: str) -> Response:
    """Serve only the public app bundle; HTML and all data APIs use full ACLs."""
    if not _is_weknora_static_asset_path(upstream_path):
        raise HTTPException(status_code=404, detail="WeKnora static resource not found")
    ctx = _verify_embed_cookie(request.cookies.get(_WEKNORA_EMBED_COOKIE))
    if not ctx.mapping_id or not ctx.weknora_id or not ctx.user_id:
        raise HTTPException(status_code=401, detail="WeKnora iframe session is invalid")
    runtime = _runtime_for_iframe_proxy(request)
    public_prefix = _request_public_prefix(request)
    cache = getattr(request.app.state, "llmwiki_static_assets", None)
    if cache is None:
        cache = StaticAssetCache()
        request.app.state.llmwiki_static_assets = cache
    key = ("static-rewrite-v1", runtime.web_base_url, public_prefix, upstream_path, str(request.query_params))

    async def load() -> StaticAsset:
        try:
            async with _proxy_http_client(request) as client:
                # No identity, cookies, Range or conditional validators are sent
                # upstream. ETags describe our rewritten/gzipped representation.
                response = await client.request(
                    "GET",
                    f"{runtime.web_base_url.rstrip('/')}{_encode_upstream_path(upstream_path)}",
                    params=request.query_params.multi_items(),
                    headers={"Accept": "*/*", "Accept-Encoding": "gzip, deflate"},
                    follow_redirects=False,
                )
        except httpx.TimeoutException as exc:
            raise HTTPException(status_code=504, detail="WeKnora static resource request timed out") from exc
        except httpx.HTTPError as exc:
            raise HTTPException(status_code=502, detail="WeKnora static resource request failed") from exc
        content_type = response.headers.get("content-type", "").lower()
        allowed_types = (
            "application/javascript",
            "application/x-javascript",
            "text/javascript",
            "text/css",
            "application/json",
            "text/json",
            "application/manifest+json",
            "image/",
            "font/",
            "application/font",
            "application/x-font",
            "application/vnd.ms-fontobject",
            "application/octet-stream",
            "application/wasm",
            "text/plain",
        )
        is_html = response.content.lstrip()[:32].lower().startswith((b"<!doctype html", b"<html"))
        if response.status_code != 200 or not content_type.startswith(allowed_types) or is_html:
            # A web-server SPA fallback is HTML, never an asset or an injected
            # account session. Do not cache/serve it under a script/CSS URL.
            raise HTTPException(status_code=404 if response.status_code in {200, 404} else 502, detail="WeKnora static resource is unavailable")
        upstream_control = response.headers.get("cache-control", "").lower()
        requires_revalidation = "no-cache" in upstream_control or re.search(r"(?:^|,)\s*(?:s-maxage|max-age)\s*=\s*\"?0\"?\s*(?:,|$)", upstream_control) is not None
        cacheable = (
            "set-cookie" not in response.headers
            and "private" not in upstream_control
            and "no-store" not in upstream_control
            and not requires_revalidation
            and all(item.strip().lower() in {"", "accept-encoding"} for item in response.headers.get("vary", "").split(","))
        )
        return await asyncio.to_thread(_prepare_weknora_static_asset, response.content, content_type, upstream_path, public_prefix, cacheable=cacheable)

    request_control = request.headers.get("cache-control", "").lower()
    refresh = "no-cache" in request_control or "max-age=0" in request_control or "no-cache" in request.headers.get("pragma", "").lower()
    return (await cache.get(key, load, refresh=refresh)).response(request)


def _rewrite_weknora_location_header(location: str, public_prefix: str, target_base_url: str) -> str:
    if not location or not public_prefix:
        return location
    try:
        current = urlsplit(location)
        upstream = urlsplit(target_base_url)
    except ValueError:
        return location

    def proxied_path(path: str) -> str:
        upstream_base_path = upstream.path.rstrip("/")
        if upstream_base_path and (path == upstream_base_path or path.startswith(f"{upstream_base_path}/")):
            path = path[len(upstream_base_path) :] or "/"
        return _without_public_prefix(public_prefix, path)

    suffix = ""
    if current.query:
        suffix += f"?{current.query}"
    if current.fragment:
        suffix += f"#{current.fragment}"

    if not current.scheme and location.startswith("/"):
        path = proxied_path(current.path or "/")
        return f"{public_prefix}{path}{suffix}"

    if current.scheme and current.netloc and current.netloc == upstream.netloc:
        path = proxied_path(current.path or "/")
        return f"{public_prefix}{path}{suffix}"
    return location


def _weknora_embed_injection(
    *,
    allowed_path: str,
    tab: str,
    session: dict[str, Any],
    ctx: WeKnoraEmbedContext,
    public_prefix: str = "",
) -> str:
    tenant_id = _tenant_id_from_session(session)
    user = session.get("user") if isinstance(session.get("user"), dict) else {}
    tenant = session.get("tenant") if isinstance(session.get("tenant"), dict) else {}
    payload = {
        "allowedPath": allowed_path,
        "tab": tab,
        "canWrite": ctx.can_write,
        "token": session.get("token") or "",
        "refreshToken": session.get("refresh_token") or "",
        "tenantId": tenant_id,
        "user": user,
        "tenant": tenant,
        "weknoraId": ctx.weknora_id,
        "allowedWeKnoraIds": list(_embed_allowed_weknora_ids(ctx)),
        "publicPrefix": public_prefix,
        "apiPrefix": _with_public_prefix(public_prefix, _WEKNORA_EMBED_API_PREFIX),
    }
    data = _escape_script_json(payload)
    return f"""
<style id="deerflow-weknora-embed-style">
html.deerflow-weknora-embed,
html.deerflow-weknora-embed body,
html.deerflow-weknora-embed #app {{
  width: 100%;
  height: 100%;
  margin: 0;
  overflow: hidden;
  background: var(--td-bg-color-page) !important;
  color: var(--td-text-color-primary) !important;
}}
html.deerflow-weknora-embed body {{
  opacity: 1 !important;
}}
html.deerflow-weknora-embed .main {{
  min-width: 0 !important;
  background: var(--td-bg-color-page) !important;
}}
html.deerflow-weknora-embed .platform-route-outlet,
html.deerflow-weknora-embed .knowledge-layout {{
  width: 100% !important;
  min-width: 0 !important;
  background: var(--td-bg-color-page) !important;
}}
html.deerflow-weknora-embed [data-deerflow-weknora-chrome="hidden"],
html.deerflow-weknora-embed #app > .main > .aside_box,
html.deerflow-weknora-embed #app > aside,
html.deerflow-weknora-embed #app > .sidebar,
html.deerflow-weknora-embed #app > .side-bar,
html.deerflow-weknora-embed #app > .side-menu,
html.deerflow-weknora-embed #app > .t-layout__sider,
html.deerflow-weknora-embed #app > .t-layout-sider,
html.deerflow-weknora-embed #app > .layout-sider,
html.deerflow-weknora-embed #app > .layout-sidebar,
html.deerflow-weknora-embed #app > .platform-sidebar,
html.deerflow-weknora-embed #app > .platform-sider,
html.deerflow-weknora-embed #app > .app-sidebar,
html.deerflow-weknora-embed #app > .app-sider,
html.deerflow-weknora-embed #app > .t-layout > aside,
html.deerflow-weknora-embed #app > .t-layout > .t-layout__sider,
html.deerflow-weknora-embed #app > .t-layout > .t-layout-sider,
html.deerflow-weknora-embed #app > .layout > aside,
html.deerflow-weknora-embed #app > .layout > .layout-sider,
html.deerflow-weknora-embed #app > .layout > .layout-sidebar {{
  display: none !important;
  visibility: hidden !important;
  pointer-events: none !important;
}}
html.deerflow-weknora-embed [data-deerflow-weknora-content="true"],
html.deerflow-weknora-embed #app,
html.deerflow-weknora-embed .t-layout,
html.deerflow-weknora-embed .t-layout__content,
html.deerflow-weknora-embed .layout-content,
html.deerflow-weknora-embed .platform-content {{
  margin-left: 0 !important;
  padding-left: 0 !important;
  left: 0 !important;
  width: 100% !important;
  max-width: none !important;
}}
html.deerflow-weknora-embed .document-header {{
  padding-left: 32px !important;
  padding-right: 32px !important;
  background: var(--td-bg-color-page) !important;
}}
html.deerflow-weknora-embed .breadcrumb-link:first-child {{
  pointer-events: none !important;
  cursor: default !important;
}}
html.deerflow-weknora-embed .breadcrumb-link:first-child {{
  color: var(--td-text-color-primary) !important;
}}
html.deerflow-weknora-embed[data-deerflow-can-write="false"] .kb-settings-button,
html.deerflow-weknora-embed[data-deerflow-can-write="false"] .document-upload-input,
html.deerflow-weknora-embed[data-deerflow-can-write="false"] .kb-upload-source-dropdown,
html.deerflow-weknora-embed[data-deerflow-can-write="false"] .t-button--theme-primary {{
  display: none !important;
}}
html.deerflow-weknora-embed .t-guide,
html.deerflow-weknora-embed .t-guide__wrapper,
html.deerflow-weknora-embed .t-guide__reference,
html.deerflow-weknora-embed .t-popup[data-popper-placement][class*="guide"],
html.deerflow-weknora-embed [data-guide-highlight],
html.deerflow-weknora-embed .contextual-guide,
html.deerflow-weknora-embed .guide__backdrop,
html.deerflow-weknora-embed [class^="guide__"],
html.deerflow-weknora-embed [class*=" guide__"] {{
  display: none !important;
}}
html.deerflow-source-detail-open [data-deerflow-source-detail-hidden="true"] {{
  display: none !important;
}}
html.deerflow-source-detail-open [data-deerflow-source-detail-panel="true"] {{
  position: fixed !important;
  inset: 0 !important;
  width: 100vw !important;
  height: 100vh !important;
  max-width: none !important;
  max-height: none !important;
  margin: 0 !important;
  transform: none !important;
  z-index: 2147483000 !important;
  background: var(--td-bg-color-page, #fff) !important;
  box-shadow: none !important;
  border-radius: 0 !important;
}}
html.deerflow-source-detail-open [data-deerflow-source-detail-panel="true"] [class*="drawer"],
html.deerflow-source-detail-open [data-deerflow-source-detail-panel="true"] [class*="Drawer"],
html.deerflow-source-detail-open [data-deerflow-source-detail-panel="true"] [class*="dialog"],
html.deerflow-source-detail-open [data-deerflow-source-detail-panel="true"] [class*="Dialog"] {{
  max-width: none !important;
}}
</style>
<script id="deerflow-weknora-embed-guard">
(() => {{
  const cfg = {data};
  const root = document.documentElement;
  root.classList.add("deerflow-weknora-embed");
  root.dataset.deerflowCanWrite = cfg.canWrite ? "true" : "false";

  function safeSet(key, value) {{
    try {{
      if (value === undefined || value === null || value === "") return;
      localStorage.setItem(key, typeof value === "string" ? value : JSON.stringify(value));
    }} catch (_) {{}}
  }}
  function safeSessionSet(key, value) {{
    try {{
      if (value === undefined || value === null || value === "") return;
      sessionStorage.setItem(key, typeof value === "string" ? value : JSON.stringify(value));
    }} catch (_) {{}}
  }}
  safeSet("weknora_token", cfg.token);
  safeSet("weknora_refresh_token", cfg.refreshToken);
  safeSet("weknora_selected_tenant_id", cfg.tenantId);
  safeSet("weknora_user", cfg.user);
  safeSet("weknora_tenant", cfg.tenant);
  [
    "weknora:contextual-guide-agent-create:v1",
    "weknora:contextual-guide-agent-list:v1",
    "weknora:contextual-guide-chat:v1",
    "weknora:contextual-guide-kb-create:v3",
    "weknora:contextual-guide-kb-detail:v1",
    "weknora:contextual-guide-kb-list:v2",
    "weknora:contextual-guide-tenant-models:v1",
    "weknora:new-user-guide-done:v1",
    "weknora:onboarding:workspace:v1",
    "weknora:onboarding:completed"
  ].forEach((key) => {{
    safeSet(key, "1");
    safeSessionSet(key, "1");
  }});

  function rewriteApiUrl(value) {{
    try {{
      const protectedUrl = protectedImageApiUrl(value);
      if (protectedUrl) return protectedUrl;
      const next = new URL(value, location.origin);
      if (next.origin !== location.origin) return value;
      const publicPrefix = cfg.publicPrefix || "";
      const apiPath = "/api/v1/";
      const publicApiPath = publicPrefix + apiPath;
      let upstreamPath = "";
      if (next.pathname.startsWith(publicApiPath)) {{
        upstreamPath = next.pathname.slice(publicPrefix.length);
      }} else if (next.pathname.startsWith(apiPath)) {{
        upstreamPath = next.pathname;
      }} else {{
        return value;
      }}
      next.pathname = cfg.apiPrefix + upstreamPath;
      return next.href;
    }} catch (_) {{
      return value;
    }}
  }}
  function protectedImageApiUrl(value) {{
    const raw = String(value || "").trim();
    if (!/^(?:storage:\\/\\/[0-9A-Za-z_-]+\\/)?(?:resource|local|minio|cos|tos|s3|oss|ks3|obs):\\/\\//i.test(raw)) {{
      return "";
    }}
    const endpoint =
      cfg.apiPrefix +
      "/api/v1/knowledge-bases/" +
      encodeURIComponent(cfg.weknoraId) +
      "/files?file_path=" +
      encodeURIComponent(raw);
    return new URL(endpoint, location.origin).href;
  }}
  const nativeFetch = window.fetch.bind(window);
  window.fetch = (input, init) => {{
    if (input instanceof Request) {{
      return nativeFetch(new Request(rewriteApiUrl(input.url), input), init);
    }}
    return nativeFetch(rewriteApiUrl(input), init);
  }};
  const nativeXhrOpen = XMLHttpRequest.prototype.open;
  XMLHttpRequest.prototype.open = function(method, url, ...rest) {{
    return nativeXhrOpen.call(this, method, rewriteApiUrl(url), ...rest);
  }};

  // WeKnora renders protected Markdown images as transparent placeholders and
  // hydrates them asynchronously.  In an embedded page that hydration can run
  // before its own API client has picked up our proxy prefix, leaving a 1x1
  // GIF forever.  Resolve the same protected source through the scoped proxy
  // as a fallback.  The backend validates the scheme and current KB again.
  function hydrateProtectedImage(image) {{
    const protectedSrc = String(image?.getAttribute?.("data-protected-src") || "").trim();
    const proxyUrl = protectedImageApiUrl(protectedSrc);
    if (!proxyUrl) return;
    if (
      image.dataset.deerflowProtectedSource === protectedSrc &&
      String(image.getAttribute("src") || "") === proxyUrl
    ) return;
    // Keep the real HTTP proxy URL on the DOM node.  WeKnora's preview/lightbox
    // reuses the thumbnail's src when the user clicks it; a short-lived blob URL
    // may render once and then become invalid before the preview opens.
    image.dataset.deerflowProtectedSource = protectedSrc;
    image.removeAttribute("data-img-loading");
    image.removeAttribute("data-wiki-image-error");
    image.src = proxyUrl;
  }}
  function hydrateProtectedImages(scope = document) {{
    if (scope?.matches?.("img[data-protected-src]")) hydrateProtectedImage(scope);
    scope?.querySelectorAll?.("img[data-protected-src]").forEach(hydrateProtectedImage);
  }}
  const protectedImageObserver = new MutationObserver((mutations) => {{
    mutations.forEach((mutation) => {{
      if (mutation.type === "attributes") hydrateProtectedImage(mutation.target);
      mutation.addedNodes?.forEach((node) => {{
        if (node?.nodeType === Node.ELEMENT_NODE) hydrateProtectedImages(node);
      }});
    }});
  }});
  protectedImageObserver.observe(document.documentElement, {{
    childList: true,
    subtree: true,
    attributes: true,
    attributeFilter: ["data-protected-src", "src"]
  }});
  hydrateProtectedImages();

  function tone(color, alpha) {{
    return `color-mix(in srgb, ${{color}} ${{Math.round(alpha * 100)}}%, transparent)`;
  }}
  let themeObserver = null;
  function applyTheme(message) {{
    if (!message || message.type !== "deerflow-weknora-theme") return;
    window.__deerflowWeKnoraTheme = message;
    const vars = message.vars || {{}};
    const dark = message.theme === "dark";
    const mode = dark ? "dark" : "light";
    if (root.classList.contains("dark") !== dark) root.classList.toggle("dark", dark);
    for (const name of ["theme-mode", "data-theme-mode", "data-theme"]) {{
      if (root.getAttribute(name) !== mode) root.setAttribute(name, mode);
    }}
    const pairs = {{
      "--td-brand-color": vars.primary,
      "--td-brand-color-hover": vars.primary,
      "--td-brand-color-active": vars.primary,
      "--td-brand-color-light": tone(vars.primary || "#0062d9", dark ? 0.22 : 0.12),
      "--td-brand-color-focus": tone(vars.primary || "#0062d9", 0.22),
      "--td-success-color": vars.primary,
      "--td-success-color-light": tone(vars.primary || "#0062d9", dark ? 0.20 : 0.10),
      "--td-bg-color-page": vars.background,
      "--td-bg-color-container": vars.card,
      "--td-bg-color-secondarycontainer": vars.muted,
      "--td-bg-color-component": vars.muted,
      "--td-bg-color-component-hover": tone(vars.primary || "#0062d9", dark ? 0.20 : 0.08),
      "--td-bg-color-container-hover": tone(vars.primary || "#0062d9", dark ? 0.18 : 0.06),
      "--td-text-color-primary": vars.foreground,
      "--td-text-color-secondary": vars.mutedForeground,
      "--td-text-color-placeholder": vars.mutedForeground,
      "--td-text-color-disabled": tone(vars.mutedForeground || "#8a8f98", 0.65),
      "--td-component-stroke": vars.border,
      "--td-component-border": vars.border,
      "--td-error-color": vars.destructive,
      "--td-font-family": "var(--app-font-family, Inter, system-ui, -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif)"
    }};
    for (const [key, value] of Object.entries(pairs)) {{
      if (value && root.style.getPropertyValue(key) !== value) root.style.setProperty(key, value);
    }}
    if (root.style.colorScheme !== mode) root.style.colorScheme = mode;
    try {{
      for (const key of ["theme-mode", "tdesign-starter-theme-mode", "weknora-theme-mode"]) {{
        if (localStorage.getItem(key) !== mode) localStorage.setItem(key, mode);
      }}
    }} catch (_) {{}}
    // Ignore our own root writes, but keep watching for WeKnora overwriting
    // them later. Content mutations must never schedule another theme write.
    themeObserver?.takeRecords();
  }}
  window.addEventListener("message", (event) => applyTheme(event.data));
  const urlTheme = new URLSearchParams(location.search).get("theme");
  if (urlTheme === "dark" || urlTheme === "light") {{
    applyTheme({{
      type: "deerflow-weknora-theme",
      theme: urlTheme,
      vars: {{
        primary: urlTheme === "dark" ? "#2b81ff" : "#0062d9",
        background: urlTheme === "dark" ? "#0e172b" : "#ffffff",
        foreground: urlTheme === "dark" ? "#eff2f5" : "#202a3a",
        card: urlTheme === "dark" ? "#1a2539" : "#ffffff",
        muted: urlTheme === "dark" ? "#232e40" : "#f3f5f8",
        mutedForeground: urlTheme === "dark" ? "#8e9aa4" : "#7f7f7f",
        border: urlTheme === "dark" ? "rgba(255,255,255,0.12)" : "#dbdbdb",
        destructive: "#d54941"
      }}
    }});
  }}

  const launchParams = new URLSearchParams(location.search);
  const requestedWikiSlug = launchParams.get("wikiSlug") || launchParams.get("slug") || "";
  const requestedDocumentId =
    launchParams.get("documentId") ||
    launchParams.get("knowledgeId") ||
    launchParams.get("knowledge_id") ||
    "";
  const sourceDetailMode = Boolean(requestedDocumentId || requestedWikiSlug);
  if (sourceDetailMode) {{
    root.classList.add("deerflow-source-detail-request");
  }}
  let requestedWikiOpened = false;
  let requestedDocumentOpened = false;
  let sourceDetailAttempts = 0;
  const maxSourceDetailAttempts = 18;
  function escapeAttr(value) {{
    if (window.CSS && typeof window.CSS.escape === "function") return window.CSS.escape(value);
    return String(value).replace(/["\\\\]/g, "\\\\$&");
  }}
  function openRequestedWikiSlug() {{
    if (!requestedWikiSlug || requestedWikiOpened || root.classList.contains("deerflow-source-detail-open")) return;
    const slug = escapeAttr(requestedWikiSlug);
    const link =
      document.querySelector(`[data-slug="${{slug}}"]`) ||
      document.querySelector(`a[href*="${{encodeURIComponent(requestedWikiSlug)}}"]`) ||
      document.querySelector(`a[href*="${{requestedWikiSlug.replace(/"/g, "%22")}}"]`);
    if (!link || typeof link.click !== "function") return;
    requestedWikiOpened = true;
    link.click();
  }}
  function openRequestedDocumentId() {{
    if (!requestedDocumentId || requestedDocumentOpened || root.classList.contains("deerflow-source-detail-open")) return;
    const id = escapeAttr(requestedDocumentId);
    const encoded = encodeURIComponent(requestedDocumentId);
    const candidates = [
      `[data-id="${{id}}"]`,
      `[data-row-key="${{id}}"]`,
      `[data-knowledge-id="${{id}}"]`,
      `[data-document-id="${{id}}"]`,
      `a[href*="${{encoded}}"]`,
      `button[href*="${{encoded}}"]`
    ];
    let node = null;
    for (const selector of candidates) {{
      node = document.querySelector(selector);
      if (node) break;
    }}
    if (!node) {{
      node = Array.from(document.querySelectorAll("[data-id],[data-row-key],[data-knowledge-id],[data-document-id]"))
        .find((item) => {{
          const values = [
            item.getAttribute("data-id"),
            item.getAttribute("data-row-key"),
            item.getAttribute("data-knowledge-id"),
            item.getAttribute("data-document-id")
          ];
          return values.includes(requestedDocumentId);
        }}) || null;
    }}
    if (!node) return;
    const clickable = node.closest("button,a,[role='button'],.t-card,.document-card,.knowledge-card,.file-card,.t-list-item,.t-table__row") || node;
    if (!clickable || typeof clickable.click !== "function") return;
    requestedDocumentOpened = true;
    clickable.click();
  }}
  function sourceDetailCandidateScore(node) {{
    if (!node || node === document.body || node === document.documentElement) return 0;
    const rect = node.getBoundingClientRect();
    if (rect.width < 360 || rect.height < 260) return 0;
    const text = (node.textContent || "").slice(0, 1200);
    const className = String(node.className || "");
    let score = 0;
    if (/基本信息|上传时间|创建时间|摘要|文档内容|查看分块|预览|引用内容|Wiki 原文|全文/.test(text)) score += 4;
    if (/drawer|Drawer|dialog|Dialog|modal|Modal|detail|Detail|preview|Preview/.test(className)) score += 3;
    if (rect.right > window.innerWidth * 0.65 && rect.width >= window.innerWidth * 0.38) score += 2;
    if (getComputedStyle(node).position === "fixed") score += 2;
    return score;
  }}
  function findSourceDetailPanel() {{
    const selectors = [
      ".t-drawer",
      ".t-drawer__content",
      ".t-dialog",
      ".t-dialog__ctx",
      ".document-detail",
      ".knowledge-detail",
      ".document-preview",
      "[class*='Drawer']",
      "[class*='drawer']",
      "[class*='Dialog']",
      "[class*='dialog']",
      "[class*='Detail']",
      "[class*='detail']",
      "[class*='Preview']",
      "[class*='preview']"
    ].join(",");
    let best = null;
    let bestScore = 0;
    document.querySelectorAll(selectors).forEach((node) => {{
      const score = sourceDetailCandidateScore(node);
      if (score > bestScore) {{
        best = node;
        bestScore = score;
      }}
    }});
    return bestScore >= 5 ? best : null;
  }}
  function isolateSourceDetailPanel() {{
    if (!sourceDetailMode) return;
    if (root.classList.contains("deerflow-source-detail-open")) return;
    const panel = findSourceDetailPanel();
    if (!panel) return;
    panel.dataset.deerflowSourceDetailPanel = "true";
    document.querySelectorAll("[data-deerflow-source-detail-hidden='true']").forEach((node) => {{
      if (node.contains(panel) || panel.contains(node)) return;
      delete node.dataset.deerflowSourceDetailHidden;
    }});
    Array.from(document.body.children).forEach((child) => {{
      if (child === panel || child.contains(panel)) return;
      child.dataset.deerflowSourceDetailHidden = "true";
    }});
    root.classList.add("deerflow-source-detail-open");
  }}
  function tickSourceDetail() {{
    if (!sourceDetailMode || root.classList.contains("deerflow-source-detail-open")) return true;
    sourceDetailAttempts += 1;
    openRequestedDocumentId();
    openRequestedWikiSlug();
    isolateSourceDetailPanel();
    if (root.classList.contains("deerflow-source-detail-open")) return true;
    if (sourceDetailAttempts >= maxSourceDetailAttempts) {{
      root.classList.add("deerflow-source-detail-open");
      return true;
    }}
    return false;
  }}

  const allowedPath = cfg.allowedPath;
  const allowedWeKnoraIds = new Set(
    [cfg.weknoraId, ...(Array.isArray(cfg.allowedWeKnoraIds) ? cfg.allowedWeKnoraIds : [])]
      .filter(Boolean)
  );
  const knowledgeBasePathPrefix = (cfg.publicPrefix || "") + "/platform/knowledge-bases/";
  function isAllowedKnowledgeBasePath(pathname) {{
    if (!pathname.startsWith(knowledgeBasePathPrefix)) return false;
    const encodedId = pathname.slice(knowledgeBasePathPrefix.length);
    if (!encodedId || encodedId.includes("/")) return false;
    try {{ return allowedWeKnoraIds.has(decodeURIComponent(encodedId)); }} catch (_) {{ return false; }}
  }}
  function allowedUrl(url) {{
    try {{
      const next = new URL(url, location.origin);
      if (next.origin !== location.origin) return null;
      if (next.pathname === allowedPath || isAllowedKnowledgeBasePath(next.pathname)) {{
        return next.pathname + next.search + next.hash;
      }}
      const query = new URLSearchParams(location.search);
      if (cfg.tab && !query.has("tab")) query.set("tab", cfg.tab);
      return allowedPath + (query.toString() ? "?" + query.toString() : "") + location.hash;
    }} catch (_) {{
      return allowedPath + location.search + location.hash;
    }}
  }}
  function enforceLocation() {{
    if (location.pathname !== allowedPath && !isAllowedKnowledgeBasePath(location.pathname)) {{
      history.replaceState(history.state, "", allowedPath + location.search + location.hash);
    }}
  }}
  function reloadForKnowledgeBaseSwitch(target, replace) {{
    try {{
      const next = new URL(target, location.origin);
      if (next.pathname === location.pathname || !isAllowedKnowledgeBasePath(next.pathname)) return false;
      if (replace) location.replace(next.href);
      else location.assign(next.href);
      return true;
    }} catch (_) {{
      return false;
    }}
  }}
  function scopeNodes(scope, selector) {{
    return [
      ...(scope?.matches?.(selector) ? [scope] : []),
      ...Array.from(scope?.querySelectorAll?.(selector) || [])
    ];
  }}
  const guideSelector = ".t-guide,.t-guide__wrapper,.t-guide__reference,.contextual-guide,.guide__backdrop,[class^='guide__'],[class*=' guide__'],[data-guide-highlight]";
  function purgeGuides(scope = document) {{
    scopeNodes(scope, guideSelector).forEach((node) => {{
        try {{ node.remove(); }} catch (_) {{}}
      }});
  }}
  let themeRefreshTimer = 0;
  function refreshThemeSoon() {{
    if (!window.__deerflowWeKnoraTheme || themeRefreshTimer) return;
    themeRefreshTimer = window.setTimeout(() => {{
      themeRefreshTimer = 0;
      applyTheme(window.__deerflowWeKnoraTheme);
    }}, 80);
  }}
  themeObserver = new MutationObserver(refreshThemeSoon);
  themeObserver.observe(root, {{ attributes: true, attributeFilter: ["class", "style", "data-theme", "data-theme-mode", "theme-mode"] }});
  const shellSidebarSelector = [
    ".main > .aside_box",
    "aside",
    ".sidebar",
    ".side-bar",
    ".side-menu",
    ".nav-sidebar",
    ".navigation-sidebar",
    ".menu-sidebar",
    ".t-layout__sider",
    ".t-layout-sider",
    ".layout-sider",
    ".layout-sidebar",
    ".platform-sidebar",
    ".platform-sider",
    ".app-sidebar",
    ".app-sider",
    "[class*='Sidebar']",
    "[class*='sidebar']",
    "[class*='SideBar']",
    "[class*='sideBar']",
    "[class*='Sider']",
    "[class*='sider']"
  ].join(",");
  function isInsideKnowledgeDetail(node) {{
    return Boolean(
      node.closest(
        ".knowledge-layout,.document-layout,.document-detail,.wiki-layout,.wiki-page,.graph-layout,.graph-page,[class*='knowledge-detail'],[class*='wiki-detail']"
      )
    );
  }}
  function isLikelyOuterSidebar(node) {{
    if (!node || node === document.body || node === document.documentElement) return false;
    if (isInsideKnowledgeDetail(node)) return false;
    const rect = node.getBoundingClientRect();
    if (!rect.width || !rect.height) return false;
    const style = getComputedStyle(node);
    const className = String(node.className || "").toLowerCase();
    const namedLikeShell =
      /sidebar|side-bar|side_menu|side-menu|sider|layout-sider|platform-sider|platform-sidebar|app-sider|app-sidebar|nav-sidebar|menu-sidebar/.test(
        className
      );
    const nearLeft = rect.left <= 4;
    const sidebarWidth = rect.width >= 40 && rect.width <= Math.min(360, Math.max(120, window.innerWidth * 0.42));
    const tallEnough = rect.height >= Math.max(240, window.innerHeight * 0.45);
    const topShellChild =
      node.parentElement === document.body ||
      node.parentElement === document.getElementById("app") ||
      node.parentElement?.parentElement === document.getElementById("app") ||
      node.closest("#app") === node.parentElement?.closest("#app");
    const positionedShell = style.position === "fixed" || style.position === "sticky";
    const text = (node.textContent || "").slice(0, 400);
    const containsAppNav = /知识库|工作台|空间|租户|模型|智能体|设置|Knowledge|Workspace|Agent|Model|Tenant/.test(text);
    return namedLikeShell && nearLeft && sidebarWidth && tallEnough && (topShellChild || positionedShell || containsAppNav);
  }}
  const shellContentSelector = "#app,.t-layout,.t-layout__content,.layout-content,.platform-content,.platform-route-outlet,.main";
  function hideWeKnoraShell(scope = document) {{
    try {{
      scopeNodes(scope, shellSidebarSelector).forEach((node) => {{
        if (isLikelyOuterSidebar(node)) {{
          if (node.dataset.deerflowWeknoraChrome !== "hidden") node.dataset.deerflowWeknoraChrome = "hidden";
          if (node.getAttribute("aria-hidden") !== "true") node.setAttribute("aria-hidden", "true");
        }}
      }});
      scopeNodes(scope, shellContentSelector).forEach((node) => {{
          if (!isInsideKnowledgeDetail(node) && node.dataset.deerflowWeknoraContent !== "true") {{
            node.dataset.deerflowWeknoraContent = "true";
          }}
      }});
    }} catch (_) {{}}
  }}
  const rawPushState = history.pushState.bind(history);
  const rawReplaceState = history.replaceState.bind(history);
  history.pushState = (state, title, url) => {{
    const target = url ? allowedUrl(url) : location.href;
    if (url && target && reloadForKnowledgeBaseSwitch(target, false)) return;
    return rawPushState(state, title, target);
  }};
  history.replaceState = (state, title, url) => {{
    const target = url ? allowedUrl(url) : location.href;
    if (url && target && reloadForKnowledgeBaseSwitch(target, true)) return;
    return rawReplaceState(state, title, target);
  }};
  window.addEventListener("popstate", enforceLocation);
  document.addEventListener("click", (event) => {{
    const anchor = event.target && event.target.closest ? event.target.closest("a[href]") : null;
    if (!anchor) return;
    const target = allowedUrl(anchor.href);
    if (target && new URL(anchor.href, location.origin).pathname !== allowedPath) {{
      event.preventDefault();
      event.stopPropagation();
      if (reloadForKnowledgeBaseSwitch(target, false)) return;
      history.replaceState(history.state, "", target);
    }}
  }}, true);
  let ready = false;
  let readyFallbackTimer = 0;
  let sourceDetailTimer = 0;
  function hasDetailShell() {{
    return Boolean(document.querySelector(".knowledge-layout") || document.querySelector(".document-header"));
  }}
  function markReady(force = false) {{
    if (ready) return;
    if (!force && (!isAllowedKnowledgeBasePath(location.pathname) && location.pathname !== allowedPath || !hasDetailShell())) return;
    ready = true;
    root.classList.add("deerflow-weknora-ready");
  }}
  function finishSourceDetail() {{
    if (sourceDetailTimer) clearInterval(sourceDetailTimer);
    sourceDetailTimer = 0;
  }}
  function refreshUi(scopes) {{
    enforceLocation();
    for (const scope of scopes) {{
      purgeGuides(scope);
      hideWeKnoraShell(scope);
    }}
    if (sourceDetailMode && tickSourceDetail()) finishSourceDetail();
    markReady(false);
    if (ready && (!sourceDetailMode || root.classList.contains("deerflow-source-detail-open"))) {{
      clearTimeout(readyFallbackTimer);
      readyFallbackTimer = 0;
    }}
  }}
  let observerTimer = 0;
  const pendingScopes = new Set();
  function scheduleRefresh(scope) {{
    for (const pending of pendingScopes) {{
      if (pending === scope || pending.contains?.(scope)) return;
      if (scope.contains?.(pending)) pendingScopes.delete(pending);
    }}
    pendingScopes.add(scope);
    if (observerTimer) return;
    observerTimer = window.setTimeout(() => {{
      observerTimer = 0;
      const scopes = Array.from(pendingScopes);
      pendingScopes.clear();
      refreshUi(scopes);
    }}, 120);
  }}
  const readyObserver = new MutationObserver((mutations) => {{
    for (const mutation of mutations) {{
      if (mutation.type === "childList") {{
        for (const node of mutation.addedNodes) {{
          if (node.nodeType === Node.ELEMENT_NODE) scheduleRefresh(node);
        }}
      }} else if (mutation.target !== root && mutation.target.matches?.(
        shellSidebarSelector + "," + shellContentSelector + "," + guideSelector + ",.knowledge-layout,.document-header"
      )) {{
        scheduleRefresh(mutation.target);
      }}
    }}
  }});
  // Root theme repair is observed separately. Scanning only inserted branches
  // avoids re-walking the whole application on every animation/style update.
  readyObserver.observe(root, {{ childList: true, subtree: true, attributes: true, attributeFilter: ["class"] }});
  readyFallbackTimer = setTimeout(() => {{
    readyFallbackTimer = 0;
    markReady(true);
    if (sourceDetailMode && !root.classList.contains("deerflow-source-detail-open")) {{
      root.classList.add("deerflow-source-detail-open");
    }}
    finishSourceDetail();
  }}, 9000);
  // Only source deep links need bounded retries while their drawer mounts.
  // Ordinary embedded pages have no recurring DOM-maintenance timer.
  if (sourceDetailMode) {{
    sourceDetailTimer = setInterval(() => refreshUi([]), 500);
  }}
  refreshUi([document]);
  if (document.readyState === "complete") {{
    scheduleRefresh(document);
  }} else {{
    window.addEventListener("load", () => scheduleRefresh(document), {{ once: true }});
  }}
  window.addEventListener("resize", () => scheduleRefresh(document));
}})();
</script>
"""


def _inject_weknora_embed_html(
    html: bytes,
    *,
    allowed_path: str,
    tab: str,
    session: dict[str, Any],
    ctx: WeKnoraEmbedContext,
    public_prefix: str = "",
) -> bytes:
    text = html.decode("utf-8", errors="replace")
    text = _rewrite_weknora_html_public_paths(text, public_prefix)
    injection = _weknora_embed_injection(
        allowed_path=allowed_path,
        tab=tab,
        session=session,
        ctx=ctx,
        public_prefix=public_prefix,
    )
    if "</head>" in text:
        text = text.replace("</head>", f"{injection}</head>", 1)
    else:
        text = f"{injection}{text}"
    return text.encode("utf-8")


async def _proxy_weknora_request(
    request: Request,
    *,
    target_base_url: str,
    upstream_path: str,
    ctx: WeKnoraEmbedContext,
    is_api: bool,
    tab: str = "wiki",
) -> Response:
    runtime = _runtime_for_iframe_proxy(request)
    session = await _get_weknora_admin_session(runtime)
    token = str(session.get("token") or "")
    body = await request.body()
    if is_api:
        await _authorize_weknora_api_proxy(request, ctx, runtime, upstream_path, body, token)

    headers = _forward_request_headers(request)
    if is_api:
        headers["Authorization"] = f"Bearer {token}"
        tenant_id = _tenant_id_from_session(session)
        if tenant_id:
            headers["X-Tenant-ID"] = tenant_id

    url = f"{target_base_url.rstrip('/')}{_encode_upstream_path(upstream_path)}"
    follow_binary_redirects = is_api and _should_follow_weknora_binary_redirect(upstream_path)
    try:
        async with _proxy_http_client(request) as client:
            response = await client.request(
                request.method,
                url,
                params=request.query_params.multi_items(),
                headers=headers,
                content=body,
                follow_redirects=follow_binary_redirects,
            )
        if is_api and response.status_code in {401, 403}:
            session = await _get_weknora_admin_session(runtime, force_refresh=True)
            headers["Authorization"] = f"Bearer {session.get('token')}"
            tenant_id = _tenant_id_from_session(session)
            if tenant_id:
                headers["X-Tenant-ID"] = tenant_id
            async with _proxy_http_client(request) as client:
                response = await client.request(
                    request.method,
                    url,
                    params=request.query_params.multi_items(),
                    headers=headers,
                    content=body,
                    follow_redirects=follow_binary_redirects,
                )
    except httpx.TimeoutException as exc:
        raise HTTPException(status_code=504, detail="WeKnora proxy request timed out") from exc
    except httpx.HTTPError as exc:
        logger.warning("WeKnora proxy request failed: url=%s error=%r", url, exc)
        raise HTTPException(status_code=502, detail="WeKnora proxy request failed") from exc

    content = response.content
    content_type = response.headers.get("content-type", "")
    public_prefix = _request_public_prefix(request)
    if is_api and request.method.upper() == "GET" and upstream_path.rstrip("/") == "/api/v1/knowledge-bases":
        content = _filter_knowledge_base_list_response(response, ctx) or content
    if is_api and ctx.is_conversation_deposit and "application/json" in content_type.lower():
        try:
            payload = json.loads(content)
        except (TypeError, ValueError, json.JSONDecodeError):
            pass
        else:
            content = json.dumps(
                _sanitize_conversation_deposit_value(payload),
                ensure_ascii=False,
                separators=(",", ":"),
            ).encode("utf-8")
    if not is_api and "text/html" in content_type.lower():
        allowed_path = _with_public_prefix(
            public_prefix,
            f"/platform/knowledge-bases/{quote(ctx.weknora_id, safe='')}",
        )
        content = _inject_weknora_embed_html(
            content,
            allowed_path=allowed_path,
            tab=tab,
            session=session,
            ctx=ctx,
            public_prefix=public_prefix,
        )
        content_type = "text/html; charset=utf-8"
    elif not is_api and _should_rewrite_weknora_static_response(content_type, upstream_path):
        text = content.decode("utf-8", errors="replace")
        lower_type = content_type.lower()
        if "javascript" in lower_type or upstream_path.lower().endswith((".js", ".mjs")):
            text = _rewrite_weknora_script_public_paths(text, public_prefix)
        else:
            text = _rewrite_weknora_html_public_paths(
                text,
                public_prefix,
                rewrite_relative=False,
            )
        content = text.encode("utf-8")

    headers_out = _safe_response_headers(response.headers)
    if "location" in headers_out:
        headers_out["location"] = _rewrite_weknora_location_header(
            headers_out["location"],
            public_prefix,
            target_base_url,
        )
    if "Location" in headers_out:
        headers_out["Location"] = _rewrite_weknora_location_header(
            headers_out["Location"],
            public_prefix,
            target_base_url,
        )
    if not is_api:
        headers_out.pop("etag", None)
        headers_out.pop("ETag", None)
    headers_out["Cache-Control"] = "no-store" if is_api or "text/html" in content_type.lower() else "private, max-age=3600"
    if (
        is_api
        and response.status_code < 400
        and _is_weknora_content_mutation(request.method, upstream_path)
    ):
        await _schedule_proxy_local_wiki_sync(request, ctx)
    return Response(
        content=content,
        status_code=response.status_code,
        media_type=content_type.split(";", 1)[0] if content_type else None,
        headers=headers_out,
    )


def _is_conversation_deposit_mapping(row: dict[str, Any]) -> bool:
    return str(row.get("owner_user_id") or "") == CONVERSATION_DEPOSIT_OWNER_USER_ID and str(row.get("name") or "").strip() == CONVERSATION_DEPOSIT_KB_NAME


_LEGACY_BRAND_PATTERN = re.compile(r"deer[\s_-]*flow", re.IGNORECASE)


def _sanitize_conversation_deposit_value(value: Any) -> Any:
    if isinstance(value, str):
        return _LEGACY_BRAND_PATTERN.sub("cmzs", value)
    if isinstance(value, dict):
        return {key: _sanitize_conversation_deposit_value(item) for key, item in value.items()}
    if isinstance(value, list):
        return [_sanitize_conversation_deposit_value(item) for item in value]
    if isinstance(value, tuple):
        return tuple(_sanitize_conversation_deposit_value(item) for item in value)
    return value


async def _sync_conversation_deposit_view(
    store: LlmWikiStore,
    client: WeKnoraClient,
    row: dict[str, Any],
    remote: dict[str, Any] | None,
) -> tuple[dict[str, Any], dict[str, Any] | None]:
    if not _is_conversation_deposit_mapping(row):
        return row, remote

    current_remote = remote
    if remote is not None and str(remote.get("description") or "") != CONVERSATION_DEPOSIT_KB_DESCRIPTION:
        try:
            updated_remote = await client.update_knowledge_base(
                str(row["weknora_id"]),
                {"description": CONVERSATION_DEPOSIT_KB_DESCRIPTION},
            )
            current_remote = {
                **remote,
                **updated_remote,
                "description": CONVERSATION_DEPOSIT_KB_DESCRIPTION,
            }
        except WeKnoraError:
            logger.warning("Failed to synchronize the cmzs conversation-deposit description", exc_info=True)

    current_row = row
    if str(row.get("description") or "") != CONVERSATION_DEPOSIT_KB_DESCRIPTION:
        updated_row = await store.update_mapping(
            str(row["id"]),
            description=CONVERSATION_DEPOSIT_KB_DESCRIPTION,
        )
        current_row = updated_row or row
    return current_row, current_remote


def _persisted_directory_paths(row: dict[str, Any]) -> list[list[str]]:
    value = row.get("wiki_directory_paths")
    if value is not None:
        return normalize_directory_paths(value)
    return paths_from_parent_directories(row.get("wiki_parent_directories"))


def _raw_wiki_text(
    row: dict[str, Any],
    remote_wiki_config: dict[str, Any],
    persisted_key: str,
    remote_key: str,
) -> str:
    persisted = row.get(persisted_key)
    if persisted is not None:
        return str(persisted)
    return str(remote_wiki_config.get(remote_key) or "").strip()


def _mapping_view(row: dict[str, Any], remote: dict[str, Any] | None, *, actor_user_id: str, is_admin: bool) -> dict[str, Any]:
    source = remote or {}
    is_conversation_deposit = _is_conversation_deposit_mapping(row)
    is_owner = row.get("owner_user_id") == actor_user_id
    description = CONVERSATION_DEPOSIT_KB_DESCRIPTION if is_conversation_deposit else source.get("description") if source.get("description") is not None else row.get("description") or ""
    display_name = str(source.get("name") or row.get("name") or "")
    profile_row = {**row, "name": display_name}
    remote_wiki_config = (
        dict(source.get("wiki_config"))
        if isinstance(source.get("wiki_config"), dict)
        else {}
    )
    directory_paths = _persisted_directory_paths(row)
    parent_directories = directory_roots(directory_paths)
    raw_content = _raw_wiki_text(
        row,
        remote_wiki_config,
        "wiki_content_instructions_raw",
        "content_instructions",
    )
    raw_extraction = _raw_wiki_text(
        row,
        remote_wiki_config,
        "wiki_extraction_instructions_raw",
        "extraction_instructions",
    )
    focus = str(row.get("wiki_focus_instructions") or "")
    try:
        compiled = compile_wiki_instructions(
            content_instructions=raw_content,
            extraction_instructions=raw_extraction,
            focus_instructions=focus,
            directory_paths=directory_paths,
        )
    except ValueError:
        # Legacy rows may contain upstream values above today's supported
        # limit. They remain readable while the next edit receives validation.
        compiled = None

    managed = row.get("wiki_generation_managed")
    if managed is None:
        strategy = source.get("indexing_strategy")
        if isinstance(strategy, dict) and isinstance(
            strategy.get("wiki_enabled"),
            bool,
        ):
            managed = bool(strategy["wiki_enabled"])
    revision = row.get("wiki_config_revision")
    apply_status = row.get("wiki_config_apply_status")
    if not apply_status:
        apply_status = (
            "active"
            if managed is True
            else "not_applicable"
            if managed is False
            else "error"
        )
    wiki_config = dict(remote_wiki_config)
    wiki_config.update({
        "content_instructions": raw_content,
        "extraction_instructions": raw_extraction,
        "parent_directories": parent_directories,
        "directory_paths": directory_paths,
        "focus_instructions": focus,
        "wiki_config_schema_version": row.get("wiki_config_schema_version"),
        "wiki_config_revision": revision,
        "applied_revision": revision if apply_status == "active" else None,
        "applied_result_id": None,
        "apply_status": apply_status,
        "apply_error": row.get("wiki_config_apply_error"),
        "directory_prompt_coverage": (
            compiled.directory_prompt_coverage if compiled else "none"
        ),
        "directory_prompt_covered_nodes": (
            compiled.directory_prompt_covered_nodes if compiled else 0
        ),
        "directory_prompt_total_nodes": (
            compiled.directory_prompt_total_nodes
            if compiled
            else len(directory_paths)
        ),
    })
    return {
        "id": row["id"],
        "remote_status": "cached" if source.get("_local_cache_only") else "available" if remote is not None else "missing",
        "name": display_name,
        "description": description,
        "test_questions": resolved_test_questions(profile_row),
        "test_questions_customized": are_test_questions_customized(profile_row),
        "retrieval_settings": resolved_retrieval_settings(row),
        "type": source.get("type") or row.get("kb_type") or "document",
        "publication_status": "published" if is_conversation_deposit else row.get("publication_status") or "private",
        "wiki_generation_managed": managed,
        "wiki_index_enabled": row.get("wiki_index_enabled", True),
        "external_search_enabled": False if is_conversation_deposit else row.get("external_search_enabled", False),
        "external_search_updated_at": row.get("external_search_updated_at"),
        "owner_user_id": row.get("owner_user_id") if is_admin else None,
        "is_owner": is_owner,
        "can_write": not is_conversation_deposit and (is_admin or is_owner),
        "is_conversation_deposit": is_conversation_deposit,
        "knowledge_count": int(source.get("knowledge_count") or 0),
        "chunk_count": int(source.get("chunk_count") or 0),
        "processing_count": int(source.get("processing_count") or 0),
        "share_count": int(source.get("share_count") or 0),
        "is_processing": bool(source.get("is_processing") or int(source.get("processing_count") or 0) > 0),
        "is_temporary": bool(source.get("is_temporary", False)),
        "embedding_model_id": source.get("embedding_model_id"),
        "vector_store_id": source.get("vector_store_id"),
        "vector_store_name": source.get("vector_store_name"),
        "vector_store_source": source.get("vector_store_source"),
        "vector_store_engine_type": source.get("vector_store_engine_type"),
        "vector_store_status": source.get("vector_store_status"),
        "chunking_config": source.get("chunking_config") if isinstance(source.get("chunking_config"), dict) else {},
        "image_processing_config": (source.get("image_processing_config") if isinstance(source.get("image_processing_config"), dict) else {}),
        "indexing_strategy": source.get("indexing_strategy") if isinstance(source.get("indexing_strategy"), dict) else {},
        "capabilities": source.get("capabilities") if isinstance(source.get("capabilities"), dict) else {},
        "extract_config": source.get("extract_config") if isinstance(source.get("extract_config"), dict) else {},
        "faq_config": source.get("faq_config") if isinstance(source.get("faq_config"), dict) else {},
        "wiki_config": wiki_config,
        "question_generation_config": (source.get("question_generation_config") if isinstance(source.get("question_generation_config"), dict) else {}),
        "auto_tag_config": source.get("auto_tag_config") if isinstance(source.get("auto_tag_config"), dict) else {},
        "vlm_config": source.get("vlm_config") if isinstance(source.get("vlm_config"), dict) else {},
        "asr_config": source.get("asr_config") if isinstance(source.get("asr_config"), dict) else {},
        "summary_model_id": source.get("summary_model_id"),
        "storage_provider": (source.get("storage_provider_config", {}).get("provider") if isinstance(source.get("storage_provider_config"), dict) else None),
        "created_at": row.get("created_at"),
        "updated_at": row.get("updated_at"),
        "published_at": row.get("published_at"),
        "remote_created_at": source.get("created_at"),
        "remote_updated_at": source.get("updated_at"),
    }


async def _authorized_mapping(
    request: Request,
    mapping_id: str,
    *,
    write: bool,
) -> tuple[dict[str, Any], str, bool]:
    user_id, is_admin = await _actor(request)
    row = await _store(request).get_authorized(mapping_id, user_id, write=write, is_admin=is_admin)
    if row is None:
        raise HTTPException(status_code=404, detail="Knowledge base not found")
    if write and _is_conversation_deposit_mapping(row):
        raise HTTPException(status_code=403, detail="Conversation deposit is managed by cmzs")
    return row, user_id, is_admin


async def _public_mapping(request: Request, mapping_id: str) -> dict[str, Any]:
    """Resolve a world-readable knowledge base without inheriting actor/admin state."""

    row = await _store(request).get_authorized(
        mapping_id,
        "__public_wiki__",
        write=False,
        is_admin=False,
    )
    if row is None or row.get("publication_status") != "published" or _is_conversation_deposit_mapping(row):
        raise HTTPException(status_code=404, detail="Published knowledge base not found")
    return row


def _validate_protected_wiki_file_path(file_path: str) -> str:
    value = file_path.strip()
    if not value or "\x00" in value or not _PROTECTED_WIKI_FILE_PATH.match(value):
        raise HTTPException(status_code=422, detail="Unsupported Wiki image path")
    return value


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:
        content, media_type = await _client_or_503().get_knowledge_base_file(
            str(row["weknora_id"]),
            safe_file_path,
        )
    except WeKnoraError as 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(
        content=content,
        media_type=media_type,
        headers={
            "Cache-Control": "public, max-age=300" if public else "private, max-age=3600",
            "Content-Disposition": f"inline; filename*=UTF-8''{quote(filename, safe='')}",
            "X-Content-Type-Options": "nosniff",
        },
    )


async def _verify_document(client: WeKnoraClient, row: dict[str, Any], document_id: str) -> dict[str, Any]:
    document = await client.get_document(document_id)
    if str(document.get("knowledge_base_id") or "") != str(row["weknora_id"]):
        raise HTTPException(status_code=404, detail="Document not found")
    return document


def _document_view(document: dict[str, Any]) -> dict[str, Any]:
    """Expose useful document state without leaking WeKnora storage paths or hashes."""

    fields = (
        "id",
        "type",
        "title",
        "description",
        "source",
        "channel",
        "parse_status",
        "pending_subtasks_count",
        "summary_status",
        "enable_status",
        "embedding_model_id",
        "file_name",
        "file_type",
        "file_size",
        "storage_size",
        "created_at",
        "updated_at",
        "processed_at",
        "error_message",
    )
    return {key: document.get(key) for key in fields}


def _related_chunk_ids(value: Any) -> list[str]:
    if not isinstance(value, list):
        return []
    result: list[str] = []
    for item in value:
        candidate = item.get("id") if isinstance(item, dict) else item
        if candidate and str(candidate) not in result:
            result.append(str(candidate))
    return result


def _chunk_view(chunk: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": str(chunk.get("id") or ""),
        "content": str(chunk.get("content") or ""),
        "chunk_index": int(chunk.get("chunk_index") or 0),
        "is_enabled": bool(chunk.get("is_enabled")),
        "status": chunk.get("status"),
        "start_at": int(chunk.get("start_at") or 0),
        "end_at": int(chunk.get("end_at") or 0),
        "chunk_type": str(chunk.get("chunk_type") or "text"),
        "parent_chunk_id": str(chunk.get("parent_chunk_id") or ""),
        "relation_chunk_ids": _related_chunk_ids(chunk.get("relation_chunks")),
        "indirect_relation_chunk_ids": _related_chunk_ids(chunk.get("indirect_relation_chunks")),
        "metadata": chunk.get("metadata") if isinstance(chunk.get("metadata"), dict) else {},
        "created_at": chunk.get("created_at"),
        "updated_at": chunk.get("updated_at"),
    }


def _string_tokens(value: Any) -> list[str]:
    tokens: list[str] = []
    if isinstance(value, str):
        raw = value.strip()
        if raw:
            tokens.append(raw)
            for separator in ("#", ":", "/", "\\"):
                if separator in raw:
                    tokens.extend(part.strip() for part in raw.split(separator) if part.strip())
    elif isinstance(value, dict):
        for key in ("id", "chunk_id", "knowledge_id", "document_id", "source_id", "ref_id", "slug"):
            tokens.extend(_string_tokens(value.get(key)))
    return list(dict.fromkeys(tokens))


def _source_ref_tokens(value: Any) -> set[str]:
    refs: set[str] = set()
    if isinstance(value, list):
        for item in value:
            refs.update(_string_tokens(item))
    else:
        refs.update(_string_tokens(value))
    return refs


def _candidate_chunk_ref_tokens(chunk: dict[str, Any], chunk_id: str) -> set[str]:
    refs = {chunk_id, str(chunk.get("id") or "")}
    metadata = chunk.get("metadata") if isinstance(chunk.get("metadata"), dict) else {}
    for source in (chunk, metadata):
        for key in (
            "knowledge_id",
            "document_id",
            "source_id",
            "source_document_id",
            "knowledge_uuid",
            "file_id",
        ):
            refs.update(_string_tokens(source.get(key)))
    return {item for item in refs if item}


def _candidate_wiki_slugs(chunk: dict[str, Any]) -> list[str]:
    metadata = chunk.get("metadata") if isinstance(chunk.get("metadata"), dict) else {}
    candidates: list[str] = []
    for source in (chunk, metadata):
        for key in ("wiki_page_slug", "page_slug", "wiki_slug", "llmwiki_slug", "slug"):
            candidates.extend(_string_tokens(source.get(key)))
    nested_wiki = metadata.get("wiki")
    if isinstance(nested_wiki, dict):
        for key in ("slug", "page_slug"):
            candidates.extend(_string_tokens(nested_wiki.get(key)))
    return list(dict.fromkeys(candidates))


def _wiki_page_view(page: dict[str, Any]) -> dict[str, Any] | None:
    slug = str(page.get("slug") or "").strip()
    title = str(page.get("title") or slug or "").strip()
    content = str(page.get("content") or "").strip()
    summary = str(page.get("summary") or "").strip()
    if not slug and not title and not content and not summary:
        return None
    return {
        "page_metadata": page.get("page_metadata") if isinstance(page.get("page_metadata"), dict) else {},
        "id": str(page.get("id") or ""),
        "slug": slug,
        "title": title,
        "page_type": str(page.get("page_type") or "page"),
        "status": str(page.get("status") or ""),
        "summary": summary,
        "content": content,
        "aliases": list(page.get("aliases") or []) if isinstance(page.get("aliases"), list) else [],
        "parent_slug": str(page.get("parent_slug") or page.get("parentSlug") or ""),
        "category_path": list(page.get("category_path") or []) if isinstance(page.get("category_path"), list) else [],
        "folder_id": str(page.get("folder_id") or page.get("folderId") or ""),
        "wiki_path": str(page.get("wiki_path") or page.get("wikiPath") or ""),
        "depth": int(page.get("depth") or 0),
        "version": int(page.get("version") or 0),
        "created_at": str(page.get("created_at") or ""),
        "updated_at": str(page.get("updated_at") or ""),
        "source_refs": list(page.get("source_refs") or []) if isinstance(page.get("source_refs"), list) else [],
        "in_links": list(page.get("in_links") or []) if isinstance(page.get("in_links"), list) else [],
        "out_links": list(page.get("out_links") or []) if isinstance(page.get("out_links"), list) else [],
    }


async def _all_wiki_pages(client: WeKnoraClient, remote_id: str) -> list[dict[str, Any]]:
    """Read every Wiki page and its processed Markdown, never raw documents."""

    summaries: dict[str, dict[str, Any]] = {}
    page_number = 1
    while True:
        listing = await client.list_wiki_pages(remote_id, page=page_number, page_size=500)
        rows = [item for item in listing.get("pages", []) if isinstance(item, dict)]
        for item in rows:
            slug = str(item.get("slug") or item.get("wiki_slug") or item.get("path") or "").strip("/")
            if slug:
                summaries[slug] = item
        total_pages = max(1, int(listing.get("total_pages") or 1))
        if page_number >= total_pages or not rows:
            break
        page_number += 1

    pages: list[dict[str, Any]] = []
    entries = list(summaries.items())
    for start in range(0, len(entries), 20):
        batch = entries[start : start + 20]
        details = await asyncio.gather(*(client.get_wiki_page(remote_id, slug) for slug, _ in batch))
        pages.extend({**summary, **detail, "slug": slug} for (slug, summary), detail in zip(batch, details, strict=True))
    return pages


def _wiki_directory_parts(page: dict[str, Any], by_slug: dict[str, dict[str, Any]]) -> list[str]:
    parent_slug = str(page.get("parent_slug") or "").strip("/")
    if parent_slug:
        ancestors: list[str] = []
        visited: set[str] = set()
        current = by_slug.get(parent_slug)
        while current is not None:
            slug = str(current.get("slug") or "").strip("/")
            if not slug or slug in visited:
                break
            visited.add(slug)
            ancestors.insert(0, str(current.get("title") or slug.rsplit("/", 1)[-1]))
            current_parent = str(current.get("parent_slug") or "").strip("/")
            current = by_slug.get(current_parent) if current_parent else None
        if ancestors:
            return ancestors

    category_path = [str(item).strip() for item in page.get("category_path") or [] if str(item).strip()]
    if category_path:
        return category_path
    wiki_path = [item.strip() for item in str(page.get("wiki_path") or "").split("/") if item.strip()]
    if len(wiki_path) > 2:
        return wiki_path[1:-1]
    slug_parts = [item for item in str(page.get("slug") or "").split("/") if item]
    return slug_parts[:-1]


def _safe_excel_text(value: Any) -> str:
    text = str(value or "")
    return f"'{text}" if text.startswith(("=", "+", "-", "@")) else text


def _build_wiki_excel(knowledge_base_name: str, raw_pages: list[dict[str, Any]]) -> BytesIO:
    from openpyxl import Workbook
    from openpyxl.styles import Alignment, Font, PatternFill

    pages = [view for raw in raw_pages if (view := _wiki_page_view(raw)) is not None]
    by_slug = {page["slug"]: page for page in pages if page["slug"]}
    directories = [_wiki_directory_parts(page, by_slug) for page in pages]
    max_depth = max((len(parts) for parts in directories), default=0)
    max_content_parts = max((max(1, (len(str(page.get("content") or "")) + 31_999) // 32_000) for page in pages), default=1)
    columns = [
        "知识库",
        "标题",
        "Slug",
        "目录完整路径",
        "所在目录",
        "父级目录完整路径",
        *(f"第{level}级目录" for level in range(1, max_depth + 1)),
        "摘要",
        *("正文" if index == 1 else f"正文（续{index - 1}）" for index in range(1, max_content_parts + 1)),
        "页面类型",
        "状态",
        "别名",
        "版本",
        "创建时间",
        "更新时间",
    ]
    workbook = Workbook()
    sheet = workbook.active
    sheet.title = "Wiki 全量导出"
    sheet.append(columns)
    for cell in sheet[1]:
        cell.font = Font(bold=True, color="FFFFFF")
        cell.fill = PatternFill("solid", fgColor="1F4E78")
        cell.alignment = Alignment(horizontal="center", vertical="center")
    for page, directory in zip(pages, directories, strict=True):
        content = str(page.get("content") or "")
        content_parts = [content[index : index + 32_000] for index in range(0, len(content), 32_000)] or [""]
        content_parts.extend([""] * (max_content_parts - len(content_parts)))
        row = [
            knowledge_base_name,
            page.get("title"),
            page.get("slug"),
            " / ".join(directory),
            directory[-1] if directory else "根目录",
            " / ".join(directory[:-1]),
            *directory,
            *([""] * (max_depth - len(directory))),
            page.get("summary"),
            *content_parts,
            page.get("page_type"),
            page.get("status"),
            "、".join(str(item) for item in page.get("aliases") or []),
            page.get("version"),
            page.get("created_at"),
            page.get("updated_at"),
        ]
        sheet.append([_safe_excel_text(value) for value in row])
    sheet.freeze_panes = "A2"
    sheet.auto_filter.ref = sheet.dimensions
    for column in sheet.columns:
        letter = column[0].column_letter
        heading = str(column[0].value or "")
        sheet.column_dimensions[letter].width = 60 if heading.startswith("正文") else 32 if heading in {"摘要", "目录完整路径", "父级目录完整路径"} else 20
        for cell in column[1:]:
            cell.alignment = Alignment(vertical="top", wrap_text=True)
    output = BytesIO()
    workbook.save(output)
    output.seek(0)
    return output


def _weknora_embed_target_path(weknora_id: str, tab: str) -> str:
    target_query = urlencode({"tab": tab}) if tab != "documents" else ""
    target = f"/platform/knowledge-bases/{quote(weknora_id, safe='')}"
    if target_query:
        target = f"{target}?{target_query}"
    return target


def _set_weknora_embed_cookie(request: Request, response: Response, payload: dict[str, Any]) -> None:
    secure = is_secure_request(request)
    signed = _sign_embed_payload(payload)
    for path in ("/", _request_public_prefix(request)):
        response.set_cookie(
            _WEKNORA_EMBED_COOKIE,
            signed,
            max_age=_WEKNORA_EMBED_TTL_SECONDS,
            httponly=True,
            secure=secure,
            samesite="none" if secure else "lax",
            path=path,
        )
    response.headers["Cache-Control"] = "no-store"


async def _find_wiki_page_for_chunk(
    client: WeKnoraClient,
    *,
    knowledge_base_id: str,
    chunk_id: str,
    chunks: list[dict[str, Any]],
) -> dict[str, Any] | None:
    focus = next((chunk for chunk in chunks if str(chunk.get("id") or "") == chunk_id), None)
    if focus is None:
        return None

    for slug in _candidate_wiki_slugs(focus):
        try:
            page = await client.get_wiki_page(knowledge_base_id, slug)
        except WeKnoraError as exc:
            if exc.status_code != 404:
                logger.debug("Failed to read WeKnora wiki page slug=%s", slug, exc_info=True)
            continue
        view = _wiki_page_view(page)
        if view and (view.get("content") or view.get("summary")):
            return view

    return None


async def _prepare_weknora_embed_session(
    request: Request,
    mapping_id: str,
) -> tuple[str, dict[str, Any]]:
    _runtime_for_iframe_proxy(request)
    row, user_id, is_admin = await _authorized_mapping(request, mapping_id, write=False)
    writable = None
    if not _is_conversation_deposit_mapping(row):
        writable = await _store(request).get_authorized(mapping_id, user_id, write=True, is_admin=is_admin)
    weknora_id = str(row["weknora_id"])
    ctx = WeKnoraEmbedContext(
        mapping_id=mapping_id,
        weknora_id=weknora_id,
        user_id=user_id,
        is_admin=is_admin,
        can_write=writable is not None,
        is_conversation_deposit=_is_conversation_deposit_mapping(row),
        exp=int(time.time()) + _WEKNORA_EMBED_TTL_SECONDS,
    )
    return weknora_id, _embed_context_payload(ctx)


@router.post("/api/llmwiki/knowledge-bases/{mapping_id}/weknora/session")
async def create_weknora_detail_session(
    request: Request,
    mapping_id: str,
    tab: Literal["documents", "wiki", "graph"] = Query(default="wiki"),
) -> JSONResponse:
    """Create a signed iframe session using the normal DeerFlow API auth path."""

    weknora_id, payload = await _prepare_weknora_embed_session(request, mapping_id)
    target = _with_public_prefix(
        _request_public_prefix(request),
        _weknora_embed_target_path(weknora_id, tab),
    )
    response = JSONResponse(
        {
            "frame_path": target,
            "expires_in": _WEKNORA_EMBED_TTL_SECONDS,
        }
    )
    _set_weknora_embed_cookie(request, response, payload)
    return response


@router.get("/api/llmwiki/knowledge-bases/{mapping_id}/weknora/frame")
async def open_weknora_detail_frame(
    request: Request,
    mapping_id: str,
    tab: Literal["documents", "wiki", "graph"] = Query(default="wiki"),
) -> RedirectResponse:
    """Start a short-lived, DeerFlow-authorized WeKnora detail iframe session."""

    weknora_id, payload = await _prepare_weknora_embed_session(request, mapping_id)
    target = _with_public_prefix(
        _request_public_prefix(request),
        _weknora_embed_target_path(weknora_id, tab),
    )
    response = RedirectResponse(target, status_code=302)
    _set_weknora_embed_cookie(request, response, payload)
    return response


@proxy_router.api_route("/platform", methods=["GET", "HEAD"])
@proxy_router.api_route("/platform/{proxied_path:path}", methods=["GET", "HEAD"])
async def proxy_weknora_platform_page(request: Request, proxied_path: str = "") -> Response:
    static_upstream_path = _weknora_static_fallback_upstream_path(request)
    if static_upstream_path is not None and _is_weknora_static_asset_path(static_upstream_path):
        return await _proxy_weknora_static_asset(request, static_upstream_path)
    ctx = await _read_embed_context(request)
    requested = f"/platform/{proxied_path}".rstrip("/")
    allowed_path = f"/platform/knowledge-bases/{quote(ctx.weknora_id, safe='')}"
    if requested != allowed_path:
        requested_weknora_id = _weknora_id_from_platform_path(requested)
        if requested_weknora_id is not None:
            selected_ctx = await _switch_embed_context(request, ctx, requested_weknora_id)
            selected_ctx = await _with_visible_embed_knowledge_bases(request, selected_ctx)
            response = await _proxy_weknora_request(
                request,
                target_base_url=_runtime_for_iframe_proxy(request).web_base_url,
                upstream_path=f"/platform/knowledge-bases/{quote(requested_weknora_id, safe='')}",
                ctx=selected_ctx,
                is_api=False,
                tab=request.query_params.get("tab") or "wiki",
            )
            _set_weknora_embed_cookie(request, response, _embed_context_payload(selected_ctx))
            return response
        static_upstream_path = _weknora_static_fallback_upstream_path(request)
        if static_upstream_path is not None:
            runtime = _runtime_for_iframe_proxy(request)
            return await _proxy_weknora_request(
                request,
                target_base_url=runtime.web_base_url,
                upstream_path=static_upstream_path,
                ctx=ctx,
                is_api=False,
            )
        target_query = urlencode({"tab": request.query_params.get("tab") or "wiki"})
        public_target = _with_public_prefix(_request_public_prefix(request), allowed_path)
        return RedirectResponse(f"{public_target}?{target_query}", status_code=302)
    runtime = _runtime_for_iframe_proxy(request)
    ctx = await _with_visible_embed_knowledge_bases(request, ctx)
    return await _proxy_weknora_request(
        request,
        target_base_url=runtime.web_base_url,
        upstream_path=allowed_path,
        ctx=ctx,
        is_api=False,
        tab=request.query_params.get("tab") or "wiki",
    )


@proxy_router.api_route("/assets/{proxied_path:path}", methods=["GET", "HEAD"])
async def proxy_weknora_assets(request: Request, proxied_path: str) -> Response:
    return await _proxy_weknora_static_asset(request, f"/assets/{proxied_path}")


@proxy_router.api_route("/locales/{proxied_path:path}", methods=["GET", "HEAD"])
async def proxy_weknora_locales(request: Request, proxied_path: str) -> Response:
    return await _proxy_weknora_static_asset(request, f"/locales/{proxied_path}")


@proxy_router.api_route("/favicon.ico", methods=["GET", "HEAD"])
@proxy_router.api_route("/favicon.svg", methods=["GET", "HEAD"])
@proxy_router.api_route("/config.js", methods=["GET", "HEAD"])
@proxy_router.api_route("/runtime-config.js", methods=["GET", "HEAD"])
@proxy_router.api_route("/logo.svg", methods=["GET", "HEAD"])
@proxy_router.api_route("/manifest.webmanifest", methods=["GET", "HEAD"])
async def proxy_weknora_root_asset(request: Request) -> Response:
    return await _proxy_weknora_static_asset(
        request,
        _without_public_prefix(
            _request_public_prefix(request),
            str(request.scope.get("path") or request.url.path),
        ),
    )


@proxy_router.api_route("/tdesign-icons/{proxied_path:path}", methods=["GET", "HEAD"])
async def proxy_weknora_tdesign_icons(request: Request, proxied_path: str) -> Response:
    return await _proxy_weknora_static_asset(request, f"/tdesign-icons/{proxied_path}")


@proxy_router.api_route(
    "/api/v1/{proxied_path:path}",
    methods=["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
)
async def proxy_weknora_api(request: Request, proxied_path: str) -> Response:
    if request.method.upper() == "OPTIONS":
        return Response(status_code=204)
    ctx = await _read_embed_context(request)
    if request.method.upper() in {"GET", "HEAD"} and proxied_path.rstrip("/") == "knowledge-bases":
        return await _local_embed_knowledge_base_directory(request, ctx)
    runtime = _runtime_for_iframe_proxy(request)
    return await _proxy_weknora_request(
        request,
        target_base_url=runtime.api_base_url,
        upstream_path=f"/api/v1/{proxied_path}",
        ctx=ctx,
        is_api=True,
    )


@router.api_route(
    "/api/llmwiki/weknora-embed/api/v1/{proxied_path:path}",
    methods=["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
)
@router.api_route(
    "/deerflow/api/llmwiki/weknora-embed/api/v1/{proxied_path:path}",
    methods=["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
)
async def proxy_weknora_embed_api(request: Request, proxied_path: str) -> Response:
    """Proxy iframe API calls under a collision-free cmzs namespace."""

    if request.method.upper() == "OPTIONS":
        return Response(status_code=204)
    ctx = await _read_embed_context(request)
    if request.method.upper() in {"GET", "HEAD"} and proxied_path.rstrip("/") == "knowledge-bases":
        return await _local_embed_knowledge_base_directory(request, ctx)
    runtime = _runtime_for_iframe_proxy(request)
    return await _proxy_weknora_request(
        request,
        target_base_url=runtime.api_base_url,
        upstream_path=f"/api/v1/{proxied_path}",
        ctx=ctx,
        is_api=True,
    )


@proxy_router.api_route("/{proxied_path:path}", methods=["GET", "HEAD"])
async def proxy_weknora_static_fallback(request: Request, proxied_path: str) -> Response:
    upstream_path = _weknora_static_fallback_upstream_path(request)
    if upstream_path is None:
        raise HTTPException(status_code=404, detail="WeKnora proxy path not found")
    if _is_weknora_static_asset_path(upstream_path):
        return await _proxy_weknora_static_asset(request, upstream_path)
    ctx = await _read_embed_context(request)
    runtime = _runtime_for_iframe_proxy(request)
    return await _proxy_weknora_request(
        request,
        target_base_url=runtime.web_base_url,
        upstream_path=upstream_path,
        ctx=ctx,
        is_api=False,
    )


@router.get("/api/llmwiki/runtime")
async def get_llmwiki_runtime() -> dict[str, Any]:
    return _runtime_or_legacy()


@router.get("/api/system-settings/llmwiki")
async def get_llmwiki_settings(request: Request) -> dict[str, Any]:
    await _require_admin(request)
    from deerflow.config import get_app_config

    app_config = get_app_config()
    runtime = get_resolved_llmwiki_runtime(app_config)
    return {
        **_runtime_or_legacy(),
        "file_api_base_url": app_config.llmwiki.weknora.api_base_url,
        "file_web_base_url": app_config.llmwiki.weknora.web_base_url,
        "effective_provider": runtime.provider,
    }


@router.put("/api/system-settings/llmwiki")
async def update_llmwiki_settings(request: Request, body: LlmWikiSettingsUpdate) -> dict[str, Any]:
    await _require_admin(request)
    try:
        override = LlmWikiRuntimeOverride(
            enabled=body.override_enabled,
            api_base_url=body.api_base_url,
            web_base_url=body.web_base_url,
        )
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from None
    current = load_system_settings()
    save_system_settings(current.model_copy(update={"llmwiki": override}))
    from app.gateway.llmwiki_index_runtime import rebuild_local_wiki_index_runtime

    await rebuild_local_wiki_index_runtime(request.app, force=True)
    return _runtime_or_legacy()


@router.delete("/api/system-settings/llmwiki/override")
async def reset_llmwiki_settings(request: Request) -> dict[str, Any]:
    await _require_admin(request)
    current = load_system_settings()
    save_system_settings(current.model_copy(update={"llmwiki": LlmWikiRuntimeOverride()}))
    from app.gateway.llmwiki_index_runtime import rebuild_local_wiki_index_runtime

    await rebuild_local_wiki_index_runtime(request.app, force=True)
    return _runtime_or_legacy()


@router.post("/api/system-settings/llmwiki/test")
async def test_llmwiki_connection(request: Request, body: LlmWikiConnectionTest) -> dict[str, Any]:
    await _require_admin(request)
    runtime = get_resolved_llmwiki_runtime()
    if body.api_base_url is not None:
        try:
            candidate = normalize_service_url(body.api_base_url)
        except ValueError as exc:
            raise HTTPException(status_code=422, detail=str(exc)) from None
        runtime = replace(runtime, provider="weknora" if candidate else "legacy", api_base_url=candidate)
    if not runtime.weknora_enabled:
        raise HTTPException(status_code=422, detail="Enter a WeKnora API address first")
    try:
        health = await build_weknora_client(runtime).health()
    except RuntimeError as exc:
        raise HTTPException(status_code=503, detail=str(exc)) from None
    except WeKnoraError as exc:
        _raise_upstream(exc)
    return {"success": True, "provider": "weknora", "health": health}


_PUBLIC_WIKI_SUMMARY_CONCURRENCY = 8
_PUBLIC_WIKI_PAGE_SIZE = 500
_PUBLIC_WIKI_PAGE_LIMIT = 100
_GENERATED_PUBLIC_WIKI_INDEX_SLUG = "__index__"


def _is_public_wiki_index_page(page: dict[str, Any]) -> bool:
    page_type = str(page.get("page_type") or "").strip().lower()
    title = str(page.get("title") or "").strip().lower()
    slug_name = str(page.get("slug") or "").strip("/").rsplit("/", 1)[-1].lower()
    return page_type == "index" or title in {"索引", "index"} or slug_name in {"index", "home", "readme"}


def _public_wiki_default_page(pages: list[dict[str, Any]]) -> dict[str, Any] | None:
    return (
        next((page for page in pages if _is_public_wiki_index_page(page)), None)
        or next((page for page in pages if str(page.get("page_type") or "").lower() == "overview"), None)
        or next(
            (page for page in pages if not str(page.get("parent_slug") or "").strip() and int(page.get("depth") or 0) == 0),
            None,
        )
        or (pages[0] if pages else None)
    )


def _count_public_wiki_values(pages: list[dict[str, Any]], key: str, fallback: str) -> dict[str, int]:
    counts: dict[str, int] = {}
    for page in pages:
        value = str(page.get(key) or fallback).strip().lower() or fallback
        counts[value] = counts.get(value, 0) + 1
    return counts


def _public_wiki_time_bound(pages: list[dict[str, Any]], key: str, *, latest: bool) -> str | None:
    values = [str(page.get(key) or "").strip() for page in pages]
    values = [value for value in values if value]
    if not values:
        return None
    return max(values) if latest else min(values)


async def _all_public_wiki_page_views(
    client: WeKnoraClient,
    remote_id: str,
) -> tuple[list[dict[str, Any]], int, bool]:
    pages_by_slug: dict[str, dict[str, Any]] = {}
    remote_total = 0
    complete = False
    for page_number in range(1, _PUBLIC_WIKI_PAGE_LIMIT + 1):
        result = await client.list_wiki_pages(
            remote_id,
            page=page_number,
            page_size=_PUBLIC_WIKI_PAGE_SIZE,
            query="",
        )
        raw_pages = result.get("pages") if isinstance(result, dict) else []
        page_items = raw_pages if isinstance(raw_pages, list) else []
        for page_item in page_items:
            if not isinstance(page_item, dict):
                continue
            page = _wiki_page_view(page_item)
            if page is None:
                continue
            slug = str(page.get("slug") or "").strip()
            pages_by_slug[slug or f"__page_{len(pages_by_slug)}"] = page
        reported_total = int(result.get("total") or 0)
        remote_total = max(remote_total, reported_total, len(pages_by_slug))
        total_pages = max(1, int(result.get("total_pages") or 1))
        if not page_items or page_number >= total_pages or (reported_total > 0 and len(pages_by_slug) >= remote_total):
            complete = True
            break
    return list(pages_by_slug.values()), max(remote_total, len(pages_by_slug)), complete


def _public_wiki_summary(
    mapping_id: str,
    pages: list[dict[str, Any]],
    *,
    total: int,
    complete: bool,
) -> dict[str, Any]:
    explicit_index = next((page for page in pages if _is_public_wiki_index_page(page)), None)
    default_page = _public_wiki_default_page(pages)
    if explicit_index is not None:
        index_mode = "explicit"
        index_slug: str | None = str(explicit_index.get("slug") or "") or None
    elif pages:
        index_mode = "generated"
        index_slug = _GENERATED_PUBLIC_WIKI_INDEX_SLUG
    else:
        index_mode = "none"
        index_slug = None
    default_slug = index_slug or (str(default_page.get("slug") or "") if default_page else None)
    route = f"/embed/knowledge/{quote(mapping_id, safe='')}"
    if default_slug:
        route = f"{route}?{urlencode({'wikiSlug': default_slug})}"
    return {
        "status": "ready" if pages else "empty",
        "page_count": total,
        "loaded_page_count": len(pages),
        "stats_complete": complete,
        "page_type_counts": _count_public_wiki_values(pages, "page_type", "page"),
        "page_status_counts": _count_public_wiki_values(pages, "status", "unknown"),
        "index_mode": index_mode,
        "has_explicit_index": explicit_index is not None,
        "index_slug": index_slug,
        "default_slug": default_slug,
        "created_at": _public_wiki_time_bound(pages, "created_at", latest=False),
        "updated_at": _public_wiki_time_bound(pages, "updated_at", latest=True),
        "embed_route": route,
    }


def _unavailable_public_wiki_summary(mapping_id: str) -> dict[str, Any]:
    return {
        "status": "unavailable",
        "page_count": 0,
        "loaded_page_count": 0,
        "stats_complete": False,
        "page_type_counts": {},
        "page_status_counts": {},
        "index_mode": "unknown",
        "has_explicit_index": False,
        "index_slug": None,
        "default_slug": None,
        "created_at": None,
        "updated_at": None,
        "embed_route": f"/embed/knowledge/{quote(mapping_id, safe='')}",
    }


def _public_local_index_summary(
    request: Request,
    mapping: dict[str, Any],
    state: dict[str, Any] | None,
) -> dict[str, Any]:
    app_state = request.app.state
    local_config = getattr(app_state, "llmwiki_effective_config", None)
    if local_config is None:
        local_config = getattr(
            getattr(getattr(app_state, "config", None), "llmwiki", None),
            "local_wiki_index",
            None,
        )
    enabled = bool(getattr(local_config, "enabled", False))
    current = dict(state or {})
    state_name = str(current.get("state") or "not_synced")
    if state_name == "idle" and int(current.get("index_revision") or 0) == 0 and not current.get("last_completed_at") and not current.get("last_success_at"):
        state_name = "not_synced"
    embedding = getattr(app_state, "llmwiki_embedding", None)
    fingerprint = embedding.fingerprint if embedding is not None else None
    fully_vectorized = bool(mapping.get("wiki_index_enabled", True)) and is_fully_vectorized(
        current,
        fingerprint,
    )
    return {
        "enabled": enabled,
        "wiki_index_enabled": bool(mapping.get("wiki_index_enabled", True)),
        "state": state_name,
        "fully_vectorized": fully_vectorized,
        "retrieval_mode": "vector" if fully_vectorized else "wiki_api",
        "remote_page_count": int(current.get("remote_page_count") or 0),
        "local_page_count": int(current.get("local_page_count") or 0),
        "ready_page_count": int(current.get("ready_page_count") or 0),
        "failed_page_count": int(current.get("failed_page_count") or 0),
        "vector_count": int(current.get("vector_count") or 0),
        "last_started_at": current.get("last_started_at"),
        "last_completed_at": current.get("last_completed_at"),
        "last_success_at": current.get("last_success_at"),
        "updated_at": current.get("updated_at"),
    }


@router.get("/api/public/llmwiki/knowledge-bases")
async def list_public_knowledge_bases(request: Request) -> dict[str, Any]:
    """List live, published Wiki libraries with safe metadata and index health."""

    rows = [
        row
        for row in await _store(request).list_visible(
            "__public_wiki__",
            scope="public",
            is_admin=False,
        )
        if row.get("publication_status") == "published" and not _is_conversation_deposit_mapping(row)
    ]
    client = _client_or_503()
    try:
        remote_items = await client.list_knowledge_bases()
    except WeKnoraError as exc:
        _raise_upstream(exc)
    remote_by_id = {str(item.get("id")): item for item in remote_items if isinstance(item, dict) and item.get("id")}
    # A successful upstream list is authoritative: stale DeerFlow mappings are
    # omitted so discovery never advertises a Wiki that its detail route cannot open.
    rows = [row for row in rows if str(row.get("weknora_id") or "") in remote_by_id]
    mapping_ids = [str(row["id"]) for row in rows]
    index_states: dict[str, dict[str, Any]] = {}
    index_store = getattr(request.app.state, "llmwiki_index_store", None)
    if index_store is not None and mapping_ids:
        try:
            index_states = {str(state.get("knowledge_base_mapping_id") or ""): state for state in await index_store.list_index_status(mapping_ids) if isinstance(state, dict)}
        except Exception:
            logger.warning("Failed to read public Wiki local-index summaries", exc_info=True)

    semaphore = asyncio.Semaphore(_PUBLIC_WIKI_SUMMARY_CONCURRENCY)

    async def build_summary(row: dict[str, Any]) -> dict[str, Any]:
        mapping_id = str(row["id"])
        remote_id = str(row["weknora_id"])
        remote = remote_by_id[remote_id]
        async with semaphore:
            try:
                remote = {**remote, **await client.get_knowledge_base(remote_id)}
            except WeKnoraError as exc:
                if exc.status_code == 404:
                    return {}
                logger.warning(
                    "Failed to load public Wiki detail metadata: mapping_id=%s",
                    mapping_id,
                    exc_info=True,
                )
            try:
                pages, page_total, complete = await _all_public_wiki_page_views(client, remote_id)
                wiki = _public_wiki_summary(
                    mapping_id,
                    pages,
                    total=page_total,
                    complete=complete,
                )
            except WeKnoraError as exc:
                logger.warning(
                    "Failed to load public Wiki page summary: mapping_id=%s",
                    mapping_id,
                    exc_info=True,
                )
                wiki = (
                    _public_wiki_summary(mapping_id, [], total=0, complete=True)
                    if exc.status_code == 404
                    else _unavailable_public_wiki_summary(mapping_id)
                )
        return {
            **_mapping_view(
                row,
                remote,
                actor_user_id="__public_wiki__",
                is_admin=False,
            ),
            "wiki": wiki,
            "local_index": _public_local_index_summary(
                request,
                row,
                index_states.get(mapping_id),
            ),
        }

    knowledge_bases = [summary for summary in await asyncio.gather(*(build_summary(row) for row in rows)) if summary]
    return {"knowledge_bases": knowledge_bases, "total": len(knowledge_bases)}


@router.get("/api/public/llmwiki/knowledge-bases/{mapping_id}")
async def get_public_knowledge_base(request: Request, mapping_id: str) -> dict[str, Any]:
    """Return the safe public metadata used by the chrome-free Wiki reader."""

    row = await _public_mapping(request, mapping_id)
    try:
        remote = await _client_or_503().get_knowledge_base(str(row["weknora_id"]))
    except WeKnoraError as exc:
        _raise_upstream(exc)
    return _mapping_view(
        row,
        remote,
        actor_user_id="__public_wiki__",
        is_admin=False,
    )


@router.get("/api/public/llmwiki/knowledge-bases/{mapping_id}/wiki/pages")
async def list_public_wiki_pages(
    request: Request,
    mapping_id: str,
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=50, ge=1, le=500),
    query: str = Query(default="", max_length=255),
) -> dict[str, Any]:
    """List Wiki pages for a published knowledge base without requiring login."""

    row = await _public_mapping(request, mapping_id)
    try:
        result = await _client_or_503().list_wiki_pages(
            str(row["weknora_id"]),
            page=page,
            page_size=page_size,
            query=query.strip(),
        )
    except WeKnoraError as exc:
        _raise_upstream(exc)
    raw_pages = result.get("pages") if isinstance(result, dict) else []
    pages = [view for page_item in (raw_pages if isinstance(raw_pages, list) else []) if isinstance(page_item, dict) for view in [_wiki_page_view(page_item)] if view is not None]
    return {**result, "pages": pages}


@router.get("/api/public/llmwiki/knowledge-bases/{mapping_id}/files")
async def get_public_wiki_file(
    request: Request,
    mapping_id: str,
    file_path: str = Query(..., min_length=1, max_length=8192),
) -> Response:
    """Serve one protected image belonging to a published Wiki."""

    row = await _public_mapping(request, mapping_id)
    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")


def _public_wiki_index_item(value: Any) -> dict[str, Any] | None:
    if not isinstance(value, dict):
        return None
    slug = str(value.get("slug") or "").strip()
    if not slug:
        return None
    return {
        "slug": slug,
        "title": str(value.get("title") or slug),
        "summary": str(value.get("summary") or ""),
        "category_path": ([str(item) for item in value.get("category_path", []) if str(item).strip()] if isinstance(value.get("category_path"), list) else []),
        "wiki_path": str(value.get("wiki_path") or ""),
        "depth": int(value.get("depth") or 0),
    }


async def _all_public_wiki_index(client: WeKnoraClient, remote_id: str) -> dict[str, Any]:
    intro = ""
    version = 0
    groups: list[dict[str, Any]] = []
    for page_type in _PUBLIC_WIKI_INDEX_TYPES:
        items: list[dict[str, Any]] = []
        cursor = ""
        seen_cursors: set[str] = set()
        total = 0
        for _page_number in range(100):
            payload = await client.get_wiki_index(
                remote_id,
                types=[page_type],
                limit=500,
                cursor=cursor,
            )
            intro = intro or str(payload.get("intro") or "")
            version = max(version, int(payload.get("version") or 0))
            raw_groups = payload.get("groups") if isinstance(payload.get("groups"), list) else []
            group = next(
                (value for value in raw_groups if isinstance(value, dict) and value.get("type") == page_type),
                {},
            )
            total = max(total, int(group.get("total") or 0))
            raw_items = group.get("items") if isinstance(group.get("items"), list) else []
            items.extend(item for value in raw_items if (item := _public_wiki_index_item(value)) is not None)
            next_cursor = str(group.get("next_cursor") or "")
            if not next_cursor or next_cursor in seen_cursors:
                break
            seen_cursors.add(next_cursor)
            cursor = next_cursor
        else:
            logger.warning(
                "Stopped public Wiki index pagination at safety limit: mapping_remote_id=%s page_type=%s",
                remote_id,
                page_type,
            )
        groups.append({"type": page_type, "total": total or len(items), "items": items})
    return {"intro": intro, "version": version, "groups": groups}


@router.get("/api/public/llmwiki/knowledge-bases/{mapping_id}/wiki/index")
async def get_public_wiki_index(request: Request, mapping_id: str) -> dict[str, Any]:
    """Return WeKnora's ordered dynamic Wiki index for the public reader."""

    row = await _public_mapping(request, mapping_id)
    try:
        return await _all_public_wiki_index(_client_or_503(), str(row["weknora_id"]))
    except WeKnoraError as exc:
        _raise_upstream(exc)


@router.get("/api/public/llmwiki/knowledge-bases/{mapping_id}/wiki/pages/{slug:path}")
async def get_public_wiki_page(
    request: Request,
    mapping_id: str,
    slug: str,
) -> dict[str, Any]:
    """Read one Wiki article from a published knowledge base."""

    row = await _public_mapping(request, mapping_id)
    try:
        page = await _client_or_503().get_wiki_page(str(row["weknora_id"]), slug)
    except WeKnoraError as exc:
        _raise_upstream(exc)
    return _wiki_page_view(page) or page


@router.get("/api/llmwiki/knowledge-bases")
async def list_knowledge_bases(
    request: Request,
    scope: Literal["personal", "public", "all"] = Query(default="all"),
    refresh: bool = False,
) -> dict[str, Any]:
    trace = trace_request(request, "llmwiki_knowledge_bases")
    try:
        stage_started = time.perf_counter()
        user_id, is_admin = await _actor(request)
        trace_stage(trace, "actor_and_access", stage_started)

        stage_started = time.perf_counter()
        store = _store(request)
        rows = await store.list_visible(user_id, scope=scope, is_admin=False)
        trace_stage(trace, "local_knowledge_base_query_db", stage_started)

        client: WeKnoraClient | None = None
        remote_items: list[dict[str, Any]] = []
        remote: dict[str, dict[str, Any]] = {}
        if refresh:
            stage_started = time.perf_counter()
            client = _client_or_503()
            try:
                remote_items = await client.list_knowledge_bases(refresh=True)
                remote = {str(item.get("id")): item for item in remote_items if item.get("id")}
            except WeKnoraError as exc:
                _raise_upstream(exc)
            trace_stage(trace, "weknora_explicit_refresh_wait", stage_started)

            deposit_remote = next(
                (item for item in remote_items if str(item.get("name") or "").strip() == CONVERSATION_DEPOSIT_KB_NAME),
                None,
            )
            if deposit_remote is not None:
                previous_mapping = await store.get_by_weknora_id(str(deposit_remote.get("id") or ""))
                deposit_mapping = await ensure_conversation_deposit_mapping(
                    store,
                    client,
                    remote_knowledge_bases=remote_items,
                )
                description_was_synchronized = previous_mapping is None or not _is_conversation_deposit_mapping(previous_mapping) or str(previous_mapping.get("description") or "") != CONVERSATION_DEPOSIT_KB_DESCRIPTION
                if deposit_mapping is not None and description_was_synchronized:
                    remote_id = str(deposit_mapping["weknora_id"])
                    remote[remote_id] = {
                        **remote.get(remote_id, deposit_remote),
                        "description": CONVERSATION_DEPOSIT_KB_DESCRIPTION,
                    }
            trace_stage(trace, "conversation_deposit_mapping_sync", stage_started)

        stage_started = time.perf_counter()
        knowledge_bases: list[dict[str, Any]] = []
        for row in rows:
            if refresh:
                remote_item = remote.get(str(row["weknora_id"]))
                # An explicit successful refresh is authoritative: do not show
                # mappings whose remote knowledge base has been deleted.
                if remote_item is None:
                    continue
                assert client is not None
                row, remote_item = await _sync_conversation_deposit_view(store, client, row, remote_item)
            else:
                # Normal page loads are served entirely from the local SQL
                # mapping. A remote check happens only after an explicit refresh.
                remote_item = {
                    "id": str(row.get("weknora_id") or ""),
                    "name": str(row.get("name") or ""),
                    "description": str(row.get("description") or ""),
                    "_local_cache_only": True,
                }
            knowledge_bases.append(_mapping_view(row, remote_item, actor_user_id=user_id, is_admin=is_admin))
        trace_stage(trace, "map_visible_knowledge_bases", stage_started)
        trace["details"].update({"scope": scope, "refresh": refresh, "remote_count": len(remote_items) if refresh else None, "local_count": len(rows), "result_count": len(knowledge_bases)})
        return {"knowledge_bases": knowledge_bases}
    finally:
        finish_trace(trace)


@router.get("/api/llmwiki/embedding-models")
async def list_embedding_models(request: Request) -> dict[str, Any]:
    """List selectable WeKnora embedding models without exposing model secrets."""

    await _actor(request)
    try:
        models = await _embedding_models(_client_or_503())
    except WeKnoraError as exc:
        _raise_upstream(exc)
    available = [item for item in models if item["available"]]
    default_model = next((item for item in available if item["is_default"]), None) or next(iter(available), None)
    default_model_id = str(default_model["id"]) if default_model is not None else None
    return {"models": models, "default_model_id": default_model_id}


@router.get("/api/llmwiki/agents/{agent_id}/knowledge-bases")
async def list_agent_knowledge_bases(request: Request, agent_id: str) -> dict[str, Any]:
    """Return the agent-bound bases still visible to the current DeerFlow user."""

    trace = trace_request(request, "llmwiki_agent_knowledge_bases")
    try:
        stage_started = time.perf_counter()
        user_id, is_admin = await _actor(request)
        agent_store = get_agent_store(request)
        agent = await (agent_store.get_any(agent_id) if is_admin else agent_store.get_visible(agent_id, user_id))
        if agent is None:
            raise HTTPException(status_code=404, detail="Agent not found")
        trace_stage(trace, "actor_and_agent_access", stage_started)

        stage_started = time.perf_counter()
        try:
            config = load_agent_config(agent_id)
        except (FileNotFoundError, ValueError):
            config = None
        bound_ids = list(config.llmwiki_knowledge_base_ids or []) if config else []
        trace_stage(trace, "load_agent_binding_config", stage_started)

        stage_started = time.perf_counter()
        store = _store(request)
        rows: list[dict[str, Any]] = []
        for mapping_id in bound_ids:
            row = await store.get_authorized(
                mapping_id,
                user_id,
                write=False,
                is_admin=is_admin,
            )
            if row is not None:
                rows.append(row)
        trace_stage(trace, "local_knowledge_base_mapping_query", stage_started)

        stage_started = time.perf_counter()
        knowledge_bases = []
        for row in rows:
            remote_item = {
                "id": str(row.get("weknora_id") or ""),
                "name": str(row.get("name") or ""),
                "description": str(row.get("description") or ""),
                "_local_cache_only": True,
            }
            knowledge_bases.append(_mapping_view(row, remote_item, actor_user_id=user_id, is_admin=is_admin))
        trace_stage(trace, "map_agent_knowledge_bases", stage_started)
        trace["details"].update({"bound_count": len(bound_ids), "remote_count": None, "result_count": len(knowledge_bases)})
        return {"knowledge_bases": knowledge_bases}
    finally:
        finish_trace(trace)


@router.post("/api/llmwiki/knowledge-bases", status_code=201)
async def create_knowledge_base(request: Request, body: KnowledgeBaseCreate) -> dict[str, Any]:
    client = _client_or_503()
    trace = trace_request(request, "llmwiki_knowledge_base_create")
    stage_started = time.perf_counter()
    user_id, is_admin = await _actor(request)
    trace_stage(trace, "actor_and_access", stage_started)
    if body.name.strip() == CONVERSATION_DEPOSIT_KB_NAME:
        raise HTTPException(status_code=409, detail="对话沉淀是 cmzs 系统保留知识库")
    wiki_config = body.wiki_config if body.wiki_enabled else None
    directory_paths_were_set = bool(
        wiki_config is not None
        and "directory_paths" in wiki_config.model_fields_set
    )
    parent_directories_were_set = bool(
        wiki_config is not None
        and "parent_directories" in wiki_config.model_fields_set
    )
    try:
        directory_paths = resolve_directory_paths(
            wiki_config.directory_paths if wiki_config else None,
            wiki_config.parent_directories if wiki_config else None,
            directory_paths_set=directory_paths_were_set,
            parent_directories_set=parent_directories_were_set,
        )
        parent_directories = directory_roots(directory_paths)
        focus_instructions = normalize_focus_instructions(
            wiki_config.focus_instructions if wiki_config else None
        )
        raw_content_instructions = (
            str(wiki_config.content_instructions or "").strip()
            if wiki_config
            else ""
        )
        raw_extraction_instructions = (
            str(wiki_config.extraction_instructions or "").strip()
            if wiki_config
            else ""
        )
        compiled = compile_wiki_instructions(
            content_instructions=raw_content_instructions,
            extraction_instructions=raw_extraction_instructions,
            focus_instructions=focus_instructions,
            directory_paths=directory_paths,
        )
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from None
    try:
        stage_started = time.perf_counter()
        upstream_models = await client.list_models()
        embedding_model_id = _select_embedding_model(
            [view for item in upstream_models if (view := _embedding_model_view(item)) is not None],
            body.embedding_model_id,
        )
        knowledge_qa_model_id = (
            _select_knowledge_qa_model(upstream_models) if body.wiki_enabled else None
        )
        trace_stage(trace, "weknora_model_catalog", stage_started)
        stage_started = time.perf_counter()
        remote = await client.create_knowledge_base(
            name=body.name.strip(),
            description=body.description,
            kb_type=body.type,
            wiki_enabled=body.wiki_enabled,
            embedding_model_id=embedding_model_id,
            knowledge_qa_model_id=knowledge_qa_model_id,
            wiki_content_instructions=(
                compiled.content_instructions if body.wiki_enabled else ""
            ),
            wiki_extraction_instructions=(
                compiled.extraction_instructions if body.wiki_enabled else ""
            ),
        )
        trace_stage(trace, "weknora_create", stage_started)
    except WeKnoraError as exc:
        finish_trace(trace)
        _raise_upstream(exc)
    try:
        # Directory settings are references for WeKnora's generator. Creating
        # folders here would turn the reference into an accidental template.
        stage_started = time.perf_counter()
        row = await _store(request).create_mapping(
            weknora_id=str(remote["id"]),
            owner_user_id=user_id,
            name=str(remote.get("name") or body.name),
            description=str(remote.get("description") or body.description),
            kb_type=str(remote.get("type") or body.type),
            retrieval_settings=dict(DEFAULT_RETRIEVAL_SETTINGS),
            wiki_parent_directories=parent_directories or None,
            wiki_directory_paths=(directory_paths if body.wiki_enabled else []),
            wiki_focus_instructions=(focus_instructions if body.wiki_enabled else ""),
            wiki_content_instructions_raw=(
                raw_content_instructions if body.wiki_enabled else ""
            ),
            wiki_extraction_instructions_raw=(
                raw_extraction_instructions if body.wiki_enabled else ""
            ),
            wiki_generation_managed=body.wiki_enabled,
            wiki_config_schema_version=(
                WIKI_CONFIG_SCHEMA_VERSION if body.wiki_enabled else None
            ),
            wiki_config_revision=1 if body.wiki_enabled else None,
            wiki_config_apply_status=(
                "active" if body.wiki_enabled else "not_applicable"
            ),
            wiki_config_apply_error=None,
        )
        trace_stage(trace, "local_mapping_create", stage_started)
    except Exception as exc:
        finish_trace(trace)
        logger.exception(
            "Failed to initialize or persist LLMWiki mapping; compensating remote create"
        )
        try:
            await client.delete_knowledge_base(str(remote["id"]))
        except Exception:
            logger.exception("Failed to compensate orphan WeKnora knowledge base")
        if isinstance(exc, WeKnoraError):
            _raise_upstream(exc)
        raise HTTPException(status_code=500, detail="Failed to register the knowledge base") from None
    _schedule_local_wiki_sync(request, row)
    finish_trace(trace)
    return _mapping_view(row, remote, actor_user_id=user_id, is_admin=is_admin)


@router.get("/api/llmwiki/knowledge-bases/{mapping_id}")
async def get_knowledge_base(request: Request, mapping_id: str) -> dict[str, Any]:
    row, user_id, is_admin = await _authorized_mapping(request, mapping_id, write=False)
    client = _client_or_503()
    try:
        remote = await client.get_knowledge_base(str(row["weknora_id"]))
    except WeKnoraError as exc:
        _raise_upstream(exc)
    row, remote = await _sync_conversation_deposit_view(_store(request), client, row, remote)
    return _mapping_view(row, remote, actor_user_id=user_id, is_admin=is_admin)


@router.patch("/api/llmwiki/knowledge-bases/{mapping_id}")
async def update_knowledge_base(request: Request, mapping_id: str, body: KnowledgeBaseUpdate) -> dict[str, Any]:
    row, user_id, is_admin = await _authorized_mapping(request, mapping_id, write=True)
    changes = body.model_dump(exclude_none=True)
    if "name" in changes:
        changes["name"] = changes["name"].strip()
        if not changes["name"]:
            raise HTTPException(status_code=422, detail="知识库名称不能为空")
        if changes["name"] == CONVERSATION_DEPOSIT_KB_NAME:
            raise HTTPException(status_code=409, detail="对话沉淀是 cmzs 系统保留知识库")
    questions_were_set = "test_questions" in body.model_fields_set
    questions = None
    if questions_were_set:
        questions = list(dict.fromkeys(str(item).strip() for item in (body.test_questions or []) if str(item).strip()))
        if any(len(item) > 500 for item in questions):
            raise HTTPException(status_code=422, detail="测试问题不能超过 500 个字符")
    retrieval_settings = None
    if body.retrieval_settings is not None:
        retrieval_settings = body.retrieval_settings.model_dump()
    wiki_config_fields = (
        set(body.wiki_config.model_fields_set)
        if body.wiki_config is not None
        else set()
    )
    directory_paths_were_set = "directory_paths" in wiki_config_fields
    parent_directories_were_set = bool(
        "parent_directories" in wiki_config_fields
    )
    directory_config_were_set = (
        directory_paths_were_set or parent_directories_were_set
    )
    client = _client_or_503()
    remote_before: dict[str, Any] | None = None
    if wiki_config_fields:
        try:
            remote_before = await client.get_knowledge_base(str(row["weknora_id"]))
        except WeKnoraError as exc:
            _raise_upstream(exc)
        strategy = remote_before.get("indexing_strategy")
        remote_wiki_enabled = (
            strategy.get("wiki_enabled")
            if isinstance(strategy, dict)
            else None
        )
        if row.get("wiki_generation_managed") is False or remote_wiki_enabled is False:
            raise HTTPException(
                status_code=409,
                detail="当前知识库未启用 Wiki 生成，不能保存 Wiki 生成配置",
            )

    current_paths = _persisted_directory_paths(row)
    directory_paths = current_paths
    parent_directories = directory_roots(current_paths)
    current_remote_wiki = (
        dict(remote_before.get("wiki_config"))
        if isinstance(remote_before, dict)
        and isinstance(remote_before.get("wiki_config"), dict)
        else {}
    )
    current_content = _raw_wiki_text(
        row,
        current_remote_wiki,
        "wiki_content_instructions_raw",
        "content_instructions",
    )
    current_extraction = _raw_wiki_text(
        row,
        current_remote_wiki,
        "wiki_extraction_instructions_raw",
        "extraction_instructions",
    )
    current_focus = str(row.get("wiki_focus_instructions") or "")
    raw_content = current_content
    raw_extraction = current_extraction
    focus_instructions = current_focus
    compiled = None
    config_changed = False
    if wiki_config_fields:
        try:
            if directory_config_were_set:
                directory_paths = resolve_directory_paths(
                    body.wiki_config.directory_paths if body.wiki_config else None,
                    body.wiki_config.parent_directories if body.wiki_config else None,
                    directory_paths_set=directory_paths_were_set,
                    parent_directories_set=parent_directories_were_set,
                )
                parent_directories = directory_roots(directory_paths)
            if "focus_instructions" in wiki_config_fields:
                focus_instructions = normalize_focus_instructions(
                    body.wiki_config.focus_instructions if body.wiki_config else None
                )
            if "content_instructions" in wiki_config_fields:
                raw_content = str(
                    body.wiki_config.content_instructions or ""
                    if body.wiki_config
                    else ""
                ).strip()
            if "extraction_instructions" in wiki_config_fields:
                raw_extraction = str(
                    body.wiki_config.extraction_instructions or ""
                    if body.wiki_config
                    else ""
                ).strip()
            compiled = compile_wiki_instructions(
                content_instructions=raw_content,
                extraction_instructions=raw_extraction,
                focus_instructions=focus_instructions,
                directory_paths=directory_paths,
            )
        except ValueError as exc:
            raise HTTPException(status_code=422, detail=str(exc)) from None
        config_changed = any(
            (
                directory_paths != current_paths,
                focus_instructions != current_focus,
                raw_content != current_content,
                raw_extraction != current_extraction,
            )
        )

    remote_changes = {
        key: value
        for key, value in changes.items()
        if key in {"name", "description"}
    }
    if config_changed and compiled is not None:
        remote_wiki_config: dict[str, str] = {}
        if (
            directory_paths
            or focus_instructions
            or directory_config_were_set
            or "focus_instructions" in wiki_config_fields
        ):
            remote_wiki_config = {
                "content_instructions": compiled.content_instructions,
                "extraction_instructions": compiled.extraction_instructions,
            }
        else:
            if "content_instructions" in wiki_config_fields:
                remote_wiki_config["content_instructions"] = raw_content
            if "extraction_instructions" in wiki_config_fields:
                remote_wiki_config["extraction_instructions"] = raw_extraction
        if remote_wiki_config:
            remote_changes["wiki_config"] = remote_wiki_config
    if (
        not remote_changes
        and not questions_were_set
        and retrieval_settings is None
        and not config_changed
    ):
        return _mapping_view(
            row,
            remote_before,
            actor_user_id=user_id,
            is_admin=is_admin,
        )
    remote = remote_before
    if remote_changes:
        try:
            remote = await client.update_knowledge_base(str(row["weknora_id"]), remote_changes)
        except WeKnoraError as exc:
            _raise_upstream(exc)
    elif remote is None:
        try:
            remote = await client.get_knowledge_base(str(row["weknora_id"]))
        except WeKnoraError as exc:
            _raise_upstream(exc)
    local_changes: dict[str, Any] = {
        "name": remote_changes.get("name"),
        "description": remote_changes.get("description"),
        "test_questions": questions,
        "test_questions_set": questions_were_set,
        "retrieval_settings": retrieval_settings,
    }
    if config_changed:
        local_changes["wiki_parent_directories"] = parent_directories
        local_changes["wiki_directory_paths"] = directory_paths
        local_changes["wiki_focus_instructions"] = focus_instructions
        local_changes["wiki_content_instructions_raw"] = raw_content
        local_changes["wiki_extraction_instructions_raw"] = raw_extraction
        local_changes["wiki_generation_managed"] = True
        local_changes["wiki_config_schema_version"] = WIKI_CONFIG_SCHEMA_VERSION
        local_changes["wiki_config_revision"] = int(
            row.get("wiki_config_revision") or 0
        ) + 1
        local_changes["wiki_config_apply_status"] = "active"
        local_changes["wiki_config_apply_error"] = None
    try:
        updated = await _store(request).update_mapping(mapping_id, **local_changes)
    except Exception:
        # A synchronous Wiki-config update reaches WeKnora before the local
        # projection is committed. Restore the provider's prior effective
        # instructions if that commit fails, so the database does not claim an
        # unapplied revision and a later edit can retry from the same raw state.
        if config_changed and remote_before is not None and remote_changes:
            rollback_changes: dict[str, Any] = {}
            if "name" in remote_changes:
                rollback_changes["name"] = str(remote_before.get("name") or row.get("name") or "")
            if "description" in remote_changes:
                rollback_changes["description"] = str(
                    remote_before.get("description")
                    if remote_before.get("description") is not None
                    else row.get("description") or ""
                )
            if "wiki_config" in remote_changes:
                previous_wiki_config = remote_before.get("wiki_config")
                previous_wiki_config = (
                    dict(previous_wiki_config)
                    if isinstance(previous_wiki_config, dict)
                    else {}
                )
                rollback_changes["wiki_config"] = {
                    "content_instructions": str(
                        previous_wiki_config.get("content_instructions") or ""
                    ),
                    "extraction_instructions": str(
                        previous_wiki_config.get("extraction_instructions") or ""
                    ),
                }
            try:
                await client.update_knowledge_base(
                    str(row["weknora_id"]),
                    rollback_changes,
                )
            except Exception:
                logger.exception(
                    "Failed to roll back WeKnora after local Wiki-config persistence failed"
                )
        raise
    effective_row = updated or row
    focus_config_was_set = "focus_instructions" in wiki_config_fields
    if focus_config_was_set and config_changed:
        _schedule_local_wiki_sync(
            request,
            effective_row,
            constraints_changed=True,
        )
    return _mapping_view(effective_row, remote, actor_user_id=user_id, is_admin=is_admin)


@router.delete("/api/llmwiki/knowledge-bases/{mapping_id}", status_code=204)
async def delete_knowledge_base(request: Request, mapping_id: str) -> None:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=True)
    try:
        await _client_or_503().delete_knowledge_base(str(row["weknora_id"]))
    except WeKnoraError as exc:
        # A 404 means the remote data is already gone. The owner/admin must still
        # be able to remove DeerFlow's stale mapping; other upstream failures stay
        # visible and never cause local-only deletion.
        if exc.status_code != 404:
            _raise_upstream(exc)
    await _store(request).delete_mapping(mapping_id)


@router.get("/api/llmwiki/knowledge-bases/{mapping_id}/documents")
async def list_documents(
    request: Request,
    mapping_id: str,
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=100, ge=1, le=200),
) -> dict[str, Any]:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=False)
    try:
        result = await _client_or_503().list_documents(str(row["weknora_id"]), page=page, page_size=page_size)
    except WeKnoraError as exc:
        _raise_upstream(exc)
    return {**result, "items": [_document_view(item) for item in result["items"]]}


@router.get("/api/llmwiki/knowledge-bases/{mapping_id}/documents/{document_id}")
async def get_document(request: Request, mapping_id: str, document_id: str) -> dict[str, Any]:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=False)
    try:
        document = await _verify_document(_client_or_503(), row, document_id)
    except WeKnoraError as exc:
        _raise_upstream(exc)
    return _document_view(document)


@router.get("/api/llmwiki/knowledge-bases/{mapping_id}/documents/{document_id}/chunks")
async def list_document_chunks(
    request: Request,
    mapping_id: str,
    document_id: str,
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=100),
) -> dict[str, Any]:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=False)
    client = _client_or_503()
    try:
        await _verify_document(client, row, document_id)
        result = await client.list_chunks(document_id, page=page, page_size=page_size)
    except WeKnoraError as exc:
        _raise_upstream(exc)
    return {**result, "items": [_chunk_view(item) for item in result["items"]]}


@router.get("/api/llmwiki/knowledge-bases/{mapping_id}/documents/{document_id}/preview")
async def preview_document(request: Request, mapping_id: str, document_id: str) -> Response:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=False)
    client = _client_or_503()
    try:
        document = await _verify_document(client, row, document_id)
        content, media_type = await client.get_document_preview(document_id)
    except WeKnoraError as exc:
        _raise_upstream(exc)
    filename = str(document.get("file_name") or document.get("title") or "document")
    return Response(
        content=content,
        media_type=media_type,
        headers={
            "Content-Disposition": f"inline; filename*=UTF-8''{quote(filename)}",
            "Cache-Control": "private, max-age=60",
        },
    )


@router.get("/api/llmwiki/knowledge-bases/{mapping_id}/files")
async def get_wiki_file(
    request: Request,
    mapping_id: str,
    file_path: str = Query(..., min_length=1, max_length=8192),
) -> Response:
    """Serve one protected Wiki image after normal knowledge-base authorization."""

    row, _, _ = await _authorized_mapping(request, mapping_id, write=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")
async def get_knowledge_base_graph(
    request: Request,
    mapping_id: str,
    limit: int = Query(default=200, ge=1, le=500),
) -> dict[str, Any]:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=False)
    client = _client_or_503()
    try:
        remote = await client.get_knowledge_base(str(row["weknora_id"]))
        strategy = remote.get("indexing_strategy") if isinstance(remote.get("indexing_strategy"), dict) else {}
        if not strategy.get("wiki_enabled"):
            return {
                "nodes": [],
                "edges": [],
                "meta": {"mode": "disabled", "total": 0, "returned": 0, "truncated": False},
            }
        graph = await client.get_wiki_graph(str(row["weknora_id"]), limit=limit)
    except WeKnoraError as exc:
        _raise_upstream(exc)
    nodes = [
        {
            "id": str(node.get("slug") or ""),
            "title": str(node.get("title") or node.get("slug") or ""),
            "type": str(node.get("page_type") or "page"),
            "link_count": int(node.get("link_count") or 0),
        }
        for node in graph.get("nodes", [])
        if isinstance(node, dict) and node.get("slug")
    ]
    edges = [{"source": str(edge.get("source") or ""), "target": str(edge.get("target") or "")} for edge in graph.get("edges", []) if isinstance(edge, dict) and edge.get("source") and edge.get("target")]
    meta = graph.get("meta") if isinstance(graph.get("meta"), dict) else {}
    return {"nodes": nodes, "edges": edges, "meta": meta}


def _knowledge_research_store(request: Request):
    store = getattr(request.app.state, "knowledge_research_store", None)
    if store is None:
        raise HTTPException(status_code=503, detail="知识库研究存储尚未就绪")
    return store


async def _knowledge_research_can_write(
    request: Request,
    mapping: dict[str, Any],
    user_id: str,
    is_admin: bool,
) -> bool:
    if _is_conversation_deposit_mapping(mapping):
        return False
    writable = await _store(request).get_authorized(
        str(mapping["id"]),
        user_id,
        write=True,
        is_admin=is_admin,
    )
    return writable is not None


def _knowledge_research_publication_view(
    row: dict[str, Any],
    *,
    user_id: str,
    can_write: bool,
    include_content: bool = False,
) -> dict[str, Any] | None:
    if not row.get("saved_hash"):
        return None
    owned = row.get("user_id") == user_id
    result: dict[str, Any] = {
        "id": row["publication_id"],
        "title": row.get("saved_title") or row.get("title") or "",
        "wiki_slug": row.get("wiki_slug") or "",
        "saved_hash": row["saved_hash"],
        "saved_at": row.get("saved_at"),
        "updated_at": row.get("saved_at") or row.get("updated_at"),
        "owned_by_current_user": owned,
        "can_continue": owned,
        "can_write": owned and can_write,
    }
    if owned:
        # Only the creator receives a bridge from the public book to the
        # private research session.  The thread id remains session-only.
        result["session_id"] = row["id"]
    if include_content:
        result["content_markdown"] = row.get("saved_markdown") or ""
    return result


def _knowledge_research_session_view(
    row: dict[str, Any],
    *,
    user_id: str,
    can_write: bool,
) -> dict[str, Any]:
    return {
        "id": row["id"],
        "thread_id": row["thread_id"],
        "title": row.get("title") or "",
        "draft_markdown": row.get("draft_markdown") or "",
        "created_at": row.get("created_at"),
        "updated_at": row.get("updated_at"),
        "can_write": can_write,
        "publication": _knowledge_research_publication_view(
            row,
            user_id=user_id,
            can_write=can_write,
        ),
    }


def _knowledge_research_default_title(question: str) -> str:
    for line in question.splitlines():
        normalized = " ".join(line.split())
        if normalized:
            return normalized[:255]
    raise HTTPException(status_code=422, detail="问题不能为空")


@router.get("/api/llmwiki/knowledge-bases/{mapping_id}/research/sessions")
async def list_knowledge_research_sessions(
    request: Request,
    mapping_id: str,
) -> dict[str, Any]:
    mapping, user_id, is_admin = await _authorized_mapping(
        request,
        mapping_id,
        write=False,
    )
    can_write = await _knowledge_research_can_write(
        request,
        mapping,
        user_id,
        is_admin,
    )
    store = _knowledge_research_store(request)
    sessions, publications = await asyncio.gather(
        store.list_owned(mapping_id, user_id),
        store.list_published(mapping_id),
    )
    return {
        "sessions": [
            _knowledge_research_session_view(
                row,
                user_id=user_id,
                can_write=can_write,
            )
            for row in sessions
        ],
        "publications": [
            view
            for row in publications
            if (
                view := _knowledge_research_publication_view(
                    row,
                    user_id=user_id,
                    can_write=can_write,
                )
            )
            is not None
        ],
    }


@router.post("/api/llmwiki/knowledge-bases/{mapping_id}/research/sessions", status_code=201)
async def create_knowledge_research_session(
    request: Request,
    mapping_id: str,
    body: KnowledgeResearchSessionCreate,
) -> dict[str, Any]:
    mapping, user_id, is_admin = await _authorized_mapping(
        request,
        mapping_id,
        write=False,
    )
    can_write = await _knowledge_research_can_write(
        request,
        mapping,
        user_id,
        is_admin,
    )
    question = body.question.strip()
    title = _knowledge_research_default_title(question)
    store = _knowledge_research_store(request)
    session_id = str(uuid4())
    publication_id = str(uuid4())
    thread_id = str(uuid4())
    session = await store.create(
        session_id=session_id,
        publication_id=publication_id,
        thread_id=thread_id,
        knowledge_base_id=mapping_id,
        user_id=user_id,
        title=title,
    )
    metadata = {
        "system": True,
        "thread_type": "knowledge_research",
        "knowledge_research_session_id": session_id,
        "knowledge_research_knowledge_base_id": mapping_id,
        "llmwiki_knowledge_base_ids": [mapping_id],
        "assistant_knowledge_base_ids": [],
        "knowledge_retrieval_mode": "classic",
        "knowledge_space_strict": True,
        "rag_mode_enabled": True,
        "rag_mode": True,
    }
    try:
        from app.gateway.routers.threads import ThreadCreateRequest, create_thread

        await create_thread(
            ThreadCreateRequest(
                thread_id=thread_id,
                assistant_id=None,
                metadata=metadata,
                initialize_checkpoint=True,
            ),
            request,
        )
        await get_thread_store(request).update_display_name(
            thread_id,
            title,
            user_id=user_id,
        )
    except Exception:
        # Thread creation writes metadata before its initial checkpoint.  Undo
        # every side on failure so no half-bound private chat remains.
        await store.delete(session_id)
        try:
            checkpointer = get_checkpointer(request)
            if hasattr(checkpointer, "adelete_thread"):
                await checkpointer.adelete_thread(thread_id)
        except Exception:
            logger.debug("Could not remove failed research checkpoint", exc_info=True)
        try:
            await get_thread_store(request).delete(thread_id, user_id=user_id)
        except Exception:
            logger.debug("Could not remove failed research thread metadata", exc_info=True)
        raise
    return _knowledge_research_session_view(
        session,
        user_id=user_id,
        can_write=can_write,
    )


async def _owned_knowledge_research_session(
    request: Request,
    mapping_id: str,
    session_id: str,
) -> tuple[dict[str, Any], dict[str, Any], str, bool, bool]:
    mapping, user_id, is_admin = await _authorized_mapping(
        request,
        mapping_id,
        write=False,
    )
    session = await _knowledge_research_store(request).get_owned(
        session_id,
        user_id,
    )
    if session is None or session["knowledge_base_id"] != mapping_id:
        raise HTTPException(status_code=404, detail="研究会话不存在或无权访问")
    can_write = await _knowledge_research_can_write(
        request,
        mapping,
        user_id,
        is_admin,
    )
    return session, mapping, user_id, is_admin, can_write


@router.get(
    "/api/llmwiki/knowledge-bases/{mapping_id}/research/sessions/{session_id}"
)
async def get_knowledge_research_session(
    request: Request,
    mapping_id: str,
    session_id: str,
) -> dict[str, Any]:
    session, _, user_id, _, can_write = await _owned_knowledge_research_session(
        request,
        mapping_id,
        session_id,
    )
    return _knowledge_research_session_view(
        session,
        user_id=user_id,
        can_write=can_write,
    )


@router.patch(
    "/api/llmwiki/knowledge-bases/{mapping_id}/research/sessions/{session_id}"
)
async def update_knowledge_research_session(
    request: Request,
    mapping_id: str,
    session_id: str,
    body: KnowledgeResearchSessionUpdate,
) -> dict[str, Any]:
    current, _, user_id, _, can_write = await _owned_knowledge_research_session(
        request,
        mapping_id,
        session_id,
    )
    fields = body.model_fields_set
    if not fields:
        return _knowledge_research_session_view(
            current,
            user_id=user_id,
            can_write=can_write,
        )
    title = body.title.strip() if "title" in fields and body.title is not None else None
    if "title" in fields and not title:
        raise HTTPException(status_code=422, detail="标题不能为空")
    updated = await _knowledge_research_store(request).update_owned(
        session_id,
        user_id,
        title=title,
        draft_markdown=(
            body.draft_markdown if "draft_markdown" in fields else None
        ),
    )
    if updated is None:
        raise HTTPException(status_code=404, detail="研究会话不存在或无权访问")
    if title is not None:
        try:
            await get_thread_store(request).update_display_name(
                updated["thread_id"],
                title,
                user_id=user_id,
            )
        except Exception:
            logger.debug("Could not update research thread title", exc_info=True)
    return _knowledge_research_session_view(
        updated,
        user_id=user_id,
        can_write=can_write,
    )


async def _upsert_knowledge_research_wiki_page(
    *,
    client: WeKnoraClient,
    knowledge_base_id: str,
    slug: str,
    payload: dict[str, Any],
    previously_saved: bool,
) -> dict[str, Any]:
    if previously_saved:
        try:
            return await client.update_wiki_page(
                knowledge_base_id,
                slug,
                payload,
            )
        except WeKnoraError as exc:
            if exc.status_code != 404:
                raise
            return await client.create_wiki_page(
                knowledge_base_id,
                {"slug": slug, **payload},
            )
    try:
        return await client.create_wiki_page(
            knowledge_base_id,
            {"slug": slug, **payload},
        )
    except WeKnoraError as create_error:
        # A prior request may have committed upstream before its local DB write
        # failed.  Probe the stable slug and turn the retry into an update.
        if create_error.status_code not in {400, 409}:
            raise
        try:
            await client.get_wiki_page(knowledge_base_id, slug)
        except WeKnoraError:
            raise create_error
        return await client.update_wiki_page(knowledge_base_id, slug, payload)


@router.put(
    "/api/llmwiki/knowledge-bases/{mapping_id}/research/sessions/{session_id}/publication"
)
async def save_knowledge_research_publication(
    request: Request,
    mapping_id: str,
    session_id: str,
    body: KnowledgeResearchPublicationSave,
) -> dict[str, Any]:
    # Write authorization and private-session ownership are independent: even
    # an administrator may not publish another user's private conversation.
    mapping, user_id, _ = await _authorized_mapping(request, mapping_id, write=True)
    if _is_conversation_deposit_mapping(mapping):
        raise HTTPException(
            status_code=403,
            detail="Conversation deposit is managed by cmzs",
        )
    title = body.title.strip()
    if not title:
        raise HTTPException(status_code=422, detail="标题不能为空")
    store = _knowledge_research_store(request)
    client = _client_or_503()
    remote_id = str(mapping["weknora_id"])
    saved: dict[str, Any] | None = None
    async with store.locked_publication(session_id, user_id) as publication:
        if publication is None:
            raise HTTPException(status_code=404, detail="研究会话不存在或无权访问")
        session = publication.snapshot()
        if session["knowledge_base_id"] != mapping_id:
            raise HTTPException(status_code=404, detail="研究会话不存在或无权访问")
        markdown = str(session.get("draft_markdown") or "")
        if not markdown.strip():
            raise HTTPException(status_code=422, detail="请先生成或填写 Markdown 正文")
        digest = hashlib.sha256(
            json.dumps(
                {"title": title, "markdown": markdown},
                ensure_ascii=False,
                sort_keys=True,
                separators=(",", ":"),
            ).encode("utf-8")
        ).hexdigest()
        current_saved_hash = session.get("saved_hash")
        if current_saved_hash == digest:
            # A prior identical request may have committed both upstream and
            # locally while its HTTP response was lost.  Treat that retry as
            # a no-op even if it carries the pre-save expected hash.
            return _knowledge_research_session_view(
                session,
                user_id=user_id,
                can_write=True,
            )
        if current_saved_hash and body.expected_saved_hash is None:
            raise HTTPException(
                status_code=409,
                detail={
                    "code": "KNOWLEDGE_RESEARCH_EXPECTED_HASH_REQUIRED",
                    "message": "覆盖已保存文档时必须携带当前版本，请刷新后再保存。",
                    "saved_hash": current_saved_hash,
                },
            )
        if (
            body.expected_saved_hash is not None
            and body.expected_saved_hash != current_saved_hash
        ):
            raise HTTPException(
                status_code=409,
                detail={
                    "code": "KNOWLEDGE_RESEARCH_PUBLICATION_CHANGED",
                    "message": "文档已被更新，请刷新后再保存。",
                    "saved_hash": current_saved_hash,
                },
            )

        slug = str(
            session.get("wiki_slug")
            or f"knowledge-research/{session['publication_id']}"
        )
        try:
            folder_ids = await ensure_directory_paths(
                client,
                remote_id,
                [[KNOWLEDGE_RESEARCH_DIRECTORY]],
            )
            folder_id = folder_ids.get(KNOWLEDGE_RESEARCH_DIRECTORY)
            if not folder_id:
                raise WeKnoraError("未能创建深度研究 Wiki 目录", status_code=502)
            page = await _upsert_knowledge_research_wiki_page(
                client=client,
                knowledge_base_id=remote_id,
                slug=slug,
                previously_saved=bool(current_saved_hash),
                payload={
                    "title": title,
                    "content": markdown,
                    "summary": "知识库问答沉淀",
                    "page_type": "summary",
                    "status": "published",
                    "category_path": [KNOWLEDGE_RESEARCH_DIRECTORY],
                    "folder_id": folder_id,
                },
            )
        except WeKnoraError as exc:
            if exc.status_code == 409:
                raise HTTPException(status_code=409, detail=str(exc)) from None
            _raise_upstream(exc)

        saved = publication.save(
            title=title,
            markdown=markdown,
            saved_hash=digest,
            wiki_slug=slug,
            wiki_page_id=str(page.get("id") or page.get("page_id") or "")
            or None,
            wiki_folder_id=folder_id,
        )
    assert saved is not None
    _schedule_local_wiki_sync(request, mapping)
    return _knowledge_research_session_view(
        saved,
        user_id=user_id,
        can_write=True,
    )


@router.get(
    "/api/llmwiki/knowledge-bases/{mapping_id}/research/publications/{publication_id}"
)
async def get_knowledge_research_publication(
    request: Request,
    mapping_id: str,
    publication_id: str,
) -> dict[str, Any]:
    mapping, user_id, is_admin = await _authorized_mapping(
        request,
        mapping_id,
        write=False,
    )
    publication = await _knowledge_research_store(request).get_publication(
        publication_id
    )
    if (
        publication is None
        or publication["knowledge_base_id"] != mapping_id
    ):
        raise HTTPException(status_code=404, detail="研究文档不存在")
    can_write = await _knowledge_research_can_write(
        request,
        mapping,
        user_id,
        is_admin,
    )
    view = _knowledge_research_publication_view(
        publication,
        user_id=user_id,
        can_write=can_write,
        include_content=True,
    )
    if view is None:
        raise HTTPException(status_code=404, detail="研究文档不存在")
    return view


@router.get("/api/llmwiki/knowledge-bases/{mapping_id}/wiki/pages")
async def list_wiki_pages(
    request: Request,
    mapping_id: str,
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=50, ge=1, le=500),
    query: str = Query(default="", max_length=255),
) -> dict[str, Any]:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=False)
    try:
        result = await _client_or_503().list_wiki_pages(
            str(row["weknora_id"]),
            page=page,
            page_size=page_size,
            query=query.strip(),
        )
        raw_pages = result.get("pages") if isinstance(result, dict) else []
        pages = [view for page_item in (raw_pages if isinstance(raw_pages, list) else []) if isinstance(page_item, dict) for view in [_wiki_page_view(page_item)] if view is not None]
        return {**result, "pages": pages}
    except WeKnoraError as exc:
        _raise_upstream(exc)


@router.get("/api/llmwiki/knowledge-bases/{mapping_id}/wiki/export.xlsx")
async def export_wiki_pages_excel(request: Request, mapping_id: str) -> StreamingResponse:
    """Export every processed Wiki page with its current and parent directories."""

    row, _, _ = await _authorized_mapping(request, mapping_id, write=False)
    client = _client_or_503()
    try:
        pages = await _all_wiki_pages(client, str(row["weknora_id"]))
    except WeKnoraError as exc:
        _raise_upstream(exc)
    try:
        output = await asyncio.to_thread(_build_wiki_excel, str(row.get("name") or "LLMWiki"), pages)
    except ImportError as exc:
        raise HTTPException(status_code=500, detail=f"openpyxl is not installed: {exc}") from exc
    timestamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S")
    safe_name = re.sub(r"[\\/:*?\"<>|\r\n]+", "_", str(row.get("name") or "LLMWiki")).strip(" ._") or "LLMWiki"
    filename = f"{safe_name}-Wiki-{timestamp}.xlsx"
    return StreamingResponse(
        output,
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": f"attachment; filename=\"wiki-export.xlsx\"; filename*=UTF-8''{quote(filename)}"},
    )


@router.post("/api/llmwiki/knowledge-bases/{mapping_id}/wiki/pages")
async def create_wiki_page(request: Request, mapping_id: str, body: WikiPageCreate) -> dict[str, Any]:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=True)
    payload = body.model_dump(exclude_none=True)
    try:
        page = await _client_or_503().create_wiki_page(str(row["weknora_id"]), payload)
        _schedule_local_wiki_sync(request, row)
        return _wiki_page_view(page) or page
    except WeKnoraError as exc:
        _raise_upstream(exc)


@router.get("/api/llmwiki/knowledge-bases/{mapping_id}/wiki/pages/{slug:path}")
async def get_wiki_page(request: Request, mapping_id: str, slug: str) -> dict[str, Any]:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=False)
    try:
        page = await _client_or_503().get_wiki_page(str(row["weknora_id"]), slug)
        return _wiki_page_view(page) or page
    except WeKnoraError as exc:
        _raise_upstream(exc)


@router.patch("/api/llmwiki/knowledge-bases/{mapping_id}/wiki/pages/{slug:path}")
async def update_wiki_page(request: Request, mapping_id: str, slug: str, body: WikiPageUpdate) -> dict[str, Any]:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=True)
    changes = body.model_dump(exclude_none=True)
    if not changes:
        try:
            page = await _client_or_503().get_wiki_page(str(row["weknora_id"]), slug)
            return _wiki_page_view(page) or page
        except WeKnoraError as exc:
            _raise_upstream(exc)
    try:
        page = await _client_or_503().update_wiki_page(str(row["weknora_id"]), slug, changes)
        _schedule_local_wiki_sync(request, row)
        return _wiki_page_view(page) or page
    except WeKnoraError as exc:
        _raise_upstream(exc)


@router.delete("/api/llmwiki/knowledge-bases/{mapping_id}/wiki/pages/{slug:path}", status_code=204)
async def delete_wiki_page(request: Request, mapping_id: str, slug: str) -> None:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=True)
    try:
        await _client_or_503().delete_wiki_page(str(row["weknora_id"]), slug)
        _schedule_local_wiki_sync(request, row)
    except WeKnoraError as exc:
        _raise_upstream(exc)


@router.post("/api/llmwiki/knowledge-bases/{mapping_id}/documents", status_code=201)
async def upload_document(request: Request, mapping_id: str, file: UploadFile = File(...)) -> dict[str, Any]:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=True)
    content = await file.read(_MAX_UPLOAD_BYTES + 1)
    if len(content) > _MAX_UPLOAD_BYTES:
        raise HTTPException(status_code=413, detail="File exceeds the 100 MB cmzs upload limit")
    if not file.filename:
        raise HTTPException(status_code=422, detail="File name is required")
    if looks_like_knowledge_package(content):
        raise HTTPException(status_code=422, detail=_KNOWLEDGE_PACKAGE_UPLOAD_DETAIL)
    try:
        result = await _client_or_503().upload_document(
            str(row["weknora_id"]),
            filename=file.filename,
            content=content,
            content_type=file.content_type or "application/octet-stream",
        )
        _schedule_local_wiki_sync(request, row)
        return result
    except WeKnoraError as exc:
        _raise_upstream(exc)


@router.post("/api/llmwiki/knowledge-bases/{mapping_id}/documents/url", status_code=201)
async def import_document_url(request: Request, mapping_id: str, body: KnowledgeUrlImport) -> dict[str, Any]:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=True)
    url = body.url.strip()
    if not url.lower().startswith(("http://", "https://")):
        raise HTTPException(status_code=422, detail="URL must start with http:// or https://")
    try:
        result = await _client_or_503().import_document_url(str(row["weknora_id"]), url=url)
        _schedule_local_wiki_sync(request, row)
        return result
    except WeKnoraError as exc:
        _raise_upstream(exc)


@router.post("/api/llmwiki/knowledge-bases/{mapping_id}/documents/manual", status_code=201)
async def create_manual_document(request: Request, mapping_id: str, body: ManualKnowledgeCreate) -> dict[str, Any]:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=True)
    try:
        result = await _client_or_503().create_manual_document(
            str(row["weknora_id"]),
            title=body.title.strip(),
            content=body.content,
        )
        _schedule_local_wiki_sync(request, row)
        return result
    except WeKnoraError as exc:
        _raise_upstream(exc)


@router.delete("/api/llmwiki/knowledge-bases/{mapping_id}/documents/{document_id}", status_code=204)
async def delete_document(request: Request, mapping_id: str, document_id: str) -> None:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=True)
    client = _client_or_503()
    try:
        await _verify_document(client, row, document_id)
        await client.delete_document(document_id)
        _schedule_local_wiki_sync(request, row)
    except WeKnoraError as exc:
        _raise_upstream(exc)


@router.post("/api/llmwiki/knowledge-bases/{mapping_id}/documents/{document_id}/reprocess")
async def reprocess_document(request: Request, mapping_id: str, document_id: str) -> dict[str, Any]:
    row, _, _ = await _authorized_mapping(request, mapping_id, write=True)
    client = _client_or_503()
    try:
        await _verify_document(client, row, document_id)
        result = await client.reprocess_document(document_id)
        _schedule_local_wiki_sync(request, row)
        return result
    except WeKnoraError as exc:
        _raise_upstream(exc)


@router.post("/api/llmwiki/search")
async def search_llmwiki(request: Request, body: SearchRequest) -> dict[str, Any]:
    user_id, is_admin = await _actor(request)
    store = _store(request)
    if body.knowledge_base_ids:
        rows: list[dict[str, Any]] = []
        for mapping_id in dict.fromkeys(body.knowledge_base_ids):
            row = await store.get_authorized(mapping_id, user_id, write=False, is_admin=is_admin)
            if row is None:
                raise HTTPException(status_code=404, detail="Knowledge base not found")
            rows.append(row)
    else:
        rows = await store.list_visible(user_id, scope=body.scope, is_admin=False)
    service = getattr(request.app.state, "llmwiki_retrieval_service", None)
    if service is None:
        raise HTTPException(status_code=503, detail={"code": "WIKI_RETRIEVAL_UNAVAILABLE"})
    include_drafts = body.include_drafts and (is_admin or all(str(row.get("owner_user_id")) == user_id for row in rows))
    try:
        result = await service.search(body.query, rows, top_k_pages=body.top_k, include_drafts=include_drafts)
    except WikiIndexError as exc:
        raise HTTPException(status_code=exc.status_code, detail={"code": exc.code, "message": str(exc)}) from None
    mapping_by_id = {str(row["id"]): row for row in rows}
    result["results"] = [_sanitize_conversation_deposit_value(item) if _is_conversation_deposit_mapping(mapping_by_id.get(str(item.get("knowledge_base_id") or ""), {})) else item for item in result.get("results") or []]
    return result


@router.post("/api/llmwiki/retrieval-debug")
async def debug_llmwiki_retrieval(request: Request, body: RetrievalDebugRequest) -> dict[str, Any]:
    """Preview the same bounded evidence that a normal Q&A turn would receive.

    The endpoint deliberately calls :func:`build_llmwiki_rag_context` instead
    of the vector-only API.  Its response therefore exposes the real
    vector/keyword/Wiki-page/original-chunk mode, thresholds, warnings and
    exact prompt-bounded evidence without starting a model run.
    """

    await _require_admin(request)
    clean_query = body.query.strip()
    if not clean_query:
        raise HTTPException(status_code=422, detail="检索问题不能为空")
    ordinary_ids = list(dict.fromkeys(item.strip() for item in body.knowledge_base_ids if item.strip()))
    assistant_ids = list(
        dict.fromkeys(item.strip() for item in body.assistant_knowledge_base_ids if item.strip())
    )
    if not ordinary_ids and not assistant_ids:
        raise HTTPException(status_code=422, detail="请至少选择一个知识空间")
    synthetic_body = SimpleNamespace(
        assistant_id=None,
        metadata=None,
        context={
            "llmwiki_knowledge_base_ids": ordinary_ids,
            "assistant_knowledge_base_ids": assistant_ids,
        },
    )
    rag = await build_llmwiki_rag_context(
        request=request,
        body=synthetic_body,
        graph_input={"messages": [HumanMessage(content=clean_query)]},
    )
    trace = dict(rag.trace)
    trace["trigger"] = "debug"
    return trace


@router.post("/api/llmwiki/conversation-deposits/turn", status_code=202)
async def create_conversation_deposit(
    request: Request,
    background_tasks: BackgroundTasks,
    body: ConversationDepositCreate,
) -> dict[str, Any]:
    runtime = get_resolved_llmwiki_runtime(getattr(request.app.state, "config", None))
    if not runtime.weknora_enabled:
        return {"queued": False, "provider": runtime.provider}
    user_id, _ = await _actor(request)
    if not await get_thread_store(request).check_access(body.thread_id, user_id, require_existing=True):
        raise HTTPException(status_code=404, detail="Thread not found")
    app = request.app
    turn = ConversationTurnDeposit(
        thread_id=body.thread_id,
        question=body.question,
        answer=body.answer,
        human_message_id=body.human_message_id,
        assistant_message_id=body.assistant_message_id,
        assistant_id=body.assistant_id,
        knowledge_base_ids=list(dict.fromkeys(item.strip() for item in body.knowledge_base_ids if item.strip())),
        created_by_user_id=user_id,
    )
    background_tasks.add_task(deposit_conversation_turn, app, turn)
    return {"queued": True, "provider": "weknora"}


@router.post("/api/llmwiki/knowledge-bases/{mapping_id}/publish")
async def request_publish(request: Request, mapping_id: str) -> dict[str, Any]:
    user_id, is_admin = await _actor(request)
    existing = await _store(request).get_authorized(
        mapping_id,
        user_id,
        write=True,
        is_admin=is_admin,
    )
    if existing is None:
        raise HTTPException(status_code=404, detail="Knowledge base not found")
    if _is_conversation_deposit_mapping(existing) and existing.get("publication_status") == "published":
        return {"id": existing["id"], "publication_status": "published"}
    try:
        await _client_or_503().get_knowledge_base(str(existing["weknora_id"]))
    except WeKnoraError as exc:
        _raise_upstream(exc)
    row = await _store(request).request_publish(mapping_id, user_id, is_admin=is_admin)
    if row is None:
        raise HTTPException(status_code=404, detail="Knowledge base not found")
    return {"id": row["id"], "publication_status": row["publication_status"]}


@router.post("/api/llmwiki/knowledge-bases/{mapping_id}/unpublish")
async def unpublish(request: Request, mapping_id: str) -> dict[str, Any]:
    user_id, is_admin = await _actor(request)
    existing = await _store(request).get_authorized(mapping_id, user_id, write=True, is_admin=is_admin)
    if existing is None:
        raise HTTPException(status_code=404, detail="Knowledge base not found")
    if _is_conversation_deposit_mapping(existing):
        raise HTTPException(status_code=403, detail="Conversation deposit is always public")
    row = await _store(request).unpublish(mapping_id, user_id, is_admin=is_admin)
    if row is None:
        raise HTTPException(status_code=404, detail="Knowledge base not found")
    return {"id": row["id"], "publication_status": row["publication_status"]}


@router.get("/api/llmwiki/sources/resolve")
async def resolve_source_knowledge_base(
    request: Request,
    knowledge_base_id: str = Query(..., min_length=1, max_length=128),
) -> dict[str, str]:
    """Resolve a skill's provider id to an authorized DeerFlow knowledge base."""
    user_id, is_admin = await _actor(request)
    store = _store(request)
    row = await store.get_authorized(knowledge_base_id, user_id, write=False, is_admin=is_admin)
    if row is None:
        # Retrieval MCPs return WeKnora ids; the original-file APIs take mapping
        # ids. Check authorization again after lookup, before exposing that id.
        candidate = await store.get_by_weknora_id(knowledge_base_id)
        if candidate is not None:
            row = await store.get_authorized(str(candidate["id"]), user_id, write=False, is_admin=is_admin)
    if row is None:
        raise HTTPException(status_code=404, detail="Knowledge base not found")
    return {"knowledge_base_id": str(row["id"])}


@router.get("/api/llmwiki/sources/context")
async def get_source_context(
    request: Request,
    knowledge_base_id: str = Query(..., min_length=1, max_length=128),
    chunk_id: str = Query(..., min_length=1, max_length=128),
) -> dict[str, Any]:
    row, _, _ = await _authorized_mapping(request, knowledge_base_id, write=False)
    client = _client_or_503()
    try:
        chunks = await client.get_chunk_context(
            chunk_id,
            expected_knowledge_base_id=str(row["weknora_id"]),
            radius=2,
        )
    except WeKnoraError as exc:
        _raise_upstream(exc)
    wiki_page = await _find_wiki_page_for_chunk(
        client,
        knowledge_base_id=str(row["weknora_id"]),
        chunk_id=chunk_id,
        chunks=chunks,
    )
    sanitize_deposit = _is_conversation_deposit_mapping(row)
    if sanitize_deposit and wiki_page is not None:
        wiki_page = _sanitize_conversation_deposit_value(wiki_page)
    focus_chunk = next((chunk for chunk in chunks if str(chunk.get("id") or "") == chunk_id), {})
    document_id = str(focus_chunk.get("knowledge_id") or focus_chunk.get("document_id") or focus_chunk.get("knowledgeId") or "")
    document_title = str(focus_chunk.get("knowledge_title") or focus_chunk.get("title") or focus_chunk.get("file_name") or focus_chunk.get("filename") or "")
    normalized_chunks = [
        {
            "id": str(chunk.get("id") or ""),
            "chunk_index": int(chunk.get("chunk_index") or 0),
            "content": str(_sanitize_conversation_deposit_value(chunk.get("content") or "") if sanitize_deposit else chunk.get("content") or ""),
            "is_focus": str(chunk.get("id") or "") == chunk_id,
        }
        for chunk in chunks
        if chunk.get("id")
    ]
    return {
        "success": True,
        "data": {
            "knowledge_base_id": row["id"],
            "knowledge_base_name": row.get("name") or "",
            "document_id": _sanitize_conversation_deposit_value(document_id) if sanitize_deposit else document_id,
            "document_title": _sanitize_conversation_deposit_value(document_title) if sanitize_deposit else document_title,
            "focus_chunk_id": chunk_id,
            "display_mode": "wiki" if wiki_page is not None else "chunks",
            "wiki_page": wiki_page,
            "chunks": normalized_chunks,
        },
    }