mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:21:58 +00:00
复核规则对子工作流里失败的那一步只报一条：子图里那一步、子工作流节点各发一条原话相同的 node.failed，以前合成两条「执行失败」（驳回同理两条「被驳回」），问数据的复核提示里同一件事写两遍。现在保留更具体的那一步，名字一层层接上写清在哪个子工作流里（「复核子流程 › 算一下」），不写内部 id；子工作流节点自己失败、原话不同的照报
This commit is contained in:
ThearchyHelios
@@ -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)
|
||||
|
||||
@@ -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("节点「复核子流程 › 算一下」执行失败：")
|
||||
Reference in New Issue
Block a user