"""安全 FAQ — the employee read surface over the security answer library.

The claims worth pinning are the ones that would be quietly wrong in
production and never raise: an entry the security team retracted staying
reachable, governance fields leaking into a payload 500 people read, a
category filter whose options come from the filtered result, an admin's
"dismissed" verdict being undone by the next person to ask the same
question, and a keyword search that only looks at the display language.

Schema comes from ``create_all`` (no migrations), so every test builds
its own rows. The Bailian call is monkeypatched — no network, no key.
"""
from __future__ import annotations

import uuid
from datetime import timedelta

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_service as faq
from app.domains.auditor.models import (
    AuditorAnswerLibraryEntry,
    AuditorSecurityFaqGap,
)

pytestmark = pytest.mark.asyncio


async def _entry(
    db, *, question_zh: str, question_en: str,
    answer_zh: str = "答案", answer_en: str = "Answer",
    category: str = "general", visible: bool = True,
    featured: bool = False, status: str = "active",
) -> 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, "is_featured": featured,
        },
        actor_id=None,
    )
    if status != "active":
        e.status = status
        await db.flush()
    await db.commit()
    return e


# ── Visibility ────────────────────────────────────────────────────────


async def test_only_active_and_visible_entries_reach_an_employee(db_session):
    shown = await _entry(
        db_session, question_zh="能用个人电脑吗?", question_en="Personal laptop?",
    )
    hidden = await _entry(
        db_session, question_zh="内部架构?", question_en="Internal topology?",
        visible=False,
    )
    archived = await _entry(
        db_session, question_zh="旧策略?", question_en="Old policy?",
        status="archived",
    )

    res = await faq.list_faq(db_session, lang="zh", page_size=50)
    ids = {i["id"] for i in res["items"]}
    assert str(shown.id) in ids
    assert str(hidden.id) not in ids
    assert str(archived.id) not in ids
    assert res["meta"]["total"] == 1


async def test_a_retracted_entry_is_not_found_by_id(db_session):
    """A deep link an employee already holds must stop working when the
    security team retracts the answer — not merely stop being listed."""
    hidden = await _entry(
        db_session, question_zh="隐藏", question_en="Hidden", visible=False,
    )
    from app.core.exceptions import NotFoundError

    with pytest.raises(NotFoundError):
        await faq.get_visible_entry(db_session, hidden.id)


async def test_backfilled_entries_land_hidden(db_session):
    """Questionnaire backfill copies one customer's prose into both
    language slots. It is raw material, not a page 500 people read."""
    e = await lib.backfill_from_answer(
        db_session, question_text="Do you have MFA?",
        answer_text="Yes, enforced for tenant X.",
        category="access_control", actor_id=None,
    )
    await db_session.commit()
    assert e.employee_visible is False
    res = await faq.list_faq(db_session, lang="zh")
    assert str(e.id) not in {i["id"] for i in res["items"]}


# ── Field narrowing ───────────────────────────────────────────────────


async def test_employee_payload_omits_every_governance_field(db_session):
    e = await _entry(
        db_session, question_zh="问题", question_en="Question",
        category="endpoint_security",
    )
    e.source_doc_ids = ["doc-1"]
    e.backing_control_keys = ["ctl-1"]
    e.usage_count = 7
    await db_session.commit()

    view = faq.faq_view(e, lang="zh")
    for leaked in (
        "owner", "owner_person_id", "usage_count", "is_stale", "stale_reason",
        "review_interval_months", "source_doc_ids", "backing_control_keys",
        "last_reviewed_at", "status", "employee_visible",
    ):
        assert leaked not in view, f"{leaked} must not reach an employee"
    assert view["question"] == "问题"
    assert view["category_label"] == "终端安全"


