循环跑满自己声明的轮数，不再被全局步数上限半路掐断

「测试 1」按校验提示改好之后仍然跑不完：while 循环配了 1005 轮，每轮是"循环节点 +
循环体"两步，而整次运行的步数上限（max_graph_steps）是 200——跑到 100 轮上下 LangGraph
抛 GraphRecursionError，报错是一段英文，叫人去调一个界面上碰不到的 recursion_limit。
图没写错，是平台自己的两个上限在打架：校验通过，跑到一半才死。

全局上限防的是没人把关的环；loop 节点的 max_iterations 是作者明说的轮数上限。现在循环
按自己的轮数另记一份预算（schema.loop_steps），加在全局上限之上，子图同理。取上界：
一个节点最多跑 Π(包住它的各层循环 max_iterations + 1) 次，嵌套相乘；done 出口之后的
节点只跑一次，不按轮数记账，否则预算虚高、等于没有上限。

没有 loop 节点把关的环（分支连回上游）照旧由全局上限拦住，报错改成中文、说清预算怎么
来的、该怎么改；子图里撞上限报子图自己的名字和步数——以前那段英文会被外层包进 NodeError
原样带出来。

实测：「测试 1」照提示改完，在真沙箱里 1000 轮跑满，25 秒，结果 501500（= 2+3+…+1001）。

测试：test_step_budget 8 条——预算的上界、嵌套相乘、done 之后不记账；全局上限压到 20 步
时循环照样跑满 30 轮；空转的环中文报错；子图里的循环有自己的预算、空转的子图报它自己的
名字。把 runner / 子图的预算换成 0，对应用例确实失败。README 补了常见问题，测试数更新为 470。

README.md

@@ -437,12 +437,17 @@ agent 侧同样会收到实情，并被告知可直接查询 `information_schema
 单次模型调用默认 300 秒没有输出即超时。多半是模型或工具没有响应；确实需要更久，调大
 `AGENTLAB_MAX_RUN_SECONDS` / `AGENTLAB_MODEL_TIMEOUT_SECONDS`。设置页「运行环境」里能看到当前值。
 
+**运行报「走满了 N 步还没跑完」。** 循环按自己 `max_iterations` 声明的轮数另算步数预算，
+配多少轮就能跑多少轮，不受全局上限约束。走到这条报错的，多半是分支连回上游、却没有
+loop 节点给它定轮数上限的环：用 loop 节点包住它。确实需要更多步，调大
+`AGENTLAB_MAX_GRAPH_STEPS`（默认 200）。
+
 ---
 
 ## 开发与验证
 
 ```bash
-# 后端：292 个测试
+# 后端：470 个测试
 cd backend && pytest tests
 
 # 前端类型检查

backend/app/engine/nodes/multi.py

@@ -6,6 +6,7 @@ import time
 from typing import Any
 
 from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
+from langgraph.errors import GraphRecursionError
 from langgraph.func import task
 
 from app.core.config import settings
