服务重启不再把运行记成"用户取消";接着跑也不再替人答掉下一个审批

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 不回归。
This commit is contained in:
ThearchyHelios
2026-09-25 16:05:14 +08:00
parent 5928cc052e
commit 03830d4815
4 changed files with 224 additions and 13 deletions
+4
View File
@@ -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`。设置页「运行环境」里能看到当前值。
+43 -12
View File
@@ -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),
+156
View File
@@ -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, "关停打断了封存"
+21 -1
View File
@@ -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() {
</div>
)}
{/* 中断了却没有待审批:服务重启时停下的,断点还在。以前这里显示"等待人工
介入",却没有任何东西可以点——看起来就是卡死了 */}
{selected.status === 'interrupted' && pendingHere.length === 0 && (
<div className="flex shrink-0 items-center gap-2 border-b px-4 py-2 text-[11.5px]"
style={{ borderColor: 'var(--warn)' }}>
<span className="flex-1 text-dim">
{selected.error || '这次运行中断了'}。前面跑完的节点不会重跑。
</span>
<button className="btn btn-sm" onClick={async () => {
try {
await api.runs.resume(selected.id, null)
toast('已从断点接着跑', 'ok')
await refreshDetail(selected.id)
} catch (e: any) { toast(e.message ?? '恢复失败', 'error') }
}}>
<Play size={11} /> 接着跑
</button>
</div>
)}
<div className="min-h-0 flex-1">
{raw ? <RawEvents events={events} /> : <AssistantStream turns={turn} />}
</div>