引擎四处：错误处理、恢复语义、幂运算炸弹

## 一条分支条件能把后端冻死求值器放行 ast.Pow，唯一护栏是 AST 节点数 <= 200。`9**9**9**9` 只有 11 个节点，却让 CPython 无限期算大整数。而求值是在事件循环里同步执行的（分支条件、 transform、口径卡、每个节点的 skip_if），没有线程池隔离也没有超时，asyncio 的 cancel 打断不了同步 CPU。给底数和指数加上界，超界报 ExpressionError。 2**10、2**0.5、10**-2 这些正常用法不受影响。 ## NodeError 绕过 retries / on_error / node.failed `except NodeError: raise` 排在重试与兜底之前，于是节点自己抛的错既不重试、也不发 node.failed、更不被 on_error=continue 接住——而 NodeError 恰恰是所有执行器表达失败的标准方式（缺必填输入、模型没配、口径卡算不出…）。同一张图配着同样的 retries=1 + on_error=continue，1/0 会重试并继续，而 JSON 模板非法直接整个 run 失败。两条路径现在行为一致，node.failed 也带上 node_id （前端靠它把出错节点从"转圈"收敛成"失败"）。 ## 没有 checkpoint 的运行被"恢复"成一份假结果重启扫描把 running 和 queued 一律标成 interrupted，而 resume 不检查有没有 checkpoint 就发 Command。queued 的运行从没写过 checkpoint，LangGraph 会拿空 state 从 START 重跑整张图——run.input 完全没注入，用字段默认值算完还落成 succeeded。现在 queued 直接标 failed（没有进度可恢复，请重新发起），resume 先 aget_state 确认有断点，没有就 409 并说清为什么。 ## 并行人工节点：一次 resume 打爆整个 run 同一 superstep 有多个 interrupt 时，resume 仍只发标量 Command，LangGraph 直接抛 RuntimeError；而所有 pending 审批已经被写成同一个回复。Approval.interrupt_id 和 ResumeIn.approval_id 这两个字段一直存在却从没被用过。现在按 interrupt_id 逐条收集：多条待办时必须指定 approval_id，回复的那条转 answered 继续等，凑齐引擎在等的全部 interrupt 才发 Command(resume={iid: value})。验证：两个并行 human 节点——不指定 approval_id 被 409 拒、回复第一条后仍 interrupted 且剩 1 条 pending、凑齐后 succeeded。单人审批与 decide 通道回归通过。
2026-11-15 01:01:59 +00:00 · 2026-09-14 15:35:47 +08:00
parent 0afffd4b9a
commit b0575a7d1b
4 changed files with 170 additions and 24 deletions
@@ -229,9 +229,17 @@ class ResumeIn(BaseModel):


@router.post("/{run_id}/resume", response_model=RunOut)
-async def resume_run(run_id: str, payload: ResumeIn) -> Run:
+async def resume_run(
+    run_id: str, payload: ResumeIn, x_actor: str | None = Header(default=None)
+) -> Run:
    try:
-        return await run_manager.resume(run_id, payload.response)
+        # approval_id 以前定义了却从没传下去——一次运行里有多条待审批时，
+        # 引擎无从知道这次回复的是哪一条
+        return await run_manager.resume(
+            run_id, payload.response,
+            approval_id=payload.approval_id,
+            actor=(x_actor or "").strip() or None,
+        )
    except KeyError as e:
        raise HTTPException(404, str(e)) from e
    except ValueError as e:
