"""提案队列本身:去重、决定、以及目标消失之后会怎样。

这一批钉住的是**队列的可信度**。一个会重复显示同一个建议的队列没人会打开;
一个接受之后告诉你「目标不存在」的队列会把评审人的注意力花光,而注意力正是
这套东西唯一的成本。
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy import func, select

from app.core.knowledge_loop import service as loop
from app.core.knowledge_loop.constants import (
    CORPUS_SUPPORT_KB,
    KIND_ADD_ALIASES,
    KIND_MERGE,
    KIND_NEW_ARTICLE,
    SOURCE_INGEST_TICKET,
    SOURCE_LINT,
    STATUS_ACCEPTED,
    STATUS_FAILED,
    STATUS_OPEN,
    STATUS_REJECTED,
    STATUS_SUPERSEDED,
)
from app.core.knowledge_loop.models import KnowledgeProposal
from tests.test_knowledge_loop.conftest import make_article, make_person

pytestmark = pytest.mark.asyncio


def _alias_proposal(target_id: str, aliases: list[str], **overrides):
    base = {
        "corpus": CORPUS_SUPPORT_KB,
        "kind": KIND_ADD_ALIASES,
        "title": "补说法",
        "source": SOURCE_LINT,
        "target_ref": target_id,
        "payload": {"aliases": aliases},
    }
    base.update(overrides)
    return base


# ── 去重 ────────────────────────────────────────────────────────────


async def test_the_same_suggestion_is_only_open_once(db_session):
    """🔴 会重复显示同一个建议的队列,没人会打开第二次。"""
    article = await make_article(db_session, title="VPN 使用说明")
    await db_session.commit()

    first = await loop.propose(db_session, **_alias_proposal(str(article.id), ["笔记本"]))
    second = await loop.propose(db_session, **_alias_proposal(str(article.id), ["笔记本"]))
    await db_session.commit()

    assert first is not None
    assert second is None, "同一个建议开了两条"
    total = (
        await db_session.execute(select(func.count()).select_from(KnowledgeProposal))
    ).scalar()
    assert total == 1


async def test_dedup_ignores_who_noticed_it(db_session):
    """同一个建议被 lint 和工单结案各发现一次,仍然只是一个建议。"""
    article = await make_article(db_session, title="VPN 使用说明")
    await db_session.commit()

    await loop.propose(
        db_session,
        **_alias_proposal(str(article.id), ["笔记本"], source=SOURCE_LINT),
    )
    duplicate = await loop.propose(
        db_session,
        **_alias_proposal(
            str(article.id),
            ["笔记本"],
            source=SOURCE_INGEST_TICKET,
            rationale="完全不同的理由",
        ),
    )
    await db_session.commit()
    assert duplicate is None


async def test_dedup_ignores_the_order_of_a_list(db_session):
    """🔴 模型这周给 [a, b]、下周给 [b, a],那是同一个建议。"""
    article = await make_article(db_session, title="VPN 使用说明")
    await db_session.commit()

    await loop.propose(
        db_session, **_alias_proposal(str(article.id), ["笔记本", "手提电脑"]),
    )
    flipped = await loop.propose(
        db_session, **_alias_proposal(str(article.id), ["手提电脑", "笔记本"]),
    )
    await db_session.commit()
    assert flipped is None


async def test_a_genuinely_different_suggestion_is_not_deduplicated(db_session):
    """护栏:去重不能宽到把不同的建议吃掉。"""
    article = await make_article(db_session, title="VPN 使用说明")
    await db_session.commit()

    await loop.propose(db_session, **_alias_proposal(str(article.id), ["笔记本"]))
    other = await loop.propose(db_session, **_alias_proposal(str(article.id), ["翻墙"]))
    await db_session.commit()
    assert other is not None


async def test_a_decided_proposal_stops_blocking_the_same_suggestion(db_session):
    """拒绝之后,同一个建议应当能再被提出来 —— 情况会变。"""
    article = await make_article(db_session, title="VPN 使用说明")
    person = await make_person(db_session)
    await db_session.commit()

    first = await loop.propose(db_session, **_alias_proposal(str(article.id), ["笔记本"]))
    assert first is not None
    await loop.decide(
        db_session, proposal=first, accept=False, actor_person_id=person.id,
    )
    await db_session.commit()

    again = await loop.propose(db_session, **_alias_proposal(str(article.id), ["笔记本"]))
    await db_session.commit()
    assert again is not None, "已决定的提案仍在挡着同一个建议"
    assert first.fingerprint == again.fingerprint
    assert first.open_fingerprint is None


async def test_a_duplicate_does_not_discard_the_batch(db_session):
    """🔴 一次 ingest 提五条,第五条重复不能把前四条一起丢掉。"""
    article = await make_article(db_session, title="VPN 使用说明")
    await db_session.commit()
    await loop.propose(db_session, **_alias_proposal(str(article.id), ["重复的"]))
    await db_session.commit()

    result = await loop.propose_many(
        db_session,
        items=[
            _alias_proposal(str(article.id), ["第一条"]),
            _alias_proposal(str(article.id), ["第二条"]),
            _alias_proposal(str(article.id), ["重复的"]),
            _alias_proposal(str(article.id), ["第四条"]),
        ],
    )
    await db_session.commit()
    assert result["created"] == 3
    assert result["duplicates"] == 1
    total = (
        await db_session.execute(select(func.count()).select_from(KnowledgeProposal))
    ).scalar()
    assert total == 4


async def test_a_batch_is_bounded(db_session):
    """一个配错的导入把三千行变成三千条提案,产出的是没人打开的队列。"""
    from app.core.knowledge_loop.constants import MAX_PROPOSALS_PER_INGEST

    article = await make_article(db_session, title="VPN 使用说明")
    await db_session.commit()
    result = await loop.propose_many(
        db_session,
        items=[
            _alias_proposal(str(article.id), [f"说法 {n}"])
            for n in range(MAX_PROPOSALS_PER_INGEST + 5)
        ],
    )
    await db_session.commit()
    assert result["created"] == MAX_PROPOSALS_PER_INGEST
    assert result["over_limit"] == 5


# ── 决定 ────────────────────────────────────────────────────────────


async def test_accepting_an_alias_proposal_writes_it_through(db_session):
    article = await make_article(db_session, title="VPN 使用说明", aliases="连不上VPN")
    person = await make_person(db_session)
    await db_session.commit()
    # id 先取成普通值:`expire_all()` 之后再读 ORM 属性会触发惰性刷新,
    # 在 async session 上就是 MissingGreenlet。
    article_id = article.id

    proposal = await loop.propose(
        db_session, **_alias_proposal(str(article.id), ["笔记本", "连不上VPN"]),
    )
    result = await loop.decide(
        db_session, proposal=proposal, accept=True, actor_person_id=person.id,
    )
    await db_session.commit()

    assert result["status"] == STATUS_ACCEPTED
    db_session.expire_all()
    from app.domains.support.models import SupportKbArticle

    fresh = await db_session.get(SupportKbArticle, article_id)
    lines = fresh.search_aliases.splitlines()
    # 并集,而且原有的那条不重复。
    assert "笔记本" in lines
    assert lines.count("连不上VPN") == 1
    # 别名是嵌入输入的一部分,不失效向量的话,这次改动想抓的那个词仍然搜不到。
    assert fresh.embedding is None


async def test_rejecting_changes_nothing(db_session):
    article = await make_article(db_session, title="VPN 使用说明", aliases="原有的")
    person = await make_person(db_session)
    await db_session.commit()
    article_id = article.id

    proposal = await loop.propose(
        db_session, **_alias_proposal(str(article.id), ["不该写进去"]),
    )
    result = await loop.decide(
        db_session,
        proposal=proposal,
        accept=False,
        actor_person_id=person.id,
        note="措辞不对",
    )
    await db_session.commit()

    assert result["status"] == STATUS_REJECTED
    # 断言用的字段先读出来,再让整个 session 过期。
    assert proposal.decision_note == "措辞不对"
    db_session.expire_all()
    from app.domains.support.models import SupportKbArticle

    fresh = await db_session.get(SupportKbArticle, article_id)
    assert fresh.search_aliases == "原有的"


async def test_deciding_twice_is_a_no_op(db_session):
    """评审人的点击是可重试的。"""
    article = await make_article(db_session, title="VPN 使用说明")
    person = await make_person(db_session)
    await db_session.commit()
    proposal = await loop.propose(db_session, **_alias_proposal(str(article.id), ["A"]))
    await loop.decide(
        db_session, proposal=proposal, accept=True, actor_person_id=person.id,
    )
    await db_session.commit()
    again = await loop.decide(
        db_session, proposal=proposal, accept=False, actor_person_id=person.id,
    )
    assert again["changed"] is False
    assert proposal.status == STATUS_ACCEPTED


async def test_a_vanished_target_is_superseded_not_rejected(db_session):
    """🔴「没人做过这个决定」和「有人决定不做」是两件事。

    把目标消失记成拒绝,等于在台账上留下一个从来没有人持有过的意见。
    """
    from app.domains.support.models import KB_ARCHIVED, SupportKbArticle

    article = await make_article(db_session, title="VPN 使用说明")
    person = await make_person(db_session)
    await db_session.commit()
    proposal = await loop.propose(db_session, **_alias_proposal(str(article.id), ["A"]))
    await db_session.commit()

    fresh = await db_session.get(SupportKbArticle, article.id)
    fresh.status = KB_ARCHIVED
    await db_session.commit()

    result = await loop.decide(
        db_session, proposal=proposal, accept=True, actor_person_id=person.id,
    )
    await db_session.commit()
    assert result["status"] == STATUS_SUPERSEDED
    assert proposal.open_fingerprint is None


async def test_an_unregistered_corpus_fails_loudly(db_session):
    """🔴 静默地什么都不做,和「接受成功」看起来一模一样。"""
    person = await make_person(db_session)
    await db_session.commit()
    proposal = await loop.propose(
        db_session,
        # 一个**确实**没有处理器的语料名。这里原本写的是 "security_faq" ——
        # 后来那个语料真的接进来了,于是这条用例安静地不再测它想测的东西。
        # 用一个不存在的名字,这条断言就不会再被别的 PR 从下面抽走。
        corpus="a_corpus_with_no_adapter",
        kind=KIND_NEW_ARTICLE,
        title="随便",
        source=SOURCE_LINT,
        payload={"title": "x", "body": "y"},
    )
    result = await loop.decide(
        db_session, proposal=proposal, accept=True, actor_person_id=person.id,
    )
    await db_session.commit()
    assert result["status"] == STATUS_FAILED
    assert "没有为语料" in (proposal.error or "")


async def test_retiring_a_page_supersedes_everything_open_about_it(db_session):
    """🔴 评审人不该读完一段仔细的理由、点了接受,才被告知页面不存在。"""
    keep = await make_article(db_session, title="保留的")
    drop = await make_article(db_session, title="被合并的")
    person = await make_person(db_session)
    await db_session.commit()

    # 一条关于「被合并的」那一条的别名提案,和一条合并提案。
    alias = await loop.propose(db_session, **_alias_proposal(str(drop.id), ["某说法"]))
    alias_id = alias.id
    merge = await loop.propose(
        db_session,
        corpus=CORPUS_SUPPORT_KB,
        kind=KIND_MERGE,
        title="合并",
        source=SOURCE_LINT,
        target_ref=str(keep.id),
        related_ref=str(drop.id),
    )
    await db_session.commit()

    await loop.decide(
        db_session, proposal=merge, accept=True, actor_person_id=person.id,
    )
    await db_session.commit()
    db_session.expire_all()

    fresh_alias = await db_session.get(KnowledgeProposal, alias_id)
    assert fresh_alias.status == STATUS_SUPERSEDED


async def test_queue_summary_counts_only_open(db_session):
    article = await make_article(db_session, title="VPN 使用说明")
    person = await make_person(db_session)
    await db_session.commit()
    a = await loop.propose(db_session, **_alias_proposal(str(article.id), ["A"]))
    await loop.propose(db_session, **_alias_proposal(str(article.id), ["B"]))
    await db_session.commit()
    await loop.decide(
        db_session, proposal=a, accept=False, actor_person_id=person.id,
    )
    await db_session.commit()

    summary = await loop.queue_summary(db_session)
    assert summary["open_total"] == 1
    assert summary["by_corpus"][CORPUS_SUPPORT_KB][KIND_ADD_ALIASES] == 1


async def test_an_unknown_kind_is_refused_at_the_door(db_session):
    with pytest.raises(ValueError, match="unknown proposal kind"):
        await loop.propose(
            db_session,
            corpus=CORPUS_SUPPORT_KB,
            kind="reorganise_everything",
            title="x",
            source=SOURCE_LINT,
        )


async def test_a_proposal_survives_its_target_being_hard_deleted(db_session):
    """target_ref 是字符串而不是外键 —— 硬删目标不能把台账一起删掉。"""
    from app.domains.support.models import SupportKbArticle

    article = await make_article(db_session, title="要被删掉的")
    await db_session.commit()
    proposal = await loop.propose(db_session, **_alias_proposal(str(article.id), ["A"]))
    await db_session.commit()
    proposal_id = proposal.id

    fresh = await db_session.get(SupportKbArticle, article.id)
    await db_session.delete(fresh)
    await db_session.commit()
    db_session.expire_all()

    survived = await db_session.get(KnowledgeProposal, proposal_id)
    assert survived is not None
    assert survived.status == STATUS_OPEN

    person = await make_person(db_session)
    await db_session.commit()
    result = await loop.decide(
        db_session, proposal=survived, accept=True, actor_person_id=person.id,
    )
    assert result["status"] == STATUS_SUPERSEDED


async def test_fingerprints_are_stable_across_whitespace_and_case(db_session):
    left = loop.fingerprint_for(
        corpus=CORPUS_SUPPORT_KB,
        kind=KIND_ADD_ALIASES,
        target_ref="t",
        related_ref=None,
        payload={"aliases": ["  笔记本  电脑 "]},
    )
    right = loop.fingerprint_for(
        corpus=CORPUS_SUPPORT_KB,
        kind=KIND_ADD_ALIASES,
        target_ref="t",
        related_ref=None,
        payload={"aliases": ["笔记本 电脑"]},
    )
    assert left == right


async def test_fingerprints_differ_by_target(db_session):
    """护栏:同一个词补给两条不同的条目,是两个建议。"""
    one = loop.fingerprint_for(
        corpus=CORPUS_SUPPORT_KB, kind=KIND_ADD_ALIASES,
        target_ref=str(uuid.uuid4()), related_ref=None,
        payload={"aliases": ["笔记本"]},
    )
    two = loop.fingerprint_for(
        corpus=CORPUS_SUPPORT_KB, kind=KIND_ADD_ALIASES,
        target_ref=str(uuid.uuid4()), related_ref=None,
        payload={"aliases": ["笔记本"]},
    )
    assert one != two


class TestTheReviewLoopCanActuallyClose:
    """Findings from the post-merge review of the four assistant PRs.

    Each of these was a queue row whose ONLY outcomes were "reject" or "accept
    and get an error" — a review queue that cannot be closed teaches the
    reviewer to stop opening it.
    """

    async def test_a_dead_link_proposal_can_be_accepted(self, db_session):
        """🔴 The lint pass produced a kind the adapter had never been taught.

        Its rationale says 「接受即删除这条关联」 and accepting one raised
        「知识库不支持这种提案:dead_link」 — every dead-link finding was
        un-closeable by construction.
        """
        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

        alive = await make_article(db_session, title="活着的那篇")
        gone = await make_article(db_session, title="被下架的那篇")
        link = await link_articles(
            db_session,
            from_id=alive.id,
            to_id=gone.id,
            kind=LINK_RELATED,
            note=None,
            actor_person_id=None,
        )
        assert link is not None
        link_id = link.id
        alive_id = alive.id
        proposal = await loop.propose(
            db_session,
            corpus=CORPUS_SUPPORT_KB,
            kind=KIND_DEAD_LINK,
            title="指向了一条已下架的条目",
            source=SOURCE_LINT,
            target_ref=str(alive_id),
            related_ref=str(gone.id),
            payload={"link_id": str(link_id), "link_kind": LINK_RELATED},
        )
        assert proposal is not None

        result = await loop.decide(
            db_session, proposal=proposal, accept=True,
            actor_person_id=None, trace_id="t-deadlink",
        )
        await db_session.commit()
        assert result["status"] == "accepted", result.get("error")
        assert await db_session.get(SupportKbArticleLink, link_id) is None

    async def test_a_failed_proposal_stops_blocking_the_same_suggestion(
        self, db_session,
    ):
        """🔴 Nothing anywhere moves a proposal back to ``open``.

        Keeping the fingerprint on a failure meant a transient error silenced
        the finding for good: the next lint pass hit the unique constraint,
        ``propose`` returned None, and it was dropped without a word.
        """
        from app.core.knowledge_loop import service as loop
        from app.core.knowledge_loop.constants import KIND_ADD_ALIASES, SOURCE_LINT

        target = str(uuid.uuid4())
        first = await loop.propose(
            db_session,
            corpus="a_corpus_with_no_handler_at_all",
            kind=KIND_ADD_ALIASES,
            title="补说法",
            source=SOURCE_LINT,
            target_ref=target,
            payload={"aliases": ["笔记本"]},
        )
        assert first is not None
        outcome = await loop.decide(
            db_session, proposal=first, accept=True,
            actor_person_id=None, trace_id="t-fail",
        )
        await db_session.commit()
        assert outcome["status"] == "failed"

        again = await loop.propose(
            db_session,
            corpus="a_corpus_with_no_handler_at_all",
            kind=KIND_ADD_ALIASES,
            title="补说法",
            source=SOURCE_LINT,
            target_ref=target,
            payload={"aliases": ["笔记本"]},
        )
        await db_session.commit()
        assert again is not None, "a failed proposal silenced the finding forever"
        assert again.id != first.id

    async def test_the_score_is_not_part_of_what_a_proposal_asks_for(
        self, db_session,
    ):
        """🔴 The cosine used to ride in ``payload``, which the fingerprint hashes.

        Accepting an alias proposal clears the page's embedding; the re-embed
        shifts the score by a thousandth; the next run filed a SECOND open
        proposal for the same pair. One row per re-embed, forever.
        """
        from app.core.knowledge_loop.service import fingerprint_for

        first = fingerprint_for(
            corpus="support_kb", kind="link", target_ref="a", related_ref="b",
            payload={},
        )
        second = fingerprint_for(
            corpus="support_kb", kind="link", target_ref="a", related_ref="b",
            payload={},
        )
        assert first == second
        # And a payload that DOES differ still separates them — the fingerprint
        # is not simply ignoring the payload.
        assert first != fingerprint_for(
            corpus="support_kb", kind="link", target_ref="a", related_ref="b",
            payload={"cosine": 0.871},
        )

    async def test_a_reviewer_can_supply_the_body_the_machine_refused_to_invent(
        self, db_session,
    ):
        """🔴 ``needs_author`` proposals could only ever be rejected.

        They are filed with ``body: ""`` on purpose — a generated security or
        IT answer is the confident, unsupported prose the queue exists to keep
        out — and accepting one hit 「标题和正文不能为空」 and failed.
        """
        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,
        )
        from app.domains.support.models import SupportKbArticle

        proposal = await loop.propose(
            db_session,
            corpus=CORPUS_SUPPORT_KB,
            kind=KIND_NEW_ARTICLE,
            title="缺一篇文档:怎么连公司 Wi-Fi",
            source=SOURCE_EVAL_FAILURE,
            payload={
                "title": "怎么连公司 Wi-Fi",
                "body": "",
                "category": "other",
                "needs_author": True,
            },
        )
        assert proposal is not None

        result = await loop.decide(
            db_session, proposal=proposal, accept=True,
            actor_person_id=None, trace_id="t-author",
            payload_override={"body": "选择 SSID「PingCAP-Corp」,用工号登录。"},
        )
        await db_session.commit()
        assert result["status"] == "accepted", result.get("error")
        article = await db_session.get(
            SupportKbArticle, uuid.UUID(result["applied_ref"]),
        )
        assert article is not None
        assert "PingCAP-Corp" in article.body
        # A draft, not published: the reviewer wrote it, but publishing is still
        # its own decision.
        assert article.status == "draft"

    async def test_an_override_cannot_smuggle_a_key_outside_the_whitelist(
        self, db_session,
    ):
        """The override fills in a blank; it is not an arbitrary payload channel.

        🔴 REFUSED, not silently dropped — and that is the round-2 correction.
        The first shape allowed "any key the proposal already declares", which
        for a ``dead_link`` proposal means ``link_id``, i.e. the target itself.
        A reviewer who typed a value is also entitled to know it went nowhere,
        rather than watching the accept fail on an unrelated validation error.
        """
        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
        proposal_id = proposal.id
        with pytest.raises(ValidationError):
            await loop.decide(
                db_session, proposal=proposal, accept=True,
                actor_person_id=None, trace_id="t-smuggle",
                payload_override={"body": "插网线。", "evil": "x"},
            )

        from app.core.knowledge_loop.models import KnowledgeProposal

        stored = await db_session.get(KnowledgeProposal, proposal_id)
        assert "evil" not in (stored.payload or {})
        # And the whitelisted key alone still goes through.
        await loop.decide(
            db_session, proposal=stored, accept=True,
            actor_person_id=None, trace_id="t-smuggle-2",
            payload_override={"body": "插网线。"},
        )
        await db_session.commit()
        assert stored.payload["body"] == "插网线。"