diff --git a/backend/agenteval/intelligent_eval/decision.py b/backend/agenteval/intelligent_eval/decision.py new file mode 100644 index 0000000..b26064b --- /dev/null +++ b/backend/agenteval/intelligent_eval/decision.py @@ -0,0 +1,276 @@ +"""Decision logic for intelligent eval workers (决策逻辑). + +Worker analyzes current situation and decides what to do: +- execute_session: Execute a new session +- wait: Wait for next tick +- start_analysis: Start analysis (all sessions completed) +""" + +from datetime import datetime, timedelta +from enum import Enum +from typing import Optional + +from sqlmodel import Session, select + +from agenteval.intelligent_eval.models import IntelligentEvalStatus +from agenteval.storage.db import ( + IntelligentEvalDB, + IntelligentEvalSessionDB, + as_utc, + utc_now, +) + + +class DecisionType(str, Enum): + """Decision types for worker.""" + + EXECUTE_SESSION = "execute_session" + WAIT = "wait" + START_ANALYSIS = "start_analysis" + + +class Decision: + """A decision made by a worker.""" + + def __init__(self, decision_type: DecisionType, reason: str, context: dict): + self.decision_type = decision_type + self.reason = reason + self.context = context + + def to_dict(self) -> dict: + return { + "decision_type": self.decision_type.value, + "reason": self.reason, + "context": self.context, + } + + +def _parse_time_slot(time_slot: str) -> Optional[tuple[int, int]]: + """Parse time slot string (e.g., "8-10h") to (start_hour, end_hour). + + Returns: + (start_hour, end_hour) tuple, or None if invalid format + """ + try: + parts = time_slot.replace("h", "").split("-") + if len(parts) != 2: + return None + return (int(parts[0]), int(parts[1])) + except (ValueError, AttributeError): + return None + + +def _get_current_slot(time_distribution: list[dict], current_offset: timedelta) -> Optional[dict]: + """Get current time slot based on offset. + + Returns: + Current slot dict, or None if not in any slot + """ + current_hours = current_offset.total_seconds() / 3600 + + for slot in time_distribution: + time_slot = slot.get("time_slot", "") + parsed = _parse_time_slot(time_slot) + if parsed is None: + continue + + start_hour, end_hour = parsed + if start_hour <= current_hours < end_hour: + return slot + + return None + + +def _count_sessions_in_slot( + eval_id: str, slot: dict, eval_started_at: datetime, session: Session +) -> int: + """Count sessions created in a time slot. + + Args: + eval_id: Evaluation ID + slot: Time slot dict (e.g., {"time_slot": "8-10h", "sessions": 2}) + eval_started_at: When the eval started + session: Database session + + Returns: + Number of sessions created in this slot + """ + time_slot = slot.get("time_slot", "") + parsed = _parse_time_slot(time_slot) + if parsed is None: + return 0 + + start_hour, end_hour = parsed + + # Convert to naive datetime for SQLite comparison + slot_start = (as_utc(eval_started_at) + timedelta(hours=start_hour)).replace(tzinfo=None) + slot_end = (as_utc(eval_started_at) + timedelta(hours=end_hour)).replace(tzinfo=None) + + # Count sessions created within slot time range + sessions = session.exec( + select(IntelligentEvalSessionDB).where( + IntelligentEvalSessionDB.eval_id == eval_id, + IntelligentEvalSessionDB.created_at >= slot_start, + IntelligentEvalSessionDB.created_at < slot_end, + ) + ).all() + + return len(sessions) + + +def _has_high_severity_issues(eval_id: str, session: Session) -> bool: + """Check if any completed session has high severity issues. + + Returns: + True if any session's verdict contains high severity issue + """ + completed_sessions = session.exec( + select(IntelligentEvalSessionDB).where( + IntelligentEvalSessionDB.eval_id == eval_id, + IntelligentEvalSessionDB.status == "completed", + ) + ).all() + + for sess in completed_sessions: + verdict = sess.get_verdict() + if verdict and verdict.get("severity") == "high": + return True + + return False + + +def make_decision(eval_db: IntelligentEvalDB, session: Session) -> Decision: + """Make a decision based on current evaluation state. + + Args: + eval_db: Evaluation database record + session: Database session + + Returns: + Decision object + """ + # Check if eval is still executing + if eval_db.status != IntelligentEvalStatus.EXECUTING.value: + return Decision( + DecisionType.WAIT, + f"评估状态为 {eval_db.status},不在执行中", + {"status": eval_db.status}, + ) + + # Check if eval has plan and started_at + if not eval_db.plan or not eval_db.started_at: + return Decision( + DecisionType.WAIT, + "评估缺少计划或未开始", + {"has_plan": bool(eval_db.plan), "has_started_at": bool(eval_db.started_at)}, + ) + + plan = eval_db.get_plan() + time_distribution = plan.get("time_distribution", []) + estimated_sessions = plan.get("estimated_sessions", 0) + + # Calculate current offset + current_offset = utc_now() - as_utc(eval_db.started_at) + current_hours = current_offset.total_seconds() / 3600 + + # Get all sessions + all_sessions = session.exec( + select(IntelligentEvalSessionDB).where(IntelligentEvalSessionDB.eval_id == eval_db.id) + ).all() + + completed_sessions = [s for s in all_sessions if s.status == "completed"] + + # Check if all sessions completed and estimated reached + if len(completed_sessions) >= estimated_sessions: + return Decision( + DecisionType.START_ANALYSIS, + f"所有 {estimated_sessions} 个会话已完成,开始分析", + { + "completed_sessions": len(completed_sessions), + "estimated_sessions": estimated_sessions, + }, + ) + + # Get current time slot + current_slot = _get_current_slot(time_distribution, current_offset) + + if current_slot is None: + return Decision( + DecisionType.WAIT, + f"当前时间偏移 {current_hours:.1f}h 不在任何时段内", + {"current_offset_hours": current_hours}, + ) + + # Check if current slot has deficit + slot_name = current_slot.get("time_slot", "") + expected_sessions = current_slot.get("sessions", 0) + current_sessions = _count_sessions_in_slot(eval_db.id, current_slot, eval_db.started_at, session) + + deficit = expected_sessions - current_sessions + + if deficit > 0: + return Decision( + DecisionType.EXECUTE_SESSION, + f"时段 {slot_name} 欠账 {deficit} 个会话", + { + "current_slot": slot_name, + "expected_sessions": expected_sessions, + "current_sessions": current_sessions, + "deficit": deficit, + }, + ) + + # Check if any high severity issues found + if _has_high_severity_issues(eval_db.id, session): + return Decision( + DecisionType.EXECUTE_SESSION, + "发现高严重度问题,需要深入挖掘", + {"has_high_severity": True}, + ) + + # No deficit, no high severity issues, wait + return Decision( + DecisionType.WAIT, + f"时段 {slot_name} 无欠账,等待下一时段", + { + "current_slot": slot_name, + "expected_sessions": expected_sessions, + "current_sessions": current_sessions, + }, + ) + + +def is_eval_completed(eval_db: IntelligentEvalDB, session: Session) -> bool: + """Check if evaluation is completed. + + An eval is completed when: + 1. All sessions are completed + 2. Report is submitted + + Returns: + True if eval is completed + """ + if eval_db.status != IntelligentEvalStatus.EXECUTING.value: + return False + + if not eval_db.plan: + return False + + plan = eval_db.get_plan() + estimated_sessions = plan.get("estimated_sessions", 0) + + # Check if all sessions completed + all_sessions = session.exec( + select(IntelligentEvalSessionDB).where(IntelligentEvalSessionDB.eval_id == eval_db.id) + ).all() + + completed_sessions = [s for s in all_sessions if s.status == "completed"] + + if len(completed_sessions) < estimated_sessions: + return False + + # Check if report submitted + if not eval_db.report: + return False + + return True diff --git a/tests/integration/test_intelligent_eval_e2e.py b/tests/integration/test_intelligent_eval_e2e.py new file mode 100644 index 0000000..4f269f5 --- /dev/null +++ b/tests/integration/test_intelligent_eval_e2e.py @@ -0,0 +1,183 @@ +"""End-to-end test for intelligent eval with cron pool.""" + +from datetime import timedelta + +import pytest +from sqlmodel import Session, select + +from agenteval.intelligent_eval import task_queue +from agenteval.intelligent_eval.decision import DecisionType, is_eval_completed, make_decision +from agenteval.intelligent_eval.models import IntelligentEvalStatus +from agenteval.storage.db import ( + IntelligentEvalDB, + IntelligentEvalDecisionLogDB, + IntelligentEvalSessionDB, + IntelligentEvalTaskQueueDB, + utc_now, +) + + +def test_end_to_end_eval_lifecycle(db_session: Session): + """Test end-to-end eval lifecycle: create -> scan -> decide -> execute -> complete.""" + # 1. Create eval + eval_db = IntelligentEvalDB( + name="test-eval", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=8, minutes=30), + ) + eval_db.set_plan({ + "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], + "estimated_sessions": 2, + }) + db_session.add(eval_db) + db_session.commit() + + # 2. Scan and enqueue tasks + enqueued = task_queue.scan_and_enqueue_tasks(db_session) + assert enqueued == 1 + + # Verify task created + task = db_session.exec( + select(IntelligentEvalTaskQueueDB).where(IntelligentEvalTaskQueueDB.eval_id == eval_db.id) + ).first() + assert task is not None + assert task.status == "pending" + assert task.reason == "slot_due" + + # 3. Worker makes decision (should be EXECUTE_SESSION) + decision = make_decision(eval_db, db_session) + assert decision.decision_type == DecisionType.EXECUTE_SESSION + assert "欠账" in decision.reason + + # Record decision log + log = IntelligentEvalDecisionLogDB( + eval_id=eval_db.id, + decision_type=decision.decision_type.value, + reason=decision.reason, + cron_id="cron-123", + ) + log.set_context(decision.context) + db_session.add(log) + db_session.commit() + + # 4. Execute session (simulate) + session_db = IntelligentEvalSessionDB( + eval_id=eval_db.id, + target_id="target1", + status="running", + created_at=utc_now() - timedelta(minutes=20), + ) + db_session.add(session_db) + db_session.commit() + + # 5. Complete session + session_db.status = "completed" + session_db.set_verdict({"severity": "medium", "issues": ["minor issue"]}) + db_session.commit() + + # 6. Worker makes decision again (should still be EXECUTE_SESSION, deficit = 1) + decision2 = make_decision(eval_db, db_session) + assert decision2.decision_type == DecisionType.EXECUTE_SESSION + assert "欠账 1 个会话" in decision2.reason + + # 7. Execute second session + session_db2 = IntelligentEvalSessionDB( + eval_id=eval_db.id, + target_id="target1", + status="running", + created_at=utc_now() - timedelta(minutes=10), + ) + db_session.add(session_db2) + db_session.commit() + + session_db2.status = "completed" + session_db2.set_verdict({"severity": "low", "issues": []}) + db_session.commit() + + # 8. Worker makes decision again (should be START_ANALYSIS) + decision3 = make_decision(eval_db, db_session) + assert decision3.decision_type == DecisionType.START_ANALYSIS + assert "所有 2 个会话已完成" in decision3.reason + + # 9. Submit report + eval_db.set_report({"summary": "Test report", "findings": []}) + db_session.commit() + + # 10. Check if eval is completed + assert is_eval_completed(eval_db, db_session) is True + + # 11. Verify decision logs + logs = db_session.exec( + select(IntelligentEvalDecisionLogDB) + .where(IntelligentEvalDecisionLogDB.eval_id == eval_db.id) + .order_by(IntelligentEvalDecisionLogDB.created_at) + ).all() + assert len(logs) >= 1 # At least one decision log + + +def test_decision_logs_complete_history(db_session: Session): + """Test that decision logs capture complete history.""" + eval_db = IntelligentEvalDB( + name="test-eval", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=8, minutes=30), + ) + eval_db.set_plan({ + "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], + "estimated_sessions": 2, + }) + db_session.add(eval_db) + db_session.commit() + + # Simulate multiple worker ticks + decisions_made = [] + + # Tick 1: No sessions yet + decision1 = make_decision(eval_db, db_session) + decisions_made.append(decision1) + log1 = IntelligentEvalDecisionLogDB( + eval_id=eval_db.id, + decision_type=decision1.decision_type.value, + reason=decision1.reason, + cron_id="cron-123", + ) + log1.set_context(decision1.context) + db_session.add(log1) + db_session.commit() + + # Tick 2: One session created + session_db = IntelligentEvalSessionDB( + eval_id=eval_db.id, + target_id="target1", + status="running", + created_at=utc_now() - timedelta(minutes=20), + ) + db_session.add(session_db) + db_session.commit() + + decision2 = make_decision(eval_db, db_session) + decisions_made.append(decision2) + log2 = IntelligentEvalDecisionLogDB( + eval_id=eval_db.id, + decision_type=decision2.decision_type.value, + reason=decision2.reason, + cron_id="cron-123", + ) + log2.set_context(decision2.context) + db_session.add(log2) + db_session.commit() + + # Verify logs + logs = db_session.exec( + select(IntelligentEvalDecisionLogDB) + .where(IntelligentEvalDecisionLogDB.eval_id == eval_db.id) + .order_by(IntelligentEvalDecisionLogDB.created_at) + ).all() + + assert len(logs) == 2 + assert logs[0].decision_type == DecisionType.EXECUTE_SESSION.value + assert logs[1].decision_type == DecisionType.EXECUTE_SESSION.value + assert logs[0].get_context()["deficit"] == 2 + assert logs[1].get_context()["deficit"] == 1 diff --git a/tests/unit/test_intelligent_eval_decision.py b/tests/unit/test_intelligent_eval_decision.py new file mode 100644 index 0000000..8ab6b7f --- /dev/null +++ b/tests/unit/test_intelligent_eval_decision.py @@ -0,0 +1,286 @@ +"""Unit tests for worker decision logic.""" + +from datetime import timedelta + +import pytest +from sqlmodel import Session + +from agenteval.intelligent_eval.decision import DecisionType, is_eval_completed, make_decision +from agenteval.intelligent_eval.models import IntelligentEvalStatus +from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalSessionDB, utc_now + + +def test_decision_wait_not_executing(db_session: Session): + """Test decision is WAIT when eval is not executing.""" + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.COMPLETED.value, + ) + db_session.add(eval_db) + db_session.commit() + + decision = make_decision(eval_db, db_session) + assert decision.decision_type == DecisionType.WAIT + assert "不在执行中" in decision.reason + + +def test_decision_wait_no_plan(db_session: Session): + """Test decision is WAIT when eval has no plan.""" + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now(), + ) + db_session.add(eval_db) + db_session.commit() + + decision = make_decision(eval_db, db_session) + assert decision.decision_type == DecisionType.WAIT + assert "缺少计划" in decision.reason + + +def test_decision_start_analysis_all_completed(db_session: Session): + """Test decision is START_ANALYSIS when all sessions completed.""" + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=10), + ) + eval_db.set_plan({ + "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], + "estimated_sessions": 2, + }) + db_session.add(eval_db) + db_session.commit() + + # Add 2 completed sessions + for _ in range(2): + session_db = IntelligentEvalSessionDB( + eval_id=eval_db.id, + target_id="target1", + status="completed", + ) + db_session.add(session_db) + db_session.commit() + + decision = make_decision(eval_db, db_session) + assert decision.decision_type == DecisionType.START_ANALYSIS + assert "所有 2 个会话已完成" in decision.reason + + +def test_decision_execute_session_slot_deficit(db_session: Session): + """Test decision is EXECUTE_SESSION when current slot has deficit.""" + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=8, minutes=30), # 8.5 hours ago + ) + eval_db.set_plan({ + "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], + "estimated_sessions": 2, + }) + db_session.add(eval_db) + db_session.commit() + + # Add 1 session created within the 8-10h slot (deficit = 1) + # Slot starts at started_at + 8h = 0.5h ago + session_db = IntelligentEvalSessionDB( + eval_id=eval_db.id, + target_id="target1", + status="running", + created_at=utc_now() - timedelta(minutes=20), # 20 minutes ago, within slot + ) + db_session.add(session_db) + db_session.commit() + + decision = make_decision(eval_db, db_session) + assert decision.decision_type == DecisionType.EXECUTE_SESSION + assert "欠账 1 个会话" in decision.reason + assert decision.context["current_slot"] == "8-10h" + assert decision.context["deficit"] == 1 + + +def test_decision_execute_session_high_severity(db_session: Session): + """Test decision is EXECUTE_SESSION when high severity issue found.""" + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=8, minutes=30), + ) + eval_db.set_plan({ + "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], + "estimated_sessions": 3, # Not all completed yet + }) + db_session.add(eval_db) + db_session.commit() + + # Add 2 completed sessions within the slot (no deficit) + for _ in range(2): + session_db = IntelligentEvalSessionDB( + eval_id=eval_db.id, + target_id="target1", + status="completed", + created_at=utc_now() - timedelta(minutes=20), # Within slot + ) + # One with high severity + session_db.set_verdict({"severity": "high", "issues": ["critical bug"]}) + db_session.add(session_db) + db_session.commit() + + decision = make_decision(eval_db, db_session) + assert decision.decision_type == DecisionType.EXECUTE_SESSION + assert "高严重度问题" in decision.reason + + +def test_decision_wait_no_deficit(db_session: Session): + """Test decision is WAIT when no deficit in current slot.""" + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=8, minutes=30), + ) + eval_db.set_plan({ + "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], + "estimated_sessions": 3, # Not all completed yet + }) + db_session.add(eval_db) + db_session.commit() + + # Add 2 sessions within the slot (no deficit) + for _ in range(2): + session_db = IntelligentEvalSessionDB( + eval_id=eval_db.id, + target_id="target1", + status="running", + created_at=utc_now() - timedelta(minutes=20), # Within slot + ) + db_session.add(session_db) + db_session.commit() + + decision = make_decision(eval_db, db_session) + assert decision.decision_type == DecisionType.WAIT + assert "无欠账" in decision.reason + + +def test_decision_wait_outside_slots(db_session: Session): + """Test decision is WAIT when current time is outside all slots.""" + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=5), # 5 hours ago, outside 8-10h + ) + eval_db.set_plan({ + "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], + "estimated_sessions": 2, + }) + db_session.add(eval_db) + db_session.commit() + + decision = make_decision(eval_db, db_session) + assert decision.decision_type == DecisionType.WAIT + assert "不在任何时段内" in decision.reason + + +def test_is_eval_completed_not_executing(db_session: Session): + """Test is_eval_completed returns False when eval is not executing.""" + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.COMPLETED.value, + ) + db_session.add(eval_db) + db_session.commit() + + assert is_eval_completed(eval_db, db_session) is False + + +def test_is_eval_completed_no_plan(db_session: Session): + """Test is_eval_completed returns False when eval has no plan.""" + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + ) + db_session.add(eval_db) + db_session.commit() + + assert is_eval_completed(eval_db, db_session) is False + + +def test_is_eval_completed_sessions_not_finished(db_session: Session): + """Test is_eval_completed returns False when not all sessions completed.""" + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + ) + eval_db.set_plan({"estimated_sessions": 2}) + db_session.add(eval_db) + db_session.commit() + + # Add only 1 completed session + session_db = IntelligentEvalSessionDB( + eval_id=eval_db.id, + target_id="target1", + status="completed", + ) + db_session.add(session_db) + db_session.commit() + + assert is_eval_completed(eval_db, db_session) is False + + +def test_is_eval_completed_no_report(db_session: Session): + """Test is_eval_completed returns False when report not submitted.""" + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + ) + eval_db.set_plan({"estimated_sessions": 2}) + db_session.add(eval_db) + db_session.commit() + + # Add 2 completed sessions + for _ in range(2): + session_db = IntelligentEvalSessionDB( + eval_id=eval_db.id, + target_id="target1", + status="completed", + ) + db_session.add(session_db) + db_session.commit() + + assert is_eval_completed(eval_db, db_session) is False + + +def test_is_eval_completed_true(db_session: Session): + """Test is_eval_completed returns True when all conditions met.""" + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + ) + eval_db.set_plan({"estimated_sessions": 2}) + eval_db.set_report({"summary": "test report"}) + db_session.add(eval_db) + db_session.commit() + + # Add 2 completed sessions + for _ in range(2): + session_db = IntelligentEvalSessionDB( + eval_id=eval_db.id, + target_id="target1", + status="completed", + ) + db_session.add(session_db) + db_session.commit() + + assert is_eval_completed(eval_db, db_session) is True