AgentEvalTool/tests/unit/test_engine.py

"""Unit tests for the async EvalEngine.

Focus on the new async behavior: cooperative cancellation, configurable
timeouts, and concurrent case execution via the semaphore.
"""

import asyncio
from typing import Any

from agenteval.channels.base import ChannelTransportError, SendResult
from agenteval.evaluation.engine import EvalEngine, TimeoutConfig
from agenteval.models import (
    Case,
    CaseOutcomeSummary,
    CaseType,
    ChannelType,
    EvalTarget,
    Expectation,
    PlatformType,
    RunStatus,
    Scenario,
    TargetStatus,
)
from agenteval.storage.db import TurnDB
from agenteval.storage.repository import RunRepository
from sqlmodel import select

from tests.unit.mock_channel import MockChannel


def _make_target() -> EvalTarget:
    return EvalTarget(
        id="t-1",
        name="mock-target",
        platform=PlatformType.AI_DIGITAL_EMPLOYEE,
        channel_type=ChannelType.TUTU_API,
        channel_config={
            "base_url": "http://mock",
            "token": "x",
            "tenant": "t",
            "chat_channel_id": "c",
            "chat_contact_id": "u",
        },
        status=TargetStatus.ACTIVE,
    )


def _build_engine(
    scenario: Scenario,
    channel: MockChannel,
    *,
    cancel_token: asyncio.Event | None = None,
    timeout_config: TimeoutConfig | None = None,
    max_concurrent_cases: int = 1,
    session=None,
) -> EvalEngine:
    """Construct an EvalEngine with a MockChannel injected.

    The factory still runs (needs a valid channel_config on the target) but
    we immediately replace the channel with the mock.
    """
    engine = EvalEngine(
        target=_make_target(),
        scenario=scenario,
        session=session,
        cancel_token=cancel_token or asyncio.Event(),
        timeout_config=timeout_config,
        max_concurrent_cases=max_concurrent_cases,
    )
    engine.channel = channel
    return engine


# ── basic happy path ─────────────────────────────────────────────────────


async def test_run_single_case_completes(db_session):
    scenario = Scenario(
        id="s-1",
        name="single",
        cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hello"])],
    )
    channel = MockChannel()
    engine = _build_engine(scenario, channel, session=db_session)

    run = await engine.run()

    assert run.status == RunStatus.COMPLETED
    assert run.summary.total_cases == 1
    assert run.summary.passed_cases == 1
    assert channel.send_calls == 1
    assert channel.poll_calls == 1


async def test_run_multi_turn_collects_all_turns(db_session):
    scenario = Scenario(
        id="s-1",
        name="multi",
        cases=[Case(id="c1", type=CaseType.MULTI_TURN, messages=["a", "b", "c"])],
    )
    channel = MockChannel()
    engine = _build_engine(scenario, channel, session=db_session)

    run = await engine.run()

    assert run.status == RunStatus.COMPLETED
    assert channel.sent == ["a", "b", "c"]
    assert channel.send_calls == 3
    # Each turn polls once.
    assert channel.poll_calls == 3


async def test_run_multiple_cases(db_session):
    scenario = Scenario(
        id="s-1",
        name="multi-case",
        cases=[
            Case(id="c1", type=CaseType.SINGLE, messages=["one"]),
            Case(id="c2", type=CaseType.SINGLE, messages=["two"]),
            Case(id="c3", type=CaseType.SINGLE, messages=["three"]),
        ],
    )
    channel = MockChannel()
    engine = _build_engine(scenario, channel, session=db_session)

    run = await engine.run()

    assert run.status == RunStatus.COMPLETED
    assert run.summary.total_cases == 3
    assert channel.sent == ["one", "two", "three"]


# ── progress callback ────────────────────────────────────────────────────


async def test_progress_callback_receives_events(db_session):
    scenario = Scenario(
        id="s-1",
        name="single",
        cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
    )
    events: list[tuple[str, dict]] = []

    async def cb(event: str, data: dict[str, Any]) -> None:
        events.append((event, data))

    channel = MockChannel()
    engine = _build_engine(scenario, channel, session=db_session)
    await engine.run(progress_callback=cb)

    event_names = [e[0] for e in events]
    assert "case_start" in event_names
    assert "turn_start" in event_names
    assert "turn_end" in event_names
    assert "case_end" in event_names
    assert "run_completed" in event_names


async def test_sync_progress_callback_also_works(db_session):
    """Engine must accept sync callbacks (CLI uses them)."""
    scenario = Scenario(
        id="s-1",
        name="single",
        cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
    )
    seen: list[str] = []

    def sync_cb(event: str, data: dict) -> None:
        seen.append(event)

    channel = MockChannel()
    engine = _build_engine(scenario, channel, session=db_session)
    await engine.run(progress_callback=sync_cb)

    assert "run_completed" in seen