async def test_language_selection_and_ja_falls_back_to_english(db_session):
    e = await _entry(
        db_session, question_zh="中文问题", question_en="English question",
        answer_zh="中文答案", answer_en="English answer",
    )
    assert faq.faq_view(e, lang="zh")["question"] == "中文问题"
    assert faq.faq_view(e, lang="en")["answer"] == "English answer"
    assert faq.resolve_lang("ja-JP") == "en"
    assert faq.resolve_lang("zh-CN") == "zh"
    assert faq.resolve_lang(None) == "zh"
    assert faq.resolve_lang("xx") == "zh"


async def test_uncatalogued_category_is_humanised_not_shown_raw(db_session):
    assert faq.category_label("shadow_it_governance", "zh") == "Shadow It Governance"
    assert faq.category_label("access_control", "en") == "Access Control"


async def test_every_category_the_library_actually_uses_has_a_label(db_session):
    """The catalogue is checked against the seeded inventory, not against
    what a plausible security taxonomy might contain.

    Written after the first cut shipped ``bcp`` / ``vulnerability_mgmt`` /
    ``privacy`` / ``compliance`` — four keys nothing uses — while seven
    real ones (``bcp_dr``, ``data_privacy``, ``identity_lifecycle``,
    ``governance``, ``risk_management``, ``certification``,
    ``vulnerability_management``) fell through to the English humaniser and
    would have rendered untranslated on a Chinese page.
    """
    seeded = {
        "access_control", "ai_governance", "application_security", "bcp_dr",
        "certification", "data_loss_prevention", "data_privacy",
        "email_security", "encryption", "endpoint_security", "governance",
        "hr_security", "identity_lifecycle", "incident_response",
        "logging_monitoring", "network_security", "physical_security",
        "risk_management", "supply_chain", "vulnerability_management",
    }
    assert seeded <= set(faq.CATEGORY_LABELS)
    # And no speculative keys: every catalogued key is one the library uses
    # (plus ``general``, the admin form's default for a new entry).
    assert set(faq.CATEGORY_LABELS) - seeded == {"general"}
    for key, labels in faq.CATEGORY_LABELS.items():
        assert set(labels) == {"zh", "en", "ja"}, key
        assert all(v.strip() for v in labels.values()), key


# ── Search ────────────────────────────────────────────────────────────


async def test_search_spans_both_languages_regardless_of_ui_language(db_session):
    await _entry(
        db_session, question_zh="是否管控 USB 存储?",
        question_en="Do you control USB storage and printing?",
    )
    # English keyword, Chinese UI — the entry must still be found.
    res = await faq.list_faq(db_session, lang="zh", search="printing")
    assert res["meta"]["total"] == 1
    # And the reverse.
    res = await faq.list_faq(db_session, lang="en", search="管控")
    assert res["meta"]["total"] == 1


async def test_like_metacharacters_are_literal_not_wildcards(db_session):
    """Measured on the real seeded library before the fix: ``search=%``
    returned all 71 entries and ``a%b`` returned 63 — a filter box that
    silently means something else than what was typed. Answers here quote
    percentages ("99.9%") and category keys ("data_loss_prevention"), so
    both metacharacters are ones a reader really types."""
    await _entry(
        db_session, question_zh="可用性承诺是多少?", question_en="What is the SLA?",
        answer_zh="99.9% 可用性。", answer_en="99.9% availability.",
    )
    await _entry(
        db_session, question_zh="数据分级怎么定?", question_en="Data classification?",
        answer_zh="按 data_loss_prevention 策略。", answer_en="Per DLP policy.",
    )

    # A bare metacharacter is a CHARACTER: it finds the one entry whose
    # text contains it, not every entry (which is what a wildcard did).
    assert (await faq.list_faq(db_session, lang="zh", search="%"))["meta"]["total"] == 1
    assert (await faq.list_faq(db_session, lang="zh", search="_"))["meta"]["total"] == 1
    # And a real percentage still finds its entry.
    assert (await faq.list_faq(db_session, lang="zh", search="99.9%"))["meta"]["total"] == 1
    # And an underscore is a character, not "any character".
    assert (await faq.list_faq(
        db_session, lang="zh", search="data_loss",
    ))["meta"]["total"] == 1
    assert (await faq.list_faq(
        db_session, lang="zh", search="dataXloss",
    ))["meta"]["total"] == 0


