AgentEvalTool/tests/unit/test_intelligent_eval_model.py

"""Lifecycle contract tests for intelligent evaluations."""

import pytest
from agenteval.channels.base import ExchangeOutcome, SendResult
from agenteval.intelligent_eval import lifecycle
from agenteval.intelligent_eval.lifecycle import (
    IntelligentEvalChannelError,
    IntelligentEvalNotFoundError,
    IntelligentEvalTransitionError,
)
from agenteval.intelligent_eval.models import IntelligentEvalSessionStatus, IntelligentEvalStatus
from agenteval.models import ChannelType, EvalTarget, PlatformType, TargetStatus
from agenteval.storage.repository import TargetRepository


@pytest.fixture()
def eval_session(db_session):
    TargetRepository(db_session).create(
        EvalTarget(
            id="target-1",
            name="测试对象",
            platform=PlatformType.AI_DIGITAL_EMPLOYEE,
            channel_type=ChannelType.TUTU_API,
            channel_config={"base_url": "http://mock", "token": "token"},
            status=TargetStatus.ACTIVE,
        )
    )
    return db_session


def _create(session, *, name: str = "智能评估"):
    return lifecycle.create_eval(
        session,
        name=name,
        target_id="target-1",
        goal="验证退货流程",
        seeds={"personas": ["老客户"]},
        intent="流程覆盖",
        role_description="模拟用户",
    )


def _start(session, *, name: str = "智能评估"):
    evaluation = _create(session, name=name)
    lifecycle.submit_plan(session, evaluation.id, {"dimensions": ["退货"]})
    return lifecycle.approve(session, evaluation.id)


def test_lifecycle_owns_complete_evaluation_state_machine(eval_session) -> None:
    evaluation = _create(eval_session)
    assert evaluation.status is IntelligentEvalStatus.PLANNING

    pending = lifecycle.submit_plan(eval_session, evaluation.id, {"dimensions": ["退货"]})
    assert pending.status is IntelligentEvalStatus.PENDING_APPROVAL
    assert pending.plan == {"dimensions": ["退货"]}

    executing = lifecycle.approve(eval_session, evaluation.id)
    assert executing.status is IntelligentEvalStatus.EXECUTING
    assert executing.started_at is not None

    completed = lifecycle.submit_report(eval_session, evaluation.id, {"summary": "done"})
    assert completed.status is IntelligentEvalStatus.COMPLETED
    assert completed.report == {"summary": "done"}
    assert completed.completed_at is not None


def test_lifecycle_rejects_invalid_or_repeated_transitions(eval_session) -> None:
    evaluation = _create(eval_session)

    with pytest.raises(IntelligentEvalTransitionError):
        lifecycle.approve(eval_session, evaluation.id)

    lifecycle.submit_plan(eval_session, evaluation.id, {"dimensions": ["退货"]})
    with pytest.raises(IntelligentEvalTransitionError):
        lifecycle.submit_plan(eval_session, evaluation.id, {"dimensions": ["重复"]})


def test_plan_transaction_failure_leaves_public_snapshot_unchanged(eval_session, monkeypatch) -> None:
    evaluation = _create(eval_session)
    monkeypatch.setattr(eval_session, "commit", lambda: (_ for _ in ()).throw(RuntimeError("commit failed")))

    with pytest.raises(RuntimeError, match="commit failed"):
        lifecycle.submit_plan(eval_session, evaluation.id, {"dimensions": ["退货"]})

    monkeypatch.undo()
    unchanged = lifecycle.get_eval(eval_session, evaluation.id)
    assert unchanged.status is IntelligentEvalStatus.PLANNING
    assert unchanged.plan is None


def test_session_ownership_and_close_invariants_are_lifecycle_rules(eval_session) -> None:
    first = _start(eval_session, name="first")
    second = _start(eval_session, name="second")
    session_obj = lifecycle.open_session(
        eval_session,
        eval_id=first.id,
        persona={"name": "老客户"},
        goal="完成退货",
        dimension="退货",
    )

    with pytest.raises(IntelligentEvalNotFoundError):
        lifecycle.close_session(
            eval_session,
            eval_id=second.id,
            session_id=session_obj.id,
            verdict={"goal_achieved": False},
        )

    closed = lifecycle.close_session(
        eval_session,
        eval_id=first.id,
        session_id=session_obj.id,
        verdict={"goal_achieved": True},
    )
    assert closed.status is IntelligentEvalSessionStatus.COMPLETED
    assert closed.verdict == {"goal_achieved": True}


class _ReplyChannel:
    async def exchange(self, _content, *, on_sent, **_kwargs):
        await on_sent(SendResult(ok=True, question_msg_id="message-1"))
        return ExchangeOutcome.succeeded(correlation_id="message-1", reply="答复", latency_ms=12)


class _TimeoutChannel:
    async def exchange(self, _content, *, on_sent, **_kwargs):
        await on_sent(SendResult(ok=True, question_msg_id="message-1"))
        return ExchangeOutcome.reply_timeout(correlation_id="message-1", latency_ms=30_000)


async def test_turn_ledger_is_observable_only_through_lifecycle(eval_session, monkeypatch) -> None:
    evaluation = _start(eval_session)
    session_obj = lifecycle.open_session(eval_session, eval_id=evaluation.id, persona={}, goal="完成退货")
    monkeypatch.setattr(lifecycle.ChannelFactory, "create", lambda _target: _ReplyChannel())

    result = await lifecycle.conduct_turn(
        eval_session,
        eval_id=evaluation.id,
        session_id=session_obj.id,
        content="你好",
    )

    messages = lifecycle.list_messages(eval_session, eval_id=evaluation.id, session_id=session_obj.id)
    sessions = lifecycle.list_sessions(eval_session, evaluation.id)
    assert result == {"reply": "答复", "latency_ms": 12, "turn_count": 1}
    assert [(item.role, item.content) for item in messages] == [("user", "你好"), ("assistant", "答复")]
    assert sessions[0].turn_count == 1


async def test_timeout_preserves_sent_message_and_turn_count(eval_session, monkeypatch) -> None:
    evaluation = _start(eval_session)
    session_obj = lifecycle.open_session(eval_session, eval_id=evaluation.id, persona={}, goal="完成退货")
    monkeypatch.setattr(lifecycle.ChannelFactory, "create", lambda _target: _TimeoutChannel())

    with pytest.raises(IntelligentEvalChannelError, match="超时"):
        await lifecycle.conduct_turn(
            eval_session,
            eval_id=evaluation.id,
            session_id=session_obj.id,
            content="你好",
        )

    messages = lifecycle.list_messages(eval_session, eval_id=evaluation.id, session_id=session_obj.id)
    sessions = lifecycle.list_sessions(eval_session, evaluation.id)
    assert [(item.role, item.content) for item in messages] == [("user", "你好")]
    assert sessions[0].turn_count == 1