from __future__ import annotations

import json
from types import SimpleNamespace
from typing import Any

import pytest
from fastapi import HTTPException
from langchain_core.messages import HumanMessage, ToolMessage
from openpyxl import load_workbook

from app.gateway.routers.llmwiki import _build_wiki_excel
from deerflow.integrations.weknora.local_index.retrieval import WikiRetrievalService, is_fully_vectorized
from deerflow.integrations.weknora.local_index.search import WikiIndexError


class _Store:
    def __init__(self, states):
        self.states = states

    async def list_index_status(self, mapping_ids=None):
        wanted = set(mapping_ids or [])
        return [state for state in self.states if not wanted or state["knowledge_base_mapping_id"] in wanted]


class _WikiClient:
    def __init__(self):
        self.list_calls = []
        self.raw_search_called = False

    async def list_wiki_pages(self, knowledge_base_id, *, page=1, page_size=50, query=""):
        self.list_calls.append((knowledge_base_id, query))
        return {"pages": [{"slug": "guides/install", "title": "安装"}], "total": 1, "total_pages": 1}

    async def get_wiki_page(self, knowledge_base_id, slug):
        return {
            "id": f"{knowledge_base_id}:{slug}",
            "slug": slug,
            "title": "安装",
            "summary": "Wiki 摘要",
            "content": "# 安装\n这是处理后的 Wiki 正文。",
            "status": "published",
            "category_path": ["指南"],
        }

    async def search(self, query, knowledge_base_ids):
        self.raw_search_called = True
        raise AssertionError("raw knowledge search must never be called")


class _OverviewWikiClient(_WikiClient):
    async def list_wiki_pages(self, knowledge_base_id, *, page=1, page_size=50, query=""):
        self.list_calls.append((knowledge_base_id, query))
        if query:
            return {"pages": [], "total": 0, "total_pages": 0}
        return {
            "pages": [
                {"slug": "index", "title": "目录", "page_type": "directory"},
                {"slug": "guides/cooling", "title": "设备散热", "page_type": "article"},
            ],
            "total": 2,
            "total_pages": 1,
        }

    async def get_wiki_page(self, knowledge_base_id, slug):
        if slug == "index":
            return {
                "id": f"{knowledge_base_id}:{slug}",
                "slug": slug,
                "title": "目录",
                "page_type": "directory",
                "content": "目录正文",
                "status": "published",
            }
        return {
            "id": f"{knowledge_base_id}:{slug}",
            "slug": slug,
            "title": "设备散热",
            "page_type": "article",
            "summary": "页面摘要",
            "content": "设备过热时应先断电，再检查风道和散热风扇。",
            "status": "published",
        }


class _VectorSearch:
    def __init__(self):
        self.embedding = SimpleNamespace(fingerprint="fp-current")
        self.calls = []

    async def search(self, query, mappings, **kwargs):
        self.calls.append([mapping["id"] for mapping in mappings])
        mapping = mappings[0]
        return {
            "results": [
                {
                    "kind": "wiki_page",
                    "id": "local-page",
                    "wiki_page_id": "local-page",
                    "wiki_slug": "local/page",
                    "knowledge_base_id": mapping["id"],
                    "title": "本地页面",
                    "content": "本地 Wiki",
                    "matched_sections": [],
                    "score": 0.95,
                }
            ],
            "warnings": [],
        }


def _ready_state(mapping_id: str):
    return {
        "knowledge_base_mapping_id": mapping_id,
        "state": "idle",
        "last_success_at": "2026-08-24T00:00:00Z",
        "embedding_fingerprint": "fp-current",
        "local_page_count": 1,
        "ready_page_count": 1,
        "failed_page_count": 0,
        "vector_count": 3,
    }


def test_fully_vectorized_requires_completed_full_consistent_scan() -> None:
    state = _ready_state("kb")
    assert is_fully_vectorized(state, "fp-current")
    assert is_fully_vectorized({**state, "state": "partial"}, "fp-current")
    assert is_fully_vectorized({**state, "state": "failed"}, "fp-current")
    assert is_fully_vectorized({**state, "state": "failed", "failed_page_count": 1}, "fp-current")
    assert not is_fully_vectorized({**state, "state": "rebuild_required"}, "fp-current")
    assert not is_fully_vectorized({**state, "ready_page_count": 0}, "fp-current")
    assert not is_fully_vectorized(state, "different-fingerprint")


class _KeywordStore(_Store):
    async def search_pages_by_keywords(self, mapping_ids, tokens, *, limit, include_drafts):
        assert mapping_ids == ["ready"]
        assert "散热" in tokens
        return [
            {
                "id": "page-1",
                "knowledge_base_mapping_id": "ready",
                "slug": "guides/cooling",
                "title": "设备散热",
                "summary": "散热指南",
                "content_md": "设备过热时应先断电并检查风道。",
                "page_type": "article",
                "status": "published",
            }
        ]


class _EmptyVectorSearch(_VectorSearch):
    async def search(self, query, mappings, **kwargs):
        self.calls.append([mapping["id"] for mapping in mappings])
        return {
            "results": [],
            "warnings": [],
            "diagnostics": {"top_raw_vector_score": 0.21, "threshold_filtered": 7},
        }


class _KeywordCandidateStore(_Store):
    def __init__(self, pages):
        super().__init__([_ready_state("ready")])
        self.pages = pages
        self.tokens: list[str] = []

    async def search_pages_by_keywords(self, mapping_ids, tokens, *, limit, include_drafts):
        assert mapping_ids == ["ready"]
        self.tokens = list(tokens)
        return self.pages[:limit]


@pytest.mark.asyncio
async def test_keyword_channel_recovers_when_vectors_are_below_threshold() -> None:
    service = WikiRetrievalService(
        _KeywordStore([_ready_state("ready")]),
        vector_search=_EmptyVectorSearch(),
        client=None,
    )

    result = await service.search("散热故障怎么处理", [{"id": "ready", "name": "设备库"}])

    assert result["mode"] == "keyword"
    assert result["results"][0]["title"] == "设备散热"
    assert result["results"][0]["retrieval_mode"] == "keyword"
    assert result["diagnostics"]["fallback_reason"] == "VECTOR_BELOW_THRESHOLD"


@pytest.mark.asyncio
async def test_keyword_channel_ignores_english_stop_words_and_single_weak_term() -> None:
    store = _KeywordCandidateStore(
        [
            {
                "id": "page-1",
                "knowledge_base_mapping_id": "ready",
                "slug": "controller",
                "title": "Controller overview",
                "summary": "Annual inspection",
                "content_md": "This controller is on standby.",
                "page_type": "article",
                "status": "published",
            }
        ]
    )
    service = WikiRetrievalService(store, vector_search=_EmptyVectorSearch(), client=None)

    result = await service.search(
        "What is the annual rainfall on Neptune north pole?",
        [{"id": "ready", "name": "Acceptance"}],
    )

    assert "what" not in store.tokens
    assert "is" not in store.tokens
    assert "the" not in store.tokens
    assert "on" not in store.tokens
    assert result["results"] == []
    assert result["diagnostics"]["keyword_candidates"] == 0
    assert result["mode"] == "vector"


@pytest.mark.asyncio
async def test_keyword_channel_keeps_precise_code_as_single_strong_match() -> None:
    store = _KeywordCandidateStore(
        [
            {
                "id": "page-1",
                "knowledge_base_mapping_id": "ready",
                "slug": "xuanwu-pump",
                "title": "XW-900 maintenance",
                "summary": "Reset procedure",
                "content_md": "Fault ZXQ-417 requires a ninety second wait.",
                "page_type": "article",
                "status": "published",
            }
        ]
    )
    service = WikiRetrievalService(store, vector_search=_EmptyVectorSearch(), client=None)

    result = await service.search(
        "How do I reset fault ZXQ-417?",
        [{"id": "ready", "name": "Acceptance"}],
    )

    assert result["mode"] == "keyword"
    assert result["results"][0]["wiki_slug"] == "xuanwu-pump"
    assert result["results"][0]["retrieval_mode"] == "keyword"


@pytest.mark.asyncio
async def test_unvectorized_library_uses_only_wiki_page_search_api() -> None:
    client = _WikiClient()
    service = WikiRetrievalService(_Store([]), vector_search=_VectorSearch(), client=client)
    mapping = {"id": "local-kb", "weknora_id": "remote-kb", "name": "测试库"}

    result = await service.search("安装", [mapping])

    assert result["mode"] == "wiki_api"
    assert result["fallback_knowledge_base_ids"] == ["local-kb"]
    assert result["results"][0]["kind"] == "wiki_page"
    assert result["results"][0]["content"] == "# 安装\n这是处理后的 Wiki 正文。"
    assert "chunk_id" not in result["results"][0]
    assert client.list_calls == [("remote-kb", "安装")]
    assert not client.raw_search_called


@pytest.mark.asyncio
async def test_mixed_search_vectors_complete_library_and_falls_back_incomplete_library() -> None:
    client = _WikiClient()
    vector = _VectorSearch()
    service = WikiRetrievalService(_Store([_ready_state("ready")]), vector_search=vector, client=client)
    mappings = [
        {"id": "ready", "weknora_id": "remote-ready", "name": "已完成"},
        {"id": "pending", "weknora_id": "remote-pending", "name": "未完成"},
    ]

    result = await service.search("安装", mappings, top_k_pages=8)

    assert result["mode"] == "mixed"
    assert vector.calls == [["ready"]]
    assert client.list_calls == [("remote-pending", "安装")]
    assert {item["knowledge_base_id"] for item in result["results"]} == {"ready", "pending"}


@pytest.mark.asyncio
async def test_overview_query_lists_representative_remote_wiki_bodies_without_query_search() -> None:
    client = _OverviewWikiClient()
    service = WikiRetrievalService(_Store([]), vector_search=_VectorSearch(), client=client)
    mapping = {"id": "local-kb", "weknora_id": "remote-kb", "name": "设备库"}

    result = await service.search("这个知识空间有哪些内容？", [mapping])

    assert result["mode"] == "overview"
    assert client.list_calls == [("remote-kb", "")]
    assert [item["wiki_slug"] for item in result["results"]] == ["guides/cooling"]
    assert result["results"][0]["retrieval_mode"] == "overview"
    assert result["results"][0]["content"].startswith("设备过热时应先断电")


def test_excel_export_contains_current_directory_and_every_parent_directory() -> None:
    long_content = "正文" * 20_000
    pages = [
        {"id": "1", "slug": "root", "title": "根专题", "content": "根", "status": "published"},
        {"id": "2", "slug": "root/child", "title": "子专题", "parent_slug": "root", "content": "子", "status": "published"},
        {
            "id": "3",
            "slug": "root/child/article",
            "title": "文章",
            "parent_slug": "root/child",
            "content": long_content,
            "status": "published",
        },
    ]

    workbook = load_workbook(_build_wiki_excel("知识库", pages), read_only=True)
    sheet = workbook["Wiki 全量导出"]
    rows = list(sheet.iter_rows(values_only=True))
    headings = list(rows[0])
    article = dict(zip(headings, rows[3], strict=True))

    assert article["目录完整路径"] == "根专题 / 子专题"
    assert article["所在目录"] == "子专题"
    assert article["父级目录完整路径"] == "根专题"
    assert article["第1级目录"] == "根专题"
    assert article["第2级目录"] == "子专题"
    assert article["正文"] + article["正文（续1）"] == long_content