async def test_the_unhelpful_tally_never_reaches_an_employee(db_session):
    """A running public count of people who marked a COMPANY SECURITY RULE
    unhelpful reads as a verdict on the rule, in front of the next 500
    readers. The button still works; the scoreboard is the security team's."""
    e = await _entry(db_session, question_zh="Q", question_en="Q")
    for _ in range(3):
        await faq.record_vote(
            db_session, entry_id=e.id, person_id=uuid.uuid4(), helpful=False,
        )
    await db_session.commit()
    await db_session.refresh(e)
    assert e.employee_unhelpful_count == 3

    view = faq.faq_view(e, lang="zh")
    assert "unhelpful_count" not in view
    assert view["helpful_count"] == 0
    vote = await faq.record_vote(
        db_session, entry_id=e.id, person_id=uuid.uuid4(), helpful=True,
    )
    assert "unhelpful_count" not in vote
    # The admin console still gets it — that is who the signal is for.
    assert (await faq.employee_overview(db_session))["needs_rewrite"][0][
        "unhelpful"
    ] == 3


async def test_a_question_hit_outranks_an_answer_hit(db_session):
    await _entry(
        db_session, question_zh="离职时设备怎么还?", question_en="Return on exit?",
        answer_zh="交回 IT。", answer_en="Return to IT.",
    )
    await _entry(
        db_session, question_zh="设备保修多久?", question_en="Warranty?",
        answer_zh="离职时不影响保修。", answer_en="Exit does not affect it.",
    )
    res = await faq.list_faq(db_session, lang="zh", search="离职")
    assert res["meta"]["total"] == 2
    assert res["items"][0]["question"] == "离职时设备怎么还?"


# ── Question matching ─────────────────────────────────────────────────


async def test_a_whole_question_still_finds_the_answer(db_session, monkeypatch):
    """The defect this pins, measured against the real seeded library:
    「钓鱼邮件」 appears nowhere in the entry that answers it — the entry
    says 「防钓鱼」 and 「钓鱼识别」 — so plain substring matching returned
    nothing and the question went to the model, or to the gap ledger, for
    an answer we already had. Three of four naturally-phrased questions
    found nothing."""
    await _entry(
        db_session,
        question_zh="邮件安全与防钓鱼有哪些措施?",
        question_en="What email security and anti-phishing controls exist?",
        answer_zh="企业邮箱基于 Google Workspace,启用反垃圾与反钓鱼能力;员工安全意识培训覆盖钓鱼识别与上报。",
        answer_en="Anti-spam and anti-phishing on Google Workspace.",
        category="email_security",
    )

    def _boom(_db):
        raise AssertionError("a question we can answer must not reach the LLM")

    monkeypatch.setattr(faq, "build_llm_client", _boom)
    res = await faq.ask(
        db_session, query="钓鱼邮件", lang="zh", person_id=uuid.uuid4(),
    )
    assert res["matched_by"] == "keyword"
    assert res["items"][0]["question"] == "邮件安全与防钓鱼有哪些措施?"


async def test_terms_are_shingled_for_chinese_and_split_for_latin(db_session):
    assert faq.query_terms("钓鱼邮件") == ["钓鱼", "鱼邮", "邮件"]
    assert faq.query_terms("MFA 是否强制?") == ["mfa", "是否", "否强", "强制"]
    # A lone character matches half the library, so it is not a term.
    assert faq.query_terms("吗") == []
    assert faq.query_terms("？！") == []
    # Bounded — every term costs four LIKEs per row.
    assert len(faq.query_terms("一二三四五六七八九十十一十二十三")) == 8


