Some checks failed
CI / test (push) Failing after 4m21s
P1 deepening (issue #7): S1: is the single source of truth for time-slot parsing, slot-due checks, session deficit, priority, attention reason, and high-severity detection. and now delegate their internal helpers to while keeping the same signatures (tests continue to pass via the thin wrappers). S4: encapsulates the stuck-cron task settlement (fail current task + enqueue retry). calls it instead of the previous runtime import of . No observable behaviour change — 873 passed + 5 xfailed unchanged.
160 lines
5.9 KiB
Python
160 lines
5.9 KiB
Python
"""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[int, int]]:
|
|
"""Parse "8-10h" -> (8, 10). Returns None on bad 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 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()
|
|
if sessions and all(s.status == "completed" 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
|