class _TraceMappingStore:
    def __init__(self, rows: list[dict[str, Any]]) -> None:
        self.rows = {str(row["id"]): dict(row) for row in rows}
        self.authorized_calls: list[tuple[str, str, bool, bool]] = []

    async def get_authorized(
        self,
        mapping_id: str,
        user_id: str,
        *,
        write: bool,
        is_admin: bool,
    ) -> dict[str, Any] | None:
        self.authorized_calls.append((mapping_id, user_id, write, is_admin))
        row = self.rows.get(mapping_id)
        return dict(row) if row is not None else None


class _TraceRawClient:
    def __init__(
        self,
        rows: list[dict[str, Any]] | None = None,
        *,
        error: Exception | None = None,
    ) -> None:
        self.rows = rows or []
        self.error = error
        self.calls: list[tuple[str, list[str]]] = []

    async def search(self, query: str, knowledge_base_ids: list[str]) -> list[dict[str, Any]]:
        self.calls.append((query, list(knowledge_base_ids)))
        if self.error is not None:
            raise self.error
        return [dict(row) for row in self.rows]


class _TraceGraphStore:
    def __init__(
        self,
        batches: list[list[dict[str, Any]]] | None = None,
        *,
        error: Exception | None = None,
    ) -> None:
        self.batches = batches or []
        self.error = error
        self.calls: list[dict[str, Any]] = []

    async def get_adjacent_pages(
        self,
        mapping_ids: list[str],
        seed_slugs: dict[str, list[str]],
        *,
        direction: str,
        limit: int,
        include_drafts: bool,
    ) -> list[dict[str, Any]]:
        self.calls.append(
            {
                "mapping_ids": list(mapping_ids),
                "seed_slugs": {key: list(value) for key, value in seed_slugs.items()},
                "direction": direction,
                "limit": limit,
                "include_drafts": include_drafts,
            }
        )
        if self.error is not None:
            raise self.error
        index = len(self.calls) - 1
        rows = self.batches[index] if index < len(self.batches) else []
        return [dict(row) for row in rows]


class _TraceRetrievalService:
    def __init__(
        self,
        payload: dict[str, Any],
        *,
        client: _TraceRawClient | None = None,
        error: Exception | None = None,
    ) -> None:
        self.payload = payload
        self.client = client
        self.error = error
        self.calls: list[dict[str, Any]] = []

    async def search(
        self,
        query: str,
        mappings: list[dict[str, Any]],
        *,
        top_k_pages: int,
    ) -> dict[str, Any]:
        self.calls.append(
            {
                "query": query,
                "mapping_ids": [str(row["id"]) for row in mappings],
                "top_k_pages": top_k_pages,
            }
        )
        if self.error is not None:
            raise self.error
        return dict(self.payload)


class _TraceAssistantKnowledgeStore:
    def __init__(self, rows: list[dict[str, Any]]) -> None:
        self.rows = rows
        self.calls: list[dict[str, Any]] = []

    async def search_pages(self, **kwargs: Any) -> list[dict[str, Any]]:
        self.calls.append(dict(kwargs))
        return [dict(row) for row in self.rows]


class _TraceAssistantGraphStore(_TraceAssistantKnowledgeStore):
    def __init__(
        self,
        rows: list[dict[str, Any]],
        *,
        bases: list[dict[str, Any]],
        graph_batches: list[list[dict[str, Any]]],
    ) -> None:
        super().__init__(rows)
        self.bases = {str(base["id"]): dict(base) for base in bases}
        self.graph = _TraceGraphStore(graph_batches)

    async def get_base(self, base_id: str) -> dict[str, Any] | None:
        base = self.bases.get(base_id)
        return dict(base) if base is not None else None

    async def get_adjacent_pages(
        self,
        base_ids: list[str],
        seed_pages: dict[str, list[str]],
        *,
        direction: str,
        limit: int,
        include_drafts: bool,
    ) -> list[dict[str, Any]]:
        return await self.graph.get_adjacent_pages(
            base_ids,
            seed_pages,
            direction=direction,
            limit=limit,
            include_drafts=include_drafts,
        )


def _trace_request(
    *,
    mapping_store: _TraceMappingStore,
    retrieval_service: _TraceRetrievalService,
    assistant_store: _TraceAssistantKnowledgeStore | None = None,
    graph_store: _TraceGraphStore | None = None,
) -> SimpleNamespace:
    local_index = SimpleNamespace(enabled=True, internal_search_enabled=True)
    state = SimpleNamespace(
        config=SimpleNamespace(
            llmwiki=SimpleNamespace(local_wiki_index=local_index),
        ),
        llmwiki_store=mapping_store,
        llmwiki_retrieval_service=retrieval_service,
        llmwiki_index_store=graph_store,
        assistant_knowledge_store=assistant_store,
        llmwiki_embedding=SimpleNamespace(fingerprint="fp-current"),
    )
    return SimpleNamespace(app=SimpleNamespace(state=state))


def _trace_body(
    *,
    llmwiki_ids: list[str] | None = None,
    assistant_ids: list[str] | None = None,
) -> SimpleNamespace:
    return SimpleNamespace(
        assistant_id=None,
        context={
            "llmwiki_knowledge_base_ids": llmwiki_ids or [],
            "assistant_knowledge_base_ids": assistant_ids or [],
        },
        metadata={},
    )


def _selected_spaces_json(trace: dict[str, Any]) -> str:
    """Allow space items to grow without fixing their exact representation."""

    return json.dumps(trace["selected_spaces"], ensure_ascii=False, sort_keys=True)


def _assert_non_negative_timings(timings: dict[str, Any]) -> None:
    assert timings
    for value in timings.values():
        if isinstance(value, (int, float)):
            assert value >= 0


def _ordinary_trace_results(count: int, *, knowledge_base_id: str = "kb-mixed") -> list[dict[str, Any]]:
    return [
        {
            "kind": "wiki_page",
            "id": f"ordinary-{index}",
            "wiki_page_id": f"ordinary-{index}",
            "wiki_slug": f"ordinary/{index}",
            "knowledge_base_id": knowledge_base_id,
            "knowledge_base_name": "普通 Wiki",
            "title": f"普通结果 {index}",
            "content": f"普通知识内容 {index}",
            "score": 0.01 * index,
            "retrieval_mode": "vector",
            "matched_sections": [
                {
                    "heading": f"普通章节 {index}",
                    "content": f"普通知识内容 {index}",
                    "score": 0.9 - index / 100,
                }
            ],
        }
        for index in range(1, count + 1)
    ]


def _graph_row(
    *,
    mapping_id: str,
    seed_slug: str,
    neighbor_slug: str,
    title: str,
    content: str,
    direction: str = "outgoing",
    edge_type: str = "wikilink",
    status: str = "published",
) -> dict[str, Any]:
    source_slug, target_slug = (
        (seed_slug, neighbor_slug)
        if direction == "outgoing"
        else (neighbor_slug, seed_slug)
    )
    return {
        "knowledge_base_mapping_id": mapping_id,
        "seed_slug": seed_slug,
        "neighbor_slug": neighbor_slug,
        "direction": direction,
        "edge_type": edge_type,
        "source_slug": source_slug,
        "target_slug": target_slug,
        "page": {
            "id": f"page:{mapping_id}:{neighbor_slug}",
            "knowledge_base_mapping_id": mapping_id,
            "slug": neighbor_slug,
            "title": title,
            "summary": f"{title}摘要",
            "content_md": content,
            "page_type": "article",
            "status": status,
            "source_refs": [{"chunk_id": f"chunk:{neighbor_slug}"}],
        },
    }


def _assistant_graph_row(
    *,
    base_id: str,
    seed_slug: str,
    neighbor_slug: str,
    title: str,
    content: str,
) -> dict[str, Any]:
    return {
        "base_id": base_id,
        "seed_slug": seed_slug,
        "neighbor_slug": neighbor_slug,
        "direction": "outgoing",
        "edge_type": "references",
        "page": {
            "id": f"assistant-page:{base_id}:{neighbor_slug}",
            "base_id": base_id,
            "slug": neighbor_slug,
            "title": title,
            "summary": f"{title}摘要",
            "content_markdown": content,
            "page_type": "article",
            "status": "active",
        },
    }


def _assistant_trace_results(count: int, *, base_id: str = "assistant-mixed") -> list[dict[str, Any]]:
    return [
        {
            "id": f"assistant-{index}",
            "base_id": base_id,
            "slug": f"assistant/{index}",
            "title": f"助手结果 {index}",
            "content": f"助手知识内容 {index}",
            "score": 0.8 - index / 100,
            "retrieval_mode": "vector",
            "matched_section": {
                "heading": f"助手章节 {index}",
                "content": f"助手知识内容 {index}",
                "score": 0.8 - index / 100,
            },
        }
        for index in range(1, count + 1)
    ]


def test_local_wiki_section_source_is_bounded_safe_and_vector_diagnostic() -> None:
    from app.gateway import llmwiki_rag

    source = llmwiki_rag._source_from_result(
        {
            "kind": "wiki_section",
            "id": "page-cooling:0",
            "wiki_page_id": "page-cooling",
            "weknora_knowledge_base_id": "remote-equipment-secret",
            "weknora_wiki_page_id": "remote-page-secret",
            "knowledge_base_id": "kb-equipment",
            "wiki_slug": "guides/cooling",
            "title": "设备散热",
            "heading": "故障处理",
            "content": "设备过热时应先断电。" + "检修步骤" * 500,
            "page_content": "不应持久化的完整页面" * 1000,
            "score": 0.016,
            "channel_score": 0.82,
            "rank_score_semantics": "rrf",
            "retrieval_mode": "vector",
            "source_refs": [
                {
                    "knowledge_id": "doc-one",
                    "knowledge_title": "设备手册.pdf",
                    "internal_metadata": "must-not-leak",
                }
            ],
            "chunk_refs": ["chunk-one"],
        },
        {"id": "kb-equipment", "name": "设备知识库"},
        1,
    )

    assert source["type"] == "llmwiki_wiki"
    assert len(source["content"]) <= 1400
    assert "knowledge_base_id=kb-equipment" in source["url"]
    assert "wiki_slug=guides%2Fcooling" in source["url"]
    serialized = json.dumps(source, ensure_ascii=False)
    assert "remote-equipment-secret" not in serialized
    assert "remote-page-secret" not in serialized
    assert "不应持久化的完整页面" not in serialized
    assert "internal_metadata" not in serialized

    trace_result = llmwiki_rag._trace_result(source)
    assert trace_result["knowledge_base_id"] == "kb-equipment"
    assert trace_result["matched_heading"] == "故障处理"
    assert trace_result["vector_score"] == pytest.approx(0.82)
    assert trace_result["rank_score_semantics"] == "rrf"


def test_prompt_keeps_partial_retrieval_diagnostics_internal_when_sources_exist() -> None:
    from app.gateway import llmwiki_rag

    prompt = llmwiki_rag._build_prompt(
        "海湖庄园",
        [
            {
                "index": 1,
                "id": "wiki-page",
                "title": "海湖庄园",
                "type": "llmwiki_wiki",
                "content": "格式化后的 Wiki 内容",
                "source": "测试知识空间",
                "retrieval_mode": "vector",
                "raw": {
                    "knowledge_base_id": "kb-test",
                    "wiki_slug": "wiki/mar-a-lago",
                },
            }
        ],
        {
            "partial": True,
            "warnings": ["VECTOR_UNAVAILABLE"],
            "fallback_reason": "VECTOR_UNAVAILABLE",
        },
    )

    assert "VECTOR_UNAVAILABLE" not in prompt
    assert "只有部分检索通道成功" not in prompt
    assert "明确说明检索结果可能不完整" not in prompt
    assert "不要在回答中展示检索通道状态" in prompt


