feat(intelligent-eval): implement decision logic and e2e flow (ticket 04)

- Add decision.py with worker decision logic (execute_session/wait/start_analysis)
- Implement time slot parsing and current slot detection
- Implement session deficit calculation per time slot
- Implement high severity issue detection
- Implement eval completion detection
- Add 12 unit tests for decision logic
- Add 2 end-to-end tests for complete lifecycle

All 798 tests passing.
This commit is contained in:
sinohqb 2026-08-12 10:10:42 +08:00
parent 30b9cac224
commit fe3399297c
3 changed files with 745 additions and 0 deletions

View File

@ -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

View File

@ -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

View File

@ -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