AgentEvalTool/backend/agenteval/intelligent_eval/task_queue.py

"""Task queue for intelligent evaluations (任务队列).

Platform scans executing evals every minute and enqueues tasks for
OpenClaw workers to pick up. Tasks are prioritized by:
1. Time slot due (时段到期)
2. Session deficit (欠账多)
3. Wait time (等待时间长)
"""

from datetime import timedelta
from typing import Optional

from sqlmodel import Session, select

from agenteval.intelligent_eval.models import IntelligentEvalStatus
from agenteval.storage.db import (
    IntelligentEvalDB,
    IntelligentEvalTaskQueueDB,
    utc_now,
)


def _is_slot_due(slot: dict, current_offset: timedelta) -> bool:
    """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:
    """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:
    """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]:
    """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:
    """Check if eval already has a pending task (去重)."""
    existing = session.exec(
        select(IntelligentEvalTaskQueueDB).where(
            IntelligentEvalTaskQueueDB.eval_id == eval_id,
            IntelligentEvalTaskQueueDB.status == "pending",
        )
    ).first()
    return existing is not None


def scan_and_enqueue_tasks(session: Session) -> int:
    """Scan all executing evals and enqueue tasks.

    Returns:
        Number of tasks enqueued
    """
    # Get all executing evals
    evals = session.exec(
        select(IntelligentEvalDB).where(IntelligentEvalDB.status == IntelligentEvalStatus.EXECUTING.value)
    ).all()

    enqueued = 0

    for eval_db in evals:
        # Check if eval needs attention
        reason = _get_attention_reason(eval_db, session)
        if reason is None:
            continue

        # Check if already has pending task (去重)
        if _has_pending_task(eval_db.id, session):
            continue

        # Calculate priority
        priority = _calculate_priority(eval_db, session)

        # Create task
        task = IntelligentEvalTaskQueueDB(
            eval_id=eval_db.id,
            status="pending",
            priority=priority,
            reason=reason,
            created_at=utc_now(),
            updated_at=utc_now(),
        )
        session.add(task)
        enqueued += 1

    session.commit()
    return enqueued


def get_next_task(session: Session) -> Optional[IntelligentEvalTaskQueueDB]:
    """Get next pending task (highest priority).

    Returns:
        Task with lowest priority value (highest priority), or None
    """
    task = session.exec(
        select(IntelligentEvalTaskQueueDB)
        .where(IntelligentEvalTaskQueueDB.status == "pending")
        .order_by(IntelligentEvalTaskQueueDB.priority, IntelligentEvalTaskQueueDB.created_at)
        .limit(1)
    ).first()
    return task


def assign_task(task_id: str, cron_id: str, session: Session) -> bool:
    """Assign a task to a cron.

    Returns:
        True if assigned successfully, False if task not found or already assigned
    """
    task = session.get(IntelligentEvalTaskQueueDB, task_id)
    if task is None or task.status != "pending":
        return False

    task.status = "assigned"
    task.assigned_cron_id = cron_id
    task.assigned_at = utc_now()
    task.updated_at = utc_now()
    session.commit()
    return True


def complete_task(task_id: str, success: bool, error: Optional[str], session: Session) -> bool:
    """Mark a task as completed or failed.

    Returns:
        True if completed successfully, False if task not found
    """
    task = session.get(IntelligentEvalTaskQueueDB, task_id)
    if task is None:
        return False

    task.status = "completed" if success else "failed"
    task.completed_at = utc_now()
    task.error = error
    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