@pytest.mark.asyncio
async def test_rag_context_exposes_successful_hybrid_retrieval_trace(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-equipment",
                "weknora_id": "remote-equipment",
                "name": "设备知识库",
                "owner_user_id": "user-a",
            }
        ]
    )
    retrieval = _TraceRetrievalService(
        {
            "query": "设备散热故障怎么处理？",
            "mode": "hybrid",
            "results": [
                {
                    "kind": "wiki_page",
                    "id": "page-cooling",
                    "wiki_page_id": "page-cooling",
                    "wiki_slug": "guides/cooling",
                    "knowledge_base_id": "kb-equipment",
                    "knowledge_base_name": "设备知识库",
                    "title": "设备散热",
                    "content": "设备过热时应先断电，再检查风道和散热风扇。",
                    "snippet": "设备过热时应先断电",
                    "matched_sections": [
                        {
                            "section_index": 0,
                            "heading": "故障处理",
                            "content": "设备过热时应先断电，再检查风道和散热风扇。",
                            "score": 0.82,
                        }
                    ],
                    "score": 0.032,
                    "channel_score": 0.82,
                    "rank_score_semantics": "rrf",
                    "retrieval_mode": "hybrid",
                    "url": "/weknora-source-preview?knowledge_base_id=kb-equipment",
                }
            ],
            "knowledge_base_count": 1,
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {
                "top_raw_vector_score": 0.82,
                "min_similarity": 0.35,
                "threshold_filtered": 0,
                "vector_candidates": 1,
                "keyword_candidates": 1,
                "wiki_api_candidates": 0,
                "fallback_reason": None,
            },
        }
    )
    request = _trace_request(mapping_store=mapping_store, retrieval_service=retrieval)

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=True),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(llmwiki_ids=["kb-equipment"]),
        graph_input={
            "messages": [
                HumanMessage(
                    id="human-one",
                    content="设备散热故障怎么处理？",
                )
            ]
        },
    )

    assert retrieval.calls == [
        {
            "query": "设备散热故障怎么处理？",
            "mapping_ids": ["kb-equipment"],
            "top_k_pages": 8,
        }
    ]
    assert mapping_store.authorized_calls == [("kb-equipment", "user-a", False, False)]
    assert rag.source_count == 1
    assert rag.sources[0]["raw"]["wiki_slug"] == "guides/cooling"

    trace = rag.trace
    assert trace["retrieval_trace_id"]
    assert trace["trigger"] == "pre_retrieval"
    assert trace["query"] == "设备散热故障怎么处理？"
    assert "kb-equipment" in _selected_spaces_json(trace)
    assert trace["authorized_space_count"] == 1
    assert trace["mode"] == "hybrid"
    assert trace["source_count"] == 1
    assert trace["partial"] is False
    assert trace["warnings"] == []
    assert trace["fallback_reason"] is None
    assert trace["fallback_knowledge_base_ids"] == []
    assert trace["diagnostics"]["top_raw_vector_score"] == pytest.approx(0.82)
    assert trace["diagnostics"]["vector_candidates"] == 1
    assert trace["results"][0]["knowledge_base_id"] == "kb-equipment"
    assert trace["results"][0]["wiki_slug"] == "guides/cooling"
    # Hybrid uses an RRF/channel score; the section score is not guaranteed
    # to be cosine similarity and must not be exposed as vector_score.
    assert trace["results"][0]["rank_score_semantics"] == "rrf"
    assert trace["results"][0]["vector_score"] is None
    assert trace["zero_result_reason"] is None
    assert isinstance(trace["capabilities"], (dict, list))
    _assert_non_negative_timings(trace["timings_ms"])


@pytest.mark.asyncio
async def test_rag_context_fuses_authorized_raw_chunks_with_wiki_results(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-equipment",
                "weknora_id": "remote-equipment",
                "name": "设备知识库",
                "owner_user_id": "user-a",
            }
        ]
    )
    long_content = "TIGER-928 关闭旁通阀后等待 75 秒，再长按绿色复位键 6 秒。" + "检修说明" * 400
    raw_client = _TraceRawClient(
        [
            {
                "chunk_id": "chunk-tiger",
                "knowledge_id": "doc-tiger",
                "knowledge_base_id": "remote-equipment",
                "title": "星河泵站编码模型验收",
                "content": long_content,
                "score": 0.91,
                "chunk_index": 0,
                "metadata": {"must_not_leak": "internal"},
            },
            {
                "chunk_id": "chunk-other",
                "knowledge_id": "doc-other",
                "knowledge_base_id": "remote-not-authorized",
                "title": "越权片段",
                "content": "SECRET-OTHER-KB",
                "score": 0.99,
            },
        ]
    )
    ordinary_results = _ordinary_trace_results(8, knowledge_base_id="kb-equipment")
    ordinary_results[0]["weknora_knowledge_base_id"] = "remote-equipment"
    ordinary_results[0]["weknora_wiki_page_id"] = "remote-page-secret"
    retrieval = _TraceRetrievalService(
        {
            "mode": "vector",
            "results": ordinary_results,
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {"vector_candidates": 8},
        },
        client=raw_client,
    )
    request = _trace_request(mapping_store=mapping_store, retrieval_service=retrieval)

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=True),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(llmwiki_ids=["kb-equipment"]),
        graph_input={"messages": [HumanMessage(content="TIGER-928 告警后如何复位？")]},
    )

    assert raw_client.calls == [("TIGER-928 告警后如何复位？", ["remote-equipment"])]
    assert rag.source_count == 8
    assert rag.sources[0]["id"] == "chunk-tiger"
    assert len(rag.sources[0]["content"]) <= 1400
    assert "TIGER-928" in rag.prompt
    assert "SECRET-OTHER-KB" not in rag.prompt
    assert "SECRET-OTHER-KB" not in json.dumps(rag.trace, ensure_ascii=False)
    assert "metadata" not in rag.sources[0]["raw"]
    assert "remote-equipment" not in json.dumps(rag.sources, ensure_ascii=False)
    assert "remote-equipment" not in json.dumps(rag.trace, ensure_ascii=False)
    assert "remote-page-secret" not in json.dumps(rag.sources, ensure_ascii=False)
    assert "remote-page-secret" not in json.dumps(rag.trace, ensure_ascii=False)

    trace = rag.trace
    assert trace["mode"] == "wiki_raw_hybrid"
    assert trace["capabilities"]["raw_document_chunk_retrieval"] is True
    assert trace["diagnostics"]["raw_chunk_candidates"] == 2
    assert trace["diagnostics"]["raw_chunk_considered"] == 1
    assert trace["diagnostics"]["raw_chunk_rejected"] == 1
    assert trace["diagnostics"]["raw_chunk_top_provider_score"] == pytest.approx(0.91)
    assert trace["diagnostics"]["final_raw_chunk_source_count"] >= 1
    assert trace["diagnostics"]["final_wiki_page_source_count"] >= 1
    assert {source["type"] for source in rag.sources} >= {"llmwiki", "llmwiki_wiki"}
    assert trace["results"][0]["retrieval_mode"] == "raw_chunk"
    assert trace["results"][0]["provider_score"] == pytest.approx(0.91)
    assert trace["results"][0]["rank_score_semantics"] == "rrf"
    assert trace["results"][0]["vector_score"] is None


@pytest.mark.asyncio
async def test_raw_chunk_hit_survives_wiki_channel_failure(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-equipment",
                "weknora_id": "remote-equipment",
                "name": "设备知识库",
                "owner_user_id": "user-a",
            }
        ]
    )
    raw_client = _TraceRawClient(
        [
            {
                "chunk_id": "chunk-tiger",
                "knowledge_id": "doc-tiger",
                "knowledge_base_id": "remote-equipment",
                "title": "复位规程",
                "content": "TIGER-928：等待 75 秒并长按绿色复位键 6 秒。",
                "score": 0.73,
            }
        ]
    )
    retrieval = _TraceRetrievalService(
        {},
        client=raw_client,
        error=WikiIndexError("WIKI_API_UNAVAILABLE", "wiki unavailable", status_code=502),
    )
    request = _trace_request(mapping_store=mapping_store, retrieval_service=retrieval)

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=True),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(llmwiki_ids=["kb-equipment"]),
        graph_input={"messages": [HumanMessage(content="TIGER-928 如何复位？")]},
    )

    assert rag.source_count == 1
    assert rag.sources[0]["id"] == "chunk-tiger"
    assert rag.trace["mode"] == "raw_chunk"
    assert rag.trace["partial"] is True
    assert "WIKI_API_UNAVAILABLE" in rag.trace["warnings"]
    assert rag.trace["zero_result_reason"] is None
    assert "TIGER-928" in rag.prompt
    assert "不能判断库内是否存在相关资料" not in rag.prompt


@pytest.mark.asyncio
async def test_raw_chunk_failure_preserves_wiki_results(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-equipment",
                "weknora_id": "remote-equipment",
                "name": "设备知识库",
                "owner_user_id": "user-a",
            }
        ]
    )
    retrieval = _TraceRetrievalService(
        {
            "mode": "keyword",
            "results": _ordinary_trace_results(1, knowledge_base_id="kb-equipment"),
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {"keyword_candidates": 1},
        },
        client=_TraceRawClient(error=RuntimeError("raw unavailable")),
    )
    request = _trace_request(mapping_store=mapping_store, retrieval_service=retrieval)

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=True),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(llmwiki_ids=["kb-equipment"]),
        graph_input={"messages": [HumanMessage(content="普通 Wiki 命中")]} ,
    )

    assert rag.source_count == 1
    assert rag.sources[0]["id"] == "ordinary-1"
    assert rag.trace["mode"] == "keyword"
    assert rag.trace["partial"] is True
    assert "RAW_CHUNK_RETRIEVAL_FAILED" in rag.trace["warnings"]


