"""一次完整的评测:问、判、落库、汇总、续跑。

助手本体和评委都被替身顶掉 —— 这里测的是**评测台**,不是模型。替身按
生产的真实形状返回(`SimulationResult` 本人,不是一个鸭子类型的字典),
所以字段名对不上会在这里红,而不是等到生产上跑第一次。
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy import func, select

from app.assistant.evaluation.constants import (
    CASE_ACTIVE,
    CAT_KB_LOOKUP,
    RUN_PARTIAL,
    RUN_SUCCEEDED,
    VERDICT_ERROR,
    VERDICT_FAIL,
)
from app.assistant.evaluation.models import (
    AssistantEvalCase,
    AssistantEvalResult,
    AssistantEvalRun,
    AssistantEvalSuite,
)

pytestmark = pytest.mark.asyncio


# ── 替身 ────────────────────────────────────────────────────────────


def _simulation(reply: str, **overrides):
    from app.assistant.conversation import SimulationResult

    payload = {
        "reply": reply,
        "tool_calls": ["search_knowledge_base"],
        "docs": [],
        "policy_conflicts": [],
        "proposed_actions": [],
        "provider": "bailian",
        "available": True,
        "turn_failed": False,
        "evidence": ["工具返回:公司 VPN 使用 Cisco AnyConnect。"],
        "latency_ms": 1234,
    }
    payload.update(overrides)
    return SimulationResult(**payload)


class _JudgeStub:
    """按用例题面决定判 pass 还是 fail,并记录被问过几次。"""

    def __init__(self, fail_questions: set[str] | None = None) -> None:
        self.fail_questions = fail_questions or set()
        self.calls: list[str] = []

    async def __call__(self, db, *, question: str, **kwargs):
        from app.assistant.evaluation.judge import JudgeVerdict

        self.calls.append(question)
        if question in self.fail_questions:
            return JudgeVerdict(
                available=True,
                scores={
                    "grounded": 1, "accurate": 2, "complete": 2,
                    "actionable": 3, "escalation": 3, "tone": 4,
                },
                verdict="fail",
                failure_type="kb_doc",
                reasoning="知识库里没有覆盖这件事。",
                model="qwen-max-latest",
            )
        return JudgeVerdict(
            available=True,
            scores={
                "grounded": 5, "accurate": 5, "complete": 4,
                "actionable": 4, "escalation": 5, "tone": 4,
            },
            verdict="pass",
            reasoning="照检索到的文档答的。",
            model="qwen-max-latest",
        )


async def _person(db, *, name="评测模拟员工"):
    from app.identity.person_models import IdentityPerson

    # 字段照生产的形状给全:`hris_person_id` 是 NOT NULL,而一个只在测试里
    # 才立得住的行,测出来的东西也只在测试里立得住。
    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"P{uuid.uuid4().hex[:8]}",
        display_name=name,
        work_email=f"eval-{uuid.uuid4().hex[:8]}@pingcap.com",
        employment_status="active",
    )
    db.add(person)
    await db.flush()
    return person


async def _suite(db, *, persona=None, questions=("怎么连公司的 VPN", "打印机怎么连")):
    suite = AssistantEvalSuite(
        slug=f"t-{uuid.uuid4().hex[:8]}",
        name="测试题库",
        status="active",
        default_persona_person_id=persona.id if persona else None,
    )
    db.add(suite)
    await db.flush()
    for index, question in enumerate(questions):
        db.add(
            AssistantEvalCase(
                suite_id=suite.id,
                ordinal=index + 1,
                question=question,
                lang="zh",
                category=CAT_KB_LOOKUP,
                status=CASE_ACTIVE,
            )
        )
    await db.flush()
    return suite


@pytest.fixture(autouse=True)
def background_sessions(db_engine, monkeypatch):
    """把 `async_session_factory` 指到测试库。

    `execute_run` 刻意为每个并发的尝试开自己的 session(AsyncSession 不能
    跨并发任务共用),而它是从 `app.core.database` 里现取工厂的。不打这个
    补丁,后台那一半会安安静静地打到真正的 `helix` 库上 —— 测试要么报表不
    存在,要么更糟:在开发库里写出真实的行。

    conftest 的 `client` fixture 里有同样一段;这里单独做一份,是为了不必
    为了几个纯后台的用例把整个 HTTP + MCP lifespan 拉起来。
    """
    from sqlalchemy.ext.asyncio import AsyncSession as _AsyncSession
    from sqlalchemy.ext.asyncio import async_sessionmaker as _maker

    from app.core import database as _db

    monkeypatch.setattr(
        _db,
        "async_session_factory",
        _maker(db_engine, class_=_AsyncSession, expire_on_commit=False),
    )


@pytest.fixture
def wire(monkeypatch):
    """把模型那一侧全部换成替身,并返回评委桩以便断言。"""

    def _wire(*, simulate, judge):
        import app.assistant.conversation as conversation
        import app.assistant.evaluation.runner as runner
        import app.domains.support.copilot as copilot

        monkeypatch.setattr(conversation, "simulate_employee_turn", simulate)
        monkeypatch.setattr(runner, "judge_attempt", judge)

        async def _available(_db):
            return True

        async def _identity(_db):
            return "bailian", "qwen-max-latest"

        monkeypatch.setattr(copilot, "copilot_available", _available)
        monkeypatch.setattr(copilot, "active_llm_identity", _identity)
        return judge

    return _wire


# ── 用例 ────────────────────────────────────────────────────────────


async def test_a_run_scores_every_case_and_aggregates(db_session, wire):
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import execute_run

    person = await _person(db_session)
    suite = await _suite(db_session, persona=person)
    await db_session.commit()

    async def simulate(db, **kwargs):
        return _simulation("按公司知识库,VPN 用 Cisco AnyConnect,步骤如下。")

    judge = wire(simulate=simulate, judge=_JudgeStub())

    run = await service.start_run(db_session, suite_id=suite.id, repeats=2)
    run_id = run.id
    await db_session.commit()

    summary = await execute_run(run_id)

    assert summary["status"] == RUN_SUCCEEDED, summary
    # 2 条用例 × 2 遍 = 4 次尝试。这个乘法是 repeats 存在的全部意义。
    assert summary["attempts_total"] == 4
    assert summary["attempts_passed"] == 4
    assert summary["pass_rate"] == 100.0
    assert summary["cases_run"] == 2
    assert summary["cases_flaky"] == 0
    assert len(judge.calls) == 4

    db_session.expire_all()
    fresh = await db_session.get(AssistantEvalRun, run_id)
    assert fresh.provider == "bailian"
    assert fresh.model == "qwen-max-latest"
    assert fresh.dimension_means["grounded"] == 5.0
    assert fresh.category_breakdown[CAT_KB_LOOKUP]["pass_rate"] == 100.0


async def test_the_run_writes_no_transcript_and_files_no_request(db_session, wire):
    """🔴 评测绝不能在员工那边留下任何痕迹。

    干跑派发器是唯一的替换点,但「它确实没写」必须被断言,而不是被相信 ——
    一次回归把 `dry_run=True` 丢掉,表现就是给同事推了一张他从没要过的
    确认卡片。
    """
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import execute_run
    from app.assistant.models import AssistantConversation, AssistantMessage

    person = await _person(db_session)
    suite = await _suite(db_session, persona=person)
    await db_session.commit()

    async def simulate(db, **kwargs):
        return _simulation("知识库里写的是这样。")

    wire(simulate=simulate, judge=_JudgeStub())
    run = await service.start_run(db_session, suite_id=suite.id, repeats=1)
    run_id = run.id
    await db_session.commit()
    await execute_run(run_id)

    conversations = (
        await db_session.execute(
            select(func.count()).select_from(AssistantConversation)
        )
    ).scalar()
    messages = (
        await db_session.execute(select(func.count()).select_from(AssistantMessage))
    ).scalar()
    assert conversations == 0, "评测在员工的会话表里留下了记录"
    assert messages == 0, "评测在员工的消息表里留下了记录"


async def test_a_failing_case_is_recorded_with_its_reasoning(db_session, wire):
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import execute_run

    person = await _person(db_session)
    suite = await _suite(db_session, persona=person)
    await db_session.commit()

    async def simulate(db, **kwargs):
        return _simulation("这个我不太清楚,建议问一下 IT。")

    wire(simulate=simulate, judge=_JudgeStub(fail_questions={"打印机怎么连"}))
    run = await service.start_run(db_session, suite_id=suite.id, repeats=1)
    run_id = run.id
    await db_session.commit()
    summary = await execute_run(run_id)

    assert summary["attempts_failed"] == 1
    assert summary["attempts_passed"] == 1
    assert summary["pass_rate"] == 50.0

    db_session.expire_all()
    failed = (
        (
            await db_session.execute(
                select(AssistantEvalResult).where(
                    AssistantEvalResult.run_id == run_id,
                    AssistantEvalResult.verdict == VERDICT_FAIL,
                )
            )
        )
        .scalars()
        .all()
    )
    assert len(failed) == 1
    assert failed[0].question == "打印机怎么连"
    assert failed[0].failure_type == "kb_doc"
    assert "知识库" in (failed[0].judge_reasoning or "")
    # 题面在结果行里冗余存了一份:之后改用例不能改写历史运行测的是什么。
    assert failed[0].category == CAT_KB_LOOKUP


async def test_an_inconsistent_case_is_counted_as_flaky(db_session, wire):
    """🔴 同一句话问两遍得到两种结果 —— 这是这套东西最有价值的产出,
    而 repeats=1 根本看不见它。"""
    from app.assistant.evaluation import service
    from app.assistant.evaluation.judge import JudgeVerdict
    from app.assistant.evaluation.runner import execute_run

    person = await _person(db_session)
    suite = await _suite(db_session, persona=person, questions=("怎么连公司的 VPN",))
    await db_session.commit()

    async def simulate(db, **kwargs):
        return _simulation("看情况。")

    seen = {"n": 0}

    async def flaky_judge(db, *, question: str, **kwargs):
        seen["n"] += 1
        passing = seen["n"] % 2 == 1
        return JudgeVerdict(
            available=True,
            scores={
                "grounded": 5 if passing else 1,
                "accurate": 5 if passing else 2,
                "complete": 4, "actionable": 4, "escalation": 4, "tone": 4,
            },
            verdict="pass" if passing else "fail",
            failure_type=None if passing else "retrieval",
            reasoning=f"第 {seen['n']} 次",
            model="qwen-max-latest",
        )

    wire(simulate=simulate, judge=flaky_judge)
    run = await service.start_run(db_session, suite_id=suite.id, repeats=2)
    run_id = run.id
    await db_session.commit()
    summary = await execute_run(run_id)

    assert summary["cases_flaky"] == 1, summary
    assert summary["pass_rate"] == 50.0


async def test_a_provider_outage_is_an_error_and_stays_out_of_the_pass_rate(
    db_session, wire,
):
    """🔴 供应商故障不能在趋势线上长得像质量回归。"""
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import execute_run

    person = await _person(db_session)
    suite = await _suite(db_session, persona=person, questions=("怎么连公司的 VPN",))
    await db_session.commit()

    async def simulate(db, **kwargs):
        return _simulation("", turn_failed=True, reply="助手暂时不可用。")

    judge = _JudgeStub()
    wire(simulate=simulate, judge=judge)
    run = await service.start_run(db_session, suite_id=suite.id, repeats=1)
    run_id = run.id
    await db_session.commit()
    summary = await execute_run(run_id)

    assert summary["attempts_errored"] == 1
    assert summary["attempts_failed"] == 0
    # 一次都没被判分,所以通过率是「无」,不是 0%。
    assert summary["pass_rate"] is None
    assert judge.calls == [], "这一轮压根没答上话,不该再花钱去判"


async def test_resuming_continues_instead_of_re_running(db_session, wire):
    """🔴 续跑要跳过已经跑过的 (用例, 第几遍),否则一次发布重启就会让
    这一跑把账单交两遍,而且计数翻倍。"""
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import execute_run

    person = await _person(db_session)
    suite = await _suite(db_session, persona=person)
    await db_session.commit()

    async def simulate(db, **kwargs):
        return _simulation("答案在这里。")

    judge = _JudgeStub()
    wire(simulate=simulate, judge=judge)
    run = await service.start_run(db_session, suite_id=suite.id, repeats=1)
    run_id = run.id
    await db_session.commit()

    first = await execute_run(run_id)
    assert first["attempts_total"] == 2
    calls_after_first = len(judge.calls)

    # 已经是终态了 —— 再调一次应当原样返回,一次模型都不该再打。
    again = await execute_run(run_id)
    assert again["status"] == RUN_SUCCEEDED
    assert len(judge.calls) == calls_after_first

    # 手动放回运行中(resume 端点做的事),此时没有欠账,仍不该重跑。
    db_session.expire_all()
    fresh = await db_session.get(AssistantEvalRun, run_id)
    fresh.status = "running"
    fresh.finished_at = None
    await db_session.commit()

    resumed = await execute_run(run_id)
    assert resumed["attempts_total"] == 2
    assert len(judge.calls) == calls_after_first, "续跑把已经跑过的又跑了一遍"


async def test_a_budget_that_runs_out_yields_partial_not_failed(db_session, wire):
    """预算用完是一种正常结局:已跑出来的数字是真的,只是没跑完。"""
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import execute_run

    person = await _person(db_session)
    suite = await _suite(
        db_session, persona=person, questions=tuple(f"问题 {i}" for i in range(6)),
    )
    await db_session.commit()

    async def simulate(db, **kwargs):
        return _simulation("答案。")

    wire(simulate=simulate, judge=_JudgeStub())
    run = await service.start_run(db_session, suite_id=suite.id, repeats=1, concurrency=1)
    run_id = run.id
    await db_session.commit()

    # 预算设成 0:第一个 worker 一进来就发现超了。
    summary = await execute_run(run_id, budget_seconds=0.0)
    assert summary["status"] == RUN_PARTIAL, summary
    assert summary["attempts_total"] == 0

    db_session.expire_all()
    fresh = await db_session.get(AssistantEvalRun, run_id)
    assert fresh.notes and "继续" in fresh.notes


async def test_a_run_refuses_to_start_twice_for_one_suite(db_session, wire):
    """双击不该产生两次半截测量,也不该把账单交两遍。"""
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import EvalRunError

    person = await _person(db_session)
    suite = await _suite(db_session, persona=person)
    await db_session.commit()

    await service.start_run(db_session, suite_id=suite.id, repeats=1)
    await db_session.commit()

    with pytest.raises(EvalRunError) as excinfo:
        await service.start_run(db_session, suite_id=suite.id, repeats=1)
    assert excinfo.value.code == "ASSISTANT_EVAL_RUN_IN_FLIGHT"


async def test_a_suite_with_no_active_cases_refuses_to_run(db_session):
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import EvalRunError

    suite = AssistantEvalSuite(slug=f"t-{uuid.uuid4().hex[:8]}", name="空题库")
    db_session.add(suite)
    await db_session.commit()

    with pytest.raises(EvalRunError) as excinfo:
        await service.start_run(db_session, suite_id=suite.id)
    assert excinfo.value.code == "ASSISTANT_EVAL_NO_ACTIVE_CASES"


async def test_a_stale_running_run_is_reaped(db_session, wire):
    """执行进程被发布干掉后,那一行不能永远停在「运行中」。"""
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import reap_stale_runs

    person = await _person(db_session)
    suite = await _suite(db_session, persona=person)
    await db_session.commit()

    run = await service.start_run(db_session, suite_id=suite.id, repeats=1)
    run_id = run.id
    await db_session.commit()

    # 立刻收尾:阈值设为 0 秒,任何还在跑的都算失联。
    reaped = await reap_stale_runs(db_session, stale_after_seconds=0.0)
    assert reaped == 1

    db_session.expire_all()
    fresh = await db_session.get(AssistantEvalRun, run_id)
    # 一次尝试都没跑成 → failed;跑成过 → partial。
    assert fresh.status == "failed"
    assert fresh.finished_at is not None


async def test_a_live_run_is_not_reaped(db_session, wire):
    """护栏:收尾闸不能把正在跑的运行掐掉。"""
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import reap_stale_runs

    person = await _person(db_session)
    suite = await _suite(db_session, persona=person)
    await db_session.commit()
    run = await service.start_run(db_session, suite_id=suite.id, repeats=1)
    # id 先取出来:`expire_all()` 之后再读 `run.id` 会触发一次惰性刷新,
    # 在 async session 上那就是 MissingGreenlet(本仓的常客)。
    run_id = run.id
    await db_session.commit()

    assert await reap_stale_runs(db_session, stale_after_seconds=3600.0) == 0
    db_session.expire_all()
    fresh = await db_session.get(AssistantEvalRun, run_id)
    assert fresh.status == "running"


async def test_a_multi_turn_case_passes_its_history_to_the_assistant(
    db_session, wire,
):
    """🔴 单轮评测测不到生产里真正的形状:「它」的所指在上一条消息里。"""
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import execute_run

    person = await _person(db_session)
    suite = AssistantEvalSuite(slug=f"t-{uuid.uuid4().hex[:8]}", name="多轮")
    suite.default_persona_person_id = person.id
    db_session.add(suite)
    await db_session.flush()
    db_session.add(
        AssistantEvalCase(
            suite_id=suite.id,
            ordinal=1,
            question="它还在保修期内吗",
            setup_turns=[
                {"role": "user", "content": "我有几台电脑"},
                {"role": "assistant", "content": "你名下有 1 台 MacBook Pro。"},
            ],
            lang="zh",
            category=CAT_KB_LOOKUP,
            status=CASE_ACTIVE,
        )
    )
    await db_session.commit()

    captured: dict = {}

    async def simulate(db, **kwargs):
        captured.update(kwargs)
        return _simulation("这台 MacBook 的保修到 2027 年 6 月。")

    wire(simulate=simulate, judge=_JudgeStub())
    run = await service.start_run(db_session, suite_id=suite.id, repeats=1)
    run_id = run.id
    await db_session.commit()
    await execute_run(run_id)

    assert captured["setup_turns"] == [
        {"role": "user", "content": "我有几台电脑"},
        {"role": "assistant", "content": "你名下有 1 台 MacBook Pro。"},
    ]
    assert captured["capture_evidence"] is True
    assert captured["person"].id == person.id


async def test_an_ungrounded_citation_fails_the_attempt_over_a_happy_judge(
    db_session, wire,
):
    """🔴 端到端地钉住那条优先级:评委满分也救不回一个编造的链接。"""
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import execute_run

    person = await _person(db_session)
    suite = await _suite(db_session, persona=person, questions=("怎么连公司的 VPN",))
    await db_session.commit()

    async def simulate(db, **kwargs):
        return _simulation(
            "详见 https://wiki.example.com/made-up-vpn-page 。",
            evidence=["工具返回:知识库里没有匹配的文章。"],
        )

    wire(simulate=simulate, judge=_JudgeStub())  # 桩默认判 pass
    run = await service.start_run(db_session, suite_id=suite.id, repeats=1)
    run_id = run.id
    await db_session.commit()
    summary = await execute_run(run_id)

    assert summary["attempts_failed"] == 1, summary
    db_session.expire_all()
    row = (
        await db_session.execute(
            select(AssistantEvalResult).where(AssistantEvalResult.run_id == run_id)
        )
    ).scalar_one()
    assert row.verdict == VERDICT_FAIL
    codes = {f["code"] for f in (row.findings or [])}
    assert "ungrounded_citation" in codes


async def test_the_run_refuses_when_the_assistant_is_switched_off(
    db_session, monkeypatch,
):
    """没配模型时,运行要明确变红并说明原因,而不是安静地 0 通过。"""
    import app.domains.support.copilot as copilot
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import execute_run

    person = await _person(db_session)
    suite = await _suite(db_session, persona=person)
    await db_session.commit()
    run = await service.start_run(db_session, suite_id=suite.id, repeats=1)
    run_id = run.id
    await db_session.commit()

    async def _unavailable(_db):
        return False

    monkeypatch.setattr(copilot, "copilot_available", _unavailable)
    summary = await execute_run(run_id)
    assert summary["status"] == "failed"

    db_session.expire_all()
    fresh = await db_session.get(AssistantEvalRun, run_id)
    assert "未启用" in (fresh.notes or "")


async def test_a_case_persona_overrides_the_suite_default(db_session, wire):
    """🔴 「以谁的身份问」是测量的一部分,不是实现细节。

    助手的工具按调用者作用域,同一句话经理问和新人问是两个正确答案。用例
    自己钉的人必须赢过题库的默认人 —— 反过来的话,一条专门为某个身份写的
    用例会安静地在别人身上跑,而且看起来完全正常。
    """
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import execute_run

    default_person = await _person(db_session, name="题库默认员工")
    pinned_person = await _person(db_session, name="用例指定员工")
    suite = await _suite(
        db_session, persona=default_person, questions=("怎么连公司的 VPN",),
    )
    case = (
        await db_session.execute(
            select(AssistantEvalCase).where(AssistantEvalCase.suite_id == suite.id)
        )
    ).scalar_one()
    case.persona_person_id = pinned_person.id
    # 同上:`expire_all()` 之后再读 ORM 实例的属性会触发惰性刷新 →
    # MissingGreenlet。断言要用的 id 一律先取成普通值。
    pinned_id = pinned_person.id
    await db_session.commit()

    asked_as: list[uuid.UUID] = []

    async def simulate(db, **kwargs):
        asked_as.append(kwargs["person"].id)
        return _simulation("答案。")

    wire(simulate=simulate, judge=_JudgeStub())
    run = await service.start_run(db_session, suite_id=suite.id, repeats=1)
    run_id = run.id
    await db_session.commit()
    await execute_run(run_id)

    assert asked_as == [pinned_id]
    db_session.expire_all()
    row = (
        await db_session.execute(
            select(AssistantEvalResult).where(AssistantEvalResult.run_id == run_id)
        )
    ).scalar_one()
    # 结果行也要记下这一次到底是以谁的身份跑的,否则事后无法复核。
    assert row.persona_person_id == pinned_id


async def test_no_resolvable_persona_errors_the_attempt_not_the_run(
    db_session, wire,
):
    """一个人都找不到时,这条用例记 error 并说明原因,而不是让整跑崩掉。"""
    from app.assistant.evaluation import service
    from app.assistant.evaluation.runner import execute_run
    from app.identity.person_models import IdentityPerson

    suite = await _suite(db_session, persona=None, questions=("怎么连公司的 VPN",))
    # 库里一个在职的人都没有(本用例没有建人,而 db_engine 每条用例前清数据)。
    remaining = (
        await db_session.execute(select(func.count()).select_from(IdentityPerson))
    ).scalar()
    assert remaining == 0, "前置条件不成立,这条断言测不到想测的东西"
    await db_session.commit()

    async def simulate(db, **kwargs):  # pragma: no cover - 不该被调用
        raise AssertionError("没有可模拟的身份时不该真的去问助手")

    wire(simulate=simulate, judge=_JudgeStub())
    run = await service.start_run(db_session, suite_id=suite.id, repeats=1)
    run_id = run.id
    await db_session.commit()
    summary = await execute_run(run_id)

    assert summary["attempts_errored"] == 1
    assert summary["pass_rate"] is None
    db_session.expire_all()
    row = (
        await db_session.execute(
            select(AssistantEvalResult).where(AssistantEvalResult.run_id == run_id)
        )
    ).scalar_one()
    assert row.verdict == VERDICT_ERROR
    assert "在职员工" in (row.error or "")