"""安全 FAQ 的检索增强(RAG)—— 召回、融合与接地.

这些用例盯的是「不会报错、只会安静地答错」的那几处:

* 语义召回**只**在词法失败的地方才值钱,所以要证明它真的能找到一条与提问
  零字面重合的条目 —— 用「问句写成问卷口吻、员工换个说法就搜不到」这个真实
  形状来构造。
* 融合按名次(RRF),因此必须证明两路都认的条目排在只有一路认的条目之前;
  否则这就只是一个语义检索,加权得再漂亮也没用。
* **接地闸**:模型写的每一句话都必须能落到这次召回的条目上。引用为空、引用
  的是没给过它的 id —— 都必须整段丢弃,而不是「至少显示一点什么」。这是整个
  改动里唯一一处会把模型的散文渲染给 500 个人看的地方。
* 降级链的每一环:没有 embeddings 后端、向量还没建、厂商挂掉、模型说答不了。
  这些都不能变成错误,页面必须还能用。

Schema 来自 ``create_all``(不跑迁移),嵌入调用全部 monkeypatch —— 不联网,
不需要 Key。
"""
from __future__ import annotations

import uuid

import pytest
from httpx import AsyncClient
from sqlalchemy import select

from app.domains.auditor import answer_library_service as lib
from app.domains.auditor import security_faq_rag as rag
from app.domains.auditor import security_faq_service as faq
from app.domains.auditor.models import (
    AuditorAnswerLibraryEntry,
    AuditorSecurityFaqGap,
)

pytestmark = pytest.mark.asyncio


@pytest.fixture(autouse=True)
def _clean_vector_cache():
    """The corpus cache is process-global. Its staleness probe would catch a
    changed corpus anyway, but a test that reads a neighbour's vectors fails
    for a reason that has nothing to do with what it asserts."""
    rag.reset_vector_cache()
    yield
    rag.reset_vector_cache()


async def _entry(
    db, *, question_zh: str, question_en: str,
    answer_zh: str = "答案", answer_en: str = "Answer",
    category: str = "general", visible: bool = True,
) -> AuditorAnswerLibraryEntry:
    e = await lib.create_entry(
        db,
        payload={
            "category": category,
            "question_zh": question_zh, "question_en": question_en,
            "answer_zh": answer_zh, "answer_en": answer_en,
            "employee_visible": visible,
        },
        actor_id=None,
    )
    await db.commit()
    return e


# ── A deterministic stand-in for the embeddings provider ──────────────
#
# Vectors are hand-written per topic rather than derived from the text: the
# point of these tests is what the retrieval DOES with a vector space, not
# whether a particular model places two sentences near each other. A fake
# that "computed" similarity from shared characters would be testing the
# lexical channel a second time, which is exactly the blind spot this whole
# feature exists to cover.

#: What ``resolve_embed_model`` returns when nothing is configured.
_ACTIVE_MODEL = "text-embedding-v3"

_LAPTOP = [1.0, 0.0, 0.0]
_PHISHING = [0.0, 1.0, 0.0]
_UNRELATED = [0.0, 0.0, 1.0]


def _embedder(mapping: dict[str, list[float]], *, calls: list | None = None):
    """``embed_texts`` stand-in: first needle found in the text wins."""

    async def _embed(_db, texts, *, timeout=60.0):
        if calls is not None:
            calls.append(list(texts))
        out = []
        for text in texts:
            vec = next(
                (v for needle, v in mapping.items() if needle in text),
                _UNRELATED,
            )
            out.append(list(vec))
        # The ACTIVE model id, not a made-up one: ``pending_clause`` treats a
        # vector produced by any other model as stale, so a fake that named
        # itself would leave every row it just embedded still pending — and
        # the backfill would re-embed the library on every run.
        return out, _ACTIVE_MODEL

    return _embed


async def _index(db, monkeypatch, mapping) -> None:
    """Embed everything visible with the fake provider."""
    monkeypatch.setattr(rag, "embed_texts", _embedder(mapping))
    await rag.backfill(db, limit=100)
    rag.reset_vector_cache()


# ── Fusion ────────────────────────────────────────────────────────────


async def test_a_document_both_channels_found_outranks_a_single_channel_winner():
    both = uuid.uuid4()
    lexical_only = uuid.uuid4()
    semantic_only = uuid.uuid4()

    fused = rag.rrf_fuse(
        {
            # ``both`` is second in each channel; the other two each top one.
            "keyword": [lexical_only, both],
            "semantic": [semantic_only, both],
        },
        limit=3,
    )
    assert [entry_id for entry_id, _, _ in fused][0] == both
    assert fused[0][2] == ["keyword", "semantic"]