@pytest.mark.asyncio
async def test_rag_expands_authorized_wiki_graph_with_bounded_per_library_quota(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-equipment",
                "weknora_id": "remote-equipment",
                "name": "设备知识库",
                "owner_user_id": "user-a",
            }
        ]
    )
    graph_rows = [
        _graph_row(
            mapping_id="kb-equipment",
            seed_slug="ordinary/1",
            neighbor_slug=f"related/{index}",
            title=f"关联页面 {index}",
            content=f"关联页面 {index} 的真实正文和操作依据。",
        )
        for index in range(1, 6)
    ]
    graph_rows.extend(
        [
            _graph_row(
                mapping_id="kb-not-authorized",
                seed_slug="ordinary/1",
                neighbor_slug="secret/page",
                title="越权页面",
                content="SECRET-GRAPH-CONTENT",
            ),
            _graph_row(
                mapping_id="kb-equipment",
                seed_slug="ordinary/1",
                neighbor_slug="empty/page",
                title="空正文",
                content="",
            ),
        ]
    )
    graph_store = _TraceGraphStore([graph_rows])
    retrieval = _TraceRetrievalService(
        {
            "mode": "hybrid",
            "results": _ordinary_trace_results(1, knowledge_base_id="kb-equipment"),
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {
                "vector_candidates": 1,
                "keyword_candidates": 1,
                "top_raw_vector_score": 0.82,
                "min_similarity": 0.35,
            },
        }
    )
    request = _trace_request(
        mapping_store=mapping_store,
        retrieval_service=retrieval,
        graph_store=graph_store,
    )

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=False),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(llmwiki_ids=["kb-equipment"]),
        graph_input={"messages": [HumanMessage(content="泵站故障还关联哪些操作？")]},
    )

    assert graph_store.calls == [
        {
            "mapping_ids": ["kb-equipment"],
            "seed_slugs": {"kb-equipment": ["ordinary/1"]},
            "direction": "both",
            "limit": 24,
            "include_drafts": False,
        }
    ]
    assert rag.trace["mode"] == "wiki_graph_hybrid"
    assert rag.trace["capabilities"]["wiki_graph_expansion"] is True
    assert rag.trace["diagnostics"]["graph_seed_count"] == 1
    assert rag.trace["diagnostics"]["graph_candidates"] == 7
    assert rag.trace["diagnostics"]["graph_unique_candidates"] == 5
    assert rag.trace["diagnostics"]["graph_accepted"] == 3
    assert rag.trace["diagnostics"]["graph_sources_used"] == 2
    assert rag.trace["diagnostics"]["graph_rejected"] == 2
    assert rag.trace["diagnostics"]["graph_hops_requested"] == 1
    assert rag.trace["diagnostics"]["graph_hops_executed"] == 1
    assert rag.trace["timings_ms"]["graph_expansion"] >= 0
    assert "SECRET-GRAPH-CONTENT" not in rag.prompt
    assert "SECRET-GRAPH-CONTENT" not in json.dumps(rag.trace, ensure_ascii=False)
    graph_results = [
        result for result in rag.trace["results"]
        if result["retrieval_mode"] == "graph_expansion"
    ]
    assert len(graph_results) == 2
    assert graph_results[0]["graph_hop"] == 1
    assert graph_results[0]["graph_score"] == pytest.approx(0.58)
    assert graph_results[0]["graph_path"] == [
        {"slug": "ordinary/1", "title": "普通结果 1"},
        {
            "slug": "related/1",
            "title": "关联页面 1",
            "direction": "outgoing",
            "edge_type": "wikilink",
        },
    ]
    assert graph_results[0]["source_refs"][0]["chunk_id"] == "chunk:related/1"


@pytest.mark.asyncio
async def test_rag_graph_expansion_uses_two_hops_only_for_explicit_or_low_confidence_queries(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-equipment",
                "weknora_id": "remote-equipment",
                "name": "设备知识库",
                "owner_user_id": "user-a",
            }
        ]
    )
    graph_store = _TraceGraphStore(
        [
            [
                _graph_row(
                    mapping_id="kb-equipment",
                    seed_slug="ordinary/1",
                    neighbor_slug="nodes/middle",
                    title="中继控制站",
                    content="中继控制站连接上游控制中心。",
                )
            ],
            [
                _graph_row(
                    mapping_id="kb-equipment",
                    seed_slug="nodes/middle",
                    neighbor_slug="nodes/upstream",
                    title="上游控制中心",
                    content="上游控制中心交接口令为 CINDER-884。",
                )
            ],
        ]
    )
    retrieval = _TraceRetrievalService(
        {
            "mode": "vector",
            "results": _ordinary_trace_results(1, knowledge_base_id="kb-equipment"),
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {
                "vector_candidates": 1,
                "top_raw_vector_score": 0.74,
                "min_similarity": 0.35,
            },
        }
    )
    request = _trace_request(
        mapping_store=mapping_store,
        retrieval_service=retrieval,
        graph_store=graph_store,
    )

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=False),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(llmwiki_ids=["kb-equipment"]),
        graph_input={"messages": [HumanMessage(content="沿完整链路做多跳追溯，找上游交接口令")]},
    )

    assert len(graph_store.calls) == 2
    assert graph_store.calls[0]["seed_slugs"] == {"kb-equipment": ["ordinary/1"]}
    assert graph_store.calls[0]["limit"] == 12
    assert graph_store.calls[1]["seed_slugs"] == {"kb-equipment": ["nodes/middle"]}
    assert graph_store.calls[1]["limit"] == 12
    assert rag.trace["diagnostics"]["graph_hops_requested"] == 2
    assert rag.trace["diagnostics"]["graph_hops_executed"] == 2
    assert rag.trace["diagnostics"]["graph_depth_reason"] == "explicit_multi_hop"
    upstream = next(
        result for result in rag.trace["results"]
        if result["wiki_slug"] == "nodes/upstream"
    )
    assert upstream["graph_hop"] == 2
    assert upstream["graph_score"] == pytest.approx(0.58 * 0.58)
    assert [node["slug"] for node in upstream["graph_path"]] == [
        "ordinary/1",
        "nodes/middle",
        "nodes/upstream",
    ]
    assert "CINDER-884" in rag.prompt
    assert llmwiki_rag._graph_hop_budget(
        "普通提问",
        {
            "top_raw_vector_score": 0.2,
            "min_similarity": 0.35,
            "vector_candidates": 0,
            "threshold_filtered": 4,
        },
    ) == (2, "low_confidence")


@pytest.mark.asyncio
async def test_rag_graph_expansion_traverses_direct_non_seed_without_duplicate_output(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-equipment",
                "weknora_id": "remote-equipment",
                "name": "设备知识库",
                "owner_user_id": "user-a",
            }
        ]
    )
    direct_results = _ordinary_trace_results(3, knowledge_base_id="kb-equipment")
    direct_results[2].update(
        {
            "id": "direct-middle",
            "wiki_page_id": "direct-middle",
            "wiki_slug": "nodes/middle",
            "title": "中继控制站",
            "content": "中继控制站也被向量通道直接召回。",
        }
    )
    graph_store = _TraceGraphStore(
        [
            [
                _graph_row(
                    mapping_id="kb-equipment",
                    seed_slug="ordinary/1",
                    neighbor_slug="nodes/middle",
                    title="中继控制站",
                    content="中继控制站连接上游控制中心。",
                )
            ],
            [
                _graph_row(
                    mapping_id="kb-equipment",
                    seed_slug="nodes/middle",
                    neighbor_slug="nodes/upstream",
                    title="上游控制中心",
                    content="上游控制中心交接口令为 NEBULA-642。",
                )
            ],
        ]
    )
    retrieval = _TraceRetrievalService(
        {
            "mode": "vector",
            "results": direct_results,
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {
                "vector_candidates": 3,
                "top_raw_vector_score": 0.74,
                "min_similarity": 0.35,
            },
        }
    )
    request = _trace_request(
        mapping_store=mapping_store,
        retrieval_service=retrieval,
        graph_store=graph_store,
    )

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=False),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(llmwiki_ids=["kb-equipment"]),
        graph_input={"messages": [HumanMessage(content="沿完整链路做二跳追溯")]},
    )

    assert len(graph_store.calls) == 2
    assert graph_store.calls[0]["seed_slugs"] == {
        "kb-equipment": ["ordinary/1", "ordinary/2"]
    }
    assert graph_store.calls[1]["seed_slugs"] == {"kb-equipment": ["nodes/middle"]}
    assert rag.trace["diagnostics"]["graph_candidates"] == 2
    assert rag.trace["diagnostics"]["graph_unique_candidates"] == 1
    assert rag.trace["diagnostics"]["graph_accepted"] == 1
    assert rag.trace["diagnostics"]["graph_hops_executed"] == 2
    middle_results = [
        result for result in rag.trace["results"]
        if result["wiki_slug"] == "nodes/middle"
    ]
    assert len(middle_results) == 1
    assert middle_results[0]["retrieval_mode"] == "vector"
    upstream = next(
        result for result in rag.trace["results"]
        if result["wiki_slug"] == "nodes/upstream"
    )
    assert upstream["retrieval_mode"] == "graph_expansion"
    assert upstream["graph_hop"] == 2
    assert [node["slug"] for node in upstream["graph_path"]] == [
        "ordinary/1",
        "nodes/middle",
        "nodes/upstream",
    ]
    assert "NEBULA-642" in rag.prompt


@pytest.mark.asyncio
async def test_rag_expands_assistant_graph_two_hops_and_reports_assistant_trace(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    base_id = "assistant-graph"
    direct_results = _assistant_trace_results(3, base_id=base_id)
    direct_results[2].update(
        {
            "id": "assistant-middle",
            "slug": "nodes/middle",
            "title": "助手中继页",
            "content": "助手中继页也被向量通道直接召回。",
            "matched_section": {
                "heading": "助手中继页",
                "content": "助手中继页也被向量通道直接召回。",
                "score": 0.61,
            },
        }
    )
    assistant_store = _TraceAssistantGraphStore(
        direct_results,
        bases=[{"id": base_id, "name": "关系助手库", "status": "active"}],
        graph_batches=[
            [
                _assistant_graph_row(
                    base_id=base_id,
                    seed_slug="assistant/1",
                    neighbor_slug="nodes/middle",
                    title="助手中继页",
                    content="助手中继页连接二跳答案页。",
                )
            ],
            [
                _assistant_graph_row(
                    base_id=base_id,
                    seed_slug="nodes/middle",
                    neighbor_slug="nodes/answer",
                    title="助手二跳答案页",
                    content="助手图二跳答案为 AURORA-915。",
                )
            ],
        ],
    )
    request = _trace_request(
        mapping_store=_TraceMappingStore([]),
        retrieval_service=_TraceRetrievalService({"results": []}),
        assistant_store=assistant_store,
    )

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=False),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(assistant_ids=[base_id]),
        graph_input={"messages": [HumanMessage(content="沿关系做二跳追溯")]},
    )

    assert rag.trace["capabilities"]["wiki_graph_expansion"] is True
    assert rag.trace["capabilities"]["raw_document_chunk_retrieval"] is True
    assert rag.trace["mode"] == "assistant_graph_hybrid"
    assert len(assistant_store.graph.calls) == 2
    assert assistant_store.graph.calls[0]["seed_slugs"] == {
        base_id: ["assistant/1", "assistant/2"]
    }
    assert assistant_store.graph.calls[1]["seed_slugs"] == {
        base_id: ["nodes/middle"]
    }
    assert rag.trace["diagnostics"]["graph_channels"] == ["assistant"]
    assert rag.trace["diagnostics"]["graph_candidates"] == 2
    assert rag.trace["diagnostics"]["graph_unique_candidates"] == 1
    assert rag.trace["diagnostics"]["graph_accepted"] == 1
    assert rag.trace["diagnostics"]["graph_hops_requested"] == 2
    assert rag.trace["diagnostics"]["graph_hops_executed"] == 2
    middle_results = [
        result for result in rag.trace["results"]
        if result["wiki_slug"] == "nodes/middle"
    ]
    assert len(middle_results) == 1
    assert middle_results[0]["kind"] == "assistant"
    assert middle_results[0]["retrieval_mode"] == "vector"
    answer = next(
        result for result in rag.trace["results"]
        if result["wiki_slug"] == "nodes/answer"
    )
    assert answer["kind"] == "assistant"
    assert answer["retrieval_mode"] == "graph_expansion"
    assert answer["graph_hop"] == 2
    assert answer["graph_score"] == pytest.approx(0.58 * 0.58)
    assert [node["slug"] for node in answer["graph_path"]] == [
        "assistant/1",
        "nodes/middle",
        "nodes/answer",
    ]
    assert "AURORA-915" in rag.prompt
    assert "知识空间标识=assistant-graph" in rag.prompt
    assert "Wiki 标识=nodes/answer" in rag.prompt
    assert "图扩展=第2跳" in rag.prompt
    assert "助手二跳答案页 [知识空间=assistant-graph; Wiki=nodes/answer]" in rag.prompt
    assert "--outgoing/references-->" in rag.prompt
    assert "应视为该末节点已检索" in rag.prompt