async def test_one_weak_term_is_not_a_match(db_session, monkeypatch):
    """Requiring two distinct terms is what stops 「公司」 — a word in
    nearly every answer — from returning the whole library."""
    await _entry(
        db_session,
        question_zh="公司数据如何分类分级?", question_en="How is data classified?",
        answer_zh="公司数据分为四级。", answer_en="Four levels.",
    )

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

    monkeypatch.setattr(faq, "build_llm_client", _no_llm)
    res = await faq.ask(
        db_session, query="公司团建怎么报销", lang="zh", person_id=None,
    )
    await db_session.commit()
    assert res["matched_by"] == "none"


async def test_a_single_term_question_still_works(db_session, monkeypatch):
    """「MFA」 is a complete question. One term is the whole query, so one
    match is the whole query matching."""
    await _entry(
        db_session,
        question_zh="管理访问是否强制 MFA?", question_en="Is MFA enforced?",
    )

    def _boom(_db):
        raise AssertionError("must not reach the LLM")

    monkeypatch.setattr(faq, "build_llm_client", _boom)
    res = await faq.ask(db_session, query="MFA", lang="zh", person_id=None)
    assert res["matched_by"] == "keyword"
    assert len(res["items"]) == 1


async def test_the_filter_box_falls_back_to_term_matching(db_session):
    """An empty result is the one thing a reader cannot act on. Substring
    first (a filter should be exact and predictable), term scoring only
    when substring finds nothing at all."""
    await _entry(
        db_session,
        question_zh="邮件安全与防钓鱼有哪些措施?", question_en="Anti-phishing?",
        answer_zh="启用反钓鱼能力。", answer_en="Anti-phishing enabled.",
    )
    # Exact substring — the ordinary path.
    assert (await faq.list_faq(db_session, lang="zh", search="防钓鱼"))["meta"]["total"] == 1
    # Not a substring anywhere, but two of its shingles are.
    assert (await faq.list_faq(db_session, lang="zh", search="钓鱼邮件"))["meta"]["total"] == 1
    # Still nothing for a genuinely unrelated filter.
    assert (await faq.list_faq(db_session, lang="zh", search="报销流程"))["meta"]["total"] == 0


# ── Facets ────────────────────────────────────────────────────────────


async def test_facets_span_the_whole_visible_library(db_session):
    await _entry(
        db_session, question_zh="A", question_en="A", category="email_security",
    )
    await _entry(
        db_session, question_zh="B", question_en="B", category="ai_governance",
    )
    await _entry(
        db_session, question_zh="C", question_en="C", category="ai_governance",
        visible=False,
    )
    featured = await _entry(
        db_session, question_zh="D", question_en="D",
        category="email_security", featured=True,
    )

    data = await faq.facets(db_session, lang="zh")
    by_key = {c["key"]: c for c in data["categories"]}
    # The hidden ai_governance row must not be counted.
    assert by_key["ai_governance"]["count"] == 1
    assert by_key["email_security"]["count"] == 2
    assert by_key["email_security"]["label"] == "邮件安全"
    assert data["total"] == 3
    assert [f["id"] for f in data["featured"]] == [str(featured.id)]


async def test_filtering_by_category_does_not_shrink_the_facets(db_session):
    """The filter options must come from the library, not from the result
    of the filter — otherwise the UI can narrow but never widen."""
    await _entry(db_session, question_zh="A", question_en="A", category="privacy")
    await _entry(db_session, question_zh="B", question_en="B", category="encryption")

    before = await faq.facets(db_session, lang="zh")
    await faq.list_faq(db_session, lang="zh", category="privacy")
    after = await faq.facets(db_session, lang="zh")
    assert before == after
    assert len(after["categories"]) == 2


# ── Engagement ────────────────────────────────────────────────────────