async def test_fusion_of_one_channel_preserves_that_channel_order():
    ids = [uuid.uuid4() for _ in range(3)]
    fused = rag.rrf_fuse({"keyword": ids}, limit=3)
    assert [entry_id for entry_id, _, _ in fused] == ids


# ── Semantic recall ───────────────────────────────────────────────────


async def test_a_question_with_no_shared_wording_is_found_by_meaning(
    db_session, monkeypatch,
):
    """The failure this whole feature exists for.

    The library is written in customer-questionnaire voice. 「我能用自己的
    电脑写公司代码吗」 shares no term with 「员工能否使用个人设备访问公司
    数据?」 — not one two-character shingle — so the lexical channel finds
    nothing and the page used to say 「没有找到」 for a question we answer.
    """
    target = await _entry(
        db_session,
        question_zh="员工能否使用个人设备访问公司数据?",
        question_en="Can staff use personal devices for company data?",
        answer_zh="公司数据只能在纳管终端上处理。",
        answer_en="Company data belongs on managed endpoints.",
        category="endpoint_security",
    )
    await _entry(
        db_session, question_zh="办公室门禁如何管理?", question_en="Badge access?",
        category="physical_security",
    )
    await _index(db_session, monkeypatch, {
        "个人设备": _LAPTOP, "personal device": _LAPTOP,
    })

    question = "我能用自己的笔记本写公司代码吗"
    # Precondition, not decoration: if the lexical channel could answer this,
    # the assertion below would pass without the semantic channel existing.
    assert await faq._keyword_hits(db_session, term=question, limit=5) == []

    monkeypatch.setattr(rag, "embed_texts", _embedder({question: _LAPTOP}))
    res = await faq.ask(
        db_session, query=question, lang="zh", person_id=uuid.uuid4(),
        compose=False,
    )
    assert res["matched_by"] == "semantic"
    assert [i["id"] for i in res["items"]] == [str(target.id)]
    # And the text shown is the stored answer, not anything generated.
    assert res["items"][0]["answer"] == "公司数据只能在纳管终端上处理。"


async def test_a_merely_least_unrelated_entry_is_not_a_semantic_hit(
    db_session, monkeypatch,
):
    """Cosine ranking always returns *something*. Without the score floor the
    library's least-unrelated row would be presented as the answer to a
    question nobody here has ever written about."""
    await _entry(
        db_session, question_zh="办公室门禁如何管理?", question_en="Badge access?",
    )
    await _index(db_session, monkeypatch, {"门禁": _UNRELATED})

    monkeypatch.setattr(rag, "embed_texts", _embedder({"__query__": _PHISHING}))
    ranked = await rag.semantic_ranked(db_session, query="__query__")
    assert ranked == []


async def test_semantic_retrieval_is_unavailable_not_empty_without_a_backend(
    db_session, monkeypatch,
):
    """``None`` (no backend / nothing indexed) and ``[]`` (searched, nothing
    close enough) are different answers to the console, even though the ask
    path degrades identically for both."""
    await _entry(db_session, question_zh="任意", question_en="Anything")

    async def _no_backend(_db, _texts, *, timeout=60.0):
        return None

    monkeypatch.setattr(rag, "embed_texts", _no_backend)
    assert await rag.semantic_ranked(db_session, query="任意问题") is None


async def test_an_entry_retracted_after_indexing_cannot_be_returned(
    db_session, monkeypatch,
):
    """The vector cache holds ids, and ids outlive visibility. Retrieval must
    re-read through the visibility gate — otherwise pulling an answer off the
    employee page leaves it reachable by asking the right question."""
    target = await _entry(
        db_session, question_zh="个人设备政策是什么?", question_en="Personal device policy?",
    )
    await _index(db_session, monkeypatch, {"个人设备": _LAPTOP})

    target.employee_visible = False
    await db_session.commit()

    monkeypatch.setattr(rag, "embed_texts", _embedder({"__q__": _LAPTOP}))
    res = await faq.ask(
        db_session, query="__q__", lang="zh", person_id=uuid.uuid4(),
        compose=False,
    )
    assert [i["id"] for i in res["items"]] == []


