From e46a3f0d62bcff035fa43a24b8a438a59803494f Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 25 Sep 2026 23:14:50 +0800 Subject: [PATCH] =?UTF-8?q?=E5=B5=8C=E5=A5=97=E5=BE=AA=E7=8E=AF=E7=9A=84?= =?UTF-8?q?=E5=86=85=E5=B1=82=E6=AF=8F=E8=BD=AE=E9=83=BD=E4=BB=8E=E5=A4=B4?= =?UTF-8?q?=E6=95=B0=EF=BC=9Amax=5Fiterations=20=E6=98=AF"=E6=AF=8F?= =?UTF-8?q?=E8=BF=9B=E6=9D=A5=E4=B8=80=E6=AC=A1"=E7=9A=84=E4=B8=8A?= =?UTF-8?q?=E9=99=90=EF=BC=8C=E4=B8=8D=E6=98=AF=E6=95=B4=E6=AC=A1=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E5=90=88=E8=AE=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 运行 87daca70：一个计时器，外层每一拍里内层跑两个 50ms 的子拍。5 秒该推进 50 拍， 结果推进了 138 拍。循环节点走 done 出口时计数原样留着，外层下一轮再把它带进来，就从 上次停下的地方接着往上数：内层 max_iterations 是 5，第一、二拍各跑 2 个子拍，第三拍 1 个，从第四拍起一进来就撞上限直接退出，子拍一次都不跑（撞上限的 warn 136 条）。 嵌套的 foreach 更糟：外层第二轮时内层游标已经在末尾，内层一项都不再跑，也不报错， 结果只是悄悄少了一大截。 改法：走 done 时记 finished，再被带进来时从 0 数。index 仍停在最后一轮，结束后读 loops.x.index 的照旧拿到轮数；画布上内层的"第 N/上限 轮"也随之每轮重新数起。 测试：test_nested_loops 两条——嵌套 while 合计 6 次（修复前 3 次）、嵌套 foreach 走完 3×2 项（修复前只有 1a、1b）。用 87daca70 的原图在临时库里真沙箱重跑：内层每拍都是 "跑、跑、结束"，撞上限 0 次。后端 473 passed ×2。 --- backend/app/engine/nodes/control.py | 10 ++- backend/tests/test_nested_loops.py | 105 ++++++++++++++++++++++++++++ 2 files changed, 113 insertions(+), 2 deletions(-) create mode 100644 backend/tests/test_nested_loops.py diff --git a/backend/app/engine/nodes/control.py b/backend/app/engine/nodes/control.py index 4b3f9f6c..f8d208bb 100644 --- a/backend/app/engine/nodes/control.py +++ b/backend/app/engine/nodes/control.py @@ -116,7 +116,12 @@ async def run_loop(state: GraphState, ctx: NodeContext) -> dict[str, Any]: mode = ctx.cfg("mode", "foreach") node_id = ctx.node.id loops = dict(state.get("loops") or {}) - cursor = int((loops.get(node_id) or {}).get("index", 0)) if isinstance(loops.get(node_id), dict) else 0 + prev = loops.get(node_id) if isinstance(loops.get(node_id), dict) else {} + # 上次已经走过 done 出口还又进来：是外层循环的下一轮把它重新带进来，从头数。 + # 以前计数原样留着接着往上加，内层的 max_iterations 于是成了整次运行合计的 + # 上限——运行 87daca70 的内层每拍该跑两个子拍，第四拍起一进来就撞上限直接 + # 退出；嵌套的 foreach 更糟，外层第二轮起内层一项都不跑，也不报错 + cursor = 0 if prev.get("finished") else int(prev.get("index", 0)) max_iter = int(ctx.cfg("max_iterations", 10) or 10) tctx = template_context(state) @@ -177,7 +182,8 @@ async def run_loop(state: GraphState, ctx: NodeContext) -> dict[str, Any]: } updates: dict[str, Any] = { "nodes": {node_id: result}, - "loops": {node_id: {"index": cursor + 1 if keep_going else cursor}}, + # 走完了 index 留在最后一轮，读 loops.x.index 的照旧拿到轮数；finished 让下次进来重数 + "loops": {node_id: {"index": cursor + 1 if keep_going else cursor, "finished": not keep_going}}, } if updates_vars: updates["vars"] = updates_vars diff --git a/backend/tests/test_nested_loops.py b/backend/tests/test_nested_loops.py new file mode 100644 index 00000000..b6e32ae1 --- /dev/null +++ b/backend/tests/test_nested_loops.py @@ -0,0 +1,105 @@ +"""嵌套循环：内层循环每被外层带进来一次，都从头数。 + +运行 87daca70：一个计时器，外层每一拍里内层跑两个 50ms 的子拍。5 秒该推进 50 拍， +结果推进了 138 拍——内层循环的计数走 done 出口时原样留着，外层下一轮再把它带进来， +就从上次停下的地方接着往上数。内层 max_iterations 是 5：第一拍用掉 2 次，第二拍 +2 次，第三拍 1 次，从第四拍起一进来就撞上限直接退出，子拍一次都不跑（撞上限的 +warn 136 条）。max_iterations 于是成了"整次运行合计多少轮"，而不是"每进来一次 +最多多少轮"。 + +嵌套的 foreach 更糟：外层第二轮时内层游标已经在末尾，内层一项都不再跑，没有报错， +结果只是悄悄少了一大截。 +""" +from __future__ import annotations + +import asyncio +import json + +import pytest + +from app.db.base import SessionLocal +from app.db.models import Run +from app.engine.runner import run_manager + + +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} + + +@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_an_inner_while_restarts_on_every_outer_round(engine_up): + """87daca70 的形状：外层 3 拍，每拍内层 2 个子拍，合计 6 次。 + + 内层上限给 3（够一拍用，不够三拍合计用）：计数不复位的话第二拍只跑 1 次、 + 第三拍一次不跑，合计 3。 + """ + row = await _finish({ + "nodes": [ + node("start", "input", fields=[]), + node("init_i", "transform", expression="0", assign_to="i"), + node("init_total", "transform", expression="0", assign_to="total"), + node("outer", "loop", mode="while", condition="vars.i < 3", max_iterations=10), + node("reset_j", "transform", expression="0", assign_to="j"), + node("inner", "loop", mode="while", condition="vars.j < 2", max_iterations=3), + node("bump_j", "transform", expression="vars.j + 1", assign_to="j"), + node("count", "transform", expression="vars.total + 1", assign_to="total"), + node("bump_i", "transform", expression="vars.i + 1", assign_to="i"), + node("out", "output", fields=[{"name": "结果", "value": "{{ vars.total }}"}]), + ], + "edges": [ + edge("start", "init_i"), edge("init_i", "init_total"), edge("init_total", "outer"), + edge("outer", "reset_j", "body"), edge("reset_j", "inner"), + edge("inner", "bump_j", "body"), edge("bump_j", "count"), edge("count", "inner"), + edge("inner", "bump_i", "done"), edge("bump_i", "outer"), + edge("outer", "out", "done"), + ], + }) + assert row.status == "succeeded", row.error + assert row.output["结果"] == "6" + + +async def test_an_inner_foreach_walks_its_items_on_every_outer_round(engine_up): + """外层 3 项 × 内层 2 项。游标不复位的话，外层第二轮起内层一项都不跑，只剩 1a、1b。""" + row = await _finish({ + "nodes": [ + node("start", "input", fields=[]), + node("init_log", "transform", expression="[]", assign_to="log"), + node("outer", "loop", mode="foreach", items="[1, 2, 3]", item_var="o", + max_iterations=10), + node("inner", "loop", mode="foreach", items='["a", "b"]', item_var="x", + max_iterations=10), + node("add", "transform", expression="vars.log + [str(vars.o) + vars.x]", + assign_to="log"), + node("out", "output", fields=[{"name": "结果", "value": "{{ vars.log | json }}"}]), + ], + "edges": [ + edge("start", "init_log"), edge("init_log", "outer"), + edge("outer", "inner", "body"), + edge("inner", "add", "body"), edge("add", "inner"), + edge("inner", "outer", "done"), + edge("outer", "out", "done"), + ], + }) + assert row.status == "succeeded", row.error + assert json.loads(row.output["结果"]) == ["1a", "1b", "2a", "2b", "3a", "3b"]