async def test_a_read_is_counted_once_per_person_per_day(db_session):
    e = await _entry(db_session, question_zh="Q", question_en="Q")
    viewer = uuid.uuid4()

    assert await faq.record_view_once(
        db_session, entry_id=e.id, viewer_id=viewer,
    ) is True
    assert await faq.record_view_once(
        db_session, entry_id=e.id, viewer_id=viewer,
    ) is False
    await db_session.commit()
    await db_session.refresh(e)
    assert e.employee_view_count == 1

    # A different reader is a different read.
    assert await faq.record_view_once(
        db_session, entry_id=e.id, viewer_id=uuid.uuid4(),
    ) is True
    await db_session.commit()
    await db_session.refresh(e)
    assert e.employee_view_count == 2


async def test_a_vote_can_be_changed_but_never_stacked(db_session):
    e = await _entry(db_session, question_zh="Q", question_en="Q")
    person = uuid.uuid4()

    await faq.record_vote(db_session, entry_id=e.id, person_id=person, helpful=True)
    await faq.record_vote(db_session, entry_id=e.id, person_id=person, helpful=True)
    await db_session.commit()
    await db_session.refresh(e)
    assert (e.employee_helpful_count, e.employee_unhelpful_count) == (1, 0)

    await faq.record_vote(db_session, entry_id=e.id, person_id=person, helpful=False)
    await db_session.commit()
    await db_session.refresh(e)
    assert (e.employee_helpful_count, e.employee_unhelpful_count) == (0, 1)

    assert await faq.my_votes(
        db_session, person_id=person, entry_ids=[e.id],
    ) == {str(e.id): False}


async def test_a_lost_vote_race_replays_instead_of_500ing(db_session, db_engine):
    """The double-tap path, driven for real rather than reasoned about.

    Two requests for the same (entry, person) race. The loser's INSERT
    hits the unique index; it must recover by re-reading the winner's row
    and returning, not by raising — and the count it hands back must be
    the winner's, which under REPEATABLE READ needs a LOCKING re-read
    (a plain refresh returns this transaction's older snapshot).

    This path had no coverage: every earlier test votes from one session,
    so the ``except IntegrityError`` branch never executed.
    """
    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

    e = await _entry(db_session, question_zh="Q", question_en="Q")
    person = uuid.uuid4()

    # Pin this session's snapshot BEFORE the winner commits — otherwise the
    # loser would start after the fact and never take the race branch.
    await db_session.execute(select(AuditorAnswerLibraryEntry.id))

    factory = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False)
    async with factory() as winner:
        await faq.record_vote(
            winner, entry_id=e.id, person_id=person, helpful=True,
        )
        await winner.commit()

    # The loser replays rather than raising, and reports the winner's count.
    result = await faq.record_vote(
        db_session, entry_id=e.id, person_id=person, helpful=True,
    )
    await db_session.commit()
    assert result["helpful"] is True
    assert result["helpful_count"] == 1

    # And the stored counters were moved exactly once.
    async with factory() as check:
        fresh = await check.get(AuditorAnswerLibraryEntry, e.id)
        assert fresh is not None
        assert (fresh.employee_helpful_count, fresh.employee_unhelpful_count) == (1, 0)


# ── Ask ───────────────────────────────────────────────────────────────


class _FakeLLM:
    def __init__(self, ids):
        self._ids = ids
        self.calls = 0

    async def chat_json(self, *, system, user, max_tokens=3000, temperature=0.3):
        self.calls += 1
        return {"ids": self._ids}