async def test_both_channels_are_reported_when_both_contributed(
    db_session, monkeypatch,
):
    entry = await _entry(
        db_session,
        question_zh="钓鱼邮件如何识别?", question_en="How to spot phishing?",
    )
    await _index(db_session, monkeypatch, {"钓鱼": _PHISHING})

    monkeypatch.setattr(rag, "embed_texts", _embedder({"钓鱼": _PHISHING}))
    res = await faq.ask(
        db_session, query="钓鱼邮件", lang="zh", person_id=uuid.uuid4(),
        compose=False,
    )
    assert res["matched_by"] == "hybrid"
    assert res["retrieval"]["channels"][str(entry.id)] == ["keyword", "semantic"]


# ── The filter box's semantic rescue ──────────────────────────────────


async def test_the_filter_box_falls_through_to_meaning_when_nothing_matches(
    db_session, monkeypatch,
):
    target = await _entry(
        db_session,
        question_zh="员工能否使用个人设备访问公司数据?",
        question_en="Can staff use personal devices for company data?",
    )
    await _index(db_session, monkeypatch, {"个人设备": _LAPTOP})

    monkeypatch.setattr(rag, "embed_texts", _embedder({"自带电脑": _LAPTOP}))
    listed = await faq.list_faq(db_session, lang="zh", search="自带电脑")
    assert [i["id"] for i in listed["items"]] == [str(target.id)]
    assert listed["meta"]["total"] == 1


async def test_the_filter_box_still_prefers_a_literal_match(
    db_session, monkeypatch,
):
    """Semantic is the rescue, not the ranking. Somebody who typed a word
    that IS in the library must get the substring match, in substring order —
    a filter box that reorders itself by a model's opinion stops being a
    filter box."""
    literal = await _entry(
        db_session, question_zh="门禁卡丢了怎么办?", question_en="Lost badge?",
    )
    await _index(db_session, monkeypatch, {"门禁": _PHISHING})

    calls: list = []
    monkeypatch.setattr(rag, "embed_texts", _embedder({"门禁": _LAPTOP}, calls=calls))
    listed = await faq.list_faq(db_session, lang="zh", search="门禁")
    assert [i["id"] for i in listed["items"]] == [str(literal.id)]
    # No embedding round trip at all: the lexical path answered.
    assert calls == []


# ── Grounded composition ──────────────────────────────────────────────


class _FakeComposer:
    def __init__(self, reply):
        self.reply = reply
        self.calls = 0
        self.last_user = ""

    async def chat_json(self, *, system, user, max_tokens=3000, temperature=0.3):
        self.calls += 1
        self.last_user = user
        return self.reply


async def test_the_composed_answer_is_returned_with_its_sources(
    db_session, monkeypatch,
):
    entry = await _entry(
        db_session, question_zh="能用生成式 AI 吗?", question_en="Generative AI?",
        answer_zh="不得投喂公司敏感信息。", answer_en="Do not feed it sensitive data.",
    )
    composer = _FakeComposer({
        "covered": True,
        "answer": "可以用于公开内容,但不要投喂公司敏感信息。",
        "cited_ids": [str(entry.id)],
    })

    # ``**_kwargs``, not ``(_db)``: the caller passes ``timeout=`` and a
    # double whose signature does not match the real one degrades into the
    # "LLM unavailable" branch — the fake would silently stop being exercised.
    async def _client(_db, **_kwargs):
        return composer

    monkeypatch.setattr(faq, "build_llm_client", _client)
    res = await faq.ask(
        db_session, query="生成式 AI", lang="zh", person_id=uuid.uuid4(),
    )

    assert res["answer"]["text"].startswith("可以用于公开内容")
    assert res["answer"]["cited_ids"] == [str(entry.id)]
    assert res["covered"] is True
    # The stored entry travels with it — always.
    assert res["items"][0]["answer"] == "不得投喂公司敏感信息。"
    # And the model saw the stored answers, so it can only paraphrase them.
    assert "不得投喂公司敏感信息。" in composer.last_user


async def test_an_answer_citing_nothing_we_gave_it_is_discarded(
    db_session, monkeypatch,
):
    """Prose with no traceable source reads exactly like a policy statement
    and has nothing behind it. The entries still render; the paragraph does
    not."""
    await _entry(
        db_session, question_zh="能用生成式 AI 吗?", question_en="Generative AI?",
    )
    composer = _FakeComposer({
        "covered": True,
        "answer": "公司允许在任何场景下使用任意 AI 工具。",
        # An id from nowhere — the shape a hallucinated citation takes.
        "cited_ids": [str(uuid.uuid4())],
    })

    # ``**_kwargs``, not ``(_db)``: the caller passes ``timeout=`` and a
    # double whose signature does not match the real one degrades into the
    # "LLM unavailable" branch — the fake would silently stop being exercised.
    async def _client(_db, **_kwargs):
        return composer

    monkeypatch.setattr(faq, "build_llm_client", _client)
    res = await faq.ask(
        db_session, query="生成式 AI", lang="zh", person_id=uuid.uuid4(),
    )
    assert res["answer"] is None
    assert len(res["items"]) == 1
    # Not a gap: we did retrieve the answer, we merely refused to narrate it.
    assert (await db_session.execute(select(AuditorSecurityFaqGap))).first() is None


