"""Findings from the post-merge review, on the knowledge-loop side.

The theme is the same as the queue's own: a review queue is only worth opening
if a click can close a row, and only safe to click fast if the rows that CANNOT
be closed fast are held back from the fast path.
"""
from __future__ import annotations

import uuid

import pytest

from tests.test_knowledge_loop.conftest import make_article

pytestmark = pytest.mark.asyncio


async def test_bulk_accept_refuses_the_kinds_that_can_make_a_page_wrong(
    client, db_session, admin_user,
):
    """🔴 The rule lived only in the console, one API call from not applying.

    Adding a word to a page cannot make it wrong; archiving one can, and
    splicing text into a live page can. Bulk exists so fifteen alias
    suggestions are one click — not so fifteen pages can be archived unread.
    """
    from app.core.knowledge_loop import service as loop
    from app.core.knowledge_loop.constants import (
        CORPUS_SUPPORT_KB,
        KIND_ADD_ALIASES,
        KIND_RETIRE,
        SOURCE_LINT,
        STATUS_OPEN,
    )
    from app.core.knowledge_loop.models import KnowledgeProposal

    _admin, token = admin_user
    article = await make_article(db_session, title="要留下的那篇")
    safe = await loop.propose(
        db_session, corpus=CORPUS_SUPPORT_KB, kind=KIND_ADD_ALIASES,
        title="补说法", source=SOURCE_LINT, target_ref=str(article.id),
        payload={"aliases": ["笔记本"]},
    )
    risky = await loop.propose(
        db_session, corpus=CORPUS_SUPPORT_KB, kind=KIND_RETIRE,
        title="下架", source=SOURCE_LINT, target_ref=str(article.id), payload={},
    )
    await db_session.commit()

    response = await client.post(
        "/api/v1/knowledge/proposals/bulk-decide",
        json={"proposal_ids": [str(safe.id), str(risky.id)], "accept": True},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 422, response.text

    # 🔴 And nothing was decided — not even the safe one. A batch the server
    # refuses must not half-apply, or the reviewer's next click works from a
    # list that no longer says what it did.
    for proposal_id in (safe.id, risky.id):
        row = await db_session.get(
            KnowledgeProposal, proposal_id, populate_existing=True,
        )
        assert row.status == STATUS_OPEN

    # Rejecting in bulk stays unrestricted: declining to act is never the
    # dangerous direction.
    rejected = await client.post(
        "/api/v1/knowledge/proposals/bulk-decide",
        json={"proposal_ids": [str(risky.id)], "accept": False},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert rejected.status_code == 200, rejected.text


async def test_an_employees_own_words_are_redacted_before_they_can_become_corpus(
    db_session, monkeypatch,
):
    """🔴 A 👍-rated answer is one employee's caller-scoped data.

    The ``new_article`` branch already treats this content as untrusted — it
    files a draft — but ``append_section`` splices it into a live article that
    every employee can read. The evaluation miner redacts at mining time for
    exactly this reason; this path had none.
    """
    from app.assistant.models import (
        ROLE_ASSISTANT,
        ROLE_USER,
        VERDICT_HELPFUL,
        AssistantConversation,
        AssistantMessage,
        AssistantMessageFeedback,
    )
    from app.domains.support import kb_lint
    from tests.test_knowledge_loop.conftest import make_person

    person = await make_person(db_session)
    conversation = AssistantConversation(
        id=uuid.uuid4(), person_id=person.id, channel="feishu", chat_id="oc-x",
    )
    db_session.add(conversation)
    await db_session.flush()
    asked = AssistantMessage(
        id=uuid.uuid4(), conversation_id=conversation.id, seq=0,
        role=ROLE_USER, content="我的邮箱 li.si@pingcap.com 收不到邮件",
    )
    answer = AssistantMessage(
        id=uuid.uuid4(), conversation_id=conversation.id, seq=1,
        role=ROLE_ASSISTANT,
        content="给 li.si@pingcap.com 重发一次即可,他的手机 13800138000 也能收。",
    )
    db_session.add_all([asked, answer])
    await db_session.flush()
    db_session.add(
        AssistantMessageFeedback(
            message_id=answer.id, person_id=person.id, verdict=VERDICT_HELPFUL,
        )
    )
    await db_session.commit()

    seen: dict[str, str] = {}

    from app.domains.support.kb_wiki import IngestOutcome

    async def _capture(db, *, title, body, **kwargs):
        seen["title"] = title
        seen["body"] = body
        seen["evidence"] = str(kwargs.get("evidence"))
        return IngestOutcome("duplicate", [], True, "已覆盖")

    # Patched on the MODULE the backfill imports it from, inside the function —
    # patching the definition site would not be the name the caller resolves.
    monkeypatch.setattr(
        "app.domains.support.kb_wiki.propose_from_knowledge", _capture,
    )
    await kb_lint.backfill_from_answers(db_session, window_days=30, limit=5)

    assert seen, "the backfill considered nothing"
    for field in ("title", "body", "evidence"):
        assert "li.si@pingcap.com" not in seen[field], field
        assert "13800138000" not in seen[field], field
    assert "[邮箱]" in seen["body"]


async def test_a_no_op_still_drains_the_gap_it_was_raised_from(db_session):
    """🔴 Success was reported while the ledger row stayed open forever.

    An early return on "the entry already says this" skipped both the gap
    closure and the audit row — and ``decide`` then marked the proposal
    accepted. The way in is ordinary: a security admin pastes the same phrasing
    on by hand before the reviewer gets to the queue.
    """
    from datetime import UTC, datetime

    from app.core.knowledge_loop import service as loop
    from app.core.knowledge_loop.constants import (
        CORPUS_SECURITY_FAQ,
        KIND_ADD_ALIASES,
        SOURCE_LINT,
    )
    from app.domains.auditor.models import (
        AuditorAnswerLibraryEntry,
        AuditorSecurityFaqGap,
    )

    phrasing = "我能用个人电脑写公司代码吗"
    entry = AuditorAnswerLibraryEntry(
        id=uuid.uuid4(), category="general",
        question_zh="贵公司是否允许员工使用个人设备处理公司数据?",
        question_en="Does your company permit BYOD?",
        answer_zh="满足 MDM 纳管即可。", answer_en="Permitted when MDM-enrolled.",
        status="active", employee_visible=True,
        # Already there — pasted on by hand between propose and review.
        employee_phrasings=phrasing,
    )
    gap = AuditorSecurityFaqGap(
        id=uuid.uuid4(), query_norm=phrasing, query_sample=phrasing,
        ask_count=3, status="open", first_asked_at=datetime.now(UTC),
        last_asked_at=datetime.now(UTC),
    )
    db_session.add_all([entry, gap])
    await db_session.flush()
    entry_id, gap_id = entry.id, gap.id

    proposal = await loop.propose(
        db_session, corpus=CORPUS_SECURITY_FAQ, kind=KIND_ADD_ALIASES,
        title="补员工问法", source=SOURCE_LINT, target_ref=str(entry_id),
        payload={"aliases": [phrasing]},
    )
    assert proposal is not None
    result = await loop.decide(
        db_session, proposal=proposal, accept=True,
        actor_person_id=None, trace_id="t-noop",
    )
    await db_session.commit()

    assert result["status"] == "accepted"
    assert result.get("noop") is True
    closed = await db_session.get(
        AuditorSecurityFaqGap, gap_id, populate_existing=True,
    )
    assert closed.status != "open", (
        "the proposal reported success and left the gap it drains wide open"
    )


async def test_a_mirrored_page_never_gets_a_proposal_that_would_edit_it(
    db_session, monkeypatch,
):
    """🔴 A queue row whose only outcomes are reject, or accept-and-error.

    Feishu is a mirrored page's only editing surface, so ``_guard_editable``
    refuses one at apply time. Producing the proposal anyway spends the
    reviewer's attention — eleven-at-a-time on the bulk path — for a row that
    was never applicable.
    """
    from app.domains.support import kb_wiki
    from app.domains.support.models import KB_SOURCE_FEISHU_DOC

    mirror = await make_article(
        db_session, title="镜像页:如何连接 VPN", source=KB_SOURCE_FEISHU_DOC,
    )
    await db_session.commit()

    async def _candidates(db, *, title, body):
        return [mirror]

    async def _model(db, system, prompt, **kwargs):
        return {
            "decision": "alias_only",
            "target_id": str(mirror.id),
            "aliases": ["连不上内网"],
            "reason": "同一件事,只是说法不同。",
        }

    monkeypatch.setattr(kb_wiki, "_candidates", _candidates)
    monkeypatch.setattr("app.domains.support.copilot.one_shot_json", _model)

    outcome = await kb_wiki.propose_from_knowledge(
        db_session,
        title="内网连不上",
        body="换了网络也不行。",
        category="network",
        source="ingest_ticket",
    )
    await db_session.commit()

    from sqlalchemy import select

    from app.core.knowledge_loop.models import KnowledgeProposal

    kinds = (
        (
            await db_session.execute(
                select(KnowledgeProposal.kind).where(
                    KnowledgeProposal.target_ref == str(mirror.id),
                )
            )
        )
        .scalars()
        .all()
    )
    assert "add_aliases" not in kinds, (
        "filed a proposal to edit a page Helix cannot write to"
    )
    assert outcome.decision == "linked_mirror"
    assert "link" in kinds


# ── round 2: reviewing the fixes themselves ─────────────────────────


async def test_a_mirror_degradation_produces_a_proposal_that_can_be_accepted(
    db_session, monkeypatch,
):
    """🔴 The round-1 fix filed a row that could only ever fail.

    ``KIND_LINK`` with no ``related_ref`` takes the adapter's "file the incoming
    knowledge as a draft and link the pair" path, and with an empty payload that
    calls ``create(body="")`` and dies on 「标题和正文不能为空」 — exactly the
    reject-or-accept-and-error row the degradation was written to prevent. The
    payload has to carry the incoming text.
    """
    from app.core.knowledge_loop import service as loop
    from app.domains.support import kb_wiki
    from app.domains.support.models import KB_SOURCE_FEISHU_DOC

    mirror = await make_article(
        db_session, title="镜像页:VPN 申请", source=KB_SOURCE_FEISHU_DOC,
    )
    await db_session.commit()

    async def _candidates(db, *, title, body):
        return [mirror]

    async def _model(db, system, prompt, **kwargs):
        return {
            "decision": "duplicate",
            "target_id": str(mirror.id),
            "aliases": ["连不上内网"],
            "reason": "同一件事。",
        }

    monkeypatch.setattr(kb_wiki, "_candidates", _candidates)
    monkeypatch.setattr("app.domains.support.copilot.one_shot_json", _model)

    outcome = await kb_wiki.propose_from_knowledge(
        db_session, title="内网连不上", body="换了网络也不行,求助。",
        category="network", source="ingest_ticket",
    )
    await db_session.commit()
    assert outcome.decision == "linked_mirror"
    assert outcome.proposals

    from app.core.knowledge_loop.models import KnowledgeProposal

    proposal = await db_session.get(
        KnowledgeProposal, uuid.UUID(outcome.proposals[0]),
    )
    result = await loop.decide(
        db_session, proposal=proposal, accept=True,
        actor_person_id=None, trace_id="t-mirror",
    )
    await db_session.commit()
    assert result["status"] == "accepted", result.get("error")


async def test_two_items_degrading_onto_one_mirror_stay_two_proposals(
    db_session, monkeypatch,
):
    """🔴 The empty payload also collapsed every degradation to one fingerprint.

    ``fingerprint_for`` hashes corpus+kind+target+related+payload, and with
    ``{}`` and no ``related_ref`` the second item onward was silently dropped
    while the caller was told a proposal had been filed.
    """
    from app.domains.support import kb_wiki
    from app.domains.support.models import KB_SOURCE_FEISHU_DOC

    mirror = await make_article(
        db_session, title="镜像页:设备申请", source=KB_SOURCE_FEISHU_DOC,
    )
    await db_session.commit()

    async def _candidates(db, *, title, body):
        return [mirror]

    async def _model(db, system, prompt, **kwargs):
        return {"decision": "duplicate", "target_id": str(mirror.id), "reason": "同"}

    monkeypatch.setattr(kb_wiki, "_candidates", _candidates)
    monkeypatch.setattr("app.domains.support.copilot.one_shot_json", _model)

    first = await kb_wiki.propose_from_knowledge(
        db_session, title="要一台显示器", body="外接显示器坏了。",
        category="hardware", source="ingest_ticket",
    )
    second = await kb_wiki.propose_from_knowledge(
        db_session, title="要一个键盘", body="键盘的空格键不回弹。",
        category="hardware", source="ingest_ticket",
    )
    await db_session.commit()
    assert first.proposals and second.proposals
    assert first.proposals[0] != second.proposals[0], (
        "two different findings collapsed onto one fingerprint"
    )


async def test_an_override_cannot_name_a_link_outside_the_proposal(db_session):
    """🔴 For ``dead_link`` the declared payload key IS the target.

    "Any key the proposal already declares" was not a sufficient constraint:
    ``link_id`` resolves straight to a row the adapter deletes, so a reviewer
    could name any link in the corpus while the decision audit named the
    legitimately-dead one.
    """
    from app.core.exceptions import ValidationError
    from app.core.knowledge_loop import service as loop
    from app.core.knowledge_loop.constants import (
        CORPUS_SUPPORT_KB,
        KIND_DEAD_LINK,
        SOURCE_LINT,
    )
    from app.domains.support.kb_wiki import link_articles
    from app.domains.support.models import LINK_RELATED, SupportKbArticleLink

    a = await make_article(db_session, title="甲")
    b = await make_article(db_session, title="乙")
    c = await make_article(db_session, title="丙")
    d = await make_article(db_session, title="丁")
    doomed = await link_articles(
        db_session, from_id=a.id, to_id=b.id, kind=LINK_RELATED,
        note=None, actor_person_id=None,
    )
    innocent = await link_articles(
        db_session, from_id=c.id, to_id=d.id, kind=LINK_RELATED,
        note=None, actor_person_id=None,
    )
    assert doomed is not None and innocent is not None
    innocent_id = innocent.id
    # Committed BEFORE the refusal: the assertion below is that the innocent
    # link survives, and an uncommitted setup would be undone by the rollback
    # rather than by anything the code did.
    await db_session.commit()

    proposal = await loop.propose(
        db_session, corpus=CORPUS_SUPPORT_KB, kind=KIND_DEAD_LINK,
        title="指向了一条已下架的条目", source=SOURCE_LINT,
        target_ref=str(a.id), related_ref=str(b.id),
        payload={"link_id": str(doomed.id), "link_kind": LINK_RELATED},
    )
    assert proposal is not None

    with pytest.raises(ValidationError):
        await loop.decide(
            db_session, proposal=proposal, accept=True,
            actor_person_id=None, trace_id="t-smuggle-link",
            payload_override={"link_id": str(innocent_id)},
        )
    await db_session.rollback()
    assert await db_session.get(SupportKbArticleLink, innocent_id) is not None


async def test_a_refused_override_key_is_reported_rather_than_dropped(db_session):
    """A reviewer who typed a value is entitled to know it went nowhere."""
    from app.core.exceptions import ValidationError
    from app.core.knowledge_loop import service as loop
    from app.core.knowledge_loop.constants import (
        CORPUS_SUPPORT_KB,
        KIND_NEW_ARTICLE,
        SOURCE_EVAL_FAILURE,
    )

    proposal = await loop.propose(
        db_session, corpus=CORPUS_SUPPORT_KB, kind=KIND_NEW_ARTICLE,
        title="缺一篇文档:打印机", source=SOURCE_EVAL_FAILURE,
        payload={"title": "打印机", "body": "", "needs_author": True},
    )
    assert proposal is not None
    with pytest.raises(ValidationError):
        await loop.decide(
            db_session, proposal=proposal, accept=True,
            actor_person_id=None, trace_id="t-refused",
            payload_override={"body": "插网线。", "needs_author": False},
        )


async def test_a_merge_onto_a_mirror_is_degraded_by_the_producer_too(db_session):
    """🔴 The producer only checked the side being ARCHIVED.

    A merge writes the survivor as well — its aliases gain the absorbed page's
    and its embedding is cleared — so a merge INTO a mirror is refused at apply
    time just the same.
    """
    from app.domains.support.kb_lint import lint_pairs
    from app.domains.support.models import KB_SOURCE_FEISHU_DOC

    keep = await make_article(
        db_session, title="怎么连接公司 Wi-Fi", source=KB_SOURCE_FEISHU_DOC,
        body="选择 SSID「PingCAP-Corp」,用工号登录。",
        embedding=[1.0, 0.0, 0.0],
    )
    drop = await make_article(
        db_session, title="公司 Wi-Fi 连接指引",
        body="选择 SSID「PingCAP-Corp」,用工号登录。",
        embedding=[1.0, 0.0, 0.0],
    )
    await db_session.commit()
    assert keep.id != drop.id

    from app.domains.support.kb_lint import LintReport

    await lint_pairs(db_session, report=LintReport())
    await db_session.commit()

    from sqlalchemy import select

    from app.core.knowledge_loop.models import KnowledgeProposal

    kinds = (
        (await db_session.execute(select(KnowledgeProposal.kind)))
        .scalars()
        .all()
    )
    assert "merge" not in kinds, (
        "proposed a merge that writes a page Helix cannot write to"
    )