async def test_a_keyword_hit_leaves_no_gap_and_shows_the_stored_text(
    db_session, monkeypatch,
):
    """A hit is a hit, with or without a model.

    This test used to assert that a keyword hit NEVER reached the LLM. That
    was true of the id-picker shape (the model's only job was to salvage a
    zero-hit query) and is deliberately no longer true: the model now writes
    a short lead answer OUT of whatever retrieval found, on every hit. What
    survives from the old contract is the part that actually protects the
    reader — the entries rendered underneath are the stored text, and a hit
    is never recorded as a gap.
    """
    await _entry(
        db_session, question_zh="能用生成式 AI 吗?", question_en="Generative AI?",
        answer_zh="不得投喂公司敏感信息。", answer_en="Do not feed it sensitive data.",
    )

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

    monkeypatch.setattr(faq, "build_llm_client", _no_key)
    res = await faq.ask(
        db_session, query="生成式 AI", lang="zh", person_id=uuid.uuid4(),
    )
    assert res["matched_by"] == "keyword"
    assert len(res["items"]) == 1
    assert res["items"][0]["answer"] == "不得投喂公司敏感信息。"
    assert res["answer"] is None
    assert (await db_session.execute(select(AuditorSecurityFaqGap))).first() is None


async def test_a_miss_with_no_llm_configured_degrades_and_records_a_gap(
    db_session, monkeypatch,
):
    await _entry(db_session, question_zh="无关", question_en="Unrelated")

    from app.core.exceptions import AppException

    async def _no_key(_db):
        raise AppException(
            code="BAILIAN_LLM_NO_KEY", message="no key", status_code=503,
        )

    monkeypatch.setattr(faq, "build_llm_client", _no_key)
    asker = uuid.uuid4()
    res = await faq.ask(
        db_session, query="办公室门禁卡丢了怎么办", lang="zh", person_id=asker,
    )
    await db_session.commit()

    assert res["matched_by"] == "none"
    assert res["ai_available"] is False
    gap = (await db_session.execute(
        select(AuditorSecurityFaqGap)
    )).scalar_one()
    assert gap.ask_count == 1
    assert gap.last_asker_person_id == asker
    assert gap.status == "open"


async def test_repeat_misses_aggregate_onto_one_gap_row(db_session, monkeypatch):
    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)
    for _ in range(3):
        await faq.ask(
            db_session, query="  门禁卡   丢了 ", lang="zh",
            person_id=uuid.uuid4(),
        )
        await db_session.commit()

    gaps = (await db_session.execute(select(AuditorSecurityFaqGap))).scalars().all()
    assert len(gaps) == 1
    assert gaps[0].ask_count == 3


async def test_the_model_may_only_pick_ids_it_was_given(db_session, monkeypatch):
    real = await _entry(
        db_session, question_zh="报销流程怎么走?", question_en="Expense flow?",
        answer_zh="走飞书审批。", answer_en="Via Feishu approval.",
    )
    fake = _FakeLLM([str(uuid.uuid4()), str(real.id)])

    async def _client(_db):
        return fake

    monkeypatch.setattr(faq, "build_llm_client", _client)
    res = await faq.ask(
        db_session, query="zzz nothing matches zzz", lang="zh",
        person_id=uuid.uuid4(),
    )
    await db_session.commit()

    assert res["matched_by"] == "ai"
    # The hallucinated id is dropped; the answer is the STORED text.
    assert [i["id"] for i in res["items"]] == [str(real.id)]
    assert res["items"][0]["answer"] == "走飞书审批。"
    # An AI match is not a gap.
    assert (await db_session.execute(select(AuditorSecurityFaqGap))).first() is None


async def test_an_llm_outage_is_not_an_error_for_the_employee(
    db_session, monkeypatch,
):
    await _entry(db_session, question_zh="无关", question_en="Unrelated")

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

    async def _client(_db):
        return _Broken()

    monkeypatch.setattr(faq, "build_llm_client", _client)
    res = await faq.ask(
        db_session, query="完全没有的问题", lang="zh", person_id=uuid.uuid4(),
    )
    await db_session.commit()
    assert res["matched_by"] == "none"
    # The provider was reachable-in-principle; only the call failed.
    assert res["ai_available"] is True


# ── Gap ledger ────────────────────────────────────────────────────────