@@ -367,7 +368,7 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
     每一步；但它有独立的状态，不会污染父图的变量池。
     """
     from app.engine.compiler import compile_graph  # 延迟导入，避免循环依赖
-    from app.engine.schema import GraphSpec
+    from app.engine.schema import GraphSpec, loop_steps
 
     if ctx.run.depth >= _MAX_DEPTH:
         raise NodeError(ctx.node.id, f"子图嵌套超过 {_MAX_DEPTH} 层，已阻止")
@@ -429,19 +430,28 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
     compiled = compile_graph(sub_spec, sub_run)
     # 子图不带 checkpointer：它的断点由父图的 checkpoint 统一承载
     app = compiled.compile()
-    result_state = await app.ainvoke(
-        {
-            "input": sub_input,
-            "vars": sub_input,
-            "nodes": {},
-            "output": {},
-            "messages": [],
-            "usage": {},
-            "trail": [],
-            "loops": {},
-        },
-        {"recursion_limit": settings.max_graph_steps},
-    )
+    limit = settings.max_graph_steps + loop_steps(sub_spec)
+    try:
+        result_state = await app.ainvoke(
+            {
+                "input": sub_input,
+                "vars": sub_input,
+                "nodes": {},
+                "output": {},
+                "messages": [],
+                "usage": {},
+                "trail": [],
+                "loops": {},
+            },
+            {"recursion_limit": limit},
+        )
+    except GraphRecursionError as e:
+        # 不接住的话，外层包装成 NodeError 时带的是 LangGraph 的英文原文
+        raise NodeError(
+            ctx.node.id,
+            f"子图「{workflow.name}」走满了 {limit} 步还没跑完，已中止。多半是子图里有个环"
+            "在空转：分支连回了上游、却没有 loop 节点给它定轮数上限",
+        ) from e
 
     output = result_state.get("output") or {}
     result = {

backend/app/engine/runner.py

@@ -9,6 +9,7 @@ from datetime import datetime, timezone
 from typing import Any
 
 from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+from langgraph.errors import GraphRecursionError
 from langgraph.types import Command
 from sqlalchemy import select, update
 
@@ -19,7 +20,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, topology_of, validate_graph
+from app.engine.schema import GraphSpec, loop_steps, topology_of, validate_graph
 
 logger = logging.getLogger(__name__)
 
@@ -466,9 +467,11 @@ class RunManager:
                     collection=collection,
                 )
                 app = compile_graph(spec, run_ctx).compile(checkpointer=self.checkpointer)
+                # 循环按自己声明的轮数另记一份预算，全局上限只管没人把关的环
+                extra_steps = loop_steps(spec)
                 config = {
                     "configurable": {"thread_id": run_id},
-                    "recursion_limit": settings.max_graph_steps,
+                    "recursion_limit": settings.max_graph_steps + extra_steps,
                 }
 
                 interrupted = False
@@ -520,6 +523,18 @@ class RunManager:
                     error = f"{type(e).__name__}: {e}"
                     logger.exception("run %s 失败", run_id)
                 await self._emit(run_id, EventType.RUN_FAILED, data={"error": error})
+            except GraphRecursionError:
+                status = "failed"
+                # LangGraph 的原文是英文，还叫人去调 recursion_limit——用户碰不到那个键。
+                # 循环的轮数已经另算了预算，走到这里的多半是不归 loop 节点管的环
+                error = (
+                    f"走满了 {settings.max_graph_steps + extra_steps} 步还没跑完，已中止"
+                    f"（图最大步数 {settings.max_graph_steps}"
+                    + (f"，另按循环轮数预留了 {extra_steps} 步" if extra_steps else "") + "）。"
+                    "多半是有个环在空转：分支连回了上游、却没有 loop 节点给它定轮数上限。"
+                    "用 loop 节点包住它；确实需要这么多步，调大 AGENTLAB_MAX_GRAPH_STEPS"
+                )
+                await self._emit(run_id, EventType.RUN_FAILED, data={"error": error})
             except Exception as e:  # noqa: BLE001
                 status = "failed"
                 # NodeError 的 message 本来就是写给人看的（"人工驳回：…"），

backend/app/engine/schema.py

@@ -211,6 +211,51 @@ def back_edges(
     return back
 
 
+def loop_steps(spec: GraphSpec) -> int:
+    """图里的循环按各自声明的轮数跑满，最多要走多少步（LangGraph 的 superstep）。
+
+    整次运行的步数上限（max_graph_steps）防的是没人把关的环；loop 节点自己带着
+    max_iterations，那是作者明说的轮数上限。以前两者打架时全局值悄悄赢：「测试 1」
+    的 while 循环配了 1005 轮、每轮 2 步，跑到 100 轮上下就被 GraphRecursionError
+    掐断——校验通过，跑到一半才死。runner 把这个数加在全局上限之上。
+
+    取上界不求精确：一个节点最多跑 Π(包住它的各层循环的 max_iterations + 1) 次，
+    逐个加起来。一步至少跑一个节点，所以步数只会比这少。
+    """
+    loops = [n for n in spec.nodes if n.type == NodeType.LOOP]
+    if not loops:
+        return 0
+    succ: dict[str, list[str]] = defaultdict(list)
+    pred: dict[str, list[str]] = defaultdict(list)
+    for e in spec.edges:
+        succ[e.source].append(e.target)
+        pred[e.target].append(e.source)
+
+    def reach(starts: Iterable[str], step: dict[str, list[str]], avoid: str) -> set[str]:
+        seen: set[str] = set()
+        stack = [s for s in starts if s != avoid]
+        while stack:
+            n = stack.pop()
+            if n not in seen:
+                seen.add(n)
+                stack.extend(x for x in step[n] if x != avoid)
+        return seen
+
+    runs: dict[str, int] = defaultdict(lambda: 1)
+    for loop in loops:
+        try:
+            cap = int(loop.config.get("max_iterations", 10) or 10)
+        except (TypeError, ValueError):
+            cap = 10    # 运行到这个节点时同样会报错，这里不替它报
+        # 循环体 = 从 body 出口出发、不经过循环节点本身、又能绕回它的那些节点。
+        # 没标出口的边路由时按 default 走，body 也会走它
+        entries = [e.target for e in spec.outgoing(loop.id) if e.sourceHandle != "done"]
+        body = reach(entries, succ, loop.id) & reach(pred[loop.id], pred, loop.id)
+        for n in body | {loop.id}:
+            runs[n] *= max(cap, 0) + 1
+    return sum(runs.values())
+
+
 def expression_fields(node: GraphNode) -> list[tuple[str, str]]:
     """这个节点上按表达式求值的字段：(给人看的名字, 原文)。
 

