From 03830d4815db6b4f294b52064c0dae2cf870cb5f Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 25 Sep 2026 16:05:14 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9C=8D=E5=8A=A1=E9=87=8D=E5=90=AF=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E6=8A=8A=E8=BF=90=E8=A1=8C=E8=AE=B0=E6=88=90"?= =?UTF-8?q?=E7=94=A8=E6=88=B7=E5=8F=96=E6=B6=88"=EF=BC=9B=E6=8E=A5?= =?UTF-8?q?=E7=9D=80=E8=B7=91=E4=B9=9F=E4=B8=8D=E5=86=8D=E6=9B=BF=E4=BA=BA?= =?UTF-8?q?=E7=AD=94=E6=8E=89=E4=B8=8B=E4=B8=80=E4=B8=AA=E5=AE=A1=E6=89=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit dev.sh 开着 --reload，改一个后端文件就是一次关停。三件事以前都是错的： 1. 关停时在跑的运行一律记成 cancelled、原因"用户取消"——之后 resume 和 continue 都拒绝它，尽管 checkpoint 完好。现在关停时收到的取消记成 interrupted （和强杀后启动扫描的处理一致），留一条 server_shutdown 日志说明原委， 不发 run.cancelled。 2. 关停会打断正在收尾的运行：状态已落、封存和 run.finished 还没写完就被取消。 现在已进入收尾的运行不取消，等它写完。 3. 因重启而中断的运行没有在等人的 interrupt，resume 却照样发 Command(resume=…)： 那个值会被后面遇到的第一个 interrupt() 当成答复——一个还没轮到的人工关卡被 自动作答（审查时的实测）；当前版本的 LangGraph 则直接在内部崩掉 （UnboundLocalError: resume_is_map），运行判 failed。现在没有待答的 interrupt 时从断点接着跑（输入 None）。 运行详情页：中断了却没有待审批的运行，以前显示"等待人工介入"却没有任何可点的 东西，现在说明原因并给「接着跑」。用真实库里的中断运行看过：无待审批的出横幅， 有待审批的仍由审批卡片处理。README 加了对应的常见问题。 test_shutdown.py 3 例，修前全部失败：关停不是用户取消；重启后接着跑，前面跑完的 节点不重跑、后面的人工关卡真的停下来等人；关停不打断收尾（封存照样写完）。 分两步修、中间跑了一次，确认第二例在关停修好之后确实撞上了 resume 的崩溃。 422 → 425 passed，连跑 8 轮全绿；check-guards / decode / stream / canvas-layout 不回归。 --- README.md | 4 + backend/app/engine/runner.py | 55 ++++++++--- backend/tests/test_shutdown.py | 156 ++++++++++++++++++++++++++++++++ frontend/src/pages/RunsPage.tsx | 22 ++++- 4 files changed, 224 insertions(+), 13 deletions(-) create mode 100644 backend/tests/test_shutdown.py diff --git a/README.md b/README.md index e7ad3ab5..11d7fe1d 100644 --- a/README.md +++ b/README.md @@ -417,6 +417,10 @@ agent 侧同样会收到实情，并被告知可直接查询 `information_schema **端口冲突。** 前端默认 5273、后端 8000，通过 `AGENTLAB_WEB_PORT` / `AGENTLAB_PORT` 修改。 +**改了后端代码，正在跑的运行显示中断。** `dev.sh` 开着热重载，改一个后端文件就是一次 +服务重启。运行会被挂起（不是取消），断点完好：在运行详情页点「接着跑」，跑完的节点不会重跑， +后面的人工审批照常停下来等人。 + **运行被中止，提示超过时限。** 单次执行（两次人工介入之间的那一段）默认最多 600 秒， 单次模型调用默认 300 秒没有输出即超时。多半是模型或工具没有响应；确实需要更久，调大 `AGENTLAB_MAX_RUN_SECONDS` / `AGENTLAB_MODEL_TIMEOUT_SECONDS`。设置页「运行环境」里能看到当前值。 diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index a3eeb2bf..1520493d 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -36,10 +36,15 @@ class RunManager: self._checkpointer: AsyncSqliteSaver | None = None self._cm: Any = None self._semaphore = asyncio.Semaphore(settings.max_concurrent_runs) + #: 正在关停。此时收到的取消来自关停而不是用户，运行要记成可恢复的中断 + self._closing = False + #: 已经跑完、正在落状态和封存的运行。关停不去打断它们 + self._finalizing: set[str] = set() # ---------------- 生命周期 ---------------- async def setup(self) -> None: + self._closing = False settings.ensure_dirs() self._cm = AsyncSqliteSaver.from_conn_string(str(settings.checkpoint_path)) self._checkpointer = await self._cm.__aenter__() @@ -61,11 +66,14 @@ class RunManager: await session.commit() async def shutdown(self) -> None: + self._closing = True # 只等还没结束的：结束了的没什么可等，而且它可能属于另一个事件循环， # gather 会直接抛 "future belongs to a different loop" pending = [t for t in self._tasks.values() if not t.done()] - for task in pending: - task.cancel() + for run_id, task in list(self._tasks.items()): + # 正在收尾的不打断：状态和封存写到一半，比哪一步都没写更难收拾 + if not task.done() and run_id not in self._finalizing: + task.cancel() await asyncio.gather(*pending, return_exceptions=True) self._tasks.clear() if self._cm is not None: @@ -300,13 +308,23 @@ class RunManager: await session.commit() await session.refresh(run) - # 单个 interrupt 用标量（LangGraph 两种都认），多个必须用 {interrupt_id: value} 映射 - command = ( - Command(resume=response) - if len(waiting) <= 1 - else Command(resume=answers) - ) - await self._emit(run_id, EventType.RUN_RESUMED, data={"response": _safe(response)}) + # 单个 interrupt 用标量（LangGraph 两种都认），多个必须用 {interrupt_id: value} 映射。 + # + # 一个都没在等：这是服务重启时停下的运行（启动扫描或关停时标成 interrupted）， + # 只能从断点接着跑（输入传 None）。以前这里照样发 Command(resume=…)——那个值 + # 会被后面遇到的第一个 interrupt() 当成答复，一个还没轮到的人工关卡就这么被 + # 自动作答了；当前版本的 LangGraph 则直接在内部崩掉。两样都不是"接着跑" + command: Command | None + if not waiting: + command = None + elif len(waiting) == 1: + command = Command(resume=response) + else: + command = Command(resume=answers) + await self._emit(run_id, EventType.RUN_RESUMED, data=( + {"message": "服务重启时中断的运行，从断点接着跑"} if command is None + else {"response": _safe(response)} + )) self._tasks[run_id] = asyncio.create_task( self._drive(run_id, spec, command, workflow_id=workflow_id) ) @@ -476,9 +494,20 @@ class RunManager: status = "interrupted" except asyncio.CancelledError: - status = "cancelled" - error = "用户取消" - await self._emit(run_id, EventType.RUN_CANCELLED, data={}) + if self._closing: + # 服务关停（dev.sh 的 --reload 改一个文件就是一次）不是用户取消。 + # checkpoint 完好，记成中断、重启后从断点接着跑——和强杀后启动 + # 扫描的处理一致。以前一律记"用户取消"，resume 和 continue 都拒绝它 + status = "interrupted" + error = "服务重启，运行已挂起，可从断点恢复" + await self._emit(run_id, EventType.LOG, data={ + "level": "warn", "code": "server_shutdown", + "message": "服务关停时这次运行还在跑，已挂起；重启后可以从断点接着跑", + }) + else: + status = "cancelled" + error = "用户取消" + await self._emit(run_id, EventType.RUN_CANCELLED, data={}) raise except TimeoutError as e: status = "failed" @@ -500,12 +529,14 @@ class RunManager: await self._emit(run_id, EventType.RUN_FAILED, data={"error": error}) finally: elapsed = int((time.perf_counter() - started) * 1000) + self._finalizing.add(run_id) try: await self._finalize(run_id, status, error, final_state, elapsed) finally: # 收尾被打断（关停时的取消、库连接出错）也得把自己摘掉。以前写在 # _finalize 后面，一被打断就漏掉，留下的任务让下一次 shutdown 的 # gather 在别的事件循环里炸开 + self._finalizing.discard(run_id) self._tasks.pop(run_id, None) # 运行到终态就把沙箱会话收掉。thread_id 每个 run 都是新的， # 不收的话每跑一次带代码节点的图就多一台常驻 microVM（几百 MB）， diff --git a/backend/tests/test_shutdown.py b/backend/tests/test_shutdown.py new file mode 100644 index 00000000..0dd9da72 --- /dev/null +++ b/backend/tests/test_shutdown.py @@ -0,0 +1,156 @@ +"""服务关停不是用户取消；重启后接着跑，也不能替人把后面的审批答了。 + +dev.sh 开着 --reload，改任何一个后端文件就是一次关停。以前关停时在跑的运行 +一律记成 cancelled、原因写"用户取消"——之后 resume 和 continue 都拒绝它，尽管 +checkpoint 完好无损。 + +更要紧的是接着跑那一步：这类运行没有在等人的 interrupt，resume 却照样发一个 +Command(resume=…)。LangGraph 会把这个值交给后面遇到的第一个 interrupt()——一个 +还没轮到的人工关卡就这么被自动作答了，审批记录从没建过，运行直接跑完。 +""" +from __future__ import annotations + +import asyncio + +import pytest +from sqlalchemy import select + +from app.db.base import SessionLocal +from app.db.models import Approval, Run, RunEvent +from app.engine import compiler +from app.engine.runner import run_manager +from app.engine.schema import NodeType + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, + "data": {"label": nid, "config": config}} + + +def chain(*nodes): + return {"nodes": list(nodes), + "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +async def _status(run_id: str) -> str: + async with SessionLocal() as session: + return (await session.get(Run, run_id)).status + + +async def _until(check, timeout: float = 15.0): + deadline = asyncio.get_event_loop().time() + timeout + while asyncio.get_event_loop().time() < deadline: + if await check(): + return + await asyncio.sleep(0.05) + raise AssertionError("等超时了") + + +async def _events(run_id: str) -> list[RunEvent]: + async with SessionLocal() as session: + return list((await session.execute( + select(RunEvent).where(RunEvent.run_id == run_id).order_by(RunEvent.seq) + )).scalars()) + + +async def _started(run_id: str, node_id: str) -> int: + return sum(1 for e in await _events(run_id) if e.type == "node.started" and e.node_id == node_id) + + +@pytest.fixture +def slow_once(monkeypatch): + """transform 节点第一次跑时卡住（好让关停撞上它），之后立刻返回。""" + calls = {"n": 0} + original = compiler.RUNNERS[NodeType.TRANSFORM] + + async def runner(state, ctx): + if ctx.node.id == "slow": + calls["n"] += 1 + if calls["n"] == 1: + await asyncio.sleep(30) + return await original(state, ctx) + + monkeypatch.setitem(compiler.RUNNERS, NodeType.TRANSFORM, runner) + return calls + + +GRAPH = chain( + node("start", "input", fields=[{"name": "question"}]), + node("a", "transform", mode="template", template="前面这步", assign_to="x"), + node("slow", "transform", mode="template", template="慢的这步"), + node("gate", "human", mode="approve", title="发出去之前看一眼"), + node("out", "output", fields=[{"name": "结果", "value": "{{ vars.x }}"}]), +) + + +async def _run_into_shutdown() -> str: + run = await run_manager.start(graph=GRAPH, input_payload={"question": "随便"}) + await _until(lambda: _started_is(run.id, "slow")) + await run_manager.shutdown() + return run.id + + +async def _started_is(run_id: str, node_id: str) -> bool: + return await _started(run_id, node_id) > 0 + + +async def test_shutdown_is_not_a_user_cancel(slow_once): + run_id = await _run_into_shutdown() + async with SessionLocal() as session: + run = await session.get(Run, run_id) + assert run.status == "interrupted", f"关停被记成了 {run.status}：{run.error}" + assert "用户取消" not in (run.error or "") + events = await _events(run_id) + assert not [e for e in events if e.type == "run.cancelled"] + assert "server_shutdown" in [e.data.get("code") for e in events if e.type == "log"] + + +async def test_resuming_after_a_restart_does_not_answer_the_next_human_gate(slow_once): + run_id = await _run_into_shutdown() + await run_manager.setup() # 重启 + await run_manager.resume(run_id, None) + + async def reached_gate() -> bool: + return await _status(run_id) in ("interrupted", "succeeded", "failed") + await _until(reached_gate) + await asyncio.sleep(0.2) + status = await _status(run_id) + assert status == "interrupted", f"人工关卡被跳过了，运行直接 {status}" + + async with SessionLocal() as session: + pending = list((await session.execute( + select(Approval).where(Approval.run_id == run_id, Approval.status == "pending") + )).scalars()) + assert [a.node_id for a in pending] == ["gate"], "关卡在等人，审批记录却没建" + assert await _started(run_id, "a") == 1, "接着跑把前面跑完的节点又跑了一遍" + + +async def test_shutdown_lets_a_finishing_run_finish(monkeypatch): + """正在收尾的运行不被关停打断：状态、封存都得落完。""" + original = run_manager._seal + + async def slow_seal(run_id): + await asyncio.sleep(0.5) + await original(run_id) + + monkeypatch.setattr(run_manager, "_seal", slow_seal) + run = await run_manager.start(graph=chain( + node("start", "input", fields=[{"name": "question"}]), + node("out", "output", fields=[]), + ), input_payload={"question": "随便"}) + + async def done() -> bool: + return await _status(run.id) == "succeeded" + await _until(done) + await run_manager.shutdown() # 撞在封存的半路上 + + async with SessionLocal() as session: + sealed = await session.get(Run, run.id) + assert sealed.manifest_hash, "关停打断了封存" diff --git a/frontend/src/pages/RunsPage.tsx b/frontend/src/pages/RunsPage.tsx index 911bac1d..71ff7fa8 100644 --- a/frontend/src/pages/RunsPage.tsx +++ b/frontend/src/pages/RunsPage.tsx @@ -1,6 +1,6 @@ import { useEffect, useMemo, useState } from 'react' import { useNavigate, useParams } from 'react-router-dom' -import { Code2, GitFork, History, RefreshCw, ShieldCheck, Trash2 } from 'lucide-react' +import { Code2, GitFork, History, Play, RefreshCw, ShieldCheck, Trash2 } from 'lucide-react' import clsx from 'clsx' import { api } from '../api/client' import { Empty, Spinner, StatusDot, useToast } from '../components/ui' @@ -217,6 +217,26 @@ export function RunsPage() { )} + {/* 中断了却没有待审批：服务重启时停下的，断点还在。以前这里显示"等待人工 + 介入"，却没有任何东西可以点——看起来就是卡死了 */} + {selected.status === 'interrupted' && pendingHere.length === 0 && ( +
+ + {selected.error || '这次运行中断了'}。前面跑完的节点不会重跑。 + + +
+ )} +
{raw ? : }