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