async def test_an_admin_verdict_survives_every_later_ask(db_session, monkeypatch):
    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)
    await faq.ask(db_session, query="停车位怎么申请", lang="zh", person_id=None)
    await db_session.commit()
    gap = (await db_session.execute(select(AuditorSecurityFaqGap))).scalar_one()

    actor = uuid.uuid4()
    await faq.resolve_gap(
        db_session, gap.id, status="dismissed", entry_id=None,
        note="不属于安全范围", actor_id=actor,
    )
    # Backdate the verdict: the column is second-precision, and a test
    # that resolves and re-asks inside one second would compare two equal
    # timestamps. Production always has the gap between them.
    gap.resolved_at = gap.resolved_at - timedelta(minutes=5)
    await db_session.commit()

    await faq.ask(db_session, query="停车位怎么申请", lang="zh", person_id=None)
    await db_session.commit()
    await db_session.refresh(gap)

    assert gap.status == "dismissed"
    assert gap.note == "不属于安全范围"
    assert gap.resolved_by_person_id == actor
    assert gap.ask_count == 2
    assert faq.gap_view(gap)["reasked_after_resolution"] is True


async def test_a_gap_cannot_be_answered_by_an_entry_employees_cannot_read(
    db_session, monkeypatch,
):
    hidden = await _entry(
        db_session, question_zh="隐藏", question_en="Hidden", visible=False,
    )

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

    monkeypatch.setattr(faq, "build_llm_client", _no_key)
    await faq.ask(db_session, query="某个没有答案的问题", lang="zh", person_id=None)
    await db_session.commit()
    gap = (await db_session.execute(select(AuditorSecurityFaqGap))).scalar_one()

    from app.core.exceptions import NotFoundError

    with pytest.raises(NotFoundError):
        await faq.resolve_gap(
            db_session, gap.id, status="answered", entry_id=hidden.id,
            note=None, actor_id=uuid.uuid4(),
        )


# ── Admin projection ──────────────────────────────────────────────────


async def test_turning_visibility_off_is_not_dropped_as_falsy(db_session):
    e = await _entry(db_session, question_zh="Q", question_en="Q")
    await lib.update_entry(db_session, e.id, payload={"employee_visible": False})
    await db_session.commit()
    await db_session.refresh(e)
    assert e.employee_visible is False
    res = await faq.list_faq(db_session, lang="zh")
    assert res["meta"]["total"] == 0


async def test_employee_overview_separates_published_from_pending(db_session):
    await _entry(db_session, question_zh="A", question_en="A")
    await _entry(db_session, question_zh="B", question_en="B", visible=False)
    await _entry(
        db_session, question_zh="C", question_en="C", featured=True,
    )

    data = await faq.employee_overview(db_session)
    # Both languages on every brief row — the console renders whichever
    # the operator's UI is in.
    for bucket in ("pending_publish", "top_viewed", "needs_rewrite"):
        for row in data[bucket]:
            assert row["question_zh"] and row["question_en"]
    assert data["visible"] == 2
    assert data["hidden"] == 1
    assert data["featured"] == 1
    assert [p["question_zh"] for p in data["pending_publish"]] == ["B"]
    assert data["gaps"] == {"open": 0, "answered": 0, "dismissed": 0}


