From d23b321225b87ef1e958daaf0959485830d92c77 Mon Sep 17 00:00:00 2001 From: sinohqb Date: Wed, 29 Jul 2026 14:48:14 +0800 Subject: [PATCH] fix(runs): mark orphaned running/pending runs failed on startup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 评测任务是进程内 asyncio 任务，服务重启会中断执行且状态永远停在 running。启动时将遗留的 running/pending 运行标记为 failed（summary 写入 interrupted 错误），清理为尽力而为，不阻断启动。另将仪表盘最近 评测记录的触发方式与版本号标签位置对调。 --- backend/agenteval/storage/repository.py | 19 +++++++++++++ backend/agenteval/web/app.py | 15 +++++++++- frontend/web/src/pages/Home.tsx | 6 ++-- tests/unit/test_orphan_runs.py | 38 +++++++++++++++++++++++++ 4 files changed, 74 insertions(+), 4 deletions(-) create mode 100644 tests/unit/test_orphan_runs.py diff --git a/backend/agenteval/storage/repository.py b/backend/agenteval/storage/repository.py index 442714c..8ffddc9 100644 --- a/backend/agenteval/storage/repository.py +++ b/backend/agenteval/storage/repository.py @@ -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: diff --git a/backend/agenteval/web/app.py b/backend/agenteval/web/app.py index 5435410..3f17752 100644 --- a/backend/agenteval/web/app.py +++ b/backend/agenteval/web/app.py @@ -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 diff --git a/frontend/web/src/pages/Home.tsx b/frontend/web/src/pages/Home.tsx index 060972b..048a842 100644 --- a/frontend/web/src/pages/Home.tsx +++ b/frontend/web/src/pages/Home.tsx @@ -216,14 +216,14 @@ function RecentRunRow({ run, onOpen }: { run: Run; onOpen: () => void }) { {run.scenario_name || run.scenario_id.slice(0, 8)} · {run.target_name || run.target_id.slice(0, 8)} + + {triggerLabels[trigger] ?? trigger} + {run.scenario_version != null && ( v{run.scenario_version} )} - - {triggerLabels[trigger] ?? trigger} - {statusLabels[run.status] ?? run.status} 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