From 975ed7a6ff79332a80bcb6c418f6ed7d19cf26e6 Mon Sep 17 00:00:00 2001 From: sinohqb Date: Thu, 13 Aug 2026 10:04:24 +0800 Subject: [PATCH] refactor(intelligent-eval): converge scheduling domain (S1) + stuck-task settlement (S4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../agenteval/intelligent_eval/cron_pool.py | 27 +-- .../agenteval/intelligent_eval/decision.py | 92 ++-------- backend/agenteval/intelligent_eval/domain.py | 159 ++++++++++++++++++ .../agenteval/intelligent_eval/task_queue.py | 156 +++++------------ 4 files changed, 219 insertions(+), 215 deletions(-) create mode 100644 backend/agenteval/intelligent_eval/domain.py diff --git a/backend/agenteval/intelligent_eval/cron_pool.py b/backend/agenteval/intelligent_eval/cron_pool.py index 1778910..bd95693 100644 --- a/backend/agenteval/intelligent_eval/cron_pool.py +++ b/backend/agenteval/intelligent_eval/cron_pool.py @@ -10,6 +10,7 @@ from datetime import timedelta from sqlmodel import Session, select from agenteval.intelligent_eval.openclaw_client import OpenClawClient +from agenteval.intelligent_eval.task_queue import requeue_stuck_task from agenteval.storage.db import OpenClawCronPoolDB, utc_now _logger = logging.getLogger("agenteval") @@ -264,8 +265,6 @@ async def handle_stuck_cron(cron: OpenClawCronPoolDB, session: Session, client: session: Database session client: OpenClaw client """ - from agenteval.intelligent_eval.task_queue import complete_task - _logger.warning(f"Handling stuck cron {cron.openclaw_cron_id}") # Mark as stuck @@ -274,28 +273,8 @@ async def handle_stuck_cron(cron: OpenClawCronPoolDB, session: Session, client: # Requeue task if any if cron.current_eval_id: - from agenteval.storage.db import IntelligentEvalTaskQueueDB - - task = session.exec( - select(IntelligentEvalTaskQueueDB).where( - IntelligentEvalTaskQueueDB.eval_id == cron.current_eval_id, - IntelligentEvalTaskQueueDB.status == "assigned", - IntelligentEvalTaskQueueDB.assigned_cron_id == cron.openclaw_cron_id, - ) - ).first() - - if task: - # Mark task as failed - complete_task(task.id, False, "Cron stuck", session) - - # Create new task for retry - new_task = IntelligentEvalTaskQueueDB( - eval_id=cron.current_eval_id, - status="pending", - priority=1, # High priority - reason="cron_stuck_retry", - ) - session.add(new_task) + # P1 deepening (S4): settlement delegated to task_queue. + requeue_stuck_task(cron.current_eval_id, cron.openclaw_cron_id, session) # Delete stuck cron try: diff --git a/backend/agenteval/intelligent_eval/decision.py b/backend/agenteval/intelligent_eval/decision.py index 0e44b13..ff0c7ad 100644 --- a/backend/agenteval/intelligent_eval/decision.py +++ b/backend/agenteval/intelligent_eval/decision.py @@ -46,97 +46,29 @@ class Decision: 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 + """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.parse_time_slot`.""" + from agenteval.intelligent_eval.domain import parse_time_slot as _impl + return _impl(time_slot) 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 + """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.get_current_slot`.""" + from agenteval.intelligent_eval.domain import get_current_slot as _impl + return _impl(time_distribution, current_offset) 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) + """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.count_sessions_in_slot`.""" + from agenteval.intelligent_eval.domain import count_sessions_in_slot as _impl + return _impl(eval_id, slot, eval_started_at, session) 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 + """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.has_high_severity_issues`.""" + from agenteval.intelligent_eval.domain import has_high_severity_issues as _impl + return _impl(eval_id, session) def make_decision(eval_db: IntelligentEvalDB, session: Session) -> Decision: diff --git a/backend/agenteval/intelligent_eval/domain.py b/backend/agenteval/intelligent_eval/domain.py new file mode 100644 index 0000000..4a05e5a --- /dev/null +++ b/backend/agenteval/intelligent_eval/domain.py @@ -0,0 +1,159 @@ +"""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 diff --git a/backend/agenteval/intelligent_eval/task_queue.py b/backend/agenteval/intelligent_eval/task_queue.py index 0fd4dcb..17f5a11 100644 --- a/backend/agenteval/intelligent_eval/task_queue.py +++ b/backend/agenteval/intelligent_eval/task_queue.py @@ -15,132 +15,33 @@ from sqlmodel import Session, select from agenteval.intelligent_eval.models import IntelligentEvalStatus from agenteval.storage.db import ( IntelligentEvalDB, - IntelligentEvalSessionDB, IntelligentEvalTaskQueueDB, - as_utc, utc_now, ) def _is_slot_due(slot: dict, current_offset: timedelta) -> bool: - """Check if a time slot is due (时段到期). - - Args: - slot: Time slot from plan.time_distribution (e.g., {"time_slot": "8-10h", "sessions": 2}) - current_offset: Time elapsed since eval started - - Returns: - True if the slot's start time has passed - """ - time_slot = slot.get("time_slot", "") - if not time_slot: - return False - - # Parse time slot (e.g., "8-10h" -> 8 hours) - try: - start_hour = int(time_slot.split("-")[0].replace("h", "")) - slot_start = timedelta(hours=start_hour) - return current_offset >= slot_start - except (ValueError, IndexError): - return False + """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.is_slot_due`.""" + from agenteval.intelligent_eval.domain import is_slot_due as _impl + return _impl(slot, current_offset) def _calculate_session_deficit(eval_db: IntelligentEvalDB, session: Session) -> int: - """Calculate session deficit (欠账). - - Returns: - Number of sessions that should exist but don't - """ - if not eval_db.plan: - return 0 - - plan = eval_db.get_plan() - time_distribution = plan.get("time_distribution", []) - - # Count current sessions - current_sessions = session.exec( - select(IntelligentEvalSessionDB).where(IntelligentEvalSessionDB.eval_id == eval_db.id) - ).all() - current_count = len(current_sessions) - - # Calculate how many sessions should exist by now - current_offset = utc_now() - as_utc(eval_db.started_at) if eval_db.started_at else timedelta(0) - should_have = 0 - - for slot in time_distribution: - if _is_slot_due(slot, current_offset): - should_have += slot.get("sessions", 0) - - # Deficit = should have - current - deficit = max(0, should_have - current_count) - return deficit + """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.calculate_session_deficit`.""" + from agenteval.intelligent_eval.domain import calculate_session_deficit as _impl + return _impl(eval_db, session) def _calculate_priority(eval_db: IntelligentEvalDB, session: Session) -> int: - """Calculate task priority (越小越优先). - - Priority rules: - - Base priority: 100 - - Time slot due: -50 - - Session deficit: -10 per session - - Wait time: -1 per 10 minutes (max -20) - """ - priority = 100 - - # Check if any time slot is due - if eval_db.plan and eval_db.started_at: - plan = eval_db.get_plan() - time_distribution = plan.get("time_distribution", []) - current_offset = utc_now() - as_utc(eval_db.started_at) - - for slot in time_distribution: - if _is_slot_due(slot, current_offset): - priority -= 50 - break - - # Session deficit - deficit = _calculate_session_deficit(eval_db, session) - priority -= deficit * 10 - - # Wait time - if eval_db.started_at: - wait_minutes = (utc_now() - as_utc(eval_db.started_at)).total_seconds() / 60 - priority -= min(int(wait_minutes / 10), 20) - - return max(priority, 1) + """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.calculate_priority`.""" + from agenteval.intelligent_eval.domain import calculate_priority as _impl + return _impl(eval_db, session) def _get_attention_reason(eval_db: IntelligentEvalDB, session: Session) -> Optional[str]: - """Determine why this eval needs attention. - - Returns: - Reason string, or None if no attention needed - """ - 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) - - # Check if any time slot is due - for slot in time_distribution: - if _is_slot_due(slot, current_offset): - deficit = _calculate_session_deficit(eval_db, session) - if deficit > 0: - return "slot_due" - - # Check if all sessions completed (need analysis) - sessions = session.exec( - select(IntelligentEvalSessionDB).where(IntelligentEvalSessionDB.eval_id == eval_db.id) - ).all() - - if sessions and all(s.status == "completed" for s in sessions): - estimated_sessions = plan.get("estimated_sessions", 0) - if len(sessions) >= estimated_sessions: - return "all_sessions_completed" - - return None + """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.get_attention_reason`.""" + from agenteval.intelligent_eval.domain import get_attention_reason as _impl + return _impl(eval_db, session) def _has_pending_task(eval_id: str, session: Session) -> bool: @@ -245,3 +146,36 @@ def complete_task(task_id: str, success: bool, error: Optional[str], session: Se task.updated_at = utc_now() session.commit() return True + +def requeue_stuck_task(eval_id: str, cron_id: str, session: Session) -> bool: + """Mark the cron-stuck task as failed and enqueue a retry task. + + P1 deepening (S4): the stuck-task settlement logic that previously lived + inside ``cron_pool.handle_stuck_cron`` (with a runtime import) now lives + here as a first-class operation. ``cron_pool`` only calls this. + + Returns: + True if a stuck task was found and requeued, False otherwise. + """ + from agenteval.storage.db import IntelligentEvalTaskQueueDB + + task = session.exec( + select(IntelligentEvalTaskQueueDB).where( + IntelligentEvalTaskQueueDB.eval_id == eval_id, + IntelligentEvalTaskQueueDB.status == "assigned", + IntelligentEvalTaskQueueDB.assigned_cron_id == cron_id, + ) + ).first() + if task is None: + return False + + complete_task(task.id, False, "Cron stuck", session) + + new_task = IntelligentEvalTaskQueueDB( + eval_id=eval_id, + status="pending", + priority=1, + reason="cron_stuck_retry", + ) + session.add(new_task) + return True