async def test_a_published_answer_whose_control_failed_is_surfaced(db_session):
    """The one place the employee face and the platform's own posture can
    disagree: ``auditor.answer_library_freshness`` marks an entry
    ``backing_control_failing`` — the fact it rests on stopped holding —
    while the entry is still being read by 500 people.

    Deliberately NOT auto-hidden: a flapping control would make answers
    vanish from the employee page, and most entries are policy statements
    a red control does not falsify. So the console has to say it.
    """
    from app.domains.auditor.answer_freshness import (
        STALE_CONTROL_FAILING,
        STALE_INTERVAL_ELAPSED,
    )

    broken = await _entry(
        db_session, question_zh="特权账号是否强制 MFA?", question_en="MFA on admins?",
    )
    merely_old = await _entry(
        db_session, question_zh="办公网如何分段?", question_en="Network segmentation?",
    )
    hidden_broken = await _entry(
        db_session, question_zh="隐藏的", question_en="Hidden one", visible=False,
    )
    broken.stale_reason = STALE_CONTROL_FAILING
    # "Nobody has re-read this in a year" is not the same claim and must
    # not raise the same alarm.
    merely_old.stale_reason = STALE_INTERVAL_ELAPSED
    hidden_broken.stale_reason = STALE_CONTROL_FAILING
    await db_session.commit()

    data = await faq.employee_overview(db_session)
    assert [r["id"] for r in data["at_risk"]] == [str(broken.id)]
    # It stays readable — the employee is not left with nothing.
    listed = await faq.list_faq(db_session, lang="zh", page_size=50)
    assert str(broken.id) in {i["id"] for i in listed["items"]}
    # …and the reader is told none of this.
    assert "stale_reason" not in faq.faq_view(broken, lang="zh")


# ── HTTP surface ──────────────────────────────────────────────────────


async def test_employee_can_read_the_faq_over_http(
    client: AsyncClient, employee_user, db_session,
):
    _user, token = employee_user
    e = await _entry(
        db_session, question_zh="邮件里的链接能点吗?", question_en="Click links?",
        category="email_security",
    )
    headers = {"Authorization": f"Bearer {token}"}

    listed = await client.get("/api/v1/me/security-faq", headers=headers)
    assert listed.status_code == 200
    body = listed.json()["data"]
    assert str(e.id) in {i["id"] for i in body["items"]}
    assert body["my_votes"] == {}

    facets = await client.get("/api/v1/me/security-faq/facets", headers=headers)
    assert facets.status_code == 200
    assert any(
        c["key"] == "email_security" for c in facets.json()["data"]["categories"]
    )

    detail = await client.get(f"/api/v1/me/security-faq/{e.id}", headers=headers)
    assert detail.status_code == 200
    assert detail.json()["data"]["my_vote"] is None

    vote = await client.post(
        f"/api/v1/me/security-faq/{e.id}/feedback",
        headers=headers, json={"helpful": True},
    )
    assert vote.status_code == 200
    assert vote.json()["data"]["helpful_count"] == 1


async def test_http_hides_a_retracted_entry_behind_404(
    client: AsyncClient, employee_user, db_session,
):
    _user, token = employee_user
    hidden = await _entry(
        db_session, question_zh="隐藏", question_en="Hidden", visible=False,
    )
    res = await client.get(
        f"/api/v1/me/security-faq/{hidden.id}",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert res.status_code == 404


async def test_the_gap_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/gaps",
        headers={"Authorization": f"Bearer {emp_token}"},
    )
    assert refused.status_code == 403

    allowed = await client.get(
        "/api/v1/auditor/security-faq/gaps",
        headers={"Authorization": f"Bearer {aud_token}"},
    )
    assert allowed.status_code == 200


async def test_ask_is_rate_limited_per_user(
    client: AsyncClient, employee_user, db_session, monkeypatch,
):
    """The tail of the ask path spends the org's LLM budget, so the cap
    has to be wired to the endpoint — not merely defined."""
    from app.core.config import settings

    monkeypatch.setattr(settings, "SECURITY_FAQ_ASK_RATE_LIMIT_MAX", 1)
    _user, token = employee_user
    await _entry(db_session, question_zh="邮件安全", question_en="Email security")
    headers = {"Authorization": f"Bearer {token}"}

    first = await client.post(
        "/api/v1/me/security-faq/ask", headers=headers,
        json={"question": "邮件安全"},
    )
    assert first.status_code == 200
    second = await client.post(
        "/api/v1/me/security-faq/ask", headers=headers,
        json={"question": "邮件安全"},
    )
    assert second.status_code == 429