async def test_a_hallucinated_citation_is_dropped_from_a_real_answer(
    db_session, monkeypatch,
):
    entry = await _entry(
        db_session, question_zh="能用生成式 AI 吗?", question_en="Generative AI?",
    )
    composer = _FakeComposer({
        "covered": True,
        "answer": "不要投喂公司敏感信息。",
        "cited_ids": [str(uuid.uuid4()), str(entry.id), str(entry.id)],
    })

    # ``**_kwargs``, not ``(_db)``: the caller passes ``timeout=`` and a
    # double whose signature does not match the real one degrades into the
    # "LLM unavailable" branch — the fake would silently stop being exercised.
    async def _client(_db, **_kwargs):
        return composer

    monkeypatch.setattr(faq, "build_llm_client", _client)
    res = await faq.ask(
        db_session, query="生成式 AI", lang="zh", person_id=uuid.uuid4(),
    )
    assert res["answer"]["cited_ids"] == [str(entry.id)]


async def test_not_covered_keeps_the_entries_and_records_a_gap(
    db_session, monkeypatch,
):
    """The case the old shape could not see. Retrieval nearly always returns
    something, so "we showed them five answers and none was theirs" used to
    be indistinguishable from a hit — and never reached the gap ledger."""
    await _entry(
        db_session, question_zh="能用生成式 AI 吗?", question_en="Generative AI?",
    )
    composer = _FakeComposer({"covered": False, "answer": "", "cited_ids": []})

    # ``**_kwargs``, not ``(_db)``: the caller passes ``timeout=`` and a
    # double whose signature does not match the real one degrades into the
    # "LLM unavailable" branch — the fake would silently stop being exercised.
    async def _client(_db, **_kwargs):
        return composer

    monkeypatch.setattr(faq, "build_llm_client", _client)
    asker = uuid.uuid4()
    res = await faq.ask(
        db_session, query="生成式 AI 的采购预算是多少", lang="zh", person_id=asker,
    )
    await db_session.commit()

    assert res["covered"] is False
    assert res["answer"] is None
    assert len(res["items"]) == 1  # the near miss is still worth reading
    gap = (await db_session.execute(select(AuditorSecurityFaqGap))).scalar_one()
    assert gap.ask_count == 1
    assert gap.last_asker_person_id == asker


async def test_a_composition_outage_still_answers_with_the_entries(
    db_session, monkeypatch,
):
    await _entry(
        db_session, question_zh="能用生成式 AI 吗?", question_en="Generative AI?",
    )

    class _Broken:
        async def chat_json(self, **_kwargs):
            raise RuntimeError("provider 500")

    async def _client(_db, **_kwargs):
        return _Broken()

    monkeypatch.setattr(faq, "build_llm_client", _client)
    res = await faq.ask(
        db_session, query="生成式 AI", lang="zh", person_id=uuid.uuid4(),
    )
    await db_session.commit()

    assert res["answer"] is None
    assert res["covered"] is True  # unknown is not "we failed you"
    assert len(res["items"]) == 1
    assert (await db_session.execute(select(AuditorSecurityFaqGap))).first() is None


async def test_composition_is_skipped_entirely_when_asked_to_skip_it(
    db_session, monkeypatch,
):
    """``compose=False`` is what the MCP twin passes: the caller IS a model
    and will write its own answer, so a second model call is pure latency."""
    await _entry(
        db_session, question_zh="能用生成式 AI 吗?", question_en="Generative AI?",
    )

    # Counted, not raised: ``_compose`` swallows exceptions by design, so a
    # raising double proves nothing here (see the composition-floor test).
    calls: list = []

    async def _client(_db, **_kwargs):
        calls.append(1)
        raise RuntimeError("should not get here")

    monkeypatch.setattr(faq, "build_llm_client", _client)
    res = await faq.ask(
        db_session, query="生成式 AI", lang="zh", person_id=uuid.uuid4(),
        compose=False,
    )
    assert calls == []
    assert res["answer"] is None
    assert len(res["items"]) == 1


