"""Pure contract tests for intelligent-evaluation read projections.""" from datetime import datetime, timezone from agenteval.intelligent_eval.models import ( IntelligentEval, IntelligentEvalSession, IntelligentEvalSessionStatus, IntelligentEvalStatus, ) from agenteval.intelligent_eval.read_model import ( IntelligentEvalReadModel, project_detail, project_execution_progress, project_list_item, ) from agenteval.intelligent_eval.repository import IntelligentEvalRepository, IntelligentEvalSessionRepository from sqlalchemy import event from sqlmodel import Session def _evaluation(eval_id: str = "eval-1") -> IntelligentEval: now = datetime(2026, 1, 1, tzinfo=timezone.utc) return IntelligentEval( id=eval_id, name="客服评估", target_id="target-1", status=IntelligentEvalStatus.EXECUTING, goal="验证退货流程", seeds={"personas": ["老客户"]}, intent="流程覆盖", role_description="模拟用户", plan={"dimensions": ["退货"]}, time_window_hours=24, created_at=now, updated_at=now, ) def _session( session_id: str, status: IntelligentEvalSessionStatus, eval_id: str = "eval-1", ) -> IntelligentEvalSession: return IntelligentEvalSession( id=session_id, eval_id=eval_id, target_id="target-1", persona={"name": session_id}, goal="完成退货", dimension="退货", status=status, turn_count=2, ) def test_list_projection_contains_stable_fields_and_counts() -> None: projection = project_list_item( _evaluation(), [ _session("s-1", IntelligentEvalSessionStatus.RUNNING), _session("s-2", IntelligentEvalSessionStatus.COMPLETED), ], ) assert projection.id == "eval-1" assert projection.session_count == 2 assert projection.completed_sessions == 1 assert projection.plan == {"dimensions": ["退货"]} def test_detail_projection_contains_session_metadata_without_messages() -> None: projection = project_detail(_evaluation(), [_session("s-1", IntelligentEvalSessionStatus.COMPLETED)]) payload = projection.model_dump(mode="json") assert payload["sessions"][0]["id"] == "s-1" assert payload["sessions"][0]["turn_count"] == 2 assert "messages" not in payload assert "content" not in payload["sessions"][0] def _count_queries(session: Session, action) -> int: count = 0 def before_cursor_execute(*_args) -> None: nonlocal count count += 1 engine = session.get_bind() event.listen(engine, "before_cursor_execute", before_cursor_execute) try: action() finally: event.remove(engine, "before_cursor_execute", before_cursor_execute) return count def test_empty_list_projection_does_not_query_sessions(db_session: Session) -> None: reader = IntelligentEvalReadModel(db_session) query_count = _count_queries(db_session, lambda: reader.list_items([])) assert query_count == 0 def test_list_projection_loads_all_sessions_with_one_query(db_session: Session) -> None: sessions = IntelligentEvalSessionRepository(db_session) sessions._create(_session("s-1", IntelligentEvalSessionStatus.RUNNING, "eval-1")) sessions._create(_session("s-2", IntelligentEvalSessionStatus.COMPLETED, "eval-2")) reader = IntelligentEvalReadModel(db_session) evaluations = [_evaluation("eval-1"), _evaluation("eval-2")] result: list = [] query_count = _count_queries(db_session, lambda: result.extend(reader.list_items(evaluations))) assert query_count == 1 assert [(item.id, item.session_count) for item in result] == [("eval-1", 1), ("eval-2", 1)] def test_detail_projection_uses_one_snapshot_query(db_session: Session) -> None: evaluations = IntelligentEvalRepository(db_session) sessions = IntelligentEvalSessionRepository(db_session) evaluations._create(_evaluation()) sessions._create(_session("s-1", IntelligentEvalSessionStatus.COMPLETED)) reader = IntelligentEvalReadModel(db_session) result: list = [] query_count = _count_queries(db_session, lambda: result.append(reader.detail_by_id("eval-1"))) assert query_count == 1 assert result[0] is not None assert result[0].id == "eval-1" assert [item.id for item in result[0].sessions] == ["s-1"] # --------------------------------------------------------------------------- # Execution progress projection # --------------------------------------------------------------------------- def _executing_eval() -> IntelligentEval: started = datetime(2026, 1, 1, 8, 0, tzinfo=timezone.utc) return IntelligentEval( id="eval-1", name="客服评估", target_id="target-1", status=IntelligentEvalStatus.EXECUTING, goal="验证退货流程", plan={ "time_distribution": [ {"time_slot": "0-1h", "sessions": 2}, {"time_slot": "1-2h", "sessions": 2}, {"time_slot": "2-3h", "sessions": 2}, ] }, time_window_hours=3, created_at=started, updated_at=started, started_at=started, ) def _slot_session( session_id: str, created_at: datetime, status: IntelligentEvalSessionStatus ) -> IntelligentEvalSession: return IntelligentEvalSession( id=session_id, eval_id="eval-1", target_id="target-1", persona={"name": session_id}, goal="完成退货", status=status, created_at=created_at, ) def test_execution_progress_slots_bucket_plan_vs_actual() -> None: now = datetime(2026, 1, 1, 9, 30, tzinfo=timezone.utc) # offset 1.5h → slot 1-2h current sessions = [ _slot_session("s-1", datetime(2026, 1, 1, 8, 10, tzinfo=timezone.utc), IntelligentEvalSessionStatus.COMPLETED), _slot_session("s-2", datetime(2026, 1, 1, 8, 20, tzinfo=timezone.utc), IntelligentEvalSessionStatus.COMPLETED), _slot_session("s-3", datetime(2026, 1, 1, 9, 5, tzinfo=timezone.utc), IntelligentEvalSessionStatus.RUNNING), ] progress = project_execution_progress(_executing_eval(), sessions, now) assert progress.current_stage == "executing" assert [(s.time_slot, s.planned, s.created, s.completed) for s in progress.slots] == [ ("0-1h", 2, 2, 2), ("1-2h", 2, 1, 0), ("2-3h", 2, 0, 0), ] assert [s.is_past for s in progress.slots] == [True, False, False] assert [s.is_current for s in progress.slots] == [False, True, False] assert progress.next_action is not None and "欠账" in progress.next_action def test_execution_progress_without_plan_or_start_has_no_slots() -> None: evaluation = _evaluation() evaluation.plan = None progress = project_execution_progress(evaluation, [], datetime(2026, 1, 1, tzinfo=timezone.utc)) assert progress.slots == [] assert progress.current_stage == "executing" def test_execution_progress_pending_approval_blocks_on_human() -> None: evaluation = _evaluation() evaluation.status = IntelligentEvalStatus.PENDING_APPROVAL progress = project_execution_progress(evaluation, [], datetime(2026, 1, 1, tzinfo=timezone.utc)) assert progress.current_stage == "approval" assert progress.blocker is not None and "审批" in progress.blocker assert progress.abnormal_outcome is None def test_execution_progress_cancelled_marks_last_reached_node() -> None: evaluation = _executing_eval() evaluation.status = IntelligentEvalStatus.CANCELLED progress = project_execution_progress(evaluation, [], datetime(2026, 1, 1, 9, 0, tzinfo=timezone.utc)) assert progress.abnormal_outcome == "cancelled" assert progress.current_stage == "executing" assert progress.blocker is not None and "取消" in progress.blocker def test_execution_progress_failed_before_plan_stays_at_planning() -> None: evaluation = _evaluation() evaluation.status = IntelligentEvalStatus.FAILED evaluation.plan = None progress = project_execution_progress(evaluation, [], datetime(2026, 1, 1, tzinfo=timezone.utc)) assert progress.abnormal_outcome == "failed" assert progress.current_stage == "planning" def test_execution_progress_completed_is_done_without_blocker() -> None: evaluation = _executing_eval() evaluation.status = IntelligentEvalStatus.COMPLETED progress = project_execution_progress(evaluation, [], datetime(2026, 1, 2, tzinfo=timezone.utc)) assert progress.current_stage == "done" assert progress.blocker is None assert progress.next_action is None def test_execution_progress_waits_for_report_when_slots_exhausted() -> None: started = datetime(2026, 1, 1, 8, 0, tzinfo=timezone.utc) now = datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc) # all slots past sessions = [_slot_session(f"s-{i}", started, IntelligentEvalSessionStatus.COMPLETED) for i in range(6)] progress = project_execution_progress(_executing_eval(), sessions, now) assert progress.next_action is not None and "报告" in progress.next_action