def test_prompt_graph_path_preserves_incoming_direction_and_space_identity() -> None:
    from app.gateway import llmwiki_rag

    rendered = llmwiki_rag._prompt_graph_path(
        {
            "graph_path": [
                {"slug": "seed", "title": "种子页"},
                {
                    "slug": "upstream",
                    "title": "上游页",
                    "direction": "incoming",
                    "edge_type": "wikilink",
                },
            ]
        },
        "kb-a",
    )

    assert rendered == (
        "种子页 [知识空间=kb-a; Wiki=seed] "
        "<--incoming/wikilink-- 上游页 [知识空间=kb-a; Wiki=upstream]"
    )


@pytest.mark.asyncio
async def test_rag_reports_wiki_graph_raw_hybrid_mode_when_all_channels_contribute(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-equipment",
                "weknora_id": "remote-equipment",
                "name": "设备知识库",
                "owner_user_id": "user-a",
            }
        ]
    )
    raw_client = _TraceRawClient(
        [
            {
                "chunk_id": "raw-one",
                "knowledge_base_id": "remote-equipment",
                "title": "原始手册",
                "content": "原始手册正文。",
                "score": 0.8,
            }
        ]
    )
    graph_store = _TraceGraphStore(
        [
            [
                _graph_row(
                    mapping_id="kb-equipment",
                    seed_slug="ordinary/1",
                    neighbor_slug="related/one",
                    title="关联规程",
                    content="关联规程正文。",
                )
            ]
        ]
    )
    retrieval = _TraceRetrievalService(
        {
            "mode": "vector",
            "results": _ordinary_trace_results(1, knowledge_base_id="kb-equipment"),
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {"vector_candidates": 1},
        },
        client=raw_client,
    )
    request = _trace_request(
        mapping_store=mapping_store,
        retrieval_service=retrieval,
        graph_store=graph_store,
    )

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=True),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(llmwiki_ids=["kb-equipment"]),
        graph_input={"messages": [HumanMessage(content="联合检索设备操作")]} ,
    )

    assert rag.trace["mode"] == "wiki_graph_raw_hybrid"
    assert {
        result["retrieval_mode"] for result in rag.trace["results"]
    } == {"raw_chunk", "vector", "graph_expansion"}
    assert rag.trace["diagnostics"]["graph_sources_used"] == 1


@pytest.mark.asyncio
async def test_graph_expansion_failure_is_partial_and_keeps_direct_wiki_results(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-equipment",
                "weknora_id": "remote-equipment",
                "name": "设备知识库",
                "owner_user_id": "user-a",
            }
        ]
    )
    graph_store = _TraceGraphStore(error=RuntimeError("graph unavailable"))
    retrieval = _TraceRetrievalService(
        {
            "mode": "vector",
            "results": _ordinary_trace_results(1, knowledge_base_id="kb-equipment"),
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {"vector_candidates": 1},
        }
    )
    request = _trace_request(
        mapping_store=mapping_store,
        retrieval_service=retrieval,
        graph_store=graph_store,
    )

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=False),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(llmwiki_ids=["kb-equipment"]),
        graph_input={"messages": [HumanMessage(content="普通 Wiki 命中，图通道失败")]},
    )

    assert rag.source_count == 1
    assert rag.sources[0]["id"] == "ordinary-1"
    assert rag.trace["mode"] == "vector"
    assert rag.trace["partial"] is True
    assert rag.trace["capabilities"]["wiki_graph_expansion"] is True
    assert "WIKI_GRAPH_EXPANSION_FAILED" in rag.trace["warnings"]
    assert rag.trace["zero_result_reason"] is None


@pytest.mark.asyncio
async def test_zero_hit_trace_preserves_vector_threshold_diagnostics(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-equipment",
                "weknora_id": "remote-equipment",
                "name": "设备知识库",
                "owner_user_id": "user-a",
            }
        ]
    )
    retrieval = _TraceRetrievalService(
        {
            "query": "冷却泵异常代码 ZX-9",
            "mode": "keyword",
            "results": [],
            "knowledge_base_count": 1,
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {
                "top_raw_vector_score": 0.21,
                "min_similarity": 0.35,
                "threshold_filtered": 7,
                "vector_candidates": 0,
                "keyword_candidates": 0,
                "wiki_api_candidates": 0,
                "fallback_reason": "VECTOR_BELOW_THRESHOLD",
            },
        }
    )
    request = _trace_request(mapping_store=mapping_store, retrieval_service=retrieval)

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=True),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(llmwiki_ids=["kb-equipment"]),
        graph_input={"messages": [HumanMessage(content="冷却泵异常代码 ZX-9")]},
    )

    assert rag.source_count == 0
    assert rag.sources == []
    assert "暂未找到相关资料" in rag.prompt
    trace = rag.trace
    assert trace["mode"] == "keyword"
    assert trace["source_count"] == 0
    assert trace["results"] == []
    assert trace["fallback_reason"] == "VECTOR_BELOW_THRESHOLD"
    assert trace["zero_result_reason"] == "VECTOR_BELOW_THRESHOLD"
    assert trace["diagnostics"]["top_raw_vector_score"] == pytest.approx(0.21)
    assert trace["diagnostics"]["min_similarity"] == pytest.approx(0.35)
    assert trace["diagnostics"]["threshold_filtered"] == 7
    assert trace["diagnostics"]["vector_candidates"] == 0
    assert trace["diagnostics"]["keyword_candidates"] == 0


@pytest.mark.asyncio
async def test_trace_records_ordinary_and_assistant_space_selection(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-ordinary",
                "weknora_id": "remote-ordinary",
                "name": "普通 Wiki",
                "owner_user_id": "user-a",
            }
        ]
    )
    retrieval = _TraceRetrievalService(
        {
            "query": "联合检索",
            "mode": "vector",
            "results": [],
            "knowledge_base_count": 1,
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {
                "top_raw_vector_score": 0.1,
                "min_similarity": 0.35,
                "threshold_filtered": 1,
                "fallback_reason": "VECTOR_BELOW_THRESHOLD",
            },
        }
    )
    assistant_store = _TraceAssistantKnowledgeStore(
        [
            {
                "id": "assistant-page",
                "base_id": "assistant-base",
                "slug": "assistant/page",
                "title": "助手知识页",
                "content": "助手库命中的内容",
                "score": 0.76,
                "retrieval_mode": "vector",
                "matched_section": {
                    "heading": "助手知识章节",
                    "content": "助手库命中的内容",
                    "score": 0.76,
                },
            }
        ]
    )
    request = _trace_request(
        mapping_store=mapping_store,
        retrieval_service=retrieval,
        assistant_store=assistant_store,
    )

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=True),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(
            llmwiki_ids=["kb-ordinary"],
            assistant_ids=["assistant-base"],
        ),
        graph_input={"messages": [HumanMessage(content="联合检索")]},
    )

    selected = _selected_spaces_json(rag.trace)
    assert "kb-ordinary" in selected
    assert "assistant-base" in selected
    assert rag.trace["authorized_space_count"] == 2
    assert rag.trace["source_count"] == 1
    assert rag.trace["results"][0]["title"] == "助手知识页"
    assert rag.trace["results"][0]["matched_heading"] == "助手知识章节"
    assert rag.trace["results"][0]["vector_score"] == pytest.approx(0.76)
    assert assistant_store.calls[0]["base_ids"] == ["assistant-base"]


@pytest.mark.asyncio
async def test_attach_and_reference_response_round_trip_retrieval_trace(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag
    from app.gateway.routers.threads import (
        ReferenceBatchesResponse,
        _extract_knowledge_retrieval_traces,
    )

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-one",
                "weknora_id": "remote-one",
                "name": "知识库一",
                "owner_user_id": "user-a",
            }
        ]
    )
    retrieval = _TraceRetrievalService(
        {
            "query": "测试追踪",
            "mode": "wiki_api",
            "results": [],
            "knowledge_base_count": 1,
            "partial": True,
            "warnings": ["kb-one:WIKI_API_SEARCH_FAILED"],
            "fallback_knowledge_base_ids": ["kb-one"],
            "diagnostics": {
                "vector_candidates": 0,
                "keyword_candidates": 0,
                "wiki_api_candidates": 0,
                "fallback_reason": "RETRIEVAL_PROVIDER_UNAVAILABLE",
            },
        }
    )
    request = _trace_request(mapping_store=mapping_store, retrieval_service=retrieval)

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=True),
    )

    original = HumanMessage(id="human-trace", content="测试追踪")
    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(llmwiki_ids=["kb-one"]),
        graph_input={"messages": [original]},
    )
    attached = llmwiki_rag.attach_llmwiki_rag_to_graph_input(
        {"messages": [original]},
        rag,
    )
    attached_message = attached["messages"][0]

    assert attached_message.additional_kwargs["knowledge_retrieval_trace"] == rag.trace
    traces = _extract_knowledge_retrieval_traces(attached["messages"])
    assert traces == [{**rag.trace, "message_id": "human-trace"}]

    # Persisted checkpoints use dictionaries. The helper must support that
    # shape, ignore malformed values and de-duplicate by trace id.
    serialized_messages = [
        {
            "type": "human",
            "content": "测试追踪",
            "additional_kwargs": {"knowledge_retrieval_trace": rag.trace},
        },
        {
            "type": "human",
            "content": "重复追踪",
            "additional_kwargs": {"knowledge_retrieval_trace": dict(rag.trace)},
        },
        {
            "type": "human",
            "content": "坏数据",
            "additional_kwargs": {"knowledge_retrieval_trace": "not-an-object"},
        },
    ]
    assert _extract_knowledge_retrieval_traces(serialized_messages) == [rag.trace]

    response = ReferenceBatchesResponse(
        batches=[],
        retrieval_traces=_extract_knowledge_retrieval_traces(serialized_messages),
    )
    payload = response.model_dump()
    assert payload["batches"] == []
    assert payload["retrieval_traces"] == [rag.trace]