async def test_record_gaps_off_leaves_the_ledger_alone(db_session, monkeypatch):
    """The MCP twin is read-only. The gap ledger means "a person asked and we
    had nothing" — an agent probing the library is not that signal."""
    await _entry(db_session, question_zh="无关", question_en="Unrelated")

    async def _no_key(_db):
        raise RuntimeError("no llm")

    monkeypatch.setattr(faq, "build_llm_client", _no_key)
    res = await faq.ask(
        db_session, query="完全没有的问题", lang="zh", person_id=uuid.uuid4(),
        record_gaps=False,
    )
    await db_session.commit()
    assert res["matched_by"] == "none"
    assert (await db_session.execute(select(AuditorSecurityFaqGap))).first() is None


# ── The index ─────────────────────────────────────────────────────────


async def test_backfill_embeds_what_is_missing_and_reports_the_backlog(
    db_session, monkeypatch,
):
    for i in range(3):
        await _entry(
            db_session, question_zh=f"问题{i}", question_en=f"Question {i}",
        )
    assert await rag.count_pending(db_session) == 3

    monkeypatch.setattr(rag, "embed_texts", _embedder({"问题": _LAPTOP}))
    result = await rag.backfill(db_session, limit=2)
    assert result == {"candidates": 2, "embedded": 2, "pending": 1}

    drained = await rag.drain(db_session)
    assert drained["pending"] == 0
    assert await rag.count_pending(db_session) == 0


async def test_a_dead_embeddings_backend_is_a_clean_no_op(db_session, monkeypatch):
    await _entry(db_session, question_zh="问题", question_en="Question")

    async def _dead(_db, _texts, *, timeout=60.0):
        return None

    monkeypatch.setattr(rag, "embed_texts", _dead)
    assert await rag.backfill(db_session) == {
        "candidates": 1, "embedded": 0, "pending": 1,
    }
    # And drain does not spin against the same wall.
    assert (await rag.drain(db_session))["rounds"] == 1


async def test_a_model_switch_marks_the_whole_library_pending(
    db_session, monkeypatch,
):
    """A vector produced by another model cannot be compared with a query
    embedded by this one. Silently keeping it would be a retrieval that
    ranks by noise and reports full coverage while doing it."""
    await _entry(db_session, question_zh="问题", question_en="Question")
    monkeypatch.setattr(rag, "embed_texts", _embedder({"问题": _LAPTOP}))
    await rag.backfill(db_session)
    assert await rag.count_pending(db_session) == 0

    async def _new_model(_db):
        return "text-embedding-v9"

    monkeypatch.setattr(rag, "resolve_embed_model", _new_model)
    assert await rag.count_pending(db_session) == 1


async def test_editing_an_entry_invalidates_its_vector(db_session, monkeypatch):
    """The vector represents the text it was computed from. An edited answer
    with its old vector is a row that answers the question it used to be."""
    entry = await _entry(db_session, question_zh="问题", question_en="Question")
    monkeypatch.setattr(rag, "embed_texts", _embedder({"问题": _LAPTOP}))
    await rag.backfill(db_session)
    assert (await db_session.get(AuditorAnswerLibraryEntry, entry.id)).embedding

    await lib.update_entry(
        db_session, entry.id, payload={"answer_zh": "完全不同的答案"},
    )
    await db_session.commit()
    refreshed = await db_session.get(AuditorAnswerLibraryEntry, entry.id)
    assert refreshed.embedding is None
    assert await rag.count_pending(db_session) == 1


async def test_a_metadata_only_edit_keeps_the_vector(db_session, monkeypatch):
    """Pinning an entry does not change what it says. Re-embedding on every
    metadata touch would pay the provider to compute the same vector."""
    entry = await _entry(db_session, question_zh="问题", question_en="Question")
    monkeypatch.setattr(rag, "embed_texts", _embedder({"问题": _LAPTOP}))
    await rag.backfill(db_session)

    await lib.update_entry(db_session, entry.id, payload={"is_featured": True})
    await db_session.commit()
    assert (await db_session.get(AuditorAnswerLibraryEntry, entry.id)).embedding


async def test_index_stats_count_only_what_employees_can_reach(
    db_session, monkeypatch,
):
    await _entry(db_session, question_zh="可见", question_en="Visible")
    await _entry(
        db_session, question_zh="隐藏", question_en="Hidden", visible=False,
    )
    monkeypatch.setattr(rag, "embed_texts", _embedder({"可见": _LAPTOP}))
    await rag.backfill(db_session)

    stats = await rag.index_stats(db_session)
    assert stats["visible"] == 1
    assert stats["indexed"] == 1
    assert stats["pending"] == 0
    assert stats["model"] == _ACTIVE_MODEL


