"""题库的播种、挖掘与两次运行的对比。

三件事在这里被钉住,它们各自对应一个具体的坏结果:

* 播种必须幂等且**不覆盖人的编辑** —— 否则没人敢再点第二次;
* 挖掘出来的问句是真实员工的原话,必须先脱敏、且一律以草稿落库;
* 对比要能说出「是哪几条从通过变成了不通过」,而不是只给一个差值 —— 一个
  差值只会开启一次调查,一份名单能结束它。
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy import func, select

from app.assistant.evaluation.constants import (
    CASE_ACTIVE,
    CASE_DRAFT,
    CAT_KB_LOOKUP,
    ORIGIN_EVAL_FAILURE,
    RUN_SUCCEEDED,
    VERDICT_FAIL,
    VERDICT_PASS,
)
from app.assistant.evaluation.models import (
    AssistantEvalCase,
    AssistantEvalResult,
    AssistantEvalRun,
    AssistantEvalSuite,
)

pytestmark = pytest.mark.asyncio


# ── 播种 ────────────────────────────────────────────────────────────


async def test_seeding_creates_the_builtin_suites(db_session):
    from app.assistant.evaluation import service
    from app.assistant.evaluation.seed_data import (
        SUITE_BROAD,
        SUITE_OBJECT_LOOKUP,
        SUITE_REGRESSION,
    )

    result = await service.seed_builtin_suites(db_session)
    await db_session.commit()

    # 三套,不是两套:2026-09-08 加了「对象定位」。它**必须是一个新套件**而不是
    # 往回归套件里追加 —— 追加进已存在的套件会以 draft 落地(见下面那条注释里的
    # 理由),而一条不跑的用例挡不住任何东西。
    assert result["suites_created"] == 3
    assert result["cases_created"] > 60, result

    slugs = {
        row[0]
        for row in (await db_session.execute(select(AssistantEvalSuite.slug))).all()
    }
    assert {SUITE_REGRESSION, SUITE_OBJECT_LOOKUP, SUITE_BROAD} <= slugs

    # 首次播种的用例直接可用 —— 逼一个人去逐条审批 90 条他没写过的题,
    # 是这类东西最后压根没人用的典型原因。
    active = (
        await db_session.execute(
            select(func.count())
            .select_from(AssistantEvalCase)
            .where(AssistantEvalCase.status == CASE_ACTIVE)
        )
    ).scalar()
    assert active == result["cases_created"]


async def test_seeding_twice_creates_nothing_new(db_session):
    """🔴 会重复播的播种,等于没人敢再点第二次。"""
    from app.assistant.evaluation import service

    first = await service.seed_builtin_suites(db_session)
    await db_session.commit()
    second = await service.seed_builtin_suites(db_session)
    await db_session.commit()

    assert second == {"suites_created": 0, "cases_created": 0}
    total = (
        await db_session.execute(select(func.count()).select_from(AssistantEvalCase))
    ).scalar()
    assert total == first["cases_created"]


async def test_seeding_does_not_overwrite_a_human_edit(db_session):
    """人改过的期望必须留着 —— 控制台存在的意义就是让人改它。"""
    from app.assistant.evaluation import service

    await service.seed_builtin_suites(db_session)
    await db_session.commit()

    case = (
        await db_session.execute(
            select(AssistantEvalCase).where(
                AssistantEvalCase.question == "怎么连公司的 VPN"
            ).limit(1)
        )
    ).scalar_one()
    case.expectations = {"notes": "这是人手写的"}
    case.status = "retired"
    case_id = case.id
    await db_session.commit()

    await service.seed_builtin_suites(db_session)
    await db_session.commit()
    db_session.expire_all()

    again = await db_session.get(AssistantEvalCase, case_id)
    assert again.expectations == {"notes": "这是人手写的"}
    assert again.status == "retired"


# ── 挖掘 ────────────────────────────────────────────────────────────


async def test_mined_cases_land_as_drafts_and_are_redacted(db_session):
    from app.assistant.evaluation import service
    from app.assistant.models import (
        CHANNEL_FEISHU,
        ROLE_USER,
        AssistantConversation,
        AssistantMessage,
    )
    from app.identity.person_models import IdentityPerson

    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"P{uuid.uuid4().hex[:8]}",
        display_name="张伟",
        work_email="zhang.wei@pingcap.com",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()

    conversation = AssistantConversation(
        person_id=person.id, channel=CHANNEL_FEISHU, chat_id="oc_test",
    )
    db_session.add(conversation)
    await db_session.flush()
    db_session.add(
        AssistantMessage(
            conversation_id=conversation.id,
            seq=1,
            role=ROLE_USER,
            content="我的邮箱 zhang.wei@pingcap.com 收不到公司的信,怎么办",
        )
    )
    suite = AssistantEvalSuite(slug=f"m-{uuid.uuid4().hex[:8]}", name="挖掘目标")
    db_session.add(suite)
    await db_session.flush()
    suite_id = suite.id
    await db_session.commit()

    result = await service.mine_candidates(
        db_session, suite_id=suite_id, days=30, limit=50,
    )
    await db_session.commit()

    assert result["created"] == 1, result
    case = (
        await db_session.execute(
            select(AssistantEvalCase).where(AssistantEvalCase.suite_id == suite_id)
        )
    ).scalar_one()
    assert case.status == CASE_DRAFT, "挖出来的必须先是草稿,由人审过再启用"
    assert "zhang.wei@pingcap.com" not in case.question
    assert "[邮箱]" in case.question
    assert case.origin_ref and case.origin_ref.startswith("message:")


async def test_mining_does_not_duplicate_an_existing_question(db_session):
    from app.assistant.evaluation import service
    from app.assistant.models import (
        CHANNEL_FEISHU,
        ROLE_USER,
        AssistantConversation,
        AssistantMessage,
    )
    from app.identity.person_models import IdentityPerson

    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"P{uuid.uuid4().hex[:8]}",
        display_name="李四",
        work_email=f"u{uuid.uuid4().hex[:6]}@pingcap.com",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()
    conversation = AssistantConversation(
        person_id=person.id, channel=CHANNEL_FEISHU, chat_id="oc_dup",
    )
    db_session.add(conversation)
    await db_session.flush()
    for seq in (1, 3):
        db_session.add(
            AssistantMessage(
                conversation_id=conversation.id,
                seq=seq,
                role=ROLE_USER,
                content="VPN 连不上 怎么办",
            )
        )
    suite = AssistantEvalSuite(slug=f"m-{uuid.uuid4().hex[:8]}", name="挖掘目标")
    db_session.add(suite)
    await db_session.flush()
    suite_id = suite.id
    await db_session.commit()

    first = await service.mine_candidates(db_session, suite_id=suite_id)
    await db_session.commit()
    assert first["created"] == 1
    assert first["skipped_duplicate"] >= 1

    # 再挖一次:同一批语料不该再产出任何东西。
    second = await service.mine_candidates(db_session, suite_id=suite_id)
    await db_session.commit()
    assert second["created"] == 0


# ── 对比 ────────────────────────────────────────────────────────────


async def _run_with(db, suite_id, *, pairs, pass_rate):
    """造一次已完成的运行:``pairs`` 是 [(case, verdict), ...]。"""
    run = AssistantEvalRun(
        suite_id=suite_id,
        status=RUN_SUCCEEDED,
        repeats=1,
        attempts_total=len(pairs),
        attempts_passed=sum(1 for _c, v in pairs if v == VERDICT_PASS),
        attempts_failed=sum(1 for _c, v in pairs if v == VERDICT_FAIL),
        pass_rate=pass_rate,
        model="qwen-max-latest",
    )
    db.add(run)
    await db.flush()
    for case, verdict in pairs:
        db.add(
            AssistantEvalResult(
                run_id=run.id,
                case_id=case.id,
                attempt=0,
                question=case.question,
                category=case.category,
                lang="zh",
                verdict=verdict,
            )
        )
    await db.flush()
    return run


async def test_compare_names_the_cases_that_regressed(db_session):
    """🔴 一个差值只会开启一次调查;一份名单能结束它。"""
    from app.assistant.evaluation import service

    suite = AssistantEvalSuite(slug=f"c-{uuid.uuid4().hex[:8]}", name="对比")
    db_session.add(suite)
    await db_session.flush()
    cases = []
    for index, question in enumerate(("A 问题", "B 问题", "C 问题")):
        case = AssistantEvalCase(
            suite_id=suite.id,
            ordinal=index,
            question=question,
            lang="zh",
            category=CAT_KB_LOOKUP,
            status=CASE_ACTIVE,
        )
        db_session.add(case)
        cases.append(case)
    await db_session.flush()

    baseline = await _run_with(
        db_session,
        suite.id,
        pairs=[(cases[0], VERDICT_PASS), (cases[1], VERDICT_PASS), (cases[2], VERDICT_FAIL)],
        pass_rate=66.67,
    )
    # 新一次:B 退了,C 好了。
    current = await _run_with(
        db_session,
        suite.id,
        pairs=[(cases[0], VERDICT_PASS), (cases[1], VERDICT_FAIL), (cases[2], VERDICT_PASS)],
        pass_rate=66.67,
    )
    await db_session.commit()

    diff = await service.compare_runs(db_session, run=current, baseline=baseline)
    regressed = {item["question"] for item in diff["regressed"]}
    improved = {item["question"] for item in diff["improved"]}
    assert regressed == {"B 问题"}
    assert improved == {"C 问题"}
    assert diff["model_changed"] is False
    # 总分没动 —— 这正是「只看差值会漏掉真实回退」的那个形状。
    assert diff["drop_pp"] == 0.0


async def test_a_tiny_run_is_not_comparable(db_session):
    """🔴 样本太少时「掉 20 个点」可以什么都没发生。会误报的告警会被无视。"""
    from app.assistant.evaluation import service

    suite = AssistantEvalSuite(slug=f"c-{uuid.uuid4().hex[:8]}", name="小样本")
    db_session.add(suite)
    await db_session.flush()
    case = AssistantEvalCase(
        suite_id=suite.id, ordinal=0, question="A", lang="zh",
        category=CAT_KB_LOOKUP, status=CASE_ACTIVE,
    )
    db_session.add(case)
    await db_session.flush()

    baseline = await _run_with(
        db_session, suite.id, pairs=[(case, VERDICT_PASS)], pass_rate=100.0,
    )
    current = await _run_with(
        db_session, suite.id, pairs=[(case, VERDICT_FAIL)], pass_rate=0.0,
    )
    await db_session.commit()

    diff = await service.compare_runs(db_session, run=current, baseline=baseline)
    assert diff["drop_pp"] == 100.0
    assert diff["comparable"] is False


# ── 失败提升为长期用例 ──────────────────────────────────────────────


async def test_promoting_failures_creates_draft_cases(db_session):
    """一次失败要变成永久用例,修好之后才有东西证明它真的修好了。"""
    from app.assistant.evaluation import service

    suite = AssistantEvalSuite(slug=f"p-{uuid.uuid4().hex[:8]}", name="源")
    target = AssistantEvalSuite(slug=f"p-{uuid.uuid4().hex[:8]}", name="目标")
    db_session.add_all([suite, target])
    await db_session.flush()
    case = AssistantEvalCase(
        suite_id=suite.id, ordinal=0, question="怎么申请 VPN 权限", lang="zh",
        category=CAT_KB_LOOKUP, status=CASE_ACTIVE,
    )
    db_session.add(case)
    await db_session.flush()
    run = AssistantEvalRun(suite_id=suite.id, status=RUN_SUCCEEDED, repeats=1)
    db_session.add(run)
    await db_session.flush()
    db_session.add(
        AssistantEvalResult(
            run_id=run.id, case_id=case.id, attempt=0,
            question=case.question, category=case.category, lang="zh",
            verdict=VERDICT_FAIL, failure_type="kb_doc",
            judge_reasoning="知识库里没有 VPN 申请的文档。",
        )
    )
    run_id, target_id = run.id, target.id
    await db_session.commit()

    result = await service.promote_failures(
        db_session, run_id=run_id, suite_id=target_id,
    )
    await db_session.commit()

    assert result == {"created": 1, "failures_seen": 1}
    promoted = (
        await db_session.execute(
            select(AssistantEvalCase).where(AssistantEvalCase.suite_id == target_id)
        )
    ).scalar_one()
    assert promoted.status == CASE_DRAFT
    assert promoted.origin == ORIGIN_EVAL_FAILURE
    assert "kb_doc" in (promoted.expectations or {}).get("notes", "")


async def test_promoting_twice_does_not_duplicate(db_session):
    from app.assistant.evaluation import service

    suite = AssistantEvalSuite(slug=f"p-{uuid.uuid4().hex[:8]}", name="源")
    db_session.add(suite)
    await db_session.flush()
    case = AssistantEvalCase(
        suite_id=suite.id, ordinal=0, question="重复的问题", lang="zh",
        category=CAT_KB_LOOKUP, status=CASE_ACTIVE,
    )
    db_session.add(case)
    await db_session.flush()
    run = AssistantEvalRun(suite_id=suite.id, status=RUN_SUCCEEDED, repeats=1)
    db_session.add(run)
    await db_session.flush()
    db_session.add(
        AssistantEvalResult(
            run_id=run.id, case_id=case.id, attempt=0,
            question=case.question, category=case.category, lang="zh",
            verdict=VERDICT_FAIL, failure_type="retrieval",
        )
    )
    run_id, suite_id = run.id, suite.id
    await db_session.commit()

    await service.promote_failures(db_session, run_id=run_id, suite_id=suite_id)
    await db_session.commit()
    second = await service.promote_failures(
        db_session, run_id=run_id, suite_id=suite_id,
    )
    await db_session.commit()
    assert second["created"] == 0


# ── 用例的删除语义 ──────────────────────────────────────────────────


async def test_a_case_with_history_is_retired_not_deleted(db_session):
    """🔴 删掉题目会把「助手过去是怎么答它的」一起删掉,而那就是趋势线。"""
    from app.assistant.evaluation import service

    suite = AssistantEvalSuite(slug=f"d-{uuid.uuid4().hex[:8]}", name="删除")
    db_session.add(suite)
    await db_session.flush()
    case = AssistantEvalCase(
        suite_id=suite.id, ordinal=0, question="有历史的题", lang="zh",
        category=CAT_KB_LOOKUP, status=CASE_ACTIVE,
    )
    db_session.add(case)
    await db_session.flush()
    run = AssistantEvalRun(suite_id=suite.id, status=RUN_SUCCEEDED, repeats=1)
    db_session.add(run)
    await db_session.flush()
    db_session.add(
        AssistantEvalResult(
            run_id=run.id, case_id=case.id, attempt=0,
            question=case.question, category=case.category, lang="zh",
            verdict=VERDICT_PASS,
        )
    )
    await db_session.commit()

    result = await service.delete_case(db_session, case)
    await db_session.commit()
    assert result == {"deleted": False, "retired": True, "results": 1}
    assert case.status == "retired"


async def test_a_case_with_no_history_is_really_deleted(db_session):
    from app.assistant.evaluation import service

    suite = AssistantEvalSuite(slug=f"d-{uuid.uuid4().hex[:8]}", name="删除")
    db_session.add(suite)
    await db_session.flush()
    case = AssistantEvalCase(
        suite_id=suite.id, ordinal=0, question="没跑过的题", lang="zh",
        category=CAT_KB_LOOKUP, status=CASE_ACTIVE,
    )
    db_session.add(case)
    await db_session.flush()
    case_id = case.id
    await db_session.commit()

    result = await service.delete_case(db_session, case)
    await db_session.commit()
    assert result["deleted"] is True
    assert await db_session.get(AssistantEvalCase, case_id) is None


# ── 对象定位套件(2026-09-08) ──────────────────────────────────────


async def test_object_lookup_cases_keep_their_deterministic_guard(db_session):
    """🔴 这一套唯一不靠评委的那部分,不能被静默删掉。

    「员工点名一个对象,回答说的却是另一个东西」这种失败**看起来完全合理** ——
    评委很容易放过它。抓住它的是一条确定性判据:``flow_grounding`` 那几句措辞
    (「尚未本人确认」/「平台根据审批记录推断」)只由那一个模块拼出,出现在一条
    「某个具体对象归谁」的回答里就等于发生了替换。

    所以每条**带编号**的用例都必须带着这条禁止短语。唯一的例外是那条护栏用例
    (「开通云账号要走什么流程」):它句子里没有编号,流程清单正是它的答案。
    """
    from app.assistant.evaluation.seed_data import OBJECT_LOOKUP_CASES

    guarded = 0
    for spec in OBJECT_LOOKUP_CASES:
        forbidden = [
            phrase
            for entry in (spec["expectations"].get("must_not_mention") or [])
            for phrase in entry
        ]
        if not forbidden:
            # 护栏用例:必须写明为什么它不带判据,否则就是漏写。
            assert "护栏" in (spec["expectations"].get("notes") or ""), spec["question"]
            continue
        assert "尚未本人确认" in forbidden, spec["question"]
        guarded += 1
    assert guarded >= 5, "带编号的用例至少五条,少了就是被删过"


async def test_every_object_lookup_case_seeds_active(db_session):
    """新套件的用例必须直接可跑。

    ``seed_builtin_suites`` 对**已存在**的套件追加的用例一律是 draft(刻意的:
    悄悄变宽的闸会移动这个部署自己的基线)。所以这一套必须以新套件落地 —— 这条
    用例就是在证明这一点,而不只是复述它。
    """
    from app.assistant.evaluation import service
    from app.assistant.evaluation.seed_data import (
        OBJECT_LOOKUP_CASES,
        SUITE_OBJECT_LOOKUP,
    )

    await service.seed_builtin_suites(db_session)
    await db_session.commit()

    suite = (
        await db_session.execute(
            select(AssistantEvalSuite).where(
                AssistantEvalSuite.slug == SUITE_OBJECT_LOOKUP,
            )
        )
    ).scalar_one()
    rows = (
        (
            await db_session.execute(
                select(AssistantEvalCase).where(
                    AssistantEvalCase.suite_id == suite.id,
                )
            )
        )
        .scalars()
        .all()
    )
    assert len(rows) == len(OBJECT_LOOKUP_CASES)
    assert {row.status for row in rows} == {CASE_ACTIVE}


async def test_the_incident_question_is_in_the_bank_verbatim(db_session):
    """事故原句原样留在题库里。

    改写它就等于放弃了这条回归证据:那次错答与这句话的具体形状(一个 12 位纯数字
    编号 + 「谁负责」)是绑在一起的 —— 「谁负责」是 ``flow_grounding`` 的字面
    分支,换成「归谁」就走不到同一条路径了。
    """
    from app.assistant.evaluation.seed_data import OBJECT_LOOKUP_CASES

    questions = [spec["question"] for spec in OBJECT_LOOKUP_CASES]
    assert "380838443567 这个云账号谁负责" in questions