"""新知识进来时问「这已经写下来了吗」,以及每周把语料自己读一遍。

模型被替身顶掉 —— 这里测的是**决策的落法**,不是模型的判断力。两条最要紧的:

* **没有模型时必须退回到今天的行为**(一篇新草稿),而不是停止 ingest。一个
  没配 LLM key 的部署不能因为这次改动而悄悄不再吸收知识。
* **模型指了一个不在候选集里的 id,不算一个决定**。那是模型在**命名**一个东西,
  按它去改一条我们从没递给它的条目,是这套东西能造成的最大破坏。
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy import select

from app.core.knowledge_loop.constants import (
    KIND_ADD_ALIASES,
    KIND_APPEND_SECTION,
    KIND_CONTRADICTION,
    KIND_NEW_ARTICLE,
    SOURCE_INGEST_TICKET,
)
from app.core.knowledge_loop.models import KnowledgeProposal
from app.domains.support import kb_wiki
from tests.test_knowledge_loop.conftest import make_article, make_mirror

pytestmark = pytest.mark.asyncio


@pytest.fixture
def model(monkeypatch):
    """Stub the schema-bound JSON call. ``None`` means "backend unavailable"."""
    state: dict = {"replies": [], "calls": []}

    async def _one_shot_json(db, system, prompt, *, timeout=60.0, max_tokens=1200):
        state["calls"].append(prompt)
        if not state["replies"]:
            return None
        # An explicit ``None`` in the list means "unavailable for THIS call" —
        # without it a test can only express "no model at all", and then
        # "stop" and "skip and carry on" are indistinguishable.
        return state["replies"].pop(0)

    import app.domains.support.copilot as copilot

    monkeypatch.setattr(copilot, "one_shot_json", _one_shot_json)
    return state


@pytest.fixture
def no_candidates(monkeypatch):
    """Force the shortlist to be empty without needing an embedding backend."""

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

    monkeypatch.setattr(kb_wiki, "_candidates", _empty)


@pytest.fixture
def candidates(monkeypatch):
    """Pin the shortlist to a fixed list of articles."""
    holder: dict = {"articles": []}

    async def _fixed(db, *, title, body):
        return holder["articles"]

    monkeypatch.setattr(kb_wiki, "_candidates", _fixed)
    return holder


async def _proposals(db) -> list[KnowledgeProposal]:
    return list(
        (await db.execute(select(KnowledgeProposal))).scalars().all()
    )


# ── ingest ──────────────────────────────────────────────────────────


async def test_no_candidates_means_a_new_page(db_session, model, no_candidates):
    outcome = await kb_wiki.propose_from_knowledge(
        db_session,
        title="怎么连打印机",
        body="步骤如下。",
        category="hardware",
        source=SOURCE_INGEST_TICKET,
    )
    await db_session.commit()
    assert outcome.decision == "new"
    rows = await _proposals(db_session)
    assert [row.kind for row in rows] == [KIND_NEW_ARTICLE]
    # 没有候选就不该去问模型 —— 那是白花的一次调用。
    assert model["calls"] == []


async def test_without_a_model_it_falls_back_to_todays_behaviour(
    db_session, model, candidates,
):
    """🔴 没配 LLM key 的部署不能因为这次改动而不再吸收知识。"""
    candidates["articles"] = [await make_article(db_session, title="打印机说明")]
    await db_session.commit()
    model["replies"] = []  # 模型不可用

    outcome = await kb_wiki.propose_from_knowledge(
        db_session,
        title="怎么连打印机",
        body="步骤如下。",
        category="hardware",
        source=SOURCE_INGEST_TICKET,
    )
    await db_session.commit()
    assert outcome.decision == "new"
    assert outcome.available is False
    rows = await _proposals(db_session)
    assert [row.kind for row in rows] == [KIND_NEW_ARTICLE]


async def test_alias_only_proposes_words_not_a_second_page(
    db_session, model, candidates,
):
    """🔴 这条是整个改动的核心:同一件事换个说法,补词,不是再开一页。"""
    article = await make_article(db_session, title="电脑坏了怎么报修")
    candidates["articles"] = [article]
    await db_session.commit()
    model["replies"] = [
        {
            "decision": "alias_only",
            "target_id": str(article.id),
            "aliases": ["笔记本坏了", "笔记本电脑故障"],
            "reason": "同一件事,只是员工用「笔记本」而条目里写「电脑」。",
        }
    ]

    outcome = await kb_wiki.propose_from_knowledge(
        db_session,
        title="笔记本坏了怎么办",
        body="送修流程同上。",
        category="hardware",
        source=SOURCE_INGEST_TICKET,
    )
    await db_session.commit()
    assert outcome.decision == "alias_only"
    rows = await _proposals(db_session)
    assert len(rows) == 1
    assert rows[0].kind == KIND_ADD_ALIASES
    assert rows[0].payload["aliases"] == ["笔记本坏了", "笔记本电脑故障"]
    assert rows[0].target_ref == str(article.id)


async def test_a_duplicate_with_nothing_to_add_files_nothing(
    db_session, model, candidates,
):
    """「已经写下来了,措辞也一样」是一个正当的结局,不是一条空提案。"""
    article = await make_article(db_session, title="电脑坏了怎么报修")
    candidates["articles"] = [article]
    await db_session.commit()
    model["replies"] = [
        {"decision": "duplicate", "target_id": str(article.id), "aliases": []}
    ]
    outcome = await kb_wiki.propose_from_knowledge(
        db_session,
        title="电脑坏了怎么报修",
        body="一样的内容。",
        category="hardware",
        source=SOURCE_INGEST_TICKET,
    )
    await db_session.commit()
    assert outcome.decision == "duplicate"
    assert await _proposals(db_session) == []


async def test_augments_proposes_a_section_on_the_existing_page(
    db_session, model, candidates,
):
    article = await make_article(db_session, title="电脑坏了怎么报修")
    candidates["articles"] = [article]
    await db_session.commit()
    model["replies"] = [
        {
            "decision": "augments",
            "target_id": str(article.id),
            "section_heading": "外接显示器不亮",
            "section_body": "先换一根线。",
            "aliases": ["显示器不亮"],
            "reason": "补了一个原条目没写的场景。",
        }
    ]
    outcome = await kb_wiki.propose_from_knowledge(
        db_session,
        title="显示器不亮怎么办",
        body="先换一根线。",
        category="hardware",
        source=SOURCE_INGEST_TICKET,
    )
    await db_session.commit()
    assert outcome.decision == "augments"
    row = (await _proposals(db_session))[0]
    assert row.kind == KIND_APPEND_SECTION
    assert row.payload["heading"] == "外接显示器不亮"
    assert row.payload["aliases"] == ["显示器不亮"]


async def test_augments_without_a_section_falls_back_to_a_new_page(
    db_session, model, candidates,
):
    """模型说要补充却没给内容 —— 那不是一个可落地的决定。"""
    article = await make_article(db_session, title="电脑坏了怎么报修")
    candidates["articles"] = [article]
    await db_session.commit()
    model["replies"] = [
        {"decision": "augments", "target_id": str(article.id), "section_body": ""}
    ]
    outcome = await kb_wiki.propose_from_knowledge(
        db_session,
        title="显示器不亮怎么办",
        body="先换一根线。",
        category="hardware",
        source=SOURCE_INGEST_TICKET,
    )
    await db_session.commit()
    assert outcome.decision == "new"
    assert (await _proposals(db_session))[0].kind == KIND_NEW_ARTICLE


async def test_an_id_outside_the_shortlist_is_refused(
    db_session, model, candidates,
):
    """🔴 模型指了一个我们没递给它的 id —— 那不是一个决定,是它在命名东西。

    照它去改一条我们从没展示过的条目,是这套东西能造成的最大破坏。
    """
    shown = await make_article(db_session, title="被展示的")
    unshown = await make_article(db_session, title="没被展示的")
    candidates["articles"] = [shown]
    await db_session.commit()
    model["replies"] = [
        {
            "decision": "alias_only",
            "target_id": str(unshown.id),
            "aliases": ["随便"],
        }
    ]
    outcome = await kb_wiki.propose_from_knowledge(
        db_session,
        title="某个问题",
        body="某段正文。",
        category="other",
        source=SOURCE_INGEST_TICKET,
    )
    await db_session.commit()
    assert outcome.decision == "new"
    rows = await _proposals(db_session)
    assert [row.kind for row in rows] == [KIND_NEW_ARTICLE]
    assert all(row.target_ref != str(unshown.id) for row in rows)


async def test_contradiction_carries_the_incoming_text(
    db_session, model, candidates,
):
    article = await make_article(db_session, title="回购说明", body="满 2 年可回购。")
    candidates["articles"] = [article]
    await db_session.commit()
    model["replies"] = [
        {
            "decision": "contradicts",
            "target_id": str(article.id),
            "reason": "一个写满 2 年,一个写 36 个月。",
        }
    ]
    outcome = await kb_wiki.propose_from_knowledge(
        db_session,
        title="电脑回购年限",
        body="折旧年限为 36 个月。",
        category="other",
        source=SOURCE_INGEST_TICKET,
    )
    await db_session.commit()
    assert outcome.decision == "contradicts"
    row = (await _proposals(db_session))[0]
    assert row.kind == KIND_CONTRADICTION
    assert "36 个月" in row.payload["incoming_body"]


async def test_empty_knowledge_is_skipped(db_session, model, no_candidates):
    outcome = await kb_wiki.propose_from_knowledge(
        db_session,
        title="   ",
        body="",
        category="other",
        source=SOURCE_INGEST_TICKET,
    )
    assert outcome.decision == "skipped"
    assert await _proposals(db_session) == []


# ── lint ────────────────────────────────────────────────────────────


async def test_pairs_below_the_floor_are_never_shown_to_the_model(
    db_session, model,
):
    """相似度闸只决定「给不给模型看」,但它得真的挡住东西。"""
    from app.domains.support.kb_lint import LintReport, lint_pairs

    await make_article(db_session, title="A", embedding=[1.0, 0.0])
    await make_article(db_session, title="B", embedding=[0.0, 1.0])
    await db_session.commit()

    report = LintReport()
    await lint_pairs(db_session, report=report)
    assert report.pairs_examined == 0
    assert model["calls"] == []


async def test_a_confirmed_duplicate_becomes_a_merge_proposal(db_session, model):
    from app.domains.support.kb_lint import LintReport, lint_pairs

    keep = await make_article(
        db_session, title="电脑报修", body="较完整", embedding=[1.0, 0.0, 0.0],
    )
    drop = await make_article(
        db_session, title="笔记本报修", body="较短", embedding=[0.99, 0.1, 0.0],
    )
    await db_session.commit()
    model["replies"] = [
        {"relation": "duplicate", "keep_id": str(keep.id), "reason": "同一件事。"}
    ]

    report = LintReport()
    await lint_pairs(db_session, report=report)
    await db_session.commit()
    assert report.duplicates == 1
    row = (await _proposals(db_session))[0]
    assert row.target_ref == str(keep.id)
    assert row.related_ref == str(drop.id)


async def test_unrelated_is_the_default_and_files_nothing(db_session, model):
    """🔴 向量分不清「同一件事」和「同一件事的反面」。"""
    await make_article(db_session, title="怎么连 VPN", embedding=[1.0, 0.0, 0.0])
    await make_article(db_session, title="VPN 连不上", embedding=[0.99, 0.1, 0.0])
    await db_session.commit()
    model["replies"] = [{"relation": "unrelated", "reason": "一个配置一个排障。"}]

    from app.domains.support.kb_lint import LintReport, lint_pairs

    report = LintReport()
    await lint_pairs(db_session, report=report)
    await db_session.commit()
    assert report.pairs_examined == 1
    assert await _proposals(db_session) == []


async def test_a_pair_that_is_already_linked_is_not_re_examined(db_session, model):
    from app.domains.support.kb_lint import LintReport, lint_pairs

    a = await make_article(db_session, title="A", embedding=[1.0, 0.0, 0.0])
    b = await make_article(db_session, title="B", embedding=[0.99, 0.1, 0.0])
    await db_session.commit()
    await kb_wiki.link_articles(db_session, from_id=a.id, to_id=b.id)
    await db_session.commit()

    report = LintReport()
    await lint_pairs(db_session, report=report)
    assert report.pairs_examined == 0
    assert model["calls"] == []


async def test_mirrored_pages_are_left_out_of_the_pair_pass(db_session, model):
    """镜像条目合并不了,提出来的提案唯一可能的结局是被拒。"""
    from app.domains.support.kb_lint import LintReport, lint_pairs

    await make_article(db_session, title="本地条目", embedding=[1.0, 0.0, 0.0])
    mirror = await make_mirror(db_session, title="镜像条目")
    mirror.embedding = [0.99, 0.1, 0.0]
    mirror.embedding_model = "text-embedding-v3"
    await db_session.commit()

    report = LintReport()
    await lint_pairs(db_session, report=report)
    assert report.pairs_examined == 0


async def test_an_unavailable_model_stops_the_pass_instead_of_skipping(
    db_session, model,
):
    """🔴 模型中途不可用时必须**停下**,而不是跳过这一对继续往下跑。

    这条用例刻意造了**三对**候选。只有一对的话,「return」和「continue」的
    行为完全一样,断言就成了空转 —— 这正是变异验证第一轮抓到的问题:改成
    continue,那一版测试依然全绿。

    三个互相相似的条目产生三对。模型第一次调用返回不可用,第二次返回一个
    「重复」判定。正确行为是第一次就停:一次调用、零条提案。跳过继续跑的话,
    第二对会拿到那个判定并落一条提案 —— 也就是把一个「无分类器」的状态变成
    了队列里的噪声。
    """
    from app.domains.support.kb_lint import LintReport, lint_pairs

    keep = await make_article(db_session, title="A", embedding=[1.0, 0.0, 0.0])
    await make_article(db_session, title="B", embedding=[0.99, 0.1, 0.0])
    await make_article(db_session, title="C", embedding=[0.98, 0.14, 0.0])
    await db_session.commit()
    model["replies"] = [
        None,  # 第一对:模型不可用
        {"relation": "duplicate", "keep_id": str(keep.id), "reason": "同一件事。"},
    ]

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

    assert report.llm_available is False
    assert report.pairs_examined == 3, "前置条件不成立:候选对不足,这条测不到停不停"
    assert len(model["calls"]) == 1, (
        "模型不可用之后仍然继续问下一对 —— 没有分类器就往队列里塞东西"
    )
    assert await _proposals(db_session) == []


async def test_a_dead_link_is_reported_once(db_session, model):
    from app.domains.support.kb_lint import LintReport, lint_dead_links
    from app.domains.support.models import KB_ARCHIVED, SupportKbArticle

    alive = await make_article(db_session, title="还在的")
    gone = await make_article(db_session, title="要下架的")
    await db_session.commit()
    alive_id, gone_id = alive.id, gone.id
    await kb_wiki.link_articles(db_session, from_id=alive_id, to_id=gone_id)
    await db_session.commit()

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

    report = LintReport()
    await lint_dead_links(db_session, report=report)
    await db_session.commit()
    assert report.dead_links == 1
    rows = await _proposals(db_session)
    assert len(rows) == 1
    assert rows[0].target_ref == str(alive_id)


async def test_a_healthy_link_reports_nothing(db_session, model):
    from app.domains.support.kb_lint import LintReport, lint_dead_links

    a = await make_article(db_session, title="A")
    b = await make_article(db_session, title="B")
    await db_session.commit()
    await kb_wiki.link_articles(db_session, from_id=a.id, to_id=b.id)
    await db_session.commit()

    report = LintReport()
    await lint_dead_links(db_session, report=report)
    assert report.dead_links == 0


async def test_eval_failures_become_the_right_kind_of_proposal(db_session, model):
    """🔴 缺文档的那条**刻意不带正文**。

    生成的正文正是这套东西要挡住的那种「读起来很像、其实没依据」的内容。
    """
    from app.assistant.evaluation.models import (
        AssistantEvalCase,
        AssistantEvalResult,
        AssistantEvalRun,
        AssistantEvalSuite,
    )
    from app.domains.support.kb_lint import backfill_from_eval_failures

    suite = AssistantEvalSuite(slug=f"s-{uuid.uuid4().hex[:8]}", name="套件")
    db_session.add(suite)
    await db_session.flush()
    case = AssistantEvalCase(
        suite_id=suite.id, ordinal=0, question="q", lang="zh",
        category="kb_lookup", status="active",
    )
    db_session.add(case)
    await db_session.flush()
    run = AssistantEvalRun(suite_id=suite.id, status="succeeded", repeats=1)
    db_session.add(run)
    await db_session.flush()
    run_id = run.id
    db_session.add(
        AssistantEvalResult(
            run_id=run_id, case_id=case.id, attempt=0,
            question="公司的密码规则是什么", category="kb_lookup", lang="zh",
            verdict="fail", failure_type="kb_doc",
            judge_reasoning="知识库里没有覆盖。",
        )
    )
    await db_session.commit()

    result = await backfill_from_eval_failures(db_session, run_id=run_id)
    await db_session.commit()
    assert result["created"] == 1
    row = (await _proposals(db_session))[0]
    assert row.kind == KIND_NEW_ARTICLE
    assert row.payload["body"] == ""
    assert row.payload["needs_author"] is True


async def test_only_corpus_failures_are_backfilled(db_session, model):
    """数据缺陷、动作缺失、产品不支持 —— 写一篇文档只会把真问题盖住。"""
    from app.assistant.evaluation.models import (
        AssistantEvalCase,
        AssistantEvalResult,
        AssistantEvalRun,
        AssistantEvalSuite,
    )
    from app.domains.support.kb_lint import backfill_from_eval_failures

    suite = AssistantEvalSuite(slug=f"s-{uuid.uuid4().hex[:8]}", name="套件")
    db_session.add(suite)
    await db_session.flush()
    case = AssistantEvalCase(
        suite_id=suite.id, ordinal=0, question="q", lang="zh",
        category="self_data", status="active",
    )
    db_session.add(case)
    await db_session.flush()
    run = AssistantEvalRun(suite_id=suite.id, status="succeeded", repeats=3)
    db_session.add(run)
    await db_session.flush()
    run_id = run.id
    # ``attempt`` 在 (run, case) 内唯一 —— 三条要各自编号,否则这条用例会红在
    # 一个和它想测的东西完全无关的约束上。
    for attempt, failure_type in enumerate(
        ("tool_field_or_data", "action", "out_of_scope"),
    ):
        db_session.add(
            AssistantEvalResult(
                run_id=run_id, case_id=case.id, attempt=attempt,
                question=f"问题 {failure_type}", category="self_data", lang="zh",
                verdict="fail", failure_type=failure_type,
            )
        )
    await db_session.commit()

    result = await backfill_from_eval_failures(db_session, run_id=run_id)
    await db_session.commit()
    assert result["failures_seen"] == 0
    assert result["created"] == 0
    assert await _proposals(db_session) == []