# ── cancellation ─────────────────────────────────────────────────────────


async def test_cancel_before_run_marks_failed(db_session):
    scenario = Scenario(
        id="s-1",
        name="single",
        cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
    )
    cancel_token = asyncio.Event()
    cancel_token.set()  # already cancelled

    channel = MockChannel()
    engine = _build_engine(scenario, channel, cancel_token=cancel_token, session=db_session)
    run = await engine.run()

    assert run.status == RunStatus.FAILED
    assert run.summary.error.code == "cancelled_by_user"
    # Engine must NOT have called the channel.
    assert channel.send_calls == 0


async def test_cancel_mid_run_stops_after_current_case(db_session):
    """Cancelling between cases should stop further cases from running."""
    scenario = Scenario(
        id="s-1",
        name="multi",
        cases=[
            Case(id="c1", type=CaseType.SINGLE, messages=["a"]),
            Case(id="c2", type=CaseType.SINGLE, messages=["b"]),
            Case(id="c3", type=CaseType.SINGLE, messages=["c"]),
        ],
    )
    cancel_token = asyncio.Event()

    # Slow channel so we have time to set the cancel token from another task.
    channel = MockChannel(reply_delay=0.05)

    async def cancel_soon() -> None:
        await asyncio.sleep(0.08)
        cancel_token.set()

    engine = _build_engine(scenario, channel, cancel_token=cancel_token, session=db_session)

    run, _ = await asyncio.gather(
        engine.run(),
        cancel_soon(),
    )

    assert run.status == RunStatus.FAILED
    assert run.summary.error.code == "cancelled_by_user"
    # At least one case ran, but not all three.
    assert 1 <= channel.send_calls < 3


# ── timeouts ─────────────────────────────────────────────────────────────


async def test_poll_timeout_records_missing_reply(db_session):
    """When the channel returns no reply within the timeout, the turn is saved
    with reply=None but the engine keeps going (doesn't crash)."""
    scenario = Scenario(
        id="s-1",
        name="single",
        cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
    )
    channel = MockChannel(missing_reply=True)
    engine = _build_engine(
        scenario,
        channel,
        timeout_config=TimeoutConfig(poll_reply=0.1),
        session=db_session,
    )

    run = await engine.run()

    # Engine completes (doesn't hang), turn has no reply.
    assert run.status == RunStatus.COMPLETED
    turns = RunRepository(db_session).get_turns(run.id)
    assert len(turns) == 1
    assert turns[0].reply is None
    assert turns[0].question_msg_id == "q-1"


async def test_sent_turn_is_durable_before_reply_polling(db_session):
    scenario = Scenario(
        id="s-1",
        name="durable-send",
        cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
    )

    class LedgerInspectingChannel(MockChannel):
        persisted_before_poll = False

        async def _poll_reply(self, question_msg_id, timeout=30.0, poll_interval=1.0):
            turns = list(db_session.exec(select(TurnDB)).all())
            self.persisted_before_poll = len(turns) == 1 and turns[0].question_msg_id == question_msg_id
            return await super()._poll_reply(question_msg_id, timeout, poll_interval)

    channel = LedgerInspectingChannel()
    engine = _build_engine(scenario, channel, session=db_session)

    await engine.run()

    assert channel.persisted_before_poll is True


async def test_poll_failure_keeps_sent_turn_and_fails_case(db_session):
    scenario = Scenario(
        id="s-1",
        name="poll-failure",
        cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
    )
    channel = MockChannel(raise_on_poll=ChannelTransportError("upstream unavailable"))
    engine = _build_engine(scenario, channel, session=db_session)

    run = await engine.run()

    assert run.status == RunStatus.COMPLETED
    assert run.summary.failed_cases == 1
    turns = RunRepository(db_session).get_turns(run.id)
    assert len(turns) == 1
    assert turns[0].question_msg_id == "q-1"
    assert turns[0].reply is None


# ── concurrency ──────────────────────────────────────────────────────────


async def test_concurrent_cases_respect_semaphore(db_session):
    """With max_concurrent_cases=2, at most 2 cases should be in-flight."""
    scenario = Scenario(
        id="s-1",
        name="concurrent",
        cases=[Case(id=f"c{i}", type=CaseType.SINGLE, messages=[f"m{i}"]) for i in range(5)],
    )
    in_flight = {"n": 0, "max": 0}
    lock = asyncio.Lock()

    class TrackedChannel(MockChannel):
        async def _send(self, content: str, **kwargs):
            async with lock:
                in_flight["n"] += 1
                in_flight["max"] = max(in_flight["max"], in_flight["n"])
            await asyncio.sleep(0.05)
            result = await super()._send(content, **kwargs)
            async with lock:
                in_flight["n"] -= 1
            return result

    channel = TrackedChannel()
    engine = _build_engine(
        scenario,
        channel,
        max_concurrent_cases=2,
        session=db_session,
    )

    run = await engine.run()

    assert run.status == RunStatus.COMPLETED
    assert channel.send_calls == 5
    assert in_flight["max"] <= 2


