eb4944a8bd
常见故障自愈有上限，超限收敛终态且可见：任务 attempts 上限、会话过期、 planning 双闸、executing 超窗兜底、触发失败计数判死、孤儿 agent 双管、 fire-and-forget 触发；open_session 预算硬闸门、settle 按终态区分、报告 scores 归一化；cron 池遗留面全删。
336 lines
12 KiB
Python
336 lines
12 KiB
Python
"""Lifespan scan-loop test (v1.1.0 defect fix).
|
||
|
||
`scan_and_enqueue_tasks` previously had no scheduler — the OpenClaw Worker
|
||
wakes every minute but could never pull a task. The lifespan now starts an
|
||
asyncio background task that scans executing evals every 60s. This test
|
||
verifies that on application startup the scan is actually invoked.
|
||
"""
|
||
|
||
from unittest.mock import MagicMock
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
|
||
def test_lifespan_starts_scan_loop(monkeypatch):
|
||
"""Lifespan startup must invoke the intelligent-eval scan loop once."""
|
||
import agenteval.intelligent_eval.task_queue as tq
|
||
import agenteval.web.app as app_mod
|
||
|
||
calls: list[int] = []
|
||
real_scan = tq.scan_and_enqueue_tasks
|
||
|
||
def fake_scan(session):
|
||
calls.append(1)
|
||
return real_scan(session)
|
||
|
||
# The scan loop calls get_session() to open a DB session; replace it with a
|
||
# no-op mock so the test does not touch the real SQLite file.
|
||
monkeypatch.setattr(app_mod, "get_session", lambda: MagicMock())
|
||
monkeypatch.setattr(tq, "scan_and_enqueue_tasks", fake_scan)
|
||
|
||
with TestClient(app_mod.app) as client:
|
||
assert client.get("/api/health").status_code == 200
|
||
|
||
# The background task runs immediately (before its first 60s sleep).
|
||
assert calls, "scan_and_enqueue_tasks should have been invoked on startup"
|
||
|
||
|
||
def _patch_worker_candidates(monkeypatch, candidates: list[str]):
|
||
"""ADR-0011：worker 触发的服务对象由冷却过滤后的候选决定（替代原 _has_pending_task）。"""
|
||
from unittest.mock import MagicMock
|
||
|
||
import agenteval.intelligent_eval.lifecycle as lifecycle_mod
|
||
import agenteval.web.app as app_mod
|
||
|
||
monkeypatch.setattr(lifecycle_mod, "worker_trigger_candidates", lambda session: candidates)
|
||
monkeypatch.setattr(lifecycle_mod, "record_worker_triggers", lambda session, ids: None)
|
||
monkeypatch.setattr(app_mod, "get_session", lambda: MagicMock())
|
||
|
||
|
||
def test_trigger_worker_skips_when_no_pending(monkeypatch):
|
||
"""No pending task → no docker exec invocation."""
|
||
import asyncio
|
||
import subprocess
|
||
from unittest.mock import MagicMock
|
||
|
||
import agenteval.web.app as app_mod
|
||
|
||
calls: list = []
|
||
_patch_worker_candidates(monkeypatch, [])
|
||
|
||
def fake_run(cmd, **kwargs):
|
||
calls.append(cmd)
|
||
return MagicMock(returncode=0, stderr="")
|
||
|
||
monkeypatch.setattr(subprocess, "run", fake_run)
|
||
|
||
assert asyncio.run(app_mod._trigger_intelligent_worker()) is False
|
||
assert not calls, "should not invoke docker exec when queue is empty"
|
||
|
||
|
||
def test_trigger_worker_calls_docker_exec(monkeypatch):
|
||
"""Pending task present → invoke `docker exec openclaw-eval openclaw agent`."""
|
||
import asyncio
|
||
import subprocess
|
||
from unittest.mock import MagicMock
|
||
|
||
import agenteval.web.app as app_mod
|
||
|
||
calls: list = []
|
||
_patch_worker_candidates(monkeypatch, ["ev-1"])
|
||
|
||
def fake_run(cmd, **kwargs):
|
||
calls.append(cmd)
|
||
return MagicMock(returncode=0, stderr="")
|
||
|
||
monkeypatch.setattr(subprocess, "run", fake_run)
|
||
|
||
assert asyncio.run(app_mod._trigger_intelligent_worker()) is True
|
||
assert calls, "docker exec should be invoked"
|
||
joined = " ".join(calls[0])
|
||
assert "docker" in joined and "openclaw" in joined
|
||
assert "agenteval-intelligent-worker" in joined
|
||
assert "--agent" in joined and "main" in joined
|
||
# 独立 session：避免 main 持久 session 上下文缓存污染导致 worker 幻觉不执行
|
||
assert "--session-id" in joined
|
||
assert "agenteval-worker-" in joined
|
||
# ADR-0011 孤儿 agent 双管：容器内 timeout 强杀
|
||
assert "timeout" in joined and "600" in joined
|
||
|
||
|
||
def test_trigger_worker_msg_has_execute_semantics(monkeypatch):
|
||
"""The trigger message must instruct immediate execution (no cron state).
|
||
|
||
`openclaw agent` has no cron state; a bare skill name makes the worker
|
||
skill "decide then wait for the next tick", deadlocking. The message must
|
||
say "立即完成当前任务 / 不要等待下一节拍".
|
||
"""
|
||
import asyncio
|
||
import subprocess
|
||
from unittest.mock import MagicMock
|
||
|
||
import agenteval.web.app as app_mod
|
||
|
||
calls: list = []
|
||
_patch_worker_candidates(monkeypatch, ["ev-1"])
|
||
|
||
def fake_run(cmd, **kwargs):
|
||
calls.append(cmd)
|
||
return MagicMock(returncode=0, stderr="")
|
||
|
||
monkeypatch.setattr(subprocess, "run", fake_run)
|
||
|
||
asyncio.run(app_mod._trigger_intelligent_worker())
|
||
joined = " ".join(calls[0])
|
||
assert "不要等待下一节拍" in joined
|
||
assert "立即完成当前任务" in joined
|
||
assert "agenteval-intelligent-worker" in joined
|
||
assert "agenteval-intelligent-analyst" in joined
|
||
|
||
|
||
def test_trigger_planner_skips_when_no_planning(monkeypatch):
|
||
"""No planning eval → no docker exec invocation."""
|
||
import asyncio
|
||
import subprocess
|
||
from unittest.mock import MagicMock
|
||
|
||
import agenteval.intelligent_eval.lifecycle as lifecycle_mod
|
||
import agenteval.web.app as app_mod
|
||
|
||
calls: list = []
|
||
# ADR-0011：planner 触发前置计数落账（record_planner_triggers），0 时不触发
|
||
monkeypatch.setattr(lifecycle_mod, "record_planner_triggers", lambda session: 0)
|
||
monkeypatch.setattr(app_mod, "get_session", lambda: MagicMock())
|
||
|
||
def fake_run(cmd, **kwargs):
|
||
calls.append(cmd)
|
||
return MagicMock(returncode=0, stderr="")
|
||
|
||
monkeypatch.setattr(subprocess, "run", fake_run)
|
||
|
||
assert asyncio.run(app_mod._trigger_intelligent_planner()) is False
|
||
assert not calls, "should not invoke docker exec when no planning eval"
|
||
|
||
|
||
def test_trigger_planner_calls_docker_exec(monkeypatch):
|
||
"""Planning eval present → invoke `docker exec openclaw-eval openclaw agent` planner skill."""
|
||
import asyncio
|
||
import subprocess
|
||
from unittest.mock import MagicMock
|
||
|
||
import agenteval.intelligent_eval.lifecycle as lifecycle_mod
|
||
import agenteval.web.app as app_mod
|
||
|
||
calls: list = []
|
||
monkeypatch.setattr(lifecycle_mod, "record_planner_triggers", lambda session: 1)
|
||
monkeypatch.setattr(app_mod, "get_session", lambda: MagicMock())
|
||
|
||
def fake_run(cmd, **kwargs):
|
||
calls.append(cmd)
|
||
return MagicMock(returncode=0, stderr="")
|
||
|
||
monkeypatch.setattr(subprocess, "run", fake_run)
|
||
|
||
assert asyncio.run(app_mod._trigger_intelligent_planner()) is True
|
||
assert calls, "docker exec should be invoked"
|
||
joined = " ".join(calls[0])
|
||
assert "docker" in joined and "openclaw" in joined
|
||
assert "agenteval-intelligent-planner" in joined
|
||
# 独立 session（同 worker 防上下文缓存污染）
|
||
assert "--session-id" in joined
|
||
assert "agenteval-planner-" in joined
|
||
assert "立即完成当前任务" in joined
|
||
assert "不要等待下一节拍" in joined
|
||
|
||
|
||
def test_supplement_execute_session_log(monkeypatch, db_session):
|
||
"""Executing eval with deficit and no execute_session log → platform backfills."""
|
||
import agenteval.web.app as app_mod
|
||
from agenteval.intelligent_eval.models import IntelligentEvalStatus
|
||
from agenteval.storage.db import IntelligentEvalDB
|
||
|
||
ev = IntelligentEvalDB(
|
||
name="supp-eval",
|
||
target_id="t1",
|
||
status=IntelligentEvalStatus.EXECUTING.value,
|
||
started_at=__import__("agenteval.storage.db", fromlist=["utc_now"]).utc_now(),
|
||
)
|
||
ev.set_plan({"time_distribution": [{"time_slot": "0-1h", "sessions": 1}], "estimated_sessions": 1})
|
||
db_session.add(ev)
|
||
db_session.commit()
|
||
|
||
assert app_mod._supplement_decision_logs(db_session) == 1
|
||
# second call: already backfilled → 0
|
||
assert app_mod._supplement_decision_logs(db_session) == 0
|
||
|
||
|
||
def test_supplement_start_analysis_log(monkeypatch, db_session):
|
||
"""Executing eval with all sessions completed and no start_analysis → backfill."""
|
||
import agenteval.web.app as app_mod
|
||
from agenteval.intelligent_eval.models import IntelligentEvalStatus
|
||
from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalSessionDB, utc_now
|
||
|
||
ev = IntelligentEvalDB(
|
||
name="supp-eval2",
|
||
target_id="t1",
|
||
status=IntelligentEvalStatus.EXECUTING.value,
|
||
started_at=utc_now(),
|
||
)
|
||
ev.set_plan({"time_distribution": [{"time_slot": "0-1h", "sessions": 1}], "estimated_sessions": 1})
|
||
db_session.add(ev)
|
||
db_session.commit()
|
||
s = IntelligentEvalSessionDB(eval_id=ev.id, target_id=ev.target_id, status="completed", goal="g")
|
||
db_session.add(s)
|
||
db_session.commit()
|
||
|
||
assert app_mod._supplement_decision_logs(db_session) == 1
|
||
|
||
|
||
def test_supplement_completed_backfill(monkeypatch, db_session):
|
||
"""Completed eval with no decision logs → backfill execute_session + start_analysis."""
|
||
import agenteval.web.app as app_mod
|
||
from agenteval.intelligent_eval.models import IntelligentEvalStatus
|
||
from agenteval.storage.db import (
|
||
IntelligentEvalDB,
|
||
IntelligentEvalDecisionLogDB,
|
||
IntelligentEvalSessionDB,
|
||
utc_now,
|
||
)
|
||
from sqlmodel import select
|
||
|
||
ev = IntelligentEvalDB(
|
||
name="supp-completed",
|
||
target_id="t1",
|
||
status=IntelligentEvalStatus.COMPLETED.value,
|
||
started_at=utc_now(),
|
||
)
|
||
ev.set_plan(
|
||
{
|
||
"time_distribution": [
|
||
{"time_slot": "0-1h", "sessions": 1},
|
||
{"time_slot": "1-2h", "sessions": 1},
|
||
],
|
||
"estimated_sessions": 2,
|
||
}
|
||
)
|
||
db_session.add(ev)
|
||
db_session.commit()
|
||
for _ in range(2):
|
||
s = IntelligentEvalSessionDB(eval_id=ev.id, target_id=ev.target_id, status="completed", goal="g")
|
||
db_session.add(s)
|
||
db_session.commit()
|
||
|
||
# 首轮：2 时段 → 2 条 execute_session + 1 条 start_analysis
|
||
assert app_mod._supplement_decision_logs(db_session) == 3
|
||
# 幂等：二次调用不再补
|
||
assert app_mod._supplement_decision_logs(db_session) == 0
|
||
|
||
logs = db_session.exec(
|
||
select(IntelligentEvalDecisionLogDB).where(IntelligentEvalDecisionLogDB.eval_id == ev.id)
|
||
).all()
|
||
types = sorted(x.decision_type for x in logs)
|
||
assert types == ["execute_session", "execute_session", "start_analysis"]
|
||
assert all(x.cron_id == "platform" for x in logs)
|
||
assert all(x.get_context().get("platform_supplemented") for x in logs)
|
||
|
||
|
||
def test_supplement_completed_skips_when_already_logged(monkeypatch, db_session):
|
||
"""Completed eval that already has worker-reported logs is not duplicated."""
|
||
import agenteval.web.app as app_mod
|
||
from agenteval.intelligent_eval.models import IntelligentEvalStatus
|
||
from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalSessionDB, utc_now
|
||
|
||
ev = IntelligentEvalDB(
|
||
name="supp-completed2",
|
||
target_id="t1",
|
||
status=IntelligentEvalStatus.COMPLETED.value,
|
||
started_at=utc_now(),
|
||
)
|
||
ev.set_plan({"time_distribution": [{"time_slot": "0-1h", "sessions": 1}], "estimated_sessions": 1})
|
||
db_session.add(ev)
|
||
db_session.commit()
|
||
s = IntelligentEvalSessionDB(eval_id=ev.id, target_id=ev.target_id, status="completed", goal="g")
|
||
db_session.add(s)
|
||
db_session.commit()
|
||
|
||
from agenteval.intelligent_eval.decision_logs import create_decision_log
|
||
|
||
# 已有 worker 正常上报的两条日志
|
||
create_decision_log(ev.id, "execute_session", "时段0-1h欠账1个会话，需要执行会话", "manual-run-1", {}, db_session)
|
||
create_decision_log(ev.id, "start_analysis", "所有会话已完成，开始分析", "manual-run-1", {}, db_session)
|
||
|
||
assert app_mod._supplement_decision_logs(db_session) == 0
|
||
|
||
|
||
def test_fire_and_forget_does_not_block_and_logs_errors(monkeypatch):
|
||
"""ADR-0011：触发派生 asyncio task 后台执行，异常在完成回调记录而不抛出。"""
|
||
import asyncio
|
||
import logging
|
||
|
||
import agenteval.web.app as app_mod
|
||
|
||
async def boom():
|
||
await asyncio.sleep(0)
|
||
raise RuntimeError("触发爆炸")
|
||
|
||
async def main():
|
||
app_mod._fire_and_forget(boom(), "Worker")
|
||
await asyncio.sleep(0.01) # 让被派生的任务及其完成回调执行
|
||
return "loop-continues"
|
||
|
||
# 直接挂 handler 到 agenteval logger：不依赖 caplog（全量跑时 alembic 迁移
|
||
# 用例的 fileConfig 会 disable_existing_loggers，把 agenteval logger 关掉）
|
||
records: list[logging.LogRecord] = []
|
||
handler = logging.Handler()
|
||
handler.emit = records.append
|
||
logger = logging.getLogger("agenteval")
|
||
was_disabled = logger.disabled
|
||
logger.disabled = False
|
||
logger.addHandler(handler)
|
||
try:
|
||
assert asyncio.run(main()) == "loop-continues"
|
||
finally:
|
||
logger.removeHandler(handler)
|
||
logger.disabled = was_disabled
|
||
assert any("触发爆炸" in r.getMessage() for r in records)
|