@pytest.mark.asyncio
async def test_retrieval_debug_reuses_answer_retrieval_and_marks_debug_trigger(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway.routers import llmwiki

    captured: dict[str, Any] = {}
    original_trace = {
        "retrieval_trace_id": "trace-debug",
        "trigger": "pre_retrieval",
        "query": "泵故障",
        "selected_spaces": [],
        "source_count": 0,
        "capabilities": {"vector_retrieval": True},
    }

    async def fake_build_llmwiki_rag_context(**kwargs: Any) -> SimpleNamespace:
        captured.update(kwargs)
        return SimpleNamespace(trace=original_trace)

    async def admin_actor(_: object) -> tuple[str, bool]:
        return "admin", True

    monkeypatch.setattr(llmwiki, "_actor", admin_actor)
    monkeypatch.setattr(
        llmwiki,
        "build_llmwiki_rag_context",
        fake_build_llmwiki_rag_context,
    )
    request = SimpleNamespace()
    body = llmwiki.RetrievalDebugRequest(
        query="  泵故障  ",
        knowledge_base_ids=[" kb-one ", "kb-one"],
        assistant_knowledge_base_ids=[" assistant-one ", "assistant-one"],
    )

    response = await llmwiki.debug_llmwiki_retrieval(request, body)

    assert captured["request"] is request
    assert captured["body"].context == {
        "llmwiki_knowledge_base_ids": ["kb-one"],
        "assistant_knowledge_base_ids": ["assistant-one"],
    }
    message = captured["graph_input"]["messages"][0]
    assert isinstance(message, HumanMessage)
    assert message.content == "泵故障"
    assert response == {**original_trace, "trigger": "debug"}
    assert original_trace["trigger"] == "pre_retrieval"


@pytest.mark.asyncio
async def test_retrieval_debug_requires_at_least_one_selected_space(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway.routers import llmwiki

    async def admin_actor(_: object) -> tuple[str, bool]:
        return "admin", True

    monkeypatch.setattr(llmwiki, "_actor", admin_actor)
    with pytest.raises(HTTPException) as exc_info:
        await llmwiki.debug_llmwiki_retrieval(
            SimpleNamespace(),
            llmwiki.RetrievalDebugRequest(
                query="泵故障",
                knowledge_base_ids=[],
                assistant_knowledge_base_ids=[],
            ),
        )

    assert exc_info.value.status_code == 422


@pytest.mark.asyncio
async def test_combined_retrieval_debug_rejects_non_admin_before_retrieval(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway.routers import llmwiki

    retrieval_started = False

    async def ordinary_actor(_: object) -> tuple[str, bool]:
        return "ordinary-user", False

    async def forbidden_retrieval(**_: Any) -> SimpleNamespace:
        nonlocal retrieval_started
        retrieval_started = True
        return SimpleNamespace(trace={})

    monkeypatch.setattr(llmwiki, "_actor", ordinary_actor)
    monkeypatch.setattr(llmwiki, "build_llmwiki_rag_context", forbidden_retrieval)

    with pytest.raises(HTTPException) as exc_info:
        await llmwiki.debug_llmwiki_retrieval(
            SimpleNamespace(),
            llmwiki.RetrievalDebugRequest(
                query="泵故障",
                knowledge_base_ids=["kb-one"],
            ),
        )

    assert exc_info.value.status_code == 403
    assert retrieval_started is False


@pytest.mark.asyncio
async def test_single_base_retrieval_debug_rejects_non_admin_before_services(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway.routers import llmwiki_index

    services_started = False

    async def ordinary_actor(_: object) -> tuple[str, bool]:
        return "ordinary-user", False

    async def forbidden_services(_: object) -> None:
        nonlocal services_started
        services_started = True

    monkeypatch.setattr(llmwiki_index, "_actor", ordinary_actor)
    monkeypatch.setattr(llmwiki_index, "_services", forbidden_services)

    with pytest.raises(HTTPException) as exc_info:
        await llmwiki_index.retrieval_debug(
            SimpleNamespace(),
            "kb-one",
            llmwiki_index.WikiRetrievalDebugRequest(query="泵故障"),
        )

    assert exc_info.value.status_code == 403
    assert services_started is False


@pytest.mark.asyncio
async def test_rag_context_restores_both_selection_kinds_from_persisted_metadata_when_keys_are_omitted(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-persisted",
                "weknora_id": "remote-persisted",
                "name": "持久化 Wiki",
                "owner_user_id": "user-a",
            }
        ]
    )
    retrieval = _TraceRetrievalService(
        {
            "mode": "vector",
            "results": [],
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {},
        }
    )
    assistant_store = _TraceAssistantKnowledgeStore([])
    request = _trace_request(
        mapping_store=mapping_store,
        retrieval_service=retrieval,
        assistant_store=assistant_store,
    )

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=True),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=SimpleNamespace(assistant_id=None, context={}, metadata={}),
        graph_input={"messages": [HumanMessage(content="继续问持久化知识")]},
        fallback_metadata={
            "llmwiki_knowledge_base_ids": ["kb-persisted"],
            "assistant_knowledge_base_ids": ["assistant-persisted"],
        },
    )

    assert rag.knowledge_base_ids == ["kb-persisted"]
    assert rag.assistant_knowledge_base_ids == ["assistant-persisted"]
    assert retrieval.calls == [
        {
            "query": "继续问持久化知识",
            "mapping_ids": ["kb-persisted"],
            "top_k_pages": 8,
        }
    ]
    assert assistant_store.calls[0]["base_ids"] == ["assistant-persisted"]
    assert rag.trace["authorized_space_count"] == 2
    selected_spaces = _selected_spaces_json(rag.trace)
    assert "kb-persisted" in selected_spaces
    assert "assistant-persisted" in selected_spaces


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("explicit_empty_key", "expected_llmwiki_ids", "expected_assistant_ids"),
    [
        ("llmwiki_knowledge_base_ids", [], ["assistant-persisted"]),
        ("assistant_knowledge_base_ids", ["kb-persisted"], []),
    ],
)
async def test_explicit_empty_selection_is_not_restored_from_persisted_metadata(
    monkeypatch: pytest.MonkeyPatch,
    explicit_empty_key: str,
    expected_llmwiki_ids: list[str],
    expected_assistant_ids: list[str],
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-persisted",
                "weknora_id": "remote-persisted",
                "name": "持久化 Wiki",
                "owner_user_id": "user-a",
            }
        ]
    )
    retrieval = _TraceRetrievalService(
        {
            "mode": "vector",
            "results": [],
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {},
        }
    )
    assistant_store = _TraceAssistantKnowledgeStore([])
    request = _trace_request(
        mapping_store=mapping_store,
        retrieval_service=retrieval,
        assistant_store=assistant_store,
    )

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=True),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=SimpleNamespace(
            assistant_id=None,
            context={explicit_empty_key: []},
            metadata={},
        ),
        graph_input={"messages": [HumanMessage(content="继续问持久化知识")]},
        fallback_metadata={
            "llmwiki_knowledge_base_ids": ["kb-persisted"],
            "assistant_knowledge_base_ids": ["assistant-persisted"],
        },
    )

    assert rag.knowledge_base_ids == expected_llmwiki_ids
    assert rag.assistant_knowledge_base_ids == expected_assistant_ids
    assert [call["mapping_ids"] for call in retrieval.calls] == (
        [["kb-persisted"]] if expected_llmwiki_ids else []
    )
    assert [call["base_ids"] for call in assistant_store.calls] == (
        [["assistant-persisted"]] if expected_assistant_ids else []
    )


@pytest.mark.asyncio
async def test_explicitly_clearing_all_spaces_emits_a_durable_no_search_trace() -> None:
    from app.gateway import llmwiki_rag

    retrieval = _TraceRetrievalService(
        {
            "mode": "vector",
            "results": [],
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {},
        }
    )
    assistant_store = _TraceAssistantKnowledgeStore([])
    request = _trace_request(
        mapping_store=_TraceMappingStore([]),
        retrieval_service=retrieval,
        assistant_store=assistant_store,
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=SimpleNamespace(
            assistant_id=None,
            context={
                "llmwiki_knowledge_base_ids": [],
                "assistant_knowledge_base_ids": [],
            },
            metadata={},
        ),
        graph_input={"messages": [HumanMessage(content="明确清空后不要检索")]},
        fallback_metadata={
            "llmwiki_knowledge_base_ids": ["kb-old"],
            "assistant_knowledge_base_ids": ["assistant-old"],
        },
    )

    assert rag.knowledge_base_ids == []
    assert rag.assistant_knowledge_base_ids == []
    assert rag.sources == []
    assert rag.prompt == ""
    assert rag.trace["mode"] == "disabled"
    assert rag.trace["selected_spaces"] == []
    assert rag.trace["source_count"] == 0
    assert rag.trace["zero_result_reason"] == "NO_KNOWLEDGE_SELECTED"
    assert retrieval.calls == []
    assert assistant_store.calls == []


def test_disabled_selection_trace_does_not_reenable_inherited_rag_mode() -> None:
    from app.gateway.llmwiki_rag import LlmWikiRagContext
    from app.gateway.services import _effective_rag_mode_after_llmwiki

    body = SimpleNamespace(
        context={
            "llmwiki_knowledge_base_ids": [],
            "assistant_knowledge_base_ids": [],
        },
        metadata={},
    )
    disabled_rag = LlmWikiRagContext(
        [],
        [],
        0,
        "",
        [],
        "明确清空后普通回答",
        {"mode": "disabled", "zero_result_reason": "NO_KNOWLEDGE_SELECTED"},
    )

    assert _effective_rag_mode_after_llmwiki(
        body=body,
        rag=disabled_rag,
        inherited=True,
    ) is False


def test_resolved_space_or_explicit_generic_rag_flag_still_enables_rag_mode() -> None:
    from app.gateway.llmwiki_rag import LlmWikiRagContext
    from app.gateway.services import _effective_rag_mode_after_llmwiki

    selected_rag = LlmWikiRagContext(["kb-1"], [], 0, "", [], "问题", {})
    assert _effective_rag_mode_after_llmwiki(
        body=SimpleNamespace(context={}, metadata={}),
        rag=selected_rag,
        inherited=False,
    ) is True

    explicit_generic_rag = LlmWikiRagContext([], [], 0, "", [], "问题", {"mode": "disabled"})
    assert _effective_rag_mode_after_llmwiki(
        body=SimpleNamespace(
            context={
                "llmwiki_knowledge_base_ids": [],
                "assistant_knowledge_base_ids": [],
                "rag_mode_enabled": True,
            },
            metadata={},
        ),
        rag=explicit_generic_rag,
        inherited=True,
    ) is True