# ── HTTP ──────────────────────────────────────────────────────────────


async def test_the_index_console_is_closed_to_employees(
    client: AsyncClient, employee_user, auditor_user,
):
    _emp, emp_token = employee_user
    _aud, aud_token = auditor_user

    refused = await client.get(
        "/api/v1/auditor/security-faq/index",
        headers={"Authorization": f"Bearer {emp_token}"},
    )
    assert refused.status_code == 403

    allowed = await client.get(
        "/api/v1/auditor/security-faq/index",
        headers={"Authorization": f"Bearer {aud_token}"},
    )
    assert allowed.status_code == 200
    assert "pending" in allowed.json()["data"]


async def test_the_employee_overview_reports_index_coverage(
    client: AsyncClient, auditor_user, db_session,
):
    """An entry with no vector is keyword-only, and that shortfall raises no
    error anywhere — employees just stop finding answers we have. It has to
    be visible on the console that owns the library."""
    _aud, token = auditor_user
    await _entry(db_session, question_zh="可见", question_en="Visible")

    res = await client.get(
        "/api/v1/auditor/security-faq/overview",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert res.status_code == 200
    index = res.json()["data"]["index"]
    assert index["visible"] == 1
    assert index["pending"] == 1


async def test_the_ask_path_bounds_its_own_model_call(db_session):
    """A timeout that is merely defined is not a timeout.

    Composition now runs on EVERY hit, synchronously, with a person watching
    a spinner — so the provider's own 90s default is the wrong bound (the
    browser gives up first, and the employee is told it failed over work the
    server was still doing). The bound has to reach the client that makes the
    call, which is a wire this test holds: ``build_llm_client`` is shared with
    the questionnaire drafting path, where 90s is still correct.
    """
    from app.domains.auditor.inbound_ai import build_llm_client
    from app.integrations.models import PlatformSetting

    db_session.add(PlatformSetting(key="ai_spending.llm.api_key", value="sk-test"))
    await db_session.flush()

    bounded = await build_llm_client(db_session, timeout=rag.COMPOSE_TIMEOUT)
    default = await build_llm_client(db_session)
    assert bounded._timeout == rag.COMPOSE_TIMEOUT
    assert default._timeout != rag.COMPOSE_TIMEOUT  # the drafting path is unchanged
    # And the bound is inside the client budget the frontend gives this call
    # (LLM_SYNC_TIMEOUT_MS = 180s) with the query-embedding hop on top.
    assert rag.COMPOSE_TIMEOUT + 15.0 < 180.0


async def test_a_dead_end_tells_the_asker_how_many_people_asked_it(
    db_session, monkeypatch,
):
    """「没有找到」 is a shrug. 「你是第 3 个问这句话的人,队列按次数排」 is an
    answer about the system's behaviour, and it is true — ``list_gaps`` orders
    by ``ask_count``. The count is an aggregate over one normalised question,
    so it names nobody."""
    await _entry(db_session, question_zh="无关", question_en="Unrelated")

    async def _no_key(_db, **_kwargs):
        raise RuntimeError("no llm")

    monkeypatch.setattr(faq, "build_llm_client", _no_key)
    counts = []
    for _ in range(3):
        res = await faq.ask(
            db_session, query="  门禁卡   丢了 ", lang="zh",
            person_id=uuid.uuid4(),
        )
        await db_session.commit()
        counts.append(res["ask_count"])

    # The normalised question is one ledger row, and each asker is told the
    # stored count — not the one their own session inserted.
    assert counts == [1, 2, 3]


async def test_the_not_covered_dead_end_carries_the_same_count(
    db_session, monkeypatch,
):
    await _entry(
        db_session, question_zh="能用生成式 AI 吗?", question_en="Generative AI?",
    )
    composer = _FakeComposer({"covered": False, "answer": "", "cited_ids": []})

    async def _client(_db, **_kwargs):
        return composer

    monkeypatch.setattr(faq, "build_llm_client", _client)
    first = await faq.ask(
        db_session, query="生成式 AI 的采购预算", lang="zh", person_id=uuid.uuid4(),
    )
    await db_session.commit()
    second = await faq.ask(
        db_session, query="生成式 AI 的采购预算", lang="zh", person_id=uuid.uuid4(),
    )
    await db_session.commit()
    assert (first["ask_count"], second["ask_count"]) == (1, 2)


async def test_a_hit_reports_no_ask_count(db_session, monkeypatch):
    """Nothing was recorded, so there is nothing to report — and a count next
    to a good answer would read as "this answer is popular", which is a
    different claim entirely."""
    await _entry(
        db_session, question_zh="能用生成式 AI 吗?", question_en="Generative AI?",
    )

    async def _no_key(_db, **_kwargs):
        raise RuntimeError("no llm")

    monkeypatch.setattr(faq, "build_llm_client", _no_key)
    res = await faq.ask(
        db_session, query="生成式 AI", lang="zh", person_id=uuid.uuid4(),
    )
    assert res["ask_count"] is None


# ── Confidence, the composition floor, and the rerank ─────────────────
#
# Retrieval already knows how sure it is; the two things done with that are
# telling the reader and NOT asking the model to summarise material the
# retrieval itself rates weak. Both are behaviour, not decoration.

#: Unit vectors at a chosen angle from _LAPTOP, so a test can put an entry in a
#: specific confidence band instead of at cosine 1.0 or 0.0.
_COS_045 = [0.45, 0.893, 0.0]   # retrievable (>= 0.4) but low confidence
_COS_055 = [0.55, 0.835, 0.0]   # medium band


async def test_two_channels_agreeing_outranks_either_channels_own_score():
    """The lexical matcher and the embedding model share no mechanism, so both
    surfacing the same row is not a coincidence either can manufacture."""
    assert rag.confidence_tier(
        channels=["keyword", "semantic"], cosine=0.41, lexical_rank=9,
    ) == rag.TIER_HIGH


async def test_confidence_bands_follow_the_cosine():
    high = rag.confidence_tier(channels=["semantic"], cosine=0.7, lexical_rank=None)
    medium = rag.confidence_tier(channels=["semantic"], cosine=0.55, lexical_rank=None)
    low = rag.confidence_tier(channels=["semantic"], cosine=0.45, lexical_rank=None)
    assert (high, medium, low) == (rag.TIER_HIGH, rag.TIER_MEDIUM, rag.TIER_LOW)


async def test_a_deep_lexical_only_hit_is_not_confident():
    """A term-overlap coincidence far down the list is the exact failure this
    feature exists to fix — it must not read as a confident answer."""
    near = rag.confidence_tier(channels=["keyword"], cosine=None, lexical_rank=2)
    deep = rag.confidence_tier(channels=["keyword"], cosine=None, lexical_rank=7)
    assert (near, deep) == (rag.TIER_MEDIUM, rag.TIER_LOW)


async def test_weak_retrieval_is_never_handed_to_the_model(
    db_session, monkeypatch,
):
    """The composition floor.

    A model asked to summarise material retrieval rates weak is exactly where
    it starts stretching, and a stretched sentence on this page is read as
    company policy. Below the floor the reader still gets the entries — with
    their confidence shown — and judges for themselves.
    """
    await _entry(
        db_session, question_zh="办公室门禁如何管理?", question_en="Badge access?",
    )
    await _index(db_session, monkeypatch, {"门禁": _LAPTOP})

    # A raise would NOT prove this: ``_compose`` catches every exception as
    # part of its degradation contract, so an AssertionError thrown here is
    # swallowed and the test passes whether the floor exists or not (verified
    # by mutation — removing the floor left the raising version green). Count
    # the calls instead.
    calls: list = []

    async def _client(_db, **_kwargs):
        calls.append(1)
        raise RuntimeError("should not get here")

    monkeypatch.setattr(faq, "build_llm_client", _client)
    monkeypatch.setattr(rag, "embed_texts", _embedder({"__q__": _COS_045}))
    res = await faq.ask(
        db_session, query="__q__", lang="zh", person_id=uuid.uuid4(),
    )

    assert res["retrieval"]["best_confidence"] == rag.TIER_LOW
    assert calls == []          # the floor held: no model call was made at all
    assert res["answer"] is None
    # The entries are still the best we have, and they still render.
    assert len(res["items"]) == 1


async def test_a_medium_hit_still_gets_a_lead_answer(db_session, monkeypatch):
    """The floor is a floor, not a ban: 0.55 is a real match and the reader
    deserves the sentence that saves them reading four paragraphs."""
    entry = await _entry(
        db_session, question_zh="门禁如何管理?", question_en="Badge access?",
    )
    await _index(db_session, monkeypatch, {"门禁": _LAPTOP})

    composer = _FakeComposer({
        "covered": True, "answer": "刷卡进入,访客登记。",
        "cited_ids": [str(entry.id)], "relevant_ids": [str(entry.id)],
    })

    async def _client(_db, **_kwargs):
        return composer

    monkeypatch.setattr(faq, "build_llm_client", _client)
    monkeypatch.setattr(rag, "embed_texts", _embedder({"__q__": _COS_055}))
    res = await faq.ask(
        db_session, query="__q__", lang="zh", person_id=uuid.uuid4(),
    )
    assert res["retrieval"]["best_confidence"] == rag.TIER_MEDIUM
    assert res["answer"]["text"] == "刷卡进入,访客登记。"


async def test_the_model_drops_the_neighbours_that_are_not_on_topic(
    db_session, monkeypatch,
):
    """RRF fuses two similarity rankings and cannot tell "closest" from
    "correct", so its tail is reliably plausible-looking noise. The same call
    that writes the lead has read every row — filtering there is free."""
    on_topic = await _entry(
        db_session, question_zh="能用生成式 AI 吗?", question_en="Generative AI?",
    )
    noise = await _entry(
        db_session, question_zh="生成式设计工具如何采购?", question_en="Design tools?",
    )
    composer = _FakeComposer({
        "covered": True,
        "answer": "公开内容可以,敏感信息不行。",
        "cited_ids": [str(on_topic.id)],
        "relevant_ids": [str(on_topic.id)],
    })

    async def _client(_db, **_kwargs):
        return composer

    monkeypatch.setattr(faq, "build_llm_client", _client)
    res = await faq.ask(
        db_session, query="生成式 AI", lang="zh", person_id=uuid.uuid4(),
    )
    assert [i["id"] for i in res["items"]] == [str(on_topic.id)]
    assert str(noise.id) not in {i["id"] for i in res["items"]}
    assert res["retrieval"]["reranked"] is True


async def test_a_rerank_that_would_empty_the_list_is_ignored(
    db_session, monkeypatch,
):
    """One model call must not be able to turn a hit into a dead end."""
    await _entry(
        db_session, question_zh="能用生成式 AI 吗?", question_en="Generative AI?",
    )
    composer = _FakeComposer({
        "covered": True, "answer": "可以。",
        "cited_ids": [], "relevant_ids": [str(uuid.uuid4())],
    })

    async def _client(_db, **_kwargs):
        return composer

    monkeypatch.setattr(faq, "build_llm_client", _client)
    res = await faq.ask(
        db_session, query="生成式 AI", lang="zh", person_id=uuid.uuid4(),
    )
    # Every id it named was invented, so the rerank is discarded…
    assert len(res["items"]) == 1
    # …and the prose went with it: nothing valid was cited.
    assert res["answer"] is None


async def test_a_cited_entry_survives_a_rerank_that_forgot_it(
    db_session, monkeypatch,
):
    """An answer whose own source was filtered out from under it would render
    a citation chip pointing at a row that is no longer on the page."""
    cited = await _entry(
        db_session, question_zh="能用生成式 AI 吗?", question_en="Generative AI?",
    )
    other = await _entry(
        db_session, question_zh="生成式 AI 工具怎么申请?", question_en="Request AI tools?",
    )
    composer = _FakeComposer({
        "covered": True, "answer": "公开内容可以。",
        "cited_ids": [str(cited.id)],
        "relevant_ids": [str(other.id)],  # forgot the one it cited
    })

    async def _client(_db, **_kwargs):
        return composer

    monkeypatch.setattr(faq, "build_llm_client", _client)
    res = await faq.ask(
        db_session, query="生成式 AI", lang="zh", person_id=uuid.uuid4(),
    )
    shown = {i["id"] for i in res["items"]}
    assert str(cited.id) in shown
    assert res["answer"]["cited_ids"] == [str(cited.id)]


async def test_confidence_reaches_the_caller_per_entry(db_session, monkeypatch):
    entry = await _entry(
        db_session, question_zh="钓鱼邮件如何识别?", question_en="Spot phishing?",
    )
    await _index(db_session, monkeypatch, {"钓鱼": _PHISHING})
    monkeypatch.setattr(rag, "embed_texts", _embedder({"钓鱼": _PHISHING}))

    res = await faq.ask(
        db_session, query="钓鱼邮件", lang="zh", person_id=uuid.uuid4(),
        compose=False,
    )
    # Both channels found it — the strongest signal retrieval has.
    assert res["retrieval"]["confidence"][str(entry.id)] == rag.TIER_HIGH
    assert res["retrieval"]["best_confidence"] == rag.TIER_HIGH