"""Shared domain logic for intelligent-eval scheduling, priority, attention. Centralises the time-slot / deficit / severity calculations that previously lived in `task_queue` and `decision` (P1 deepening, S1). Both modules now delegate here so a single source of truth governs "8-10h"-style semantics. """ from datetime import datetime, timedelta from typing import Optional from sqlmodel import Session, select from agenteval.storage.db import ( IntelligentEvalDB, IntelligentEvalSessionDB, as_utc, utc_now, ) # --------------------------------------------------------------------------- # Time-slot parsing # --------------------------------------------------------------------------- def parse_time_slot(time_slot: str) -> Optional[tuple[float, float]]: """Parse a slot into (start_hours, end_hours), both in hours. Supports hour-level ("8-10h" -> (8, 10)) and minute-level ("0-20min" -> (0, 1/3)) slots. planner 对短窗口(如 1h)会用分钟级时段,长窗口用 小时级;统一换算成小时(float)供上层判断。Returns None on bad format. """ try: raw = time_slot.strip().lower() factor = 1.0 if raw.endswith("min"): raw = raw[:-3] factor = 1.0 / 60 elif raw.endswith("h"): raw = raw[:-1] factor = 1.0 parts = raw.split("-") if len(parts) != 2: return None return (int(parts[0]) * factor, int(parts[1]) * factor) except (ValueError, AttributeError): return None def is_slot_due(slot: dict, current_offset: timedelta) -> bool: """True if the slot's start hour has been reached.""" parsed = parse_time_slot(slot.get("time_slot", "")) if parsed is None: return False start_hour, _ = parsed return current_offset >= timedelta(hours=start_hour) def get_current_slot(time_distribution: list[dict], current_offset: timedelta) -> Optional[dict]: """Return the slot dict whose [start, end) contains the current offset.""" current_hours = current_offset.total_seconds() / 3600 for slot in time_distribution: parsed = parse_time_slot(slot.get("time_slot", "")) if parsed is None: continue start_hour, end_hour = parsed if start_hour <= current_hours < end_hour: return slot return None # --------------------------------------------------------------------------- # Session accounting # --------------------------------------------------------------------------- def count_sessions_in_slot( eval_id: str, slot: dict, eval_started_at: datetime, session: Session ) -> int: """Count sessions of this eval whose created_at falls in the slot window.""" parsed = parse_time_slot(slot.get("time_slot", "")) if parsed is None: return 0 start_hour, end_hour = parsed 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) rows = session.exec( select(IntelligentEvalSessionDB).where( IntelligentEvalSessionDB.eval_id == eval_id, IntelligentEvalSessionDB.created_at >= slot_start, IntelligentEvalSessionDB.created_at < slot_end, ) ).all() return len(rows) def count_total_sessions(eval_id: str, session: Session) -> int: rows = session.exec( select(IntelligentEvalSessionDB).where(IntelligentEvalSessionDB.eval_id == eval_id) ).all() return len(rows) def calculate_session_deficit(eval_db: IntelligentEvalDB, session: Session) -> int: """Total sessions that should exist by now minus what actually exists.""" if not eval_db.plan: return 0 plan = eval_db.get_plan() time_distribution = plan.get("time_distribution", []) current_offset = ( utc_now() - as_utc(eval_db.started_at) if eval_db.started_at else timedelta(0) ) should_have = sum( slot.get("sessions", 0) for slot in time_distribution if is_slot_due(slot, current_offset) ) return max(0, should_have - count_total_sessions(eval_db.id, session)) # --------------------------------------------------------------------------- # Priority # --------------------------------------------------------------------------- def calculate_priority(eval_db: IntelligentEvalDB, session: Session) -> int: """Smaller value = higher priority. Base 100 minus slot-due, deficit, wait.""" priority = 100 if eval_db.plan and eval_db.started_at: plan = eval_db.get_plan() current_offset = utc_now() - as_utc(eval_db.started_at) if any(is_slot_due(s, current_offset) for s in plan.get("time_distribution", [])): priority -= 50 priority -= calculate_session_deficit(eval_db, session) * 10 if eval_db.started_at: wait_min = (utc_now() - as_utc(eval_db.started_at)).total_seconds() / 60 priority -= min(int(wait_min / 10), 20) return max(priority, 1) # --------------------------------------------------------------------------- # Attention / severity # --------------------------------------------------------------------------- def get_attention_reason(eval_db: IntelligentEvalDB, session: Session) -> Optional[str]: """Returns 'slot_due' / 'all_sessions_completed' / None.""" if not eval_db.plan or not eval_db.started_at: return None plan = eval_db.get_plan() time_distribution = plan.get("time_distribution", []) current_offset = utc_now() - as_utc(eval_db.started_at) if any(is_slot_due(s, current_offset) for s in time_distribution): if calculate_session_deficit(eval_db, session) > 0: return "slot_due" sessions = session.exec( select(IntelligentEvalSessionDB).where(IntelligentEvalSessionDB.eval_id == eval_db.id) ).all() # ADR-0011:expired/failed 同为会话终态——存在过期会话时也必须触发 analyst, # 否则评估永远等不到"全部完成"而卡在 executing if sessions and all(s.status in ("completed", "failed", "expired") for s in sessions): if len(sessions) >= plan.get("estimated_sessions", 0): return "all_sessions_completed" return None def has_high_severity_issues(eval_id: str, session: Session) -> bool: completed = session.exec( select(IntelligentEvalSessionDB).where( IntelligentEvalSessionDB.eval_id == eval_id, IntelligentEvalSessionDB.status == "completed", ) ).all() for s in completed: verdict = s.get_verdict() if verdict and verdict.get("severity") == "high": return True return False