@pytest.mark.asyncio
async def test_llmwiki_actor_uses_regular_runtime_user_without_admin_escalation(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag
    from app.gateway.auth.models import User
    from deerflow.runtime.user_context import reset_current_user, set_current_user

    async def no_http_user(_: object) -> None:
        return None

    context_user = User(email="regular@example.com", system_role="user")
    token = set_current_user(context_user)
    try:
        monkeypatch.setattr(llmwiki_rag, "get_optional_user_from_request", no_http_user)
        # Even disabled-auth mode must not override a real request-scoped user.
        monkeypatch.setattr(llmwiki_rag, "is_auth_disabled", lambda: True)

        user_id, is_admin = await llmwiki_rag._actor(SimpleNamespace())
    finally:
        reset_current_user(token)

    assert user_id == str(context_user.id)
    assert is_admin is False


@pytest.mark.asyncio
async def test_llmwiki_actor_without_any_user_is_not_admin_when_auth_is_enabled(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    async def no_http_user(_: object) -> None:
        return None

    monkeypatch.setattr(llmwiki_rag, "get_optional_user_from_request", no_http_user)
    monkeypatch.setattr(llmwiki_rag, "get_current_user", lambda: None)
    monkeypatch.setattr(llmwiki_rag, "get_effective_user_id", lambda: "default-user")
    monkeypatch.setattr(llmwiki_rag, "is_auth_disabled", lambda: False)

    assert await llmwiki_rag._actor(SimpleNamespace()) == ("default-user", False)


@pytest.mark.asyncio
async def test_fail_unstarted_run_marks_run_and_thread_error_and_releases_lease_once() -> None:
    from app.gateway.services import _fail_unstarted_run
    from deerflow.runtime.runs.manager import RunManager
    from deerflow.runtime.runs.schemas import RunStatus

    class ThreadStore:
        def __init__(self) -> None:
            self.calls: list[tuple[str, str]] = []

        async def update_status(self, thread_id: str, status: str) -> None:
            self.calls.append((thread_id, status))

    class ConcurrencyGate:
        def __init__(self) -> None:
            self.releases: list[dict[str, Any]] = []

        async def release(self, **kwargs: Any) -> None:
            self.releases.append(dict(kwargs))

    run_mgr = RunManager()
    record = await run_mgr.create("thread-unstarted")
    thread_store = ThreadStore()
    gate = ConcurrencyGate()
    lease_metadata = {
        "run_id": record.run_id,
        "thread_id": record.thread_id,
        "model_name": "test-model",
        "user_id": "user-a",
    }

    await _fail_unstarted_run(
        run_mgr=run_mgr,
        run_ctx=SimpleNamespace(thread_store=thread_store),
        record=record,
        thread_id=record.thread_id,
        concurrency_gate=gate,
        redis_lease_acquired=True,
        lease_metadata=lease_metadata,
        exc=RuntimeError("retrieval setup failed"),
    )

    assert record.status == RunStatus.error
    assert record.error == "RuntimeError: retrieval setup failed"
    assert thread_store.calls == [("thread-unstarted", "error")]
    assert gate.releases == [lease_metadata]


@pytest.mark.asyncio
async def test_mixed_selection_backfills_all_eight_ordinary_results_when_assistant_has_no_hits(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-mixed",
                "weknora_id": "remote-mixed",
                "name": "普通 Wiki",
                "owner_user_id": "user-a",
            }
        ]
    )
    retrieval = _TraceRetrievalService(
        {
            "mode": "vector",
            "results": _ordinary_trace_results(8),
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {},
        }
    )
    assistant_store = _TraceAssistantKnowledgeStore([])
    request = _trace_request(
        mapping_store=mapping_store,
        retrieval_service=retrieval,
        assistant_store=assistant_store,
    )

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=True),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(
            llmwiki_ids=["kb-mixed"],
            assistant_ids=["assistant-mixed"],
        ),
        graph_input={"messages": [HumanMessage(content="混合检索无助手命中")]},
    )

    assert retrieval.calls[0]["top_k_pages"] == 8
    assert assistant_store.calls[0]["limit"] == 8
    assert rag.source_count == 8
    assert [source["id"] for source in rag.sources] == [
        f"ordinary-{index}" for index in range(1, 9)
    ]
    assert [source["index"] for source in rag.sources] == list(range(1, 9))
    assert [result["rank"] for result in rag.trace["results"]] == list(range(1, 9))


@pytest.mark.asyncio
async def test_mixed_selection_reserves_four_results_per_populated_channel_and_renumbers(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-mixed",
                "weknora_id": "remote-mixed",
                "name": "普通 Wiki",
                "owner_user_id": "user-a",
            }
        ]
    )
    retrieval = _TraceRetrievalService(
        {
            "mode": "vector",
            "results": _ordinary_trace_results(8),
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {},
        }
    )
    assistant_store = _TraceAssistantKnowledgeStore(_assistant_trace_results(8))
    request = _trace_request(
        mapping_store=mapping_store,
        retrieval_service=retrieval,
        assistant_store=assistant_store,
    )

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=True),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(
            llmwiki_ids=["kb-mixed"],
            assistant_ids=["assistant-mixed"],
        ),
        graph_input={"messages": [HumanMessage(content="混合检索双通道命中")]},
    )

    assert rag.trace["mode"] == "mixed"
    assert rag.source_count == 8
    assert [source["id"] for source in rag.sources] == [
        "ordinary-1",
        "ordinary-2",
        "ordinary-3",
        "ordinary-4",
        "assistant-1",
        "assistant-2",
        "assistant-3",
        "assistant-4",
    ]
    assert [source["index"] for source in rag.sources] == list(range(1, 9))
    assert [result["rank"] for result in rag.trace["results"]] == list(range(1, 9))
    assert [result["kind"] for result in rag.trace["results"]] == [
        "llmwiki",
        "llmwiki",
        "llmwiki",
        "llmwiki",
        "assistant",
        "assistant",
        "assistant",
        "assistant",
    ]


@pytest.mark.asyncio
async def test_mixed_selection_keeps_raw_and_formatted_wiki_in_ordinary_quota(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway import llmwiki_rag

    mapping_store = _TraceMappingStore(
        [
            {
                "id": "kb-mixed",
                "weknora_id": "remote-mixed",
                "name": "普通 Wiki",
                "owner_user_id": "user-a",
            }
        ]
    )
    raw_client = _TraceRawClient(
        [
            {
                "chunk_id": f"raw-{index}",
                "knowledge_id": f"document-{index}",
                "knowledge_base_id": "remote-mixed",
                "title": "海湖庄园资料",
                "content": f"原始文档切片 {index}",
                "score": 0.99 - index / 100,
            }
            for index in range(1, 9)
        ]
    )
    retrieval = _TraceRetrievalService(
        {
            "mode": "vector",
            "results": _ordinary_trace_results(8),
            "partial": False,
            "warnings": [],
            "fallback_knowledge_base_ids": [],
            "diagnostics": {},
        },
        client=raw_client,
    )
    assistant_store = _TraceAssistantKnowledgeStore(_assistant_trace_results(8))
    request = _trace_request(
        mapping_store=mapping_store,
        retrieval_service=retrieval,
        assistant_store=assistant_store,
    )

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=True),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(
            llmwiki_ids=["kb-mixed"],
            assistant_ids=["assistant-mixed"],
        ),
        graph_input={"messages": [HumanMessage(content="海湖庄园")]},
    )

    ordinary_head = rag.sources[:4]
    assert [source["type"] for source in ordinary_head] == [
        "llmwiki",
        "llmwiki_wiki",
        "llmwiki",
        "llmwiki_wiki",
    ]
    assert [source["type"] for source in rag.sources[4:]] == [
        "assistant_knowledge"
    ] * 4
    assert rag.trace["diagnostics"]["final_raw_chunk_source_count"] == 2
    assert sum(source["type"] == "llmwiki_wiki" for source in rag.sources) == 2


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("failure_kind", "expected_warning"),
    [
        ("missing", "ASSISTANT_RETRIEVAL_PROVIDER_UNAVAILABLE"),
        ("raises", "ASSISTANT_RETRIEVAL_FAILED"),
    ],
)
async def test_assistant_provider_failure_uses_unavailable_prompt_instead_of_no_match_prompt(
    monkeypatch: pytest.MonkeyPatch,
    failure_kind: str,
    expected_warning: str,
) -> None:
    from app.gateway import llmwiki_rag

    class FailingAssistantStore:
        async def search_pages(self, **_: Any) -> list[dict[str, Any]]:
            raise RuntimeError("assistant index unavailable")

    mapping_store = _TraceMappingStore([])
    retrieval = _TraceRetrievalService({"results": []})
    assistant_store: Any | None = None if failure_kind == "missing" else FailingAssistantStore()
    request = _trace_request(
        mapping_store=mapping_store,
        retrieval_service=retrieval,
        assistant_store=assistant_store,
    )

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_rag, "_actor", actor)
    monkeypatch.setattr(
        llmwiki_rag,
        "get_resolved_llmwiki_runtime",
        lambda _config: SimpleNamespace(weknora_enabled=True),
    )

    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=request,
        body=_trace_body(assistant_ids=["assistant-failing"]),
        graph_input={"messages": [HumanMessage(content="助手知识检索失败")]},
    )

    assert rag.sources == []
    assert rag.trace["partial"] is True
    assert expected_warning in rag.trace["warnings"]
    assert rag.trace["zero_result_reason"] == "RETRIEVAL_PROVIDER_UNAVAILABLE"
    assert "检索不可用或未完整执行" not in rag.prompt
    assert "不能判断库内是否存在相关资料" not in rag.prompt
    assert "暂未找到相关资料" in rag.prompt
    assert expected_warning not in rag.prompt
    assert "不要向用户展示检索通道状态" in rag.prompt


@pytest.mark.asyncio
async def test_query_empty_trace_only_context_attaches_to_latest_human_and_round_trips() -> None:
    from app.gateway import llmwiki_rag
    from app.gateway.routers.threads import _extract_knowledge_retrieval_traces

    graph_input = {
        "messages": [
            HumanMessage(id="human-old", content="旧问题"),
            HumanMessage(
                id="human-latest",
                content="   ",
                additional_kwargs={"client_marker": "keep"},
            ),
        ]
    }
    rag = await llmwiki_rag.build_llmwiki_rag_context(
        request=SimpleNamespace(),
        body=_trace_body(llmwiki_ids=["kb-query-empty"]),
        graph_input=graph_input,
    )

    assert rag.prompt == ""
    assert rag.sources == []
    assert rag.trace["zero_result_reason"] == "QUERY_EMPTY"

    attached = llmwiki_rag.attach_llmwiki_rag_to_graph_input(graph_input, rag)
    old_message, latest_message = attached["messages"]
    assert "knowledge_retrieval_trace" not in old_message.additional_kwargs
    assert latest_message.additional_kwargs["client_marker"] == "keep"
    assert latest_message.additional_kwargs["knowledge_retrieval_trace"] == rag.trace
    assert "prompt_prefix" not in latest_message.additional_kwargs
    assert _extract_knowledge_retrieval_traces(attached["messages"]) == [
        {**rag.trace, "message_id": "human-latest"}
    ]


def test_tool_trace_parses_zero_result_error_and_clears_stale_reason_for_hits() -> None:
    from app.gateway.routers.threads import _tool_trace_from_message

    error_message = ToolMessage(
        name="llmwiki_search",
        tool_call_id="tool-error",
        content=json.dumps(
            {
                "retrieval_trace_id": "trace-tool-error",
                "query": "故障代码",
                "selection": ["kb-one"],
                "assistant_selection": [],
                "mode": "empty",
                "knowledge_base_count": 1,
                "results": [],
                "partial": True,
                "warnings": ["RETRIEVAL_PROVIDER_UNAVAILABLE"],
                "zero_result_reason": "RETRIEVAL_PROVIDER_UNAVAILABLE",
            },
            ensure_ascii=False,
        ),
    )
    error_trace = _tool_trace_from_message(error_message)

    assert error_trace is not None
    assert error_trace["trigger"] == "tool"
    assert error_trace["source_count"] == 0
    assert error_trace["results"] == []
    assert error_trace["zero_result_reason"] == "RETRIEVAL_PROVIDER_UNAVAILABLE"

    success_message = ToolMessage(
        name="llmwiki_search",
        tool_call_id="tool-success",
        content=json.dumps(
            {
                "retrieval_trace_id": "trace-tool-success",
                "query": "设备散热",
                "selection": ["kb-one"],
                "mode": "vector",
                "knowledge_base_count": 1,
                "results": [
                    {
                        "id": "page-one",
                        "knowledge_base_id": "kb-one",
                        "title": "设备散热",
                        "content": "先断电，再检查风道。",
                        "retrieval_mode": "vector",
                        "score": 0.91,
                    }
                ],
                # A defensive parser must not carry a stale failure reason
                # into a trace that contains usable evidence.
                "zero_result_reason": "RETRIEVAL_PROVIDER_UNAVAILABLE",
            },
            ensure_ascii=False,
        ),
    )
    success_trace = _tool_trace_from_message(success_message)

    assert success_trace is not None
    assert success_trace["source_count"] == 1
    assert success_trace["results"][0]["id"] == "page-one"
    assert success_trace["zero_result_reason"] is None


class _ToolTraceMappingStore:
    def __init__(self, rows: dict[str, dict[str, Any] | None]) -> None:
        self.rows = rows

    async def get_authorized(self, mapping_id: str, *_: Any, **__: Any) -> dict[str, Any] | None:
        return self.rows.get(mapping_id)


