From d1ad670fb3169b23fc411d87614629795f6e54cc Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 09:10:27 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=8D=E6=A0=B8=E8=A7=84=E5=88=99=E5=AF=B9?= =?UTF-8?q?=E5=AD=90=E5=B7=A5=E4=BD=9C=E6=B5=81=E9=87=8C=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E7=9A=84=E9=82=A3=E4=B8=80=E6=AD=A5=E5=8F=AA=E6=8A=A5=E4=B8=80?= =?UTF-8?q?=E6=9D=A1=EF=BC=9A=E5=AD=90=E5=9B=BE=E9=87=8C=E9=82=A3=E4=B8=80?= =?UTF-8?q?=E6=AD=A5=E3=80=81=E5=AD=90=E5=B7=A5=E4=BD=9C=E6=B5=81=E8=8A=82?= =?UTF-8?q?=E7=82=B9=E5=90=84=E5=8F=91=E4=B8=80=E6=9D=A1=E5=8E=9F=E8=AF=9D?= =?UTF-8?q?=E7=9B=B8=E5=90=8C=E7=9A=84=20node.failed=EF=BC=8C=E4=BB=A5?= =?UTF-8?q?=E5=89=8D=E5=90=88=E6=88=90=E4=B8=A4=E6=9D=A1=E3=80=8C=E6=89=A7?= =?UTF-8?q?=E8=A1=8C=E5=A4=B1=E8=B4=A5=E3=80=8D=EF=BC=88=E9=A9=B3=E5=9B=9E?= =?UTF-8?q?=E5=90=8C=E7=90=86=E4=B8=A4=E6=9D=A1=E3=80=8C=E8=A2=AB=E9=A9=B3?= =?UTF-8?q?=E5=9B=9E=E3=80=8D=EF=BC=89=EF=BC=8C=E9=97=AE=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E7=9A=84=E5=A4=8D=E6=A0=B8=E6=8F=90=E7=A4=BA=E9=87=8C=E5=90=8C?= =?UTF-8?q?=E4=B8=80=E4=BB=B6=E4=BA=8B=E5=86=99=E4=B8=A4=E9=81=8D=E3=80=82?= =?UTF-8?q?=E7=8E=B0=E5=9C=A8=E4=BF=9D=E7=95=99=E6=9B=B4=E5=85=B7=E4=BD=93?= =?UTF-8?q?=E7=9A=84=E9=82=A3=E4=B8=80=E6=AD=A5=EF=BC=8C=E5=90=8D=E5=AD=97?= =?UTF-8?q?=E4=B8=80=E5=B1=82=E5=B1=82=E6=8E=A5=E4=B8=8A=E5=86=99=E6=B8=85?= =?UTF-8?q?=E5=9C=A8=E5=93=AA=E4=B8=AA=E5=AD=90=E5=B7=A5=E4=BD=9C=E6=B5=81?= =?UTF-8?q?=E9=87=8C=EF=BC=88=E3=80=8C=E5=A4=8D=E6=A0=B8=E5=AD=90=E6=B5=81?= =?UTF-8?q?=E7=A8=8B=20=E2=80=BA=20=E7=AE=97=E4=B8=80=E4=B8=8B=E3=80=8D?= =?UTF-8?q?=EF=BC=89=EF=BC=8C=E4=B8=8D=E5=86=99=E5=86=85=E9=83=A8=20id?= =?UTF-8?q?=EF=BC=9B=E5=AD=90=E5=B7=A5=E4=BD=9C=E6=B5=81=E8=8A=82=E7=82=B9?= =?UTF-8?q?=E8=87=AA=E5=B7=B1=E5=A4=B1=E8=B4=A5=E3=80=81=E5=8E=9F=E8=AF=9D?= =?UTF-8?q?=E4=B8=8D=E5=90=8C=E7=9A=84=E7=85=A7=E6=8A=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/engine/review.py | 22 +++++- backend/tests/test_review_subgraph.py | 105 ++++++++++++++++++++++++++ 2 files changed, 125 insertions(+), 2 deletions(-) create mode 100644 backend/tests/test_review_subgraph.py diff --git a/backend/app/engine/review.py b/backend/app/engine/review.py index 98f349ee..26b10939 100644 --- a/backend/app/engine/review.py +++ b/backend/app/engine/review.py @@ -206,6 +206,8 @@ def scan(events: Iterable[Any], output: Mapping[str, Any] | None = None) -> list elif etype == EventType.TOOL_END and tool and tool in failed_at: recovered.add(tool) + #: 报过的节点失败 (节点 id, 原话)：子工作流节点跟着子图里那一步失败时认得出是同一件事 + failures: list[tuple[str, str]] = [] for e in rows: etype = str(_event_field(e, "type", "")) data = _event_field(e, "data") or {} @@ -224,11 +226,18 @@ def scan(events: Iterable[Any], output: Mapping[str, Any] | None = None) -> list # 失败」；也不进 RETRYABLE——重跑只会把同一件事再问一遍人。答案没走完是缺口，记 DEGRADED（没有成果时 # empty_output 另按 BROKEN 说） err = str(data.get("error") or "").strip()[:200] - add("node_rejected", f"节点「{titles.get(node_id) or node_id}」被驳回：{err}", DEGRADED) + if any(inner.startswith(f"{node_id}/") and said == err for inner, said in failures): + continue # 子工作流里那一步被驳回、子工作流节点跟着收尾：同一件事，只报那一步（同下面的执行失败） + failures.append((str(node_id), err)) + add("node_rejected", f"节点「{_step_name(str(node_id), titles)}」被驳回：{err}", DEGRADED) elif etype == EventType.NODE_FAILED: err = str(data.get("error") or "").strip()[:200] - add("node_failed", f"节点「{titles.get(node_id) or node_id}」执行失败：{err}", BROKEN) + if any(inner.startswith(f"{node_id}/") and said == err for inner, said in failures): + # 子工作流里那一步失败了，子工作流节点（「sub」）跟着失败、原话相同：同一件事，只报更具体的那一步（上面已报） + continue + failures.append((str(node_id), err)) + add("node_failed", f"节点「{_step_name(str(node_id), titles)}」执行失败：{err}", BROKEN) elif etype == EventType.RUN_FAILED: err = str(data.get("error") or "").strip()[:200] @@ -264,6 +273,15 @@ def scan(events: Iterable[Any], output: Mapping[str, Any] | None = None) -> list return signals +def _step_name(node_id: str, titles: Mapping[str, str]) -> str: + """失败的那一步给人看的名字。子工作流里的（「子工作流节点 id/…」）一层层接上 node.started 记的名字：「复核子流程 › 算一下」， + 不写内部 id；没记名字的那一层照写 id 的那一段。""" + if "/" not in node_id: + return titles.get(node_id) or node_id + parts = node_id.split("/") + return " › ".join(titles.get("/".join(parts[:i + 1])) or parts[i] for i in range(len(parts))) + + def worst(signals: Iterable[Signal]) -> str: """一组信号里最重的那档。没有信号时返回空串。""" ranked = sorted(signals, key=lambda s: _SEVERITY_RANK.get(s.severity, 1), reverse=True) diff --git a/backend/tests/test_review_subgraph.py b/backend/tests/test_review_subgraph.py new file mode 100644 index 00000000..6ea79e4a --- /dev/null +++ b/backend/tests/test_review_subgraph.py @@ -0,0 +1,105 @@ +"""复核规则（engine/review.scan）对子工作流里失败的那一步只报一条：子图里那一步一条 node.failed、子工作流节点又一条，原话 +相同，以前合成两条「执行失败」，问数据页的复核提示里同一件事写两遍。保留更具体的那一步，名字写清在哪个子工作流里 +（「子工作流 › 那一步」，不写内部 id）。图和数据都是虚构的。 +""" +from __future__ import annotations + +import asyncio + +import pytest +from sqlalchemy import select + +from app.core.events import EventType +from app.db.base import SessionLocal +from app.db.models import Run, RunEvent +from app.engine import review as rv +from app.engine.runner import run_manager +from tests import ledger_samples as S + +OK = {"结果": "x"} + + +def ev(etype, **data): + return {"type": str(etype), "node_id": data.pop("node_id", ""), "data": data} + + +def test_a_failure_inside_a_subworkflow_is_reported_once_with_where_it_happened(): + signals = rv.scan([ + ev(EventType.NODE_STARTED, node_id="sub", label="复核子流程"), + ev(EventType.NODE_STARTED, node_id="sub/calc", label="算一下"), + ev(EventType.NODE_FAILED, node_id="sub/calc", error="表达式写错了"), + ev(EventType.NODE_FAILED, node_id="sub", error="表达式写错了"), + ], OK) + [signal] = [s for s in signals if s.kind == "node_failed"] + assert signal.detail == "节点「复核子流程 › 算一下」执行失败：表达式写错了" + assert "sub/" not in signal.detail and signal.severity == rv.BROKEN + + +def test_nested_two_levels_and_different_errors_stay_apart(): + signals = rv.scan([ + ev(EventType.NODE_STARTED, node_id="a", label="外层"), + ev(EventType.NODE_STARTED, node_id="a/b", label="内层"), + ev(EventType.NODE_STARTED, node_id="a/b/c", label="取数"), + ev(EventType.NODE_FAILED, node_id="a/b/c", error="连不上"), + ev(EventType.NODE_FAILED, node_id="a/b", error="连不上"), + ev(EventType.NODE_FAILED, node_id="a", error="连不上"), + # 父图里另一个节点原话不同的失败照报 + ev(EventType.NODE_STARTED, node_id="other", label="另一步"), + ev(EventType.NODE_FAILED, node_id="other", error="超时"), + ], OK) + assert [s.detail for s in signals if s.kind == "node_failed"] == [ + "节点「外层 › 内层 › 取数」执行失败：连不上", "节点「另一步」执行失败：超时"] + + +def test_a_rejection_inside_a_subworkflow_is_reported_once_too(): + signals = rv.scan([ + ev(EventType.NODE_STARTED, node_id="sub", label="复核子流程"), + ev(EventType.NODE_STARTED, node_id="sub/gate", label="把关"), + ev(EventType.NODE_FAILED, node_id="sub/gate", error="人工驳回：不对", reason="rejected"), + ev(EventType.NODE_FAILED, node_id="sub", error="人工驳回：不对", reason="rejected"), + ], OK) + assert [s.detail for s in signals if s.kind == "node_rejected"] == ["节点「复核子流程 › 把关」被驳回：人工驳回：不对"] + + +def test_a_subworkflow_failing_on_its_own_still_reports(): + """子工作流节点自己失败（没选工作流、版本不存在）：子图里没有哪一步失败，照报这一条。""" + signals = rv.scan([ + ev(EventType.NODE_STARTED, node_id="sub", label="复核子流程"), + ev(EventType.NODE_FAILED, node_id="sub", error="「子工作流」节点尚未选择要嵌套的工作流"), + ], OK) + assert [s.detail for s in signals if s.kind == "node_failed"] == [ + "节点「复核子流程」执行失败：「子工作流」节点尚未选择要嵌套的工作流"] + + +@pytest.fixture +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +def node(nid, ntype, label=None, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}} + + +async def test_a_real_run_failing_inside_a_subworkflow_gives_one_signal(engine_up): + inner = S.chain(node("s", "input"), node("calc", "transform", "算一下", mode="expression", expression="1 / 0"), + node("o", "output", fields=[{"name": "r", "value": "{{ nodes.calc.text }}"}])) + sub_id = await S.publish("复核子流程", graph=inner, level="draft") + graph = S.chain(node("start", "input"), node("sub", "subgraph", "复核子流程", workflow_id=sub_id, workflow_version=1), + node("out", "output", fields=[{"name": "r", "value": "{{ nodes.sub.output.r }}"}])) + run = await run_manager.start(graph=graph, input_payload={}) + for _ in range(400): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run.id) + if row.status in ("failed", "succeeded") and not run_manager.is_active(run.id): + break + assert row.status == "failed" + async with SessionLocal() as session: + rows = list((await session.execute(select(RunEvent).where(RunEvent.run_id == run.id) + .order_by(RunEvent.seq))).scalars()) + events = [{"type": r.type, "node_id": r.node_id, "data": r.data} for r in rows] + failed = [s for s in rv.scan(events, {}) if s.kind == "node_failed"] + assert len(failed) == 1, [s.detail for s in failed] + assert failed[0].detail.startswith("节点「复核子流程 › 算一下」执行失败：")