# ── send failure ─────────────────────────────────────────────────────────


async def test_send_failure_aborts_case(db_session):
    """When channel.send fails, the case is marked failed but the engine
    continues with the next case."""
    scenario = Scenario(
        id="s-1",
        name="mixed",
        cases=[
            Case(id="c1", type=CaseType.SINGLE, messages=["a"]),
            Case(id="c2", type=CaseType.SINGLE, messages=["b"]),
        ],
    )
    call_count = {"n": 0}

    class FlakeyChannel(MockChannel):
        async def _send(self, content, **kwargs):
            call_count["n"] += 1
            if call_count["n"] == 1:
                return SendResult(ok=False, error="boom")
            return await super()._send(content, **kwargs)

    channel = FlakeyChannel()
    engine = _build_engine(scenario, channel, session=db_session)

    run = await engine.run()

    assert run.status == RunStatus.COMPLETED
    assert run.summary.failed_cases == 1
    assert run.summary.passed_cases == 1


# ── dynamic case generation failure ──────────────────────────────────────


async def test_dynamic_generation_failure_records_case_error(db_session):
    """A dynamic case whose message generation fails must persist the reason
    into run.summary.case_errors — not just emit it transiently. Otherwise a
    run shows 0 rules / failed with no discoverable cause."""
    scenario = Scenario(
        id="s-1",
        name="dynamic",
        cases=[Case(id="dyn-1", type=CaseType.DYNAMIC, prompt="生成问题", turns=3)],
        llm_config=None,  # 缺 llm_config → 生成消息立即失败
    )
    channel = MockChannel()
    engine = _build_engine(scenario, channel, session=db_session)

    run = await engine.run()

    assert run.status == RunStatus.COMPLETED
    assert run.summary.failed_cases == 1
    assert run.summary.total_rules == 0
    # 关键：失败原因被持久化到 summary，可在报告 / DB 查看
    assert run.summary.case_errors
    assert run.summary.case_errors[0]["case_id"] == "dyn-1"
    assert "llm_config" in run.summary.case_errors[0]["error"]
    # 被测通道不应被调用（生成阶段就失败了）
    assert channel.send_calls == 0


# ── expectation + explicit rules are additive (ticket 01) ────────────────
# MockChannel replies "echo: q-1", so keyword "echo" passes, "__NOPE__" fails.

from agenteval.models import EvalRuleConfig, RuleLogic  # noqa: E402


def _case_with(
    *,
    rules: list[EvalRuleConfig] | None = None,
    expectations: Expectation | None = None,
    rule_logic: RuleLogic = RuleLogic.ALL,
    rule_pass_threshold: float = 0.6,
) -> Case:
    return Case(
        id="c1",
        type=CaseType.SINGLE,
        messages=["hi"],
        eval_rules=rules or [],
        expectations=expectations or Expectation(),
        rule_logic=rule_logic,
        rule_pass_threshold=rule_pass_threshold,
    )


async def test_expectation_fails_case_even_when_rules_pass(db_session):
    """期望不满足 → 用例不通过，即使显式规则全部通过。"""
    scenario = Scenario(
        id="s1",
        name="s",
        cases=[
            _case_with(
                rules=[EvalRuleConfig(type="keyword_match", params={"keywords": ["echo"]})],
                expectations=Expectation(keywords_include=["__NOPE__"]),
            )
        ],
    )
    channel = MockChannel()
    engine = _build_engine(scenario, channel, session=db_session)

    run = await engine.run()

    assert run.status == RunStatus.COMPLETED
    assert run.summary.failed_cases == 1


async def test_expectation_and_rules_both_pass(db_session):
    """期望与规则都满足 → 通过，且期望派生判定同构落库、reason 可辨识来源。"""
    scenario = Scenario(
        id="s1",
        name="s",
        cases=[
            _case_with(
                rules=[EvalRuleConfig(type="keyword_match", params={"keywords": ["echo"]})],
                expectations=Expectation(keywords_include=["echo"], response_time_max_ms=99999),
            )
        ],
    )
    channel = MockChannel()
    engine = _build_engine(scenario, channel, session=db_session)

    run = await engine.run()

    assert run.summary.passed_cases == 1
    results = RunRepository(db_session).get_results(run.id)
    # 1 显式规则 + 2 期望派生（keyword + response_time）
    assert len(results) == 3
    implicit = [r for r in results if "期望" in r.reason]
    assert len(implicit) == 2
    assert all(r.passed for r in results)


