eb4944a8bd
常见故障自愈有上限，超限收敛终态且可见：任务 attempts 上限、会话过期、 planning 双闸、executing 超窗兜底、触发失败计数判死、孤儿 agent 双管、 fire-and-forget 触发；open_session 预算硬闸门、settle 按终态区分、报告 scores 归一化；cron 池遗留面全删。
87 lines
2.5 KiB
Python
87 lines
2.5 KiB
Python
"""Worker task resilience tests (Gitea issue #3 / P0).
|
||
|
||
T1 — concurrent assign: only one worker wins; complete×stuck does not lose task.
|
||
"""
|
||
|
||
from agenteval.intelligent_eval.task_queue import assign_task, complete_task, get_next_task
|
||
from agenteval.storage.db import (
|
||
IntelligentEvalTaskQueueDB,
|
||
utc_now,
|
||
)
|
||
from sqlmodel import Session, create_engine
|
||
|
||
# T1 — concurrent assign: only one worker wins
|
||
|
||
|
||
def test_concurrent_assign_only_one_succeeds(db_session):
|
||
task = IntelligentEvalTaskQueueDB(
|
||
eval_id="eval1", status="pending", priority=1, reason="slot_due"
|
||
)
|
||
db_session.add(task)
|
||
db_session.commit()
|
||
|
||
ok1 = assign_task(task.id, "cron-A", db_session)
|
||
ok2 = assign_task(task.id, "cron-B", db_session)
|
||
|
||
assert ok1 is True
|
||
assert ok2 is False
|
||
|
||
db_session.refresh(task)
|
||
assert task.status == "assigned"
|
||
assert task.assigned_cron_id == "cron-A"
|
||
|
||
|
||
def test_concurrent_assign_via_two_sessions(tmp_db_path, db_session):
|
||
task = IntelligentEvalTaskQueueDB(
|
||
eval_id="eval1", status="pending", priority=1, reason="slot_due"
|
||
)
|
||
db_session.add(task)
|
||
db_session.commit()
|
||
|
||
engine2 = create_engine(
|
||
f"sqlite:///{tmp_db_path}",
|
||
connect_args={"check_same_thread": False},
|
||
)
|
||
session2 = Session(engine2)
|
||
try:
|
||
t_main = get_next_task(db_session)
|
||
t_other = get_next_task(session2)
|
||
assert t_main is not None and t_other is not None
|
||
assert t_main.id == t_other.id
|
||
|
||
results = (
|
||
assign_task(t_main.id, "cron-main", db_session),
|
||
assign_task(t_other.id, "cron-other", session2),
|
||
)
|
||
assert sorted(results) == [False, True]
|
||
|
||
session2.expire_all()
|
||
final = db_session.get(IntelligentEvalTaskQueueDB, t_main.id)
|
||
assert final.status == "assigned"
|
||
assert final.assigned_cron_id in {"cron-main", "cron-other"}
|
||
finally:
|
||
session2.close()
|
||
engine2.dispose()
|
||
|
||
|
||
# T1 — complete × stuck-handler convergence: task is never lost
|
||
|
||
|
||
def test_complete_and_stuck_converge_to_terminal_state(db_session):
|
||
task = IntelligentEvalTaskQueueDB(
|
||
eval_id="eval1",
|
||
status="assigned",
|
||
priority=1,
|
||
reason="slot_due",
|
||
assigned_cron_id="cron-1",
|
||
assigned_at=utc_now(),
|
||
)
|
||
db_session.add(task)
|
||
db_session.commit()
|
||
|
||
complete_task(task.id, True, None, db_session)
|
||
complete_task(task.id, False, "Cron stuck", db_session)
|
||
|
||
db_session.refresh(task)
|
||
assert task.status in {"completed", "failed"}
|