backend/tests/test_step_budget.py

@@ -0,0 +1,164 @@
+"""循环跑满自己声明的轮数，不被全局步数上限半路掐断。
+
+开发库里的「测试 1」：while 循环 `vars.state.count <= 1000`，max_iterations 配了
+1005。每轮是「循环节点 + 循环体」两步，而整次运行的步数上限（max_graph_steps）
+是 200——跑到 100 轮上下 LangGraph 抛 GraphRecursionError，报错是一段英文，
+叫人去调一个界面上根本碰不到的 recursion_limit。校验是通过的：这张图没写错，
+是平台自己的两个上限在打架。
+
+全局上限防的是没人把关的环；loop 节点的 max_iterations 是作者明说的轮数上限。
+所以循环按自己的轮数另记一份预算（loop_steps），全局上限照旧管其余的环。
+"""
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from app.core.config import settings
+from app.db.base import SessionLocal
+from app.db.models import Run
+from app.engine.runner import run_manager
+from app.engine.schema import GraphSpec, loop_steps
+
+
+def node(nid, ntype, **config):
+    return {"id": nid, "type": ntype, "data": {"label": nid, "config": config}}
+
+
+def edge(source, target, handle=None):
+    return {"source": source, "target": target, "sourceHandle": handle}
+
+
+def counting_loop(rounds: int, cap: int) -> dict:
+    """「测试 1」的骨架：while 数到 rounds，循环体一个节点。"""
+    return {
+        "nodes": [
+            node("start", "input", fields=[]),
+            node("init", "transform", expression="0", assign_to="n"),
+            node("lp", "loop", mode="while", condition=f"vars.n < {rounds}", max_iterations=cap),
+            node("inc", "transform", expression="vars.n + 1", assign_to="n"),
+            node("out", "output", fields=[{"name": "结果", "value": "{{ vars.n }}"}]),
+        ],
+        "edges": [edge("start", "init"), edge("init", "lp"), edge("lp", "inc", "body"),
+                  edge("inc", "lp"), edge("lp", "out", "done")],
+    }
+
+
+def test_no_loop_no_extra_budget():
+    spec = GraphSpec.model_validate({
+        "nodes": [node("a", "input"), node("b", "output")], "edges": [edge("a", "b")]})
+    assert loop_steps(spec) == 0
+
+
+def test_budget_covers_the_declared_rounds():
+    # 1005 轮 × 每轮 2 步，外加收尾的一次判定
+    assert loop_steps(GraphSpec.model_validate(counting_loop(1000, 1005))) >= 1005 * 2 + 1
+
+
+def test_nested_loops_multiply():
+    """内层循环每一轮外层都要从头跑一遍，预算得乘起来，不是加起来。"""
+    spec = GraphSpec.model_validate({
+        "nodes": [node("start", "input"),
+                  node("outer", "loop", mode="foreach", items="[1,2,3]", max_iterations=3),
+                  node("inner", "loop", mode="foreach", items="[1,2,3,4]", max_iterations=4),
+                  node("work", "transform", expression="1"),
+                  node("out", "output")],
+        "edges": [edge("start", "outer"), edge("outer", "inner", "body"),
+                  edge("inner", "work", "body"), edge("work", "inner"),
+                  edge("inner", "outer", "done"), edge("outer", "out", "done")],
+    })
+    # 实际：外层 4 次判定 + 3 轮 ×（内层 5 次判定 + 4 次干活）= 31 步
+    assert loop_steps(spec) >= 31
+
+
+def test_nodes_after_the_loop_are_not_charged_per_round():
+    """done 出口之后的节点只跑一次，不该按轮数记账——否则预算虚高，等于没有上限。"""
+    spec = GraphSpec.model_validate(counting_loop(3, 3))
+    assert loop_steps(spec) == (3 + 1) * 2   # 只有 lp 和 inc
+
+
+@pytest.fixture
+async def engine_up():
+    await run_manager.setup()
+    yield
+    await run_manager.shutdown()
+
+
+async def _finish(graph) -> Run:
+    run = await run_manager.start(graph=graph, input_payload={})
+    for _ in range(600):
+        async with SessionLocal() as session:
+            row = await session.get(Run, run.id)
+        if row.status in ("succeeded", "failed", "cancelled", "interrupted"):
+            return row
+        await asyncio.sleep(0.05)
+    raise AssertionError(f"没跑完：{row.status}")
+
+
+async def test_a_loop_runs_past_the_global_step_limit(engine_up, monkeypatch):
+    """全局上限压到 20 步，循环照样跑满它声明的 30 轮。"""
+    monkeypatch.setattr(settings, "max_graph_steps", 20)
+    row = await _finish(counting_loop(30, 35))
+    assert row.status == "succeeded", row.error
+    assert row.output["结果"] == "30"
+
+
+async def test_an_unguarded_cycle_still_stops_and_says_why(engine_up, monkeypatch):
+    """分支连回上游、没有 loop 节点把关的环，照旧由全局上限拦住——报错得是人话。"""
+    monkeypatch.setattr(settings, "max_graph_steps", 12)
+    row = await _finish({
+        "nodes": [node("start", "input", fields=[]),
+                  node("t", "transform", expression="1"),
+                  node("br", "branch", cases=[{"key": "again", "condition": "true"}]),
+                  node("out", "output", fields=[])],
+        "edges": [edge("start", "t"), edge("t", "br"), edge("br", "t", "again"),
+                  edge("br", "out", "default")],
+    })
+    assert row.status == "failed"
+    assert "走满了 12 步" in row.error and "loop 节点" in row.error, row.error
+    assert "recursion_limit" not in row.error
+
+
+# --------------------------------------------------------------------------
+# 子图：自己的循环自己算预算；撞上限时报子图自己的名字和步数
+# --------------------------------------------------------------------------
+
+
+async def _parent_of(sub_graph: dict, name: str) -> dict:
+    from app.db.models import Workflow
+
+    async with SessionLocal() as session:
+        wf = Workflow(name=name, graph=sub_graph)
+        session.add(wf)
+        await session.commit()
+        wf_id = wf.id
+    return {
+        "nodes": [node("start", "input", fields=[]),
+                  node("sub", "subgraph", workflow_id=wf_id),
+                  node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.sub.output.结果 }}"}])],
+        "edges": [edge("start", "sub"), edge("sub", "out")],
+    }
+
+
+async def test_a_loop_inside_a_subgraph_gets_its_own_budget(engine_up, monkeypatch):
+    monkeypatch.setattr(settings, "max_graph_steps", 20)
+    row = await _finish(await _parent_of(counting_loop(30, 35), "数到 30"))
+    assert row.status == "succeeded", row.error
+    assert row.output["结果"] == "30"
+
+
+async def test_an_unguarded_cycle_inside_a_subgraph_says_which_one(engine_up, monkeypatch):
+    monkeypatch.setattr(settings, "max_graph_steps", 12)
+    spinning = {
+        "nodes": [node("start", "input", fields=[]),
+                  node("t", "transform", expression="1"),
+                  node("br", "branch", cases=[{"key": "again", "condition": "true"}]),
+                  node("out", "output", fields=[])],
+        "edges": [edge("start", "t"), edge("t", "br"), edge("br", "t", "again"),
+                  edge("br", "out", "default")],
+    }
+    row = await _finish(await _parent_of(spinning, "空转的子图"))
+    assert row.status == "failed"
+    assert "子图「空转的子图」走满了 12 步" in row.error, row.error
+    assert "recursion_limit" not in row.error