mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 03:11:59 +00:00
失败的运行能从断点接着跑了，前面跑过的节点不重来
库里 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 事件会说成不是续跑。
This commit is contained in:
ThearchyHelios
@@ -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 时很有用。"""
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
@@ -108,6 +108,9 @@ export const api = {
|
||||
run_class?: 'formal' | 'exploratory'; version?: number
|
||||
}) => post<Run>('/runs', body),
|
||||
artifacts: (id: string) => get<any[]>(`/runs/${id}/artifacts`),
|
||||
/** 从失败的节点接着跑。graph 只能带改过配置的同一张图，结构必须一致 */
|
||||
continue: (id: string, graph?: GraphSpec | null) =>
|
||||
post<Run>(`/runs/${id}/continue`, { graph: graph ?? null }),
|
||||
cancel: (id: string) => post<{ ok: boolean }>(`/runs/${id}/cancel`),
|
||||
resume: (id: string, response: any) => post<Run>(`/runs/${id}/resume`, { response }),
|
||||
events: (id: string, after = 0) => get<RunEvent[]>(`/runs/${id}/events?after=${after}`),
|
||||
|
||||
@@ -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<Record<string, string>>({})
|
||||
@@ -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 (
|
||||
<div className="relative" ref={wrap}>
|
||||
<div className="relative flex items-center gap-1.5" ref={wrap}>
|
||||
{/* 图挂在半路时，"接着跑"才是用户接下来最想做的事：他刚在画布上把那个
|
||||
写错的模型 id / 循环条件改好，重跑一遍意味着前面查过的表、跑过的 SQL
|
||||
全部重来。断点一直在 checkpoint 里躺着，只是以前没有入口 */}
|
||||
{run?.status === 'failed' && !blocked && (
|
||||
<button className="btn gap-1.5" disabled={busy} onClick={resume}
|
||||
title="从失败的那个节点接着跑，前面跑过的节点不重来。只能改节点配置，改了结构要重新发起">
|
||||
{busy ? <Spinner size={11} /> : <RotateCw size={11} />}
|
||||
接着跑
|
||||
</button>
|
||||
)}
|
||||
<button
|
||||
className="btn btn-primary gap-1.5"
|
||||
disabled={busy || blocked}
|
||||
|
||||
@@ -185,6 +185,8 @@ interface StudioState {
|
||||
retryCopilot: () => void
|
||||
clearCopilot: () => void
|
||||
attachRun: (runId: string) => Promise<void>
|
||||
/** 失败的运行从断点接着跑，把画布上改过的配置一起带过去 */
|
||||
continueRun: () => Promise<void>
|
||||
stopRun: () => Promise<void>
|
||||
clearRun: () => void
|
||||
applyEvent: (event: RunEvent) => void
|
||||
@@ -716,6 +718,22 @@ export const useStudio = create<StudioState>((set, get) => ({
|
||||
set({ unsubscribe: stop, streaming: true })
|
||||
},
|
||||
|
||||
continueRun: async () => {
|
||||
// 从失败的那个节点接着跑，前面跑过的不重来。
|
||||
//
|
||||
// 把画布**当前**这张图带过去，而不是运行时那份快照——失败的多半是配置错了
|
||||
// （模型 id 写错、循环条件语法错、缺必填输入），用户正是在画布上改完它才
|
||||
// 点的这里。带的是运行时那份的话，改完还会再撞同一个错。
|
||||
//
|
||||
// 后端只接受骨架一致的图：增删节点或改连线就拒绝。checkpoint 是按节点名
|
||||
// 存的，结构变了续下去会拿着错位的状态跑出一份似是而非的结果。
|
||||
const { run, nodes, edges } = get()
|
||||
if (!run) return
|
||||
const continued = await api.runs.continue(run.id, toGraph(nodes, edges))
|
||||
set({ run: continued, streaming: true })
|
||||
await get().attachRun(run.id)
|
||||
},
|
||||
|
||||
stopRun: async () => {
|
||||
const run = get().run
|
||||
if (!run) return
|
||||
|
||||
Reference in New Issue
Block a user