diff --git a/backend/agenteval/intelligent_eval/task_queue.py b/backend/agenteval/intelligent_eval/task_queue.py index 2c01ef9..114f8e2 100644 --- a/backend/agenteval/intelligent_eval/task_queue.py +++ b/backend/agenteval/intelligent_eval/task_queue.py @@ -226,3 +226,47 @@ def get_next_task_with_eval(session: Session) -> Optional[dict]: }, }, } + + +# --------------------------------------------------------------------------- +# 卡死恢复(方案③遗留):assigned 超时重新入队 +# --------------------------------------------------------------------------- + +STALE_ASSIGNED_MINUTES = 10 + + +def requeue_stale_assigned_tasks(session: Session) -> int: + """Requeue tasks that stayed `assigned` too long without completing. + + 方案③ worker 由平台触发 openclaw agent(cron=manual-run-...,非真实 cron), + fault_tolerance 的 stuck 检测(基于 cron pool busy + last_active_at)不适用。 + 若 agent 中断/失败,任务会永久卡在 `assigned`,scan 只查 pending 不会再入队。 + + 这里把「assigned 超过 STALE_ASSIGNED_MINUTES 且对应评估仍 executing」的任务 + 重置为 pending(清空认领信息),平台 scan 循环随后会重新触发 worker 重试。 + + Returns: + 重新入队的任务数。 + """ + threshold = utc_now() - timedelta(minutes=STALE_ASSIGNED_MINUTES) + stale = session.exec( + select(IntelligentEvalTaskQueueDB).where( + IntelligentEvalTaskQueueDB.status == "assigned", + IntelligentEvalTaskQueueDB.assigned_at < threshold, + ) + ).all() + + requeued = 0 + for task in stale: + ev = session.get(IntelligentEvalDB, task.eval_id) + if ev is None or ev.status != IntelligentEvalStatus.EXECUTING.value: + continue + task.status = "pending" + task.assigned_cron_id = None + task.assigned_at = None + task.updated_at = utc_now() + requeued += 1 + + if requeued: + session.commit() + return requeued diff --git a/backend/agenteval/web/app.py b/backend/agenteval/web/app.py index 4708625..8ae5edb 100644 --- a/backend/agenteval/web/app.py +++ b/backend/agenteval/web/app.py @@ -100,8 +100,14 @@ async def _intelligent_eval_scan_loop() -> None: try: session = get_session() try: - from agenteval.intelligent_eval.task_queue import scan_and_enqueue_tasks + from agenteval.intelligent_eval.task_queue import ( + requeue_stale_assigned_tasks, + scan_and_enqueue_tasks, + ) + r = requeue_stale_assigned_tasks(session) + if r: + _logger.info("卡死恢复:%d 个 assigned 任务重新入队", r) n = scan_and_enqueue_tasks(session) if n: _logger.info("智能评估扫描:入队 %d 个 Worker 任务", n) diff --git a/tests/unit/test_task_queue_requeue.py b/tests/unit/test_task_queue_requeue.py new file mode 100644 index 0000000..43b42c0 --- /dev/null +++ b/tests/unit/test_task_queue_requeue.py @@ -0,0 +1,94 @@ +"""Stale-assigned requeue tests (worker crash recovery). + +方案③ worker 由平台触发 openclaw agent(cron=manual-run,非真实 cron), +fault_tolerance 的 stuck 检测不适用;任务可能永久卡 assigned。requeue +把「assigned 超时且评估仍 executing」的任务重置为 pending。 +""" +from datetime import timedelta + +from agenteval.intelligent_eval.models import IntelligentEvalStatus +from agenteval.intelligent_eval.task_queue import ( + STALE_ASSIGNED_MINUTES, + requeue_stale_assigned_tasks, +) +from agenteval.storage.db import ( + IntelligentEvalDB, + IntelligentEvalTaskQueueDB, + utc_now, +) +from sqlmodel import Session + + +def _make_executing_eval(db_session: Session) -> IntelligentEvalDB: + ev = IntelligentEvalDB( + name="req-eval", + target_id="t1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now(), + ) + db_session.add(ev) + db_session.commit() + return ev + + +def test_requeue_stale_assigned_task(db_session: Session): + ev = _make_executing_eval(db_session) + task = IntelligentEvalTaskQueueDB( + eval_id=ev.id, + status="assigned", + priority=1, + reason="slot_due", + assigned_cron_id="manual-run-x", + assigned_at=utc_now() - timedelta(minutes=STALE_ASSIGNED_MINUTES + 5), + ) + db_session.add(task) + db_session.commit() + + assert requeue_stale_assigned_tasks(db_session) == 1 + + db_session.refresh(task) + assert task.status == "pending" + assert task.assigned_cron_id is None + assert task.assigned_at is None + + +def test_fresh_assigned_not_requeued(db_session: Session): + ev = _make_executing_eval(db_session) + task = IntelligentEvalTaskQueueDB( + eval_id=ev.id, + status="assigned", + priority=1, + reason="slot_due", + assigned_cron_id="manual-run-y", + assigned_at=utc_now(), # fresh + ) + db_session.add(task) + db_session.commit() + + assert requeue_stale_assigned_tasks(db_session) == 0 + db_session.refresh(task) + assert task.status == "assigned" + + +def test_stale_assigned_non_executing_not_requeued(db_session: Session): + ev = IntelligentEvalDB( + name="cancelled-eval", + target_id="t1", + status=IntelligentEvalStatus.CANCELLED.value, + ) + db_session.add(ev) + db_session.commit() + task = IntelligentEvalTaskQueueDB( + eval_id=ev.id, + status="assigned", + priority=1, + reason="slot_due", + assigned_cron_id="manual-run-z", + assigned_at=utc_now() - timedelta(minutes=STALE_ASSIGNED_MINUTES + 5), + ) + db_session.add(task) + db_session.commit() + + assert requeue_stale_assigned_tasks(db_session) == 0 + db_session.refresh(task) + assert task.status == "assigned"