async def test_implicit_expectation_not_in_any_combination(db_session):
    """rule_logic=ANY 只组合显式规则：期望通过不能救活全败的显式规则组。"""
    scenario = Scenario(
        id="s1",
        name="s",
        cases=[
            _case_with(
                rules=[EvalRuleConfig(type="keyword_match", params={"keywords": ["__NOPE__"]})],
                expectations=Expectation(keywords_include=["echo"]),  # 通过
                rule_logic=RuleLogic.ANY,
            )
        ],
    )
    channel = MockChannel()
    engine = _build_engine(scenario, channel, session=db_session)

    run = await engine.run()

    assert run.summary.failed_cases == 1


async def test_implicit_expectation_is_hard_constraint_over_weighted(db_session):
    """rule_logic=WEIGHTED 达标但期望不满足 → 仍不通过（期望是硬约束）。"""
    scenario = Scenario(
        id="s1",
        name="s",
        cases=[
            _case_with(
                rules=[EvalRuleConfig(type="keyword_match", params={"keywords": ["echo"]}, weight=1.0)],
                expectations=Expectation(keywords_include=["__NOPE__"]),
                rule_logic=RuleLogic.WEIGHTED,
                rule_pass_threshold=0.5,  # 显式加权得分 1.0 ≥ 0.5
            )
        ],
    )
    channel = MockChannel()
    engine = _build_engine(scenario, channel, session=db_session)

    run = await engine.run()

    assert run.summary.failed_cases == 1


async def test_pure_expectation_case_behavior_unchanged(db_session):
    """纯期望用例（无显式规则）：满足通过、不满足失败，与升级前一致。"""
    scenario = Scenario(
        id="s1",
        name="s",
        cases=[
            Case(id="ok", type=CaseType.SINGLE, messages=["hi"], expectations=Expectation(keywords_include=["echo"])),
            Case(
                id="bad", type=CaseType.SINGLE, messages=["hi"], expectations=Expectation(keywords_include=["__NOPE__"])
            ),
        ],
    )
    channel = MockChannel()
    engine = _build_engine(scenario, channel, session=db_session)

    run = await engine.run()

    assert run.summary.passed_cases == 1
    assert run.summary.failed_cases == 1


# ── scenario_version snapshot (ticket 04) ────────────────────────────────


async def test_engine_run_snapshots_scenario_version(db_session):
    """引擎直启（CLI 路径）创建的运行快照场景当前版本。"""
    scenario = Scenario(
        id="s-1",
        name="versioned",
        version=3,
        cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
    )
    channel = MockChannel()
    engine = _build_engine(scenario, channel, session=db_session)

    run = await engine.run()

    assert run.scenario_version == 3
    persisted = RunRepository(db_session).get(run.id)
    assert persisted.scenario_version == 3


# ── 权威判定写入 summary.case_outcomes（判定语义收敛） ────────────────────


async def test_summary_contains_case_outcomes_and_case_level_pass_rate(db_session):
    scenario = Scenario(
        id="s-1",
        name="outcomes",
        cases=[
            # 连通用例（无规则无期望）
            Case(id="conn", type=CaseType.SINGLE, messages=["ping"]),
            # 判定失败用例（期望不满足）
            Case(
                id="bad", type=CaseType.SINGLE, messages=["hi"], expectations=Expectation(keywords_include=["__NOPE__"])
            ),
        ],
    )
    channel = MockChannel()
    engine = _build_engine(scenario, channel, session=db_session)

    run = await engine.run()

    outcomes = run.summary.case_outcomes
    assert outcomes["conn"] == CaseOutcomeSummary(passed=True, connectivity=True)
    assert outcomes["bad"] == CaseOutcomeSummary(passed=False, connectivity=False)
    # 通过率为用例级口径（CONTEXT.md），不再是规则级
    assert run.summary.pass_rate == 0.5


async def test_connectivity_case_without_reply_fails(db_session):
    """连通用例没收到回复=故障=不通过（此前无条件判通过的 bug）。"""
    scenario = Scenario(
        id="s-1",
        name="conn-fail",
        cases=[
            Case(id="conn", type=CaseType.SINGLE, messages=["ping"]),
        ],
    )
    channel = MockChannel(missing_reply=True)
    engine = _build_engine(
        scenario,
        channel,
        session=db_session,
        timeout_config=TimeoutConfig(poll_reply=0.2),
    )

    run = await engine.run()

    assert run.summary.passed_cases == 0
    assert run.summary.case_outcomes["conn"] == CaseOutcomeSummary(passed=False, connectivity=False)