From c980da65d0bad75f92db06af63a47291f8ff8981 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 20 Sep 2026 11:54:16 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=B1=E8=B4=A5=E7=9A=84=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E8=83=BD=E4=BB=8E=E6=96=AD=E7=82=B9=E6=8E=A5=E7=9D=80=E8=B7=91?= =?UTF-8?q?=E4=BA=86=EF=BC=8C=E5=89=8D=E9=9D=A2=E8=B7=91=E8=BF=87=E7=9A=84?= =?UTF-8?q?=E8=8A=82=E7=82=B9=E4=B8=8D=E9=87=8D=E6=9D=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 库里 28 条 failed 的构成说明了为什么值得单独开这条路： 模型 id 写错（404） 7 循环条件语法错 3 401 3 工具参数名不对 3 缺必填输入 2 大半是**配置错了**，改一下就能跑。而"改一下再跑"原来意味着整张图从头来 一遍——前面花两分钟查完的表结构、跑完的 SQL 全部重跑，尽管那些结果一直 躺在 checkpoint 里。checkpointer 本来就一直开着（thread_id = run_id）， 挡住这条路的只是 resume 里的一行状态判定：它只认 interrupted，failed 进不去。 **带着改过的图续。** 不让改配置的话这功能基本没用：上面那几类续下去会 立刻再撞同一个错。所以 /runs/{id}/continue 接受一张改过的图，画布上点 「接着跑」带的就是当前编辑中的那张——用户正是改完它才点的。 **但只准改配置。** checkpoint 是按节点名存的（编译时 add_node(node.id)）， 而节点配置是运行时从 spec 现读的（NodeContext.cfg）。所以只改配置对得上， 增删节点或改连线就对不上——续下去会拿着错位的通道状态跑出一份似是而非的 结果，比直接报错糟得多。加了 schema.topology_of() 做这道判据：节点 id + 类型 + 边，不含任何配置；对不上就 409，让用户重新发起。 这正是 resume 那边踩过的坑的同一类：不看有没有 checkpoint 就发 Command， LangGraph 拿空 state 从 START 重跑，最后还落成 succeeded。所以这里也先问 一次 aget_state，没有待跑节点就拒绝，而不是硬发一个 None 下去。 时间线上要说清楚是接着跑、保了几步（"从「lp」接着跑，前面 2 个节点的结果 保留"）——否则用户看到时间线又动起来，分不清这次是续的还是整张图又跑了一遍。 验证：后端 249 passed（新增 test_continue_run.py 7 条，核心那条断言的是 node.started 事件——前面的节点确实没有再出现）。端到端真跑了一遍： 第一遍 start → prep → lp 挂在 lp，改好条件后第二遍只跑 lp → out， 成果里带着 prep 那步的结果；改了结构的图被 409 拒绝；没有失败的运行时 画布上不显示这个按钮。 顺带：_drive 里 resumed 的判定漏了 payload is None 这种（从 checkpoint 直接往下走），run.started 事件会说成不是续跑。 --- backend/app/api/runs.py | 27 ++++ backend/app/engine/runner.py | 87 ++++++++++- backend/app/engine/schema.py | 19 +++ backend/tests/test_continue_run.py | 227 +++++++++++++++++++++++++++++ frontend/src/api/client.ts | 3 + frontend/src/run/RunControl.tsx | 28 +++- frontend/src/store/studio.ts | 18 +++ 7 files changed, 404 insertions(+), 5 deletions(-) create mode 100644 backend/tests/test_continue_run.py diff --git a/backend/app/api/runs.py b/backend/app/api/runs.py index d128ab85..94d9f56e 100644 --- a/backend/app/api/runs.py +++ b/backend/app/api/runs.py @@ -246,6 +246,33 @@ async def resume_run( raise HTTPException(409, str(e)) from e +class ContinueIn(BaseModel): + """接着跑。graph 只能带改过配置的同一张图，结构必须一致。""" + + graph: dict[str, Any] | None = None + + +@router.post("/{run_id}/continue", response_model=RunOut) +async def continue_run( + run_id: str, payload: ContinueIn, x_actor: str | None = Header(default=None) +) -> Run: + """从失败的节点接着跑，前面跑过的不重来。 + + 和 /resume 不是一回事：那条是人工介入的断点（status=interrupted），这条是 + 真的挂了（status=failed）。挡住后者的原来只是 resume 里的一行状态判定， + 而 checkpoint 一直都在——改个模型 id 就得把前面两分钟的取数重跑一遍， + 纯属白等。 + """ + try: + return await run_manager.continue_failed( + run_id, graph=payload.graph, actor=(x_actor or "").strip() or None, + ) + except KeyError as e: + raise HTTPException(404, str(e)) from e + except ValueError as e: + raise HTTPException(409, str(e)) from e + + @router.get("/{run_id}/state") async def get_state(run_id: str) -> dict[str, Any]: """读取 checkpoint 里的当前状态。调试中断中的 run 时很有用。""" diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index f99c7c67..77be9b35 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -18,7 +18,7 @@ from app.db.base import SessionLocal from app.db.models import Approval, Run, RunEvent, Workflow from app.engine.compiler import compile_graph, initial_state from app.engine.context import NodeError, RunContext -from app.engine.schema import GraphSpec, validate_graph +from app.engine.schema import GraphSpec, topology_of, validate_graph logger = logging.getLogger(__name__) @@ -302,6 +302,88 @@ class RunManager: ) return run + async def continue_failed( + self, run_id: str, *, graph: dict[str, Any] | None = None, actor: str | None = None + ) -> Run: + """从失败的地方接着跑，而不是整张图从头来一遍。 + + 为什么值得单独开这条路：库里 28 条 failed 里，模型 id 写错、401、循环 + 条件语法错、缺必填输入加起来占了大半——全是**配置错了**，改一下就能跑。 + 而现在"改一下再跑"意味着前面花两分钟查完的表结构、跑完的 SQL 全部重来。 + checkpointer 一直开着（thread_id = run_id），那些结果本来就在库里躺着。 + + graph 可以带一张改过的图，但只准改配置：节点 id、类型、连线必须和原来 + 一模一样。checkpoint 是按节点名存的，改了结构就对不上，续下去会拿着 + 错位的通道状态跑出一份似是而非的结果——那比直接报错糟得多。 + """ + async with SessionLocal() as session: + run = await session.get(Run, run_id) + if not run: + raise KeyError(f"找不到运行 {run_id}") + if run.status != "failed": + raise ValueError( + f"当前状态是 {run.status}，只有失败的运行能接着跑。" + "（等人工介入的用恢复，已完成的重新发起一次。）" + ) + spec = GraphSpec.model_validate(run.graph) + workflow_id = run.workflow_id + + if graph is not None: + revised = GraphSpec.model_validate(graph) + if topology_of(revised) != topology_of(spec): + raise ValueError( + "接着跑只能改节点配置，不能增删节点或改连线——" + "断点是按节点名存的，结构变了就对不上了。" + "要改结构请重新发起一次运行。" + ) + report = validate_graph(revised) + if not report.ok: + first = next((i.message for i in report.issues if i.level == "error"), "") + raise ValueError(f"改过的图没有通过校验：{first}") + spec = revised + + # 先问引擎到底停在哪。没有待跑的节点就不是"能接着跑"的情形—— + # 硬发一个 None 下去，LangGraph 会从 START 重跑整张图 + run_ctx = RunContext(run_id=run_id, thread_id=run_id, spec=spec, workflow_id=workflow_id) + app = compile_graph(spec, run_ctx).compile(checkpointer=self.checkpointer) + config = {"configurable": {"thread_id": run_id}} + snapshot = await app.aget_state(config) + pending = [str(n) for n in (getattr(snapshot, "next", None) or [])] + if not pending: + raise ValueError( + "这次运行没有留下可以接着跑的断点（多半是还没跑到第一个节点就挂了）。" + "请重新发起一次运行——继续下去只会从头跑一遍，还看不出来。" + ) + + done = [n for n in (snapshot.values or {}).get("nodes", {}) if n not in pending] + async with SessionLocal() as session: + run = await session.get(Run, run_id) + bus.set_seq(run_id, run.last_seq) + run.status = "running" + run.error = None + if graph is not None: + run.graph = spec.model_dump(mode="json") + if actor: + run.started_by = actor + await session.commit() + await session.refresh(run) + + labels = {n.id: n.title for n in spec.nodes} + await self._emit( + run_id, EventType.RUN_RESUMED, + data={ + "from": pending, + # 说清楚"哪些没重来"——否则用户看到时间线又动起来， + # 分不清这次是接着跑还是整张图又跑了一遍 + "message": f"从「{labels.get(pending[0], pending[0])}」接着跑" + + (f"，前面 {len(done)} 个节点的结果保留" if done else ""), + }, + ) + self._tasks[run_id] = asyncio.create_task( + self._drive(run_id, spec, None, workflow_id=workflow_id) + ) + return run + async def cancel(self, run_id: str) -> bool: task = self._tasks.get(run_id) if task and not task.done(): @@ -342,7 +424,8 @@ class RunManager: await self._emit( run_id, EventType.RUN_STARTED, - data={"nodes": len(spec.nodes), "resumed": isinstance(payload, Command)}, + data={"nodes": len(spec.nodes), + "resumed": payload is None or isinstance(payload, Command)}, ) run_ctx = RunContext( diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py index b183bdae..07023364 100644 --- a/backend/app/engine/schema.py +++ b/backend/app/engine/schema.py @@ -148,6 +148,25 @@ class ValidationResult(BaseModel): self.ok = False +def topology_of(spec: GraphSpec) -> tuple[tuple, tuple]: + """图的骨架：节点的 id 与类型、加上所有的边。**不含任何配置。** + + 断点续跑要用它。LangGraph 的 checkpoint 是按节点名存的（编译时 + `builder.add_node(node.id, …)`），而节点配置是运行时从 spec 现读的 + （`NodeContext.cfg`）。所以拿一张只改了配置的图去续，checkpoint 对得上， + 前面跑过的节点结果照样能用；一旦增删节点或改连线，checkpoint 里的 + 通道状态就和新图对不上了——续下去跑出来的是一份似是而非的结果， + 比直接报错糟得多。 + + 这正是 resume 那边踩过的坑：不看有没有 checkpoint 就发 Command， + LangGraph 拿空 state 从 START 重跑，最后还落成 succeeded。 + """ + return ( + tuple(sorted((n.id, str(n.type)) for n in spec.nodes)), + tuple(sorted((e.source, e.target, e.sourceHandle or "") for e in spec.edges)), + ) + + def validate_graph(spec: GraphSpec) -> ValidationResult: """编译前的静态检查。错误会挡住运行，警告只在画布上提示。""" result = ValidationResult() diff --git a/backend/tests/test_continue_run.py b/backend/tests/test_continue_run.py new file mode 100644 index 00000000..1b05be6c --- /dev/null +++ b/backend/tests/test_continue_run.py @@ -0,0 +1,227 @@ +"""失败的运行接着跑，而不是整张图从头来一遍。 + +库里 28 条 failed 的构成说明了为什么值得单独开这条路：模型 id 写错、401、 +循环条件语法错、缺必填输入加起来占了大半——全是**配置错了**，改一下就能跑。 +而"改一下再跑"原来意味着前面花两分钟查完的表结构、跑完的 SQL 全部重来， +尽管那些结果一直躺在 checkpoint 里。 + +这里最要紧的不是接口通不通，是两条： +1. 前面跑过的节点**真的没有重跑**（不然"接着跑"只是个说法）； +2. 结构变了就必须拒绝。checkpoint 是按节点名存的，拿一张改了结构的图去续， + 会拿着错位的通道状态跑出一份似是而非的结果——比直接报错糟得多。 +""" + +from __future__ import annotations + +import asyncio + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.db.base import SessionLocal +from app.db.models import Run, RunEvent +from app.engine.runner import run_manager +from app.engine.schema import GraphSpec, topology_of +from app.main import app + + +@pytest.fixture(autouse=True) +async def engine_up(): + """这组测试真的跑图，需要 checkpointer——平时是 FastAPI 的 lifespan 建的。 + + 不开的话 continue_failed 连 aget_state 都调不了，而"能不能接着跑"整件事 + 就建立在 checkpoint 上。 + """ + await run_manager.setup() + yield + await run_manager.shutdown() + + +@pytest.fixture +async def client(): + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as c: + yield c + + +def node(nid, ntype, **config): + return { + "id": nid, "type": ntype, "position": {"x": 0, "y": 0}, + "data": {"label": nid, "config": config}, + } + + +def graph_with(*, condition: str): + """input → 整形 → 循环（条件由参数给）→ 成果。 + + 循环条件写错会抛 NodeError，而它前面的整形节点已经跑完了——正好用来验 + "接着跑"到底有没有把前面那步保下来。库里真有 4 条运行是这么挂的。 + """ + return { + "nodes": [ + node("start", "input", fields=[{"name": "question"}]), + node("prep", "transform", mode="template", template="备好了", assign_to="prepped"), + node("lp", "loop", mode="while", condition=condition, max_iterations=1), + node("body", "transform", mode="template", template="转一圈"), + node("out", "output", fields=[{"name": "结果", "value": "{{ vars.prepped }}"}]), + ], + "edges": [ + {"source": "start", "target": "prep"}, + {"source": "prep", "target": "lp"}, + {"source": "lp", "target": "body", "sourceHandle": "body"}, + {"source": "body", "target": "lp"}, + {"source": "lp", "target": "out", "sourceHandle": "done"}, + ], + } + + +BROKEN = "这不是表达式 >>> 啊" +FIXED = "false" + + +async def _run_until_done(graph, *, timeout=20.0): + run = await run_manager.start(graph=graph, input_payload={"question": "随便"}) + await _wait(run.id, ("failed", "succeeded", "cancelled"), timeout) + return run.id + + +async def _wait(run_id, statuses, timeout=20.0): + deadline = asyncio.get_event_loop().time() + timeout + while asyncio.get_event_loop().time() < deadline: + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row and row.status in statuses: + return row.status + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run_id) + raise AssertionError(f"等超时了，run {run_id} 还是 {row.status if row else '不存在'}") + + +async def _node_starts(run_id: str) -> list[str]: + """这次运行里哪些节点真的开跑了。接着跑之后不该再出现前面那些。""" + async with SessionLocal() as session: + rows = list((await session.execute( + RunEvent.__table__.select().where(RunEvent.run_id == run_id) + )).mappings()) + return [r["node_id"] for r in rows if r["type"] == "node.started" and r["node_id"]] + + +def test_topology_ignores_config_but_not_structure(): + """改配置骨架不变，改结构就变——这是能不能续的唯一判据。""" + a = GraphSpec.model_validate(graph_with(condition=BROKEN)) + b = GraphSpec.model_validate(graph_with(condition=FIXED)) + assert topology_of(a) == topology_of(b) + + moved = graph_with(condition=FIXED) + moved["nodes"][1]["position"] = {"x": 500, "y": 900} + assert topology_of(GraphSpec.model_validate(moved)) == topology_of(a) + + extra = graph_with(condition=FIXED) + extra["nodes"].append(node("zzz", "transform", mode="template", template="多出来的")) + assert topology_of(GraphSpec.model_validate(extra)) != topology_of(a) + + +@pytest.mark.asyncio +async def test_continue_skips_the_nodes_that_already_ran(): + """核心不变量：接着跑不重跑前面的节点。""" + run_id = await _run_until_done(graph_with(condition=BROKEN)) + async with SessionLocal() as session: + assert (await session.get(Run, run_id)).status == "failed" + + first_pass = await _node_starts(run_id) + assert "prep" in first_pass, f"第一遍应该跑到整形节点：{first_pass}" + before = len(first_pass) + + await run_manager.continue_failed(run_id, graph=graph_with(condition=FIXED)) + await _wait(run_id, ("succeeded", "failed")) + + async with SessionLocal() as session: + run = await session.get(Run, run_id) + assert run.status == "succeeded", run.error + # 前面那步的结果还在，说明续的是同一条线程而不是重开了一张图 + assert run.output.get("结果") == "备好了" + + second_pass = (await _node_starts(run_id))[before:] + assert "prep" not in second_pass, f"整形节点被重跑了：{second_pass}" + assert "start" not in second_pass, f"输入节点被重跑了：{second_pass}" + assert "out" in second_pass, f"应该跑完剩下的：{second_pass}" + + +@pytest.mark.asyncio +async def test_continue_refuses_a_restructured_graph(): + """结构变了就对不上。宁可报错，也不能跑出一份似是而非的结果。""" + run_id = await _run_until_done(graph_with(condition=BROKEN)) + + restructured = graph_with(condition=FIXED) + restructured["nodes"].append(node("extra", "transform", mode="template", template="新加的")) + restructured["edges"].append({"source": "prep", "target": "extra"}) + + with pytest.raises(ValueError, match="不能增删节点"): + await run_manager.continue_failed(run_id, graph=restructured) + + async with SessionLocal() as session: + assert (await session.get(Run, run_id)).status == "failed" # 没被动过 + + +@pytest.mark.asyncio +async def test_continue_refuses_a_graph_that_fails_validation(): + """骨架没动，但改出来的配置本身就是非法的——也得拦。 + + 用出具契约缺 metrics_from：那是纯配置错误，结构一个字没改，所以它必定 + 走过了拓扑那一关，考的就是后面那道校验。 + """ + run_id = await _run_until_done(graph_with(condition=BROKEN)) + bad = graph_with(condition=FIXED) + bad["nodes"][4]["data"]["config"]["contract"] = {"narrative": "{{ vars.prepped }}"} + assert topology_of(GraphSpec.model_validate(bad)) == topology_of( + GraphSpec.model_validate(graph_with(condition=BROKEN)) + ), "这个用例的前提是骨架不变" + + with pytest.raises(ValueError, match="没有通过校验"): + await run_manager.continue_failed(run_id, graph=bad) + + +@pytest.mark.asyncio +async def test_only_failed_runs_can_continue(client): + """跑成功的接着跑没有意义——断点在终点，续下去等于从 START 重来一遍。""" + run_id = await _run_until_done(graph_with(condition=FIXED)) + async with SessionLocal() as session: + assert (await session.get(Run, run_id)).status == "succeeded" + + resp = await client.post(f"/api/runs/{run_id}/continue", json={}) + assert resp.status_code == 409 + assert "只有失败的运行" in resp.json()["detail"] + + +@pytest.mark.asyncio +async def test_continue_unknown_run_is_404(client): + resp = await client.post("/api/runs/没这个/continue", json={}) + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_continue_through_the_api(client): + """走接口跑一遍，并确认时间线上说清楚了是接着跑、保了几步。""" + run_id = await _run_until_done(graph_with(condition=BROKEN)) + + resp = await client.post( + f"/api/runs/{run_id}/continue", json={"graph": graph_with(condition=FIXED)} + ) + assert resp.status_code == 200, resp.text + assert resp.json()["status"] == "running" + await _wait(run_id, ("succeeded", "failed")) + + async with SessionLocal() as session: + run = await session.get(Run, run_id) + rows = list((await session.execute( + RunEvent.__table__.select().where(RunEvent.run_id == run_id) + )).mappings()) + assert run.status == "succeeded", run.error + # 改过的图要落库，否则下次再看这次运行，图和实际跑的对不上 + assert run.graph["nodes"][2]["data"]["config"]["condition"] == FIXED + + resumed = [r for r in rows if r["type"] == "run.resumed"] + assert resumed, "接着跑必须在时间线上留下痕迹" + msg = resumed[-1]["data"].get("message", "") + assert "接着跑" in msg and "保留" in msg, msg diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index f4f0f3d6..d3512674 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -108,6 +108,9 @@ export const api = { run_class?: 'formal' | 'exploratory'; version?: number }) => post('/runs', body), artifacts: (id: string) => get(`/runs/${id}/artifacts`), + /** 从失败的节点接着跑。graph 只能带改过配置的同一张图，结构必须一致 */ + continue: (id: string, graph?: GraphSpec | null) => + post(`/runs/${id}/continue`, { graph: graph ?? null }), cancel: (id: string) => post<{ ok: boolean }>(`/runs/${id}/cancel`), resume: (id: string, response: any) => post(`/runs/${id}/resume`, { response }), events: (id: string, after = 0) => get(`/runs/${id}/events?after=${after}`), diff --git a/frontend/src/run/RunControl.tsx b/frontend/src/run/RunControl.tsx index b338db09..1608948a 100644 --- a/frontend/src/run/RunControl.tsx +++ b/frontend/src/run/RunControl.tsx @@ -1,5 +1,5 @@ import { useEffect, useMemo, useRef, useState } from 'react' -import { ChevronDown, Play, ShieldCheck, Square } from 'lucide-react' +import { ChevronDown, Play, RotateCw, ShieldCheck, Square } from 'lucide-react' import { useStudio } from '../store/studio' import { Spinner, useToast } from '../components/ui' @@ -19,7 +19,7 @@ export function RunControl() { const issues = useStudio((s) => s.issues) const workflow = useStudio((s) => s.workflow) const dirty = useStudio((s) => s.dirty) - const { startRun, startFormalRun, stopRun } = useStudio() + const { startRun, startFormalRun, stopRun, continueRun } = useStudio() const toast = useToast() const [open, setOpen] = useState(false) const [values, setValues] = useState>({}) @@ -86,8 +86,30 @@ export function RunControl() { const blocked = !!errors.length || !nodes.length + const resume = async () => { + setBusy(true) + try { + await continueRun() + toast('从断点接着跑，前面跑过的节点不重来', 'ok') + } catch (e: any) { + toast(e.message ?? '接着跑失败', 'error') + } finally { + setBusy(false) + } + } + return ( -
+
+ {/* 图挂在半路时，"接着跑"才是用户接下来最想做的事：他刚在画布上把那个 + 写错的模型 id / 循环条件改好，重跑一遍意味着前面查过的表、跑过的 SQL + 全部重来。断点一直在 checkpoint 里躺着，只是以前没有入口 */} + {run?.status === 'failed' && !blocked && ( + + )}