fix(runs): mark orphaned running/pending runs failed on startup
Some checks failed
CI / test (push) Failing after 59s
Some checks failed
CI / test (push) Failing after 59s
评测任务是进程内 asyncio 任务,服务重启会中断执行且状态永远停在 running。启动时将遗留的 running/pending 运行标记为 failed(summary 写入 interrupted 错误),清理为尽力而为,不阻断启动。另将仪表盘最近 评测记录的触发方式与版本号标签位置对调。
This commit is contained in:
parent
fbac28bc7e
commit
d23b321225
@ -277,6 +277,25 @@ class RunRepository:
|
||||
self.session.refresh(db)
|
||||
return _run_from_db(db)
|
||||
|
||||
def mark_orphans_failed(self) -> int:
|
||||
"""服务启动时清理:把遗留的 running/pending 运行标记为 failed。
|
||||
|
||||
评测任务是进程内 asyncio 任务,服务重启后不会恢复;不清理则这些
|
||||
运行永远停留在 running(僵尸运行)。
|
||||
"""
|
||||
statement = select(EvalRunDB).where(EvalRunDB.status.in_(["running", "pending"])) # type: ignore[attr-defined]
|
||||
orphans = self.session.exec(statement).all()
|
||||
for db in orphans:
|
||||
db.status = "failed"
|
||||
db.completed_at = db.completed_at or utc_now()
|
||||
summary = db.get_summary() or {}
|
||||
summary["error"] = {"code": "interrupted", "message": "服务重启导致评测中断"}
|
||||
db.set_summary(summary)
|
||||
self.session.add(db)
|
||||
if orphans:
|
||||
self.session.commit()
|
||||
return len(orphans)
|
||||
|
||||
def update(self, run: EvalRun) -> Optional[EvalRun]:
|
||||
existing = self.session.get(EvalRunDB, run.id)
|
||||
if not existing:
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
"""FastAPI web backend for AgentEvalTool."""
|
||||
|
||||
import logging
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
|
||||
@ -8,7 +9,8 @@ from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import FileResponse, JSONResponse
|
||||
|
||||
from agenteval.config import get_settings
|
||||
from agenteval.storage.db import init_db
|
||||
from agenteval.storage.db import get_session, init_db
|
||||
from agenteval.storage.repository import RunRepository
|
||||
from agenteval.version import get_build_info, get_version
|
||||
from agenteval.web.deps import require_api_key
|
||||
from agenteval.web.routers import auth, files, model_configs, proxy, reports, runs, scenarios, stats, targets
|
||||
@ -18,6 +20,17 @@ from agenteval.web.websocket import ws_manager
|
||||
@asynccontextmanager
|
||||
async def lifespan(_: FastAPI):
|
||||
init_db()
|
||||
# 评测任务是进程内 asyncio 任务,重启后不会恢复——清理僵尸运行(尽力而为,不阻断启动)
|
||||
try:
|
||||
session = get_session()
|
||||
try:
|
||||
count = RunRepository(session).mark_orphans_failed()
|
||||
if count:
|
||||
logging.getLogger("agenteval").warning("启动清理:%d 个中断的运行已标记为 failed", count)
|
||||
finally:
|
||||
session.close()
|
||||
except Exception as exc:
|
||||
logging.getLogger("agenteval").warning("启动清理失败(忽略): %s", exc)
|
||||
yield
|
||||
|
||||
|
||||
|
||||
@ -216,14 +216,14 @@ function RecentRunRow({ run, onOpen }: { run: Run; onOpen: () => void }) {
|
||||
<span style={{ fontWeight: 500 }}>{run.scenario_name || run.scenario_id.slice(0, 8)}</span>
|
||||
<span style={{ color: colors.textMuted }}> · {run.target_name || run.target_id.slice(0, 8)}</span>
|
||||
</div>
|
||||
<Tag color={triggerColors[trigger] ?? 'default'} style={{ margin: 0, fontSize: 10, lineHeight: '16px', padding: '0 4px' }}>
|
||||
{triggerLabels[trigger] ?? trigger}
|
||||
</Tag>
|
||||
{run.scenario_version != null && (
|
||||
<Tag style={{ margin: 0, fontSize: 10, lineHeight: '16px', padding: '0 4px' }}>
|
||||
v{run.scenario_version}
|
||||
</Tag>
|
||||
)}
|
||||
<Tag color={triggerColors[trigger] ?? 'default'} style={{ margin: 0, fontSize: 10, lineHeight: '16px', padding: '0 4px' }}>
|
||||
{triggerLabels[trigger] ?? trigger}
|
||||
</Tag>
|
||||
<span style={{ fontSize: 12, color: dotColor, width: 48, flexShrink: 0 }}>{statusLabels[run.status] ?? run.status}</span>
|
||||
<span style={{
|
||||
fontSize: 12, fontWeight: 600, width: 44, textAlign: 'right', flexShrink: 0,
|
||||
|
||||
38
tests/unit/test_orphan_runs.py
Normal file
38
tests/unit/test_orphan_runs.py
Normal file
@ -0,0 +1,38 @@
|
||||
"""Startup cleanup of orphan runs (interrupted by server restart)."""
|
||||
|
||||
from agenteval.models import EvalRun, RunStatus
|
||||
from agenteval.storage.repository import RunRepository
|
||||
|
||||
|
||||
def _make_run(session, status: RunStatus) -> str:
|
||||
run = RunRepository(session).create(EvalRun(
|
||||
target_id="t-1", scenario_id="s-1", status=status,
|
||||
))
|
||||
return run.id
|
||||
|
||||
|
||||
def test_mark_orphans_failed(db_session):
|
||||
repo = RunRepository(db_session)
|
||||
running_id = _make_run(db_session, RunStatus.RUNNING)
|
||||
pending_id = _make_run(db_session, RunStatus.PENDING)
|
||||
completed_id = _make_run(db_session, RunStatus.COMPLETED)
|
||||
failed_id = _make_run(db_session, RunStatus.FAILED)
|
||||
|
||||
count = repo.mark_orphans_failed()
|
||||
|
||||
assert count == 2
|
||||
for rid in (running_id, pending_id):
|
||||
run = repo.get(rid)
|
||||
assert run.status == RunStatus.FAILED
|
||||
assert run.summary["error"]["code"] == "interrupted"
|
||||
assert run.completed_at is not None
|
||||
# 已完结的运行不受影响
|
||||
assert repo.get(completed_id).status == RunStatus.COMPLETED
|
||||
assert repo.get(failed_id).status == RunStatus.FAILED
|
||||
assert repo.get(completed_id).summary is None
|
||||
|
||||
|
||||
def test_mark_orphans_failed_noop_when_clean(db_session):
|
||||
repo = RunRepository(db_session)
|
||||
_make_run(db_session, RunStatus.COMPLETED)
|
||||
assert repo.mark_orphans_failed() == 0
|
||||
Loading…
Reference in New Issue
Block a user