From 5c34ac6bc55cf5dbff3634d5e5537804204c8b8a Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 12:46:35 +0800 Subject: [PATCH] =?UTF-8?q?=E9=A9=B3=E5=9B=9E=E4=B8=8D=E9=87=8D=E8=AF=95?= =?UTF-8?q?=E3=80=81=E4=B8=8D=E8=A2=AB=E3=80=8C=E5=87=BA=E9=94=99=E6=97=B6?= =?UTF-8?q?=E7=BB=A7=E7=BB=AD=E3=80=8D=E5=90=9E=E6=8E=89=EF=BC=9A=E4=BA=BA?= =?UTF-8?q?=E7=9A=84=E5=86=B3=E5=AE=9A=E7=85=A7=E6=80=BB=E6=96=B9=E6=A1=88?= =?UTF-8?q?=20X9=20=E6=94=B6=E6=88=90=E8=A2=AB=E9=A9=B3=E5=9B=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：NodeRejected 是 NodeError 的子类，compiler._execute 的通用 except 照常接住它。节点配了失败重试的，驳回之后 会重试这个节点——再开一次待办、把同一件事再问一遍人；配了 on_error=continue 的，驳回被当成节点失败吞掉，图接着往下 走，最后以 succeeded 结束，通知、定时推送、发布单都看不到这次驳回。这和 X9 定下的「驳回 = cancelled + end_reason= rejected」直接矛盾。「失败重试」「出错时继续」是给故障准备的，驳回不是故障。 做了什么： - _execute 在通用 except 之前单独接住 NodeRejected：不重试，直接跳出；容错分支多一条「不是驳回」才走。节点这一步 照旧记一条 node.failed（实例进度、时间线、存量判断都按它收尾这一步），再原样交给 runner 收成被驳回。 - NodeRejected 的说明改成现在的样子（以前写的是 on_error=continue 照旧认它）。 原有行为变了：配了重试或出错时继续的关卡，驳回后不再重问、不再接着走。发布说明里要写明。 测试：test_run_rejected 加五种关卡 ×（出错时继续、失败重试）的参数化，断言终态是 cancelled + rejected、没有重试日志、 驳回之后没往下走、节点只有一条 node.failed。先跑：出错时继续的五条以 succeeded 结束，失败重试的五条又停回 interrupted （重问了人），10 条失败；改完通过。 --- backend/app/engine/compiler.py | 13 ++++++++++--- backend/app/engine/context.py | 3 ++- backend/tests/test_run_rejected.py | 25 +++++++++++++++++++++++++ 3 files changed, 37 insertions(+), 4 deletions(-) diff --git a/backend/app/engine/compiler.py b/backend/app/engine/compiler.py index c9c115de..d0599d58 100644 --- a/backend/app/engine/compiler.py +++ b/backend/app/engine/compiler.py @@ -10,7 +10,7 @@ from langgraph.graph import END, START, StateGraph from app.core.errors import describe_exception, raw_detail from app.core.events import EventType -from app.engine.context import NodeContext, NodeError, RunContext +from app.engine.context import NodeContext, NodeError, NodeRejected, RunContext from app.engine.expressions import open_template_scope, template_scope from app.engine.nodes import ( challenge, control, dual_check, human, io, knowledge, llm, merge, metrics, multi, report, tools, @@ -144,6 +144,13 @@ def _wrap( raise except asyncio.CancelledError: raise + except NodeRejected as e: + # 人的决定（驳回、超时驳回），不是故障（总方案 X9）：「失败重试」「出错时继续」都是给故障准备的。重试只会把 + # 同一件事再问一遍人；on_error=continue 会把驳回吞掉接着往下走，最后以 succeeded 结束，通知、定时推送、 + # 发布单都看不到这次驳回。所以不重试、不被容错接住；节点这一步照旧记一条 node.failed（时间线、实例进度 + # 按它收尾这一步），交给 runner 收成 cancelled + end_reason=rejected + last_error = e + break except Exception as e: # noqa: BLE001 # NodeError 以前在这里直接 raise，于是节点自己抛的错既不重试、 # 也不发 node.failed、更不被 on_error=continue 接住。而 NodeError @@ -168,8 +175,8 @@ def _wrap( ctx.emit(EventType.NODE_FAILED, error=message, duration_ms=elapsed, node_id=node.id, detail=_detail(last_error)) - # 容错模式：记下错误继续往下走，而不是让整张图挂掉 - if node.config.get("on_error") == "continue": + # 容错模式：记下错误继续往下走，而不是让整张图挂掉。驳回不算错误，不走这里（见上） + if node.config.get("on_error") == "continue" and not isinstance(last_error, NodeRejected): return { "nodes": {node.id: {"error": message, "failed": True}}, "trail": [{"node_id": node.id, "error": message, "ts": time.time()}], diff --git a/backend/app/engine/context.py b/backend/app/engine/context.py index 9e4654a3..7cc915ea 100644 --- a/backend/app/engine/context.py +++ b/backend/app/engine/context.py @@ -119,7 +119,8 @@ class NodeRejected(NodeError): 驳回过的试运行说成「试运行失败了」，记录页还给「继续运行」——接着跑只会读到同一个驳回答复、再停一次。现在 runner 单独接住它，收成 cancelled + end_reason=rejected，run.cancelled 带上在哪一步、谁、结论和备注，消费方都读这一个字段。 - 是 NodeError 的子类：节点失败的那一套（node.failed、on_error=continue）照旧认它；原话（str）照旧写给人看， + 是 NodeError 的子类：节点这一步照旧记 node.failed（时间线、实例进度按它收尾）；但不重试、不被 on_error=continue + 接住（engine/compiler._execute：那两项是给故障准备的，驳回被吞掉运行就以 succeeded 结束了）。原话（str）照旧写给人看， Run.error 和以前一字不差。kind 见 REJECTION_KINDS；decision 是 reject 或 expired_reject（同待办的结论）。""" def __init__(self, node_id: str, message: str, *, kind: str, decision: str = "reject", diff --git a/backend/tests/test_run_rejected.py b/backend/tests/test_run_rejected.py index 690db845..bf00fda6 100644 --- a/backend/tests/test_run_rejected.py +++ b/backend/tests/test_run_rejected.py @@ -212,3 +212,28 @@ async def test_a_run_still_queued_at_restart_gets_its_run_failed_event(): assert row.status == "failed" and row.started_at is None and row.finished_at is None [failed] = await _events(run_id, "run.failed") assert failed.seq == 3 and failed.data == {"error": row.error} + + +@pytest.mark.parametrize("kind", list(GATES)) +@pytest.mark.parametrize("tolerance", [{"on_error": "continue"}, {"retries": 1, "retry_backoff": 0.01}]) +async def test_a_rejection_is_neither_retried_nor_swallowed_by_on_error(kind, tolerance): + """「失败重试」「出错时继续」是给故障准备的。驳回是人的决定：重试只会把同一件事再问一遍人，on_error=continue 会把 + 驳回吞掉、接着往下走、最后以 succeeded 结束——通知、定时推送、发布单都看不到这次驳回（总方案 X9）。""" + gate, said = GATES[kind] + gate = {**gate, "data": {**gate["data"], "config": {**gate["data"]["config"], **tolerance}}} + run = await run_manager.start(graph=chain(START, gate, OUT), input_payload={"q": "x"}) + await _wait(run.id, ("interrupted",)) + await run_manager.wait_idle(run.id) + await run_manager.resume(run.id, {"approved": False, "note": "不对"}, actor="张工") + final = await _wait(run.id, ("cancelled", "failed", "succeeded", "interrupted")) + if final.status == "interrupted": + # 重试会再问一遍人：又停在同一个关卡上 + await run_manager.wait_idle(run.id) + final = await _wait(run.id, ("cancelled", "failed", "succeeded", "interrupted")) + assert (final.status, final.end_reason) == ("cancelled", "rejected"), (final.status, final.error) + assert final.error == said and final.error_node_id == "gate" + events = await _events(run.id) + assert not [e for e in events if e.type == "log" and (e.data or {}).get("code") == "node_retry"], "驳回不该重试" + assert not [e for e in events if e.type == "node.finished" and e.node_id == "out"], "驳回之后不该接着往下走" + # 节点这一步照旧留一条 node.failed（实例进度、时间线按它收尾这一步），只有一条 + assert len([e for e in events if e.type == "node.failed" and e.node_id == "gate"]) == 1