eb4944a8bd
常见故障自愈有上限，超限收敛终态且可见：任务 attempts 上限、会话过期、 planning 双闸、executing 超窗兜底、触发失败计数判死、孤儿 agent 双管、 fire-and-forget 触发；open_session 预算硬闸门、settle 按终态区分、报告 scores 归一化；cron 池遗留面全删。
131 lines
3.9 KiB
Python
131 lines
3.9 KiB
Python
"""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"
|
||
|
||
|
||
def test_stale_assigned_abandoned_at_attempts_cap(db_session: Session):
|
||
"""ADR-0011：attempts 达到上限后任务置 failed（不再重入队）并补录决策日志。"""
|
||
from agenteval.intelligent_eval.task_queue import MAX_TASK_ATTEMPTS
|
||
from agenteval.storage.db import IntelligentEvalDecisionLogDB
|
||
from sqlmodel import select
|
||
|
||
ev = _make_executing_eval(db_session)
|
||
task = IntelligentEvalTaskQueueDB(
|
||
eval_id=ev.id,
|
||
status="assigned",
|
||
priority=1,
|
||
reason="slot_due",
|
||
assigned_cron_id="manual-run-cap",
|
||
assigned_at=utc_now() - timedelta(minutes=STALE_ASSIGNED_MINUTES + 5),
|
||
attempts=MAX_TASK_ATTEMPTS - 1,
|
||
)
|
||
db_session.add(task)
|
||
db_session.commit()
|
||
|
||
assert requeue_stale_assigned_tasks(db_session) == 0
|
||
|
||
db_session.refresh(task)
|
||
assert task.status == "failed"
|
||
assert task.attempts == MAX_TASK_ATTEMPTS
|
||
assert task.completed_at is not None
|
||
assert "放弃" in task.error
|
||
|
||
logs = db_session.exec(
|
||
select(IntelligentEvalDecisionLogDB).where(
|
||
IntelligentEvalDecisionLogDB.eval_id == ev.id,
|
||
IntelligentEvalDecisionLogDB.decision_type == "task_abandoned",
|
||
)
|
||
).all()
|
||
assert len(logs) == 1
|