@@ -443,7 +451,11 @@ async def decide(
        response["args"] = payload.args

    try:
-        return await run_manager.resume(approval.run_id, response)
+        return await run_manager.resume(
+            approval.run_id, response,
+            approval_id=approval.id,
+            actor=approval.resolved_by,
+        )
    except ValueError as e:
        raise HTTPException(409, str(e)) from e

@@ -114,23 +114,26 @@ def _wrap(node: GraphNode, run_ctx: RunContext) -> Callable[[GraphState], Awaita
                raise
            except asyncio.CancelledError:
                raise
-            except NodeError:
-                raise
            except Exception as e:  # noqa: BLE001
+                # NodeError 以前在这里直接 raise，于是节点自己抛的错既不重试、
+                # 也不发 node.failed、更不被 on_error=continue 接住。而 NodeError
+                # 恰恰是所有执行器表达失败的标准方式（缺必填输入、模型没配、
+                # 口径卡算不出…），等于这两个配置对最常见的报错全都不生效。
                last_error = e
                if attempt < retries:
                    ctx.emit(
                        EventType.LOG,
                        level="warn",
-                        message=f"第 {attempt + 1} 次失败（{type(e).__name__}: {e}），准备重试",
+                        message=f"第 {attempt + 1} 次失败（{_describe(e)}），准备重试",
                    )
                    await asyncio.sleep(backoff * (2**attempt))
                    continue
                break

        elapsed = int((time.perf_counter() - started) * 1000)
-        message = f"{type(last_error).__name__}: {last_error}"
-        ctx.emit(EventType.NODE_FAILED, error=message, duration_ms=elapsed)
+        message = _describe(last_error)
+        # node_id 要带上：前端靠它把出错的节点从"转圈"收敛成"失败"
+        ctx.emit(EventType.NODE_FAILED, error=message, duration_ms=elapsed, node_id=node.id)

        # 容错模式：记下错误继续往下走，而不是让整张图挂掉
        if node.config.get("on_error") == "continue":
@@ -138,6 +141,10 @@ def _wrap(node: GraphNode, run_ctx: RunContext) -> Callable[[GraphState], Awaita
                "nodes": {node.id: {"error": message, "failed": True}},
                "trail": [{"node_id": node.id, "error": message, "ts": time.time()}],
            }
+        # 已经是 NodeError 就别再套一层，否则消息会变成
+        # "NodeError: NodeError: 真正的原因"
+        if isinstance(last_error, NodeError):
+            raise last_error
        raise NodeError(node.id, message) from last_error

    async def _entry(state: GraphState) -> dict[str, Any]:
@@ -146,6 +153,15 @@ def _wrap(node: GraphNode, run_ctx: RunContext) -> Callable[[GraphState], Awaita
    return _entry


+def _describe(error: BaseException | None) -> str:
+    """错误消息。NodeError 的 message 本来就是写给人看的，不用再前缀类型名。"""
+    if error is None:
+        return "未知错误"
+    if isinstance(error, NodeError):
+        return str(error)
+    return f"{type(error).__name__}: {error}"
+
+
 def _preview(updates: dict[str, Any], node_id: str) -> Any:
    """给前端卡片用的输出摘要，避免把整坨数据塞进事件流。"""
    payload = (updates.get("nodes") or {}).get(node_id)
@@ -107,6 +107,27 @@ def render_deep(value: Any, ctx: dict[str, Any]) -> Any:
 # 安全表达式求值（分支条件用）
 # --------------------------------------------------------------------------

+# 幂运算的上界。`9**9**9**9` 只有 11 个 AST 节点，轻松过掉节点数护栏，
+# 却会让 CPython 无限期地算一个天文数字的大整数 —— 而求值是在事件循环里
+# 同步执行的（分支条件、transform、口径卡、每个节点的 skip_if 都走这里），
+# 既没有线程池隔离也没有超时，asyncio 的 cancel 打断不了同步 CPU。
+# 一条画布上的分支条件就能把整个后端冻死。
+_MAX_POW_EXPONENT = 1024
+_MAX_POW_BASE = 1e15
+
+
+def _safe_pow(base: Any, exponent: Any) -> Any:
+    """带上界的幂运算：超界直接报错，而不是让进程算到天荒地老。"""
+    try:
+        if abs(exponent) > _MAX_POW_EXPONENT:
+            raise ExpressionError(f"指数超过上限 {_MAX_POW_EXPONENT}")
+        if abs(base) > _MAX_POW_BASE and exponent > 1:
+            raise ExpressionError(f"底数超过上限 {_MAX_POW_BASE:g}")
+    except TypeError as e:  # 非数值类型交给 operator.pow 自己报
+        raise ExpressionError(f"幂运算的操作数必须是数值：{e}") from e
+    return operator.pow(base, exponent)
+
+
 _BIN_OPS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
@@ -114,7 +135,7 @@ _BIN_OPS = {
    ast.Div: operator.truediv,
    ast.FloorDiv: operator.floordiv,
    ast.Mod: operator.mod,
-    ast.Pow: operator.pow,
+    ast.Pow: _safe_pow,
 }
 _CMP_OPS = {
    ast.Eq: operator.eq,
@@ -42,13 +42,21 @@ class RunManager:
        settings.ensure_dirs()
        self._cm = AsyncSqliteSaver.from_conn_string(str(settings.checkpoint_path))
        self._checkpointer = await self._cm.__aenter__()
-        # 上次进程被强杀时留下的 running 记录，标成中断态而不是永远转圈
+        # 上次进程被强杀时留下的记录。running 和 queued 要分开处理：
+        # running 至少跑过一步、有 checkpoint，可以续；queued 还在排队，
+        # 一个 checkpoint 都没写过，"恢复"它只会从 START 用默认值重跑一遍，
+        # 还丢掉 run.input——那不是恢复，是伪造一份结果。
        async with SessionLocal() as session:
            await session.execute(
                update(Run)
-                .where(Run.status.in_(["running", "queued"]))
+                .where(Run.status == "running")
                .values(status="interrupted", error="服务重启，运行已挂起，可从断点恢复")
            )
+            await session.execute(
+                update(Run)
+                .where(Run.status == "queued")
+                .values(status="failed", error="服务重启时这次运行还在排队，没有可恢复的进度，请重新发起")
+            )
            await session.commit()

    async def shutdown(self) -> None:
@@ -160,8 +168,23 @@ class RunManager:
        )
        return run

-    async def resume(self, run_id: str, response: Any) -> Run:
-        """从人工介入的断点继续。"""
+    async def resume(
+        self, run_id: str, response: Any, *, approval_id: str | None = None,
+        actor: str | None = None,
+    ) -> Run:
+        """从人工介入的断点继续。
+
+        两件事以前是错的：
+
+        一是不看有没有 checkpoint 就发 Command。重启扫描会把 queued（从没跑过、
+        没有任何 checkpoint）也标成 interrupted，对这种 run 发 Command，LangGraph
+        会拿一个空 state 从 START 重跑整张图——run.input 完全没被注入，最后还落成
+        succeeded。跑出来的是一份用默认值算的假结果。
+
+        二是同一 superstep 里有多个 interrupt 时，只发标量 Command 会让 LangGraph
+        直接抛 RuntimeError 打爆整个 run，而所有 pending 审批还都被写成同一个回复。
+        现在按 interrupt_id 逐条收集，凑齐了才继续，没凑齐就继续等下一个人回复。
+        """
        async with SessionLocal() as session:
            run = await session.get(Run, run_id)
            if not run:
@@ -169,29 +192,103 @@ class RunManager:
            if run.status not in ("interrupted",):
                raise ValueError(f"当前状态是 {run.status}，无法恢复")
            spec = GraphSpec.model_validate(run.graph)
-            bus.set_seq(run_id, run.last_seq)
-            run.status = "running"
-            run.error = None
+            workflow_id = run.workflow_id

-            # 把这次恢复对应的待审批记录结掉，前端的待办列表才不会一直挂着
-            pending = (
+        # 先问引擎：这条线程到底停在哪、还等着哪些 interrupt
+        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)
+        waiting = list(getattr(snapshot, "interrupts", None) or [])
+
+        if not waiting and not getattr(snapshot, "next", None):
+            # 没有断点可续。多半是重启时被标成 interrupted 的 queued 运行。
+            raise ValueError(
+                "这次运行没有可恢复的断点（很可能重启前还没真正开始跑）。"
+                "请重新发起一次运行，而不是恢复——继续下去只会用默认值跑出一份假结果。"
+            )
+
+        answers: dict[str, Any] = {}
+        async with SessionLocal() as session:
+            pending = list((
                await session.execute(
                    select(Approval).where(
                        Approval.run_id == run_id, Approval.status == "pending"
                    )
                )
-            ).scalars()
-            for approval in pending:
-                approval.status = "resolved"
-                approval.response = _safe(response) if isinstance(response, dict) else {"value": _safe(response)}
-                approval.resolved_at = datetime.now(timezone.utc)
+            ).scalars())

+            # 定位这次回复的是哪一条：显式指定优先，否则只在唯一 pending 时才敢猜
+            target = None
+            if approval_id:
+                target = next((a for a in pending if a.id == approval_id), None)
+                if target is None:
+                    raise ValueError(f"审批 {approval_id} 不在待处理列表里")
+            elif len(pending) == 1:
+                target = pending[0]
+            elif len(pending) > 1:
+                raise ValueError(
+                    f"这次运行有 {len(pending)} 条待处理的人工介入，"
+                    "恢复时必须指定 approval_id 说明回复的是哪一条"
+                )
+
+            now = datetime.now(timezone.utc)
+            if target is not None:
+                target.status = "answered"  # 已回复，但还没提交给引擎
+                target.response = (
+                    _safe(response) if isinstance(response, dict) else {"value": _safe(response)}
+                )
+                target.resolved_at = now
+                if actor:
+                    target.resolved_by = actor
+                await session.commit()
+
+            # 凑齐没有？引擎还在等的每一个 interrupt 都得有答案
+            answered = list((
+                await session.execute(
+                    select(Approval).where(
+                        Approval.run_id == run_id,
+                        Approval.status.in_(["answered", "pending"]),
+                    )
+                )
+            ).scalars())
+            by_iid = {a.interrupt_id: a for a in answered if a.interrupt_id}
+            missing = []
+            for item in waiting:
+                iid = getattr(item, "id", None)
+                rec = by_iid.get(iid)
+                if rec is None or rec.status != "answered":
+                    missing.append(iid)
+                    continue
+                payload = rec.response or {}
+                answers[iid] = payload.get("value", payload) if isinstance(payload, dict) else payload
+
+            if missing:
+                # 还差人没回。保持 interrupted，把已回复的那条留在 answered 上等齐
+                await session.commit()
+                run = await session.get(Run, run_id)
+                return run
+
+            # 齐了：正式落 resolved，然后驱动引擎
+            for rec in answered:
+                if rec.status == "answered":
+                    rec.status = "resolved"
+            run = await session.get(Run, run_id)
+            bus.set_seq(run_id, run.last_seq)
+            run.status = "running"
+            run.error = None
            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)})
        self._tasks[run_id] = asyncio.create_task(
-            self._drive(run_id, spec, Command(resume=response), workflow_id=run.workflow_id)
+            self._drive(run_id, spec, command, workflow_id=workflow_id)
        )
        return run