fix(intelligent-eval): add lifespan scan-loop for worker task enqueue
All checks were successful
CI / test (push) Successful in 3m53s
All checks were successful
CI / test (push) Successful in 3m53s
scan_and_enqueue_tasks had no scheduler: the OpenClaw Worker wakes every minute but the platform never enqueued executing evals, so the queue was always empty. lifespan now starts an asyncio background task that scans executing intelligent evals every 60s (aligned with the Worker wake), cancelled cleanly on shutdown. Verified by a new startup test (879 total).
This commit is contained in:
parent
6309b6abca
commit
5836b84681
@ -1,5 +1,6 @@
|
|||||||
"""FastAPI web backend for AgentEvalTool."""
|
"""FastAPI web backend for AgentEvalTool."""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
import logging
|
import logging
|
||||||
from contextlib import asynccontextmanager
|
from contextlib import asynccontextmanager
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
@ -30,9 +31,34 @@ from agenteval.web.routers import (
|
|||||||
from agenteval.web.websocket import ws_manager
|
from agenteval.web.websocket import ws_manager
|
||||||
|
|
||||||
|
|
||||||
|
async def _intelligent_eval_scan_loop() -> None:
|
||||||
|
"""Scan executing intelligent evals and enqueue worker tasks every minute.
|
||||||
|
|
||||||
|
v1.1.0 缺陷修复:`scan_and_enqueue_tasks` 此前没有调度点,OpenClaw Worker
|
||||||
|
每分钟唤醒却永远取不到任务。平台启动后每 60s 扫描一次 executing 的评估,
|
||||||
|
与 Worker 的每分钟唤醒对齐。失败不阻断(下次循环继续)。
|
||||||
|
"""
|
||||||
|
_logger = logging.getLogger("agenteval")
|
||||||
|
while True:
|
||||||
|
try:
|
||||||
|
session = get_session()
|
||||||
|
try:
|
||||||
|
from agenteval.intelligent_eval.task_queue import scan_and_enqueue_tasks
|
||||||
|
|
||||||
|
n = scan_and_enqueue_tasks(session)
|
||||||
|
if n:
|
||||||
|
_logger.info("智能评估扫描:入队 %d 个 Worker 任务", n)
|
||||||
|
finally:
|
||||||
|
session.close()
|
||||||
|
except Exception as exc:
|
||||||
|
logging.getLogger("agenteval").warning("智能评估扫描失败(忽略): %s", exc)
|
||||||
|
await asyncio.sleep(60)
|
||||||
|
|
||||||
|
|
||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
async def lifespan(_: FastAPI):
|
async def lifespan(_: FastAPI):
|
||||||
init_db()
|
init_db()
|
||||||
|
scan_task = asyncio.create_task(_intelligent_eval_scan_loop())
|
||||||
# 恢复耐久 Campaign/智能作业,并清理无法安全重放的中断运行(尽力而为,不阻断启动)。
|
# 恢复耐久 Campaign/智能作业,并清理无法安全重放的中断运行(尽力而为,不阻断启动)。
|
||||||
try:
|
try:
|
||||||
session = get_session()
|
session = get_session()
|
||||||
@ -65,6 +91,12 @@ async def lifespan(_: FastAPI):
|
|||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
logging.getLogger("agenteval").warning("启动清理失败(忽略): %s", exc)
|
logging.getLogger("agenteval").warning("启动清理失败(忽略): %s", exc)
|
||||||
yield
|
yield
|
||||||
|
# 优雅停止后台扫描,再停其他进程内任务。
|
||||||
|
scan_task.cancel()
|
||||||
|
try:
|
||||||
|
await scan_task
|
||||||
|
except asyncio.CancelledError:
|
||||||
|
pass
|
||||||
# 优雅停止所有进程内任务:先停活动调度循环,再停在跑的评测运行,
|
# 优雅停止所有进程内任务:先停活动调度循环,再停在跑的评测运行,
|
||||||
# 最后停活动智能作业(分析 / 周期对比)和 judge 复核。
|
# 最后停活动智能作业(分析 / 周期对比)和 judge 复核。
|
||||||
try:
|
try:
|
||||||
|
|||||||
34
tests/integration/test_intelligent_eval_scan_scheduler.py
Normal file
34
tests/integration/test_intelligent_eval_scan_scheduler.py
Normal file
@ -0,0 +1,34 @@
|
|||||||
|
"""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"
|
||||||
Loading…
Reference in New Issue
Block a user