class _ToolTraceOrdinaryService:
    def __init__(self, result: dict[str, Any] | Exception) -> None:
        self.result = result
        self.calls: list[list[str]] = []

    async def search(self, _query: str, rows: list[dict[str, Any]]) -> dict[str, Any]:
        self.calls.append([str(row["id"]) for row in rows])
        if isinstance(self.result, Exception):
            raise self.result
        return self.result


class _ToolTraceAssistantStore:
    def __init__(
        self,
        *,
        bases: dict[str, dict[str, Any] | None],
        result: list[dict[str, Any]] | Exception,
    ) -> None:
        self.bases = bases
        self.result = result

    async def get_base(self, base_id: str) -> dict[str, Any] | None:
        return self.bases.get(base_id)

    async def search_pages(self, *, base_ids: list[str], **_: Any) -> list[dict[str, Any]]:
        if isinstance(self.result, Exception):
            raise self.result
        return self.result


def _tool_trace_ordinary_hit() -> dict[str, Any]:
    return {
        "id": "ordinary-page",
        "knowledge_base_id": "ordinary-kb",
        "title": "普通 Wiki 命中",
        "content": "普通 Wiki 检索证据",
        "retrieval_mode": "vector",
        "score": 0.91,
    }


def _tool_trace_assistant_hit() -> dict[str, Any]:
    return {
        "id": "assistant-page",
        "base_id": "assistant-kb",
        "title": "助手库命中",
        "content": "助手知识检索证据",
        "retrieval_mode": "vector",
        "score": 0.87,
    }


def _configure_search_tool_trace(
    monkeypatch: pytest.MonkeyPatch,
    *,
    mapping_store: _ToolTraceMappingStore,
    ordinary_service: _ToolTraceOrdinaryService,
    assistant_store: _ToolTraceAssistantStore | None,
) -> Any:
    from deerflow.persistence import assistant_knowledge
    from deerflow.tools.builtins import llmwiki_search_tool as tool_module

    monkeypatch.setattr(
        tool_module,
        "get_app_config",
        lambda: SimpleNamespace(
            llmwiki=SimpleNamespace(
                local_wiki_index=SimpleNamespace(enabled=True, internal_search_enabled=True)
            )
        ),
    )
    monkeypatch.setattr(
        tool_module,
        "get_resolved_llmwiki_runtime",
        lambda: SimpleNamespace(weknora_enabled=True),
    )
    monkeypatch.setattr(tool_module, "get_effective_user_id", lambda: "user-a")
    monkeypatch.setattr(
        tool_module,
        "get_current_user",
        lambda: SimpleNamespace(system_role="user"),
    )
    monkeypatch.setattr(tool_module, "_agent_allowed_ids", lambda _context: None)
    monkeypatch.setattr(tool_module, "get_llmwiki_store", lambda: mapping_store)
    monkeypatch.setattr(
        tool_module,
        "get_local_wiki_retrieval_service",
        lambda: ordinary_service,
    )
    monkeypatch.setattr(
        tool_module,
        "get_local_wiki_search_service",
        lambda: SimpleNamespace(embedding=object()),
    )
    monkeypatch.setattr(
        assistant_knowledge,
        "get_assistant_knowledge_store",
        lambda: assistant_store,
    )
    return tool_module


async def _run_mixed_search_tool(tool_module: Any) -> dict[str, Any]:
    runtime = SimpleNamespace(
        context={
            "llmwiki_knowledge_base_ids": ["ordinary-kb"],
            "assistant_knowledge_base_ids": ["assistant-kb"],
        }
    )
    content = await tool_module.llmwiki_search_tool.coroutine("散热故障", runtime)
    return json.loads(content)


@pytest.mark.asyncio
async def test_mixed_tool_keeps_assistant_hit_when_ordinary_authorization_fails(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    ordinary_service = _ToolTraceOrdinaryService({"mode": "vector", "results": []})
    tool_module = _configure_search_tool_trace(
        monkeypatch,
        mapping_store=_ToolTraceMappingStore({"ordinary-kb": None}),
        ordinary_service=ordinary_service,
        assistant_store=_ToolTraceAssistantStore(
            bases={"assistant-kb": {"id": "assistant-kb"}},
            result=[_tool_trace_assistant_hit()],
        ),
    )

    trace = await _run_mixed_search_tool(tool_module)

    assert ordinary_service.calls == []
    assert trace["selection_count"] == 2
    assert trace["authorized_space_count"] == 1
    assert trace["partial"] is True
    assert trace["warnings"] == ["SPACE_NOT_FOUND_OR_UNAUTHORIZED"]
    assert [result["id"] for result in trace["results"]] == ["assistant-page"]
    assert trace["zero_result_reason"] is None


@pytest.mark.asyncio
async def test_mixed_tool_keeps_assistant_hit_when_ordinary_search_fails(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    tool_module = _configure_search_tool_trace(
        monkeypatch,
        mapping_store=_ToolTraceMappingStore(
            {"ordinary-kb": {"id": "ordinary-kb", "name": "普通 Wiki"}}
        ),
        ordinary_service=_ToolTraceOrdinaryService(
            WikiIndexError("WIKI_NOT_FINALIZED", "not ready", status_code=409)
        ),
        assistant_store=_ToolTraceAssistantStore(
            bases={"assistant-kb": {"id": "assistant-kb"}},
            result=[_tool_trace_assistant_hit()],
        ),
    )

    trace = await _run_mixed_search_tool(tool_module)

    assert trace["selection_count"] == 2
    assert trace["authorized_space_count"] == 2
    assert trace["partial"] is True
    assert "WIKI_NOT_FINALIZED" in trace["warnings"]
    assert [result["id"] for result in trace["results"]] == ["assistant-page"]
    assert trace["zero_result_reason"] is None


@pytest.mark.asyncio
async def test_classic_tool_scopes_one_library_and_assigns_stable_global_ranks(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    ordinary_service = _ToolTraceOrdinaryService(
        {
            "mode": "vector",
            "results": [
                {
                    "id": "lower-score",
                    "knowledge_base_id": "ordinary-two",
                    "title": "次相关页面",
                    "content": "次相关证据",
                    "score": 0.71,
                },
                {
                    "id": "higher-score",
                    "knowledge_base_id": "ordinary-two",
                    "title": "最相关页面",
                    "content": "最相关证据",
                    "score": 0.95,
                },
            ],
            "partial": False,
            "warnings": [],
        }
    )
    tool_module = _configure_search_tool_trace(
        monkeypatch,
        mapping_store=_ToolTraceMappingStore(
            {
                "ordinary-one": {"id": "ordinary-one", "name": "一号库"},
                "ordinary-two": {"id": "ordinary-two", "name": "二号库"},
            }
        ),
        ordinary_service=ordinary_service,
        assistant_store=None,
    )
    runtime = SimpleNamespace(
        context={
            "knowledge_retrieval_mode": "classic",
            "llmwiki_knowledge_base_ids": ["ordinary-one", "ordinary-two"],
            "assistant_knowledge_base_ids": [],
        }
    )

    first = json.loads(
        await tool_module.llmwiki_search_tool.coroutine(
            "治理进展",
            runtime,
            knowledge_base_id="ordinary-two",
        )
    )
    second = json.loads(
        await tool_module.llmwiki_search_tool.coroutine(
            "治理进展",
            runtime,
            knowledge_base_id="ordinary-two",
        )
    )

    assert ordinary_service.calls == [["ordinary-two"], ["ordinary-two"]]
    assert first["selection"] == ["ordinary-two"]
    assert [item["id"] for item in first["results"]] == [
        "higher-score",
        "lower-score",
    ]
    assert [item["relevance_rank"] for item in first["results"]] == [1, 2]
    assert [item["citation_id"] for item in first["results"]] == [
        item["citation_id"] for item in second["results"]
    ]
    assert all(
        item["citation_id"].startswith("llmwiki:") for item in first["results"]
    )


@pytest.mark.asyncio
async def test_classic_search_tool_rejects_smart_mode_even_if_called_directly(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    ordinary_service = _ToolTraceOrdinaryService(
        {"mode": "vector", "results": [_tool_trace_ordinary_hit()]}
    )
    tool_module = _configure_search_tool_trace(
        monkeypatch,
        mapping_store=_ToolTraceMappingStore(
            {"ordinary-kb": {"id": "ordinary-kb", "name": "普通 Wiki"}}
        ),
        ordinary_service=ordinary_service,
        assistant_store=None,
    )
    runtime = SimpleNamespace(
        context={
            "knowledge_retrieval_mode": "smart_reasoning",
            "llmwiki_knowledge_base_ids": ["ordinary-kb"],
            "assistant_knowledge_base_ids": [],
        }
    )

    trace = json.loads(
        await tool_module.llmwiki_search_tool.coroutine("散热故障", runtime)
    )

    assert trace["results"] == []
    assert trace["zero_result_reason"] == "CLASSIC_MODE_REQUIRED"
    assert ordinary_service.calls == []


@pytest.mark.asyncio
@pytest.mark.parametrize("provider_missing", [False, True])
async def test_mixed_tool_keeps_ordinary_hit_when_assistant_channel_fails(
    monkeypatch: pytest.MonkeyPatch,
    provider_missing: bool,
) -> None:
    assistant_store = (
        None
        if provider_missing
        else _ToolTraceAssistantStore(
            bases={"assistant-kb": {"id": "assistant-kb"}},
            result=RuntimeError("assistant index unavailable"),
        )
    )
    tool_module = _configure_search_tool_trace(
        monkeypatch,
        mapping_store=_ToolTraceMappingStore(
            {"ordinary-kb": {"id": "ordinary-kb", "name": "普通 Wiki"}}
        ),
        ordinary_service=_ToolTraceOrdinaryService(
            {
                "mode": "vector",
                "results": [_tool_trace_ordinary_hit()],
                "partial": False,
                "warnings": [],
            }
        ),
        assistant_store=assistant_store,
    )

    trace = await _run_mixed_search_tool(tool_module)

    expected_warning = (
        "ASSISTANT_RETRIEVAL_PROVIDER_UNAVAILABLE"
        if provider_missing
        else "ASSISTANT_RETRIEVAL_FAILED"
    )
    assert trace["selection_count"] == 2
    assert trace["authorized_space_count"] == (1 if provider_missing else 2)
    assert trace["partial"] is True
    assert trace["warnings"] == [expected_warning]
    assert [result["id"] for result in trace["results"]] == ["ordinary-page"]
    assert trace["zero_result_reason"] is None


@pytest.mark.parametrize(
    ("extra_fields", "expected_authorized"),
    [
        ({"selection_count": 2, "authorized_space_count": 1}, 1),
        ({"knowledge_base_count": 2}, 2),
    ],
)
def test_tool_trace_parser_prefers_explicit_authorized_count_and_supports_legacy(
    extra_fields: dict[str, int],
    expected_authorized: int,
) -> None:
    from app.gateway.routers.threads import _tool_trace_from_message

    message = ToolMessage(
        name="llmwiki_search",
        tool_call_id="call-counts",
        content=json.dumps(
            {
                "query": "散热故障",
                "selection": ["ordinary-kb"],
                "assistant_selection": ["assistant-kb"],
                "results": [],
                **extra_fields,
            },
            ensure_ascii=False,
        ),
    )

    trace = _tool_trace_from_message(message)

    assert trace is not None
    assert trace["selection_count"] == 2
    assert trace["authorized_space_count"] == expected_authorized