From d57cdcaffcf35f7fd80d959e42cf9f1f503eb745 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 18 Oct 2026 15:56:20 +0800 Subject: [PATCH] =?UTF-8?q?=E9=A9=B3=E5=9B=9E=E5=87=BA=E5=8F=A3=EF=BC=88?= =?UTF-8?q?=E5=90=8E=E7=AB=AF=EF=BC=89=EF=BC=9A=E5=AE=A1=E6=89=B9=E3=80=81?= =?UTF-8?q?=E8=A1=A5=E5=85=85=E8=BE=93=E5=85=A5=E3=80=81=E7=BC=96=E8=BE=91?= =?UTF-8?q?=E7=A8=BF=E4=BB=B6=E3=80=81=E5=8A=A8=E4=BD=9C=E7=A1=AE=E8=AE=A4?= =?UTF-8?q?=E3=80=81=E4=BB=A3=E7=A0=81=E5=92=8C=E5=B7=A5=E5=85=B7=E5=AE=A1?= =?UTF-8?q?=E6=89=B9=E3=80=81=E5=8F=8C=E7=9B=B2=E8=A3=81=E5=86=B3=E6=8E=A5?= =?UTF-8?q?=E4=BA=86=E3=80=8C=E9=A9=B3=E5=9B=9E=E3=80=8D=E5=87=BA=E5=8F=A3?= =?UTF-8?q?=E6=97=B6=E6=B2=BF=E5=AE=83=E5=BE=80=E4=B8=8B=E8=B5=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 以前这六类关卡被驳回一律抛 NodeRejected，运行收成 cancelled + end_reason=rejected，没有「驳回之后走另一条路」 （docs/design/orchestration-round/00-总方案.md §3）。现在画布上多一个可选的「驳回」出口（sourceHandle="rejected"，和 分支、审批同一套连线机制）： - engine/schema：GATE_TYPES（人工、调用工具、沙箱代码、双盲复算）、reject_targets（驳回出口连到哪几步）、approve_mode； 画布校验：别的节点接驳回出口报错（graph.reject_exit_invalid，条件分支的 case 标识可以叫 rejected，不算）；审批同时设了 「驳回即终止运行」和驳回边的给一条提示（human.stop_on_reject_exit） - engine/compiler：接了驳回出口的关卡按出口路由（驳回出口 / 其余是普通出口，连线没写 sourceHandle 的照样是普通出口）； NodeRejected 被接住时这一步正常结束：产出 NodeRejected.exit_payload()（rejected、decision、actor、note、kind、text， 审批另带 approved=false、「结果存为变量」照旧写 false），node.finished 带 outcome=rejected、哪种关卡、结论、谁、备注、 转到哪几步（to），trail 记 outcome=rejected；不重试、不被「出错时继续」接住，照旧 - engine/context：NodeRejected 加 stop（审批明说了驳回即终止的，驳回边照旧走不到）和 exit_payload；动作确认抛在发送之前， 走出口时什么都没发 - engine/runner：成功结束的运行途经驳回出口时，run.finished 带 rejected_via（在哪一步、谁、结论、转到哪），照实说「途经 驳回」；没有途经的终态事件一字不变 - engine/governance：发布检查各级都拦「驳回出口那条路走不到成果节点或结束」（reject_exit.no_end）；退回上游再问一次的放过 - engine/layout：驳回出口排在普通出口下面 - apps/resolve：实例进度写「被驳回，转到「X」」（动作确认写「未确认（被驳回），转到…」，超时驳回写「超时驳回，转到…」）； 实例状态（实例页头、列表、通知同一句）写「「x」被驳回，转到「y」」，走到另一个成果节点的照旧写那个成果节点；退回修改后 又批准了的按批准算；子工作流里的步骤不进实例进度 - 没接出口：行为一字不变（node.failed reason=rejected、运行 cancelled + rejected）；审批没设驳回即终止、没接驳回边的老规矩 照旧。接了驳回边的审批，实例进度那一步的说法从「已驳回」改成「被驳回，转到「已退回」」（test_app_instance_page 跟着改） 另：engine/context 的 RunContext 加 prefix / global_id、事件经 stream_writer 发（子工作流把父运行的写入口交下来）， 这一步本身不改变最外层的事件，下一个提交接上子工作流 测试：test_reject_exit（五类关卡接了出口各走一遍、批准走普通出口、超时驳回走出口、没接一字不变、驳回即终止优先、 重试和出错时继续不管驳回、退回修改再问一次、校验与发布检查、布局、实例进度）；test_process_actions（动作确认走出口 不发送）；test_dual_check_node（裁决走出口） --- backend/app/apps/resolve.py | 69 ++++- backend/app/engine/compiler.py | 63 ++++- backend/app/engine/context.py | 68 ++++- backend/app/engine/governance.py | 36 +++ backend/app/engine/layout.py | 13 +- backend/app/engine/nodes/dual_check.py | 4 +- backend/app/engine/nodes/human.py | 14 +- backend/app/engine/nodes/tools.py | 5 +- backend/app/engine/runner.py | 14 +- backend/app/engine/schema.py | 44 +++ backend/tests/test_app_instance_page.py | 5 +- backend/tests/test_dual_check_node.py | 21 ++ backend/tests/test_process_actions.py | 24 ++ backend/tests/test_reject_exit.py | 340 ++++++++++++++++++++++++ 14 files changed, 684 insertions(+), 36 deletions(-) create mode 100644 backend/tests/test_reject_exit.py diff --git a/backend/app/apps/resolve.py b/backend/app/apps/resolve.py index ae394c3b..ea7abd3c 100644 --- a/backend/app/apps/resolve.py +++ b/backend/app/apps/resolve.py @@ -1503,6 +1503,9 @@ class _Trail: tasks: list[Any] = field(default_factory=list) actions: list[Any] = field(default_factory=list) events: list[tuple[str, str]] = field(default_factory=list) + #: 走了「驳回」出口、而且最后一次执行就是走的驳回出口的关卡：节点 id → 那条 node.finished 的数据（outcome=rejected， + #: 带 to：转到哪几步。engine/compiler._rejected_exit）。退回修改之后又批准了的不算：结局是批准 + rejected_out: dict[str, dict[str, Any]] = field(default_factory=dict) async def _trails(session: AsyncSession, runs: list[Run]) -> dict[str, _Trail]: @@ -1528,9 +1531,46 @@ async def _trails(session: AsyncSession, runs: list[Run]) -> dict[str, _Trail]: .order_by(RunEvent.seq))).all(): if node_id: out[run_id].events.append((etype, node_id)) + await _rejected_exits(session, [r for r in runs if r.status == "succeeded"], out) return out +async def _rejected_exits(session: AsyncSession, runs: list[Run], out: dict[str, _Trail]) -> None: + """成功结束、途经驳回出口的运行（run.finished 带 rejected_via，engine/runner._rejected_via）：每个走过驳回出口的关卡， + 最后一次执行是不是走的驳回出口（退回修改之后又批准的，结局是批准）。只翻途经驳回的运行里那几个关卡的 node.finished。""" + ids = [r.id for r in runs] + if not ids: + return + gates: dict[str, set[str]] = {} + for run_id, data in (await session.execute(select(RunEvent.run_id, RunEvent.data).where( + RunEvent.run_id.in_(ids), RunEvent.type == "run.finished"))).all(): + via = data.get("rejected_via") if isinstance(data, dict) else None + names = {str(v.get("node_id")) for v in via or [] if isinstance(v, dict) and v.get("node_id")} + if names: + gates[run_id] = names + if not gates: + return + wanted = sorted({n for names in gates.values() for n in names}) + for run_id, node_id, data in (await session.execute(select(RunEvent.run_id, RunEvent.node_id, RunEvent.data).where( + RunEvent.run_id.in_(list(gates)), RunEvent.type == "node.finished", RunEvent.node_id.in_(wanted)) + .order_by(RunEvent.seq))).all(): + if node_id not in gates[run_id]: + continue + if isinstance(data, dict) and data.get("outcome") == "rejected": + out[run_id].rejected_out[node_id] = data + else: + out[run_id].rejected_out.pop(node_id, None) + + +def rejected_exit_text(data: dict[str, Any]) -> str: + """走驳回出口的那一步怎么说：「被驳回，转到「通知申请人」」；动作确认写「未确认（被驳回），转到…」，超时驳回写「超时驳回， + 转到…」。转到哪几步取 node.finished 上记的 to（图上的名字）。""" + head = "未确认（被驳回）" if data.get("kind") == "action_confirm" and data.get("decision") == "reject" \ + else "超时驳回" if data.get("decision") == "expired_reject" else "被驳回" + to = [str(t.get("label") or t.get("node_id")) for t in data.get("to") or [] if isinstance(t, dict)] + return f"{head}，转到{'、'.join(f'「{t}」' for t in to)}" if to else head + + def _decision(trail: _Trail, node_id: str) -> tuple[str | None, str | None]: """这个节点最近一条处理完的待办：(种类, 结论 approve / reject / expired_reject / submit)。""" done = [t for t in trail.tasks if t.node_id == node_id and t.status in ("resolving", "resolved")] @@ -1545,15 +1585,18 @@ _REJECTIONS = ("reject", "expired_reject") def rejected_outputs(graph: Any) -> dict[str, set[str]]: - """成果节点 → 从哪几个人工节点的「驳回」出口一路走到它（中间不经过别的人工节点：再遇到人工节点，结局由那一个人 - 的决定说了算）。按图认，不按成果节点的标题猜。""" + """成果节点 → 从哪几个关卡（人工节点，以及接了「驳回」出口的调用工具、沙箱代码、双盲复算，engine/schema.GATE_TYPES） + 的「驳回」出口一路走到它（中间不经过别的人工节点：再遇到人工节点，结局由那一个人的决定说了算）。按图认，不按成果 + 节点的标题猜。""" from app.apps.tree import node_map + from app.engine.schema import GATE_TYPES + gate_types = {str(t) for t in GATE_TYPES} nodes = node_map(graph) edges = [e for e in (graph or {}).get("edges") or [] if isinstance(e, dict)] out: dict[str, set[str]] = {} for gate in (e for e in edges if e.get("sourceHandle") == REJECTED_HANDLE - and (nodes.get(str(e.get("source"))) or {}).get("type") == "human"): + and (nodes.get(str(e.get("source"))) or {}).get("type") in gate_types): seen: set[str] = set() todo = [str(gate.get("target"))] while todo: @@ -1594,7 +1637,8 @@ def instance_state(run: Run, trail: _Trail) -> dict[str, str]: - 走完了：图里有几个成果节点（分支的几种结局，如「已通过」「已退回」）、恰好走到了其中一个的，写走到的那一个；只有 一个成果节点的、几个成果节点并行都走到了的（「周报」「明细」一起出）写「已完成」——那不是哪一种结局。走到的是 否定的那种结局——这次运行里人工节点驳回了、从它的「驳回」出口一路走到的成果节点（rejected_outputs）——code 记 - cancelled（同「未确认（被驳回）」，是人的决定，不画成成功），文字照旧写成果节点的标题； + cancelled（同「未确认（被驳回）」，是人的决定，不画成成功），文字照旧写成果节点的标题；别的走了关卡「驳回」出口、 + 最后一次就是走的驳回出口的（_Trail.rejected_out），写「「x」被驳回，转到「y」」，code 同样记 cancelled； - 停着：有已发出、结果未知的动作，写「等待核实写入结果」；否则按在等的待办说（等待审批、等待确认发送……）； - 被驳回（cancelled + end_reason=rejected，engine/context.NodeRejected）：动作确认被驳回写「「付款」未确认（被驳回）」， 审批设了驳回即终止、补充输入被驳回等写「「x」被驳回」，code 记 cancelled；手动停止的写「已取消」； @@ -1621,8 +1665,13 @@ def instance_state(run: Run, trail: _Trail) -> dict[str, str]: reached = [nid for etype, nid in trail.events if etype == "node.finished" and nid in outputs] if len(outputs) > 1 and len(set(reached)) == 1: gates = rejected_outputs(run.graph).get(reached[0], set()) - negative = any(_decision(trail, gate)[1] in _REJECTIONS for gate in gates) + # 驳回了：按待办的结论认（人工节点），或者这一步最后一次就是走的驳回出口（工具调用审批、代码执行审批没开待办） + negative = any(_decision(trail, gate)[1] in _REJECTIONS or gate in trail.rejected_out for gate in gates) return {"code": "cancelled" if negative else "succeeded", "text": title(reached[0])} + if trail.rejected_out: + # 走了「驳回」出口（总方案 §3）：运行照驳回那条路结束，照实说被驳回、转到了哪一步；是人的决定，不画成成功 + node_id, data = list(trail.rejected_out.items())[-1] + return {"code": "cancelled", "text": f"「{title(node_id)}」{rejected_exit_text(data)}"} return {"code": "succeeded", "text": STATUS_TEXT["succeeded"]} if run.status == "cancelled" and run.end_reason == "rejected": failed = [nid for etype, nid in trail.events if etype == "node.failed"] @@ -1745,8 +1794,11 @@ async def _timeline(session: AsyncSession, ctx: _Page) -> dict[str, Any]: #: 被驳回的那一步：node.failed 上带着 reason=rejected 和结论（engine/compiler，总方案 X9）。工具调用审批、代码执行审批 #: 这类驳回没开待办，以前只能落成「失败」；老运行的 node.failed 没有 reason，照旧按待办认 rejected: dict[str, dict[str, Any]] = {} + #: 走了「驳回」出口的那一步：node.finished 带 outcome=rejected（engine/compiler._rejected_exit），节点照常结束 + exits: dict[str, dict[str, Any]] = {} for etype, node_id, ts, data in rows: - if not node_id: + if not node_id or (node_id not in titles and "/" in node_id): + # 子工作流里的那几步（节点 id 带「子工作流节点 id/」前缀）：实例进度只说这张图上的步骤，子工作流算一步 continue step = steps.setdefault(node_id, {"node": node_id, "title": titles.get(node_id, node_id), "state": "running", "started_at": datetime.fromtimestamp(ts, timezone.utc).isoformat()}) @@ -1756,6 +1808,9 @@ async def _timeline(session: AsyncSession, ctx: _Page) -> dict[str, Any]: rejected[node_id] = data elif etype == "node.started": rejected.pop(node_id, None) + exits.pop(node_id, None) + if etype == "node.finished" and isinstance(data, dict) and data.get("outcome") == "rejected": + exits[node_id] = data for node_id, step in steps.items(): text = None record = next((a for a in reversed(trail.actions) if a.node_id == node_id), None) @@ -1764,6 +1819,8 @@ async def _timeline(session: AsyncSession, ctx: _Page) -> dict[str, Any]: said = rejected.get(node_id) if step["state"] == "failed" else None if record is not None and record.state in ("sent", "unknown"): step["state"] = "unknown" + elif step["state"] == "done" and node_id in exits: + step["state"], text = "rejected", rejected_exit_text(exits[node_id]) elif said is not None: step["state"] = "rejected" text = "未确认（被驳回）" if said.get("kind") == "action_confirm" and said.get("decision") == "reject" \ diff --git a/backend/app/engine/compiler.py b/backend/app/engine/compiler.py index 9d6fbcdc..9e3a1f72 100644 --- a/backend/app/engine/compiler.py +++ b/backend/app/engine/compiler.py @@ -15,7 +15,10 @@ 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, ) -from app.engine.schema import GraphNode, GraphSpec, NodeType, back_edges, innermost_loops +from app.engine.schema import ( + GATE_TYPES, REJECT_HANDLE, GraphNode, GraphSpec, NodeType, approve_mode, back_edges, innermost_loops, + reject_targets, +) from app.engine.state import GraphState from app.process.actions import ActionPending @@ -103,7 +106,7 @@ def _wrap( payload, kind="node_output", run_id=run_ctx.run_id, - node_id=node.id, + node_id=run_ctx.global_id(node.id), meta={"type": str(node.type), "attempt": attempt + 1}, ) except Exception: # noqa: BLE001 - 工件写失败不该毁掉运行本身 @@ -124,7 +127,7 @@ def _wrap( # 这一步新记进证据台账的条目（口径卡的指标集工件等）。事件在封存范围 # 内，证据接口从这里出发找工件，不去读可以事后插行的 artifacts 表。 # 没产出证据的节点不带这个键，事件和以前一字不差 - **({"evidence": updates["evidence"]} if updates.get("evidence") else {}), + **_own_evidence(updates, node.id), ) trail = { "node_id": node.id, @@ -169,6 +172,11 @@ def _wrap( break elapsed = int((time.perf_counter() - started) * 1000) + if isinstance(last_error, NodeRejected) and not last_error.stop \ + and (targets := reject_targets(run_ctx.spec, node)): + # 接了「驳回」出口：驳回不让运行结束，这一步正常结束、沿驳回那条边往下走（总方案 §3）。没接的照旧往下收成 + # node.failed（reason=rejected）+ 运行 cancelled，一字不变 + return await _rejected_exit(node, ctx, last_error, targets, elapsed=elapsed, attempt=attempt + 1) message = _describe(last_error) # node_id 要带上：前端靠它把出错的节点从"转圈"收敛成"失败"。 # error 是给人看的那句，原始异常放 detail，排查的人照样拿得到。 @@ -196,6 +204,48 @@ def _wrap( return _entry +def _own_evidence(updates: dict[str, Any], node_id: str) -> dict[str, Any]: + """node.finished 上带的台账条目：这一步自己新记的。子工作流节点交回父图的条目（节点 id 是「子工作流节点 id/…」， + multi.run_subgraph）不带：子工作流里那几个节点自己的 node.finished 已经带着它们（节点 id 同样带前缀），再记一遍， + 证据接口会把查询认成子工作流节点自己查的。""" + entries = [e for e in updates.get("evidence") or [] + if not (isinstance(e, dict) and str(e.get("node_id") or "").startswith(f"{node_id}/"))] + return {"evidence": entries} if entries else {} + + +async def _rejected_exit(node: GraphNode, ctx: NodeContext, error: NodeRejected, targets: list[str], *, + elapsed: int, attempt: int) -> dict[str, Any]: + """被驳回、接了「驳回」出口：这一步正常结束（node.finished 带 outcome=rejected 和哪种关卡、结论、谁、备注、转到哪几步）， + 产出 NodeRejected.exit_payload()，路由按它的 __decision__ 走驳回那条边。 + + 动作确认被驳回时 NodeRejected 抛在发送之前，走到这里时什么都没发（照旧「未确认（被驳回）」）。trail 上这一步记 + outcome=rejected：运行收尾据此在 run.finished 上照实写「途经驳回」（runner._rejected_via）。审批模式写了「结果存为 + 变量」的照旧写 false（以前接了驳回边的审批就是这么写的）；别的关卡这一步没有产出值，变量不写。""" + from app.core.artifact_store import put_json + + run_ctx = ctx.run + payload = error.exit_payload() + nodes = run_ctx.spec.node_map() + to = [{"node_id": run_ctx.global_id(t), "label": nodes[t].title if t in nodes else t} for t in dict.fromkeys(targets)] + try: + artifact_id: str | None = await put_json(payload, kind="node_output", run_id=run_ctx.run_id, + node_id=run_ctx.global_id(node.id), + meta={"type": str(node.type), "attempt": attempt, "outcome": "rejected"}) + except Exception: # noqa: BLE001 - 工件写失败不该毁掉运行本身 + artifact_id = None + said = {"kind": error.kind, "decision": error.decision, "actor": error.actor, "note": error.note} + ctx.emit(EventType.NODE_FINISHED, duration_ms=elapsed, attempt=attempt, + preview=_preview({"nodes": {node.id: payload}}, node.id), artifact=artifact_id, + outcome="rejected", **said, to=to) + updates: dict[str, Any] = {"nodes": {node.id: payload}, "trail": [{ + "node_id": node.id, "type": str(node.type), "duration_ms": elapsed, "attempt": attempt, "ts": time.time(), + "outcome": "rejected", "label": node.title, **said, "to": to}]} + var_name = str(node.config.get("assign_to") or "").strip() + if var_name and error.kind == "approve": + updates["vars"] = {var_name: False} + return updates + + def _describe(error: BaseException | None) -> str: """错误消息。NodeError 的 message 本来就是写给人看的；别的异常翻成一句原因， 不带类型名和内部符号——界面上出现过「_make_query_tool.._run() got an @@ -294,8 +344,13 @@ def _make_router( 所以真正的判断逻辑放在节点执行阶段，这里只查表。 """ mapping: dict[str, list[str]] = {} + # 审批模式以外的关卡节点（接了驳回出口才走到这里）只有两条路：驳回出口，和其余的（普通出口 out）。正常结束时产出里 + # 没有 __decision__（调用工具、沙箱代码、双盲复算），或者是补充输入、编辑稿件写的 approved，都落到 default + gate = node.type in GATE_TYPES and not approve_mode(spec, node) for edge in spec.outgoing(node.id): key = edge.sourceHandle or "default" + if gate: + key = REJECT_HANDLE if key == REJECT_HANDLE else "default" mapping.setdefault(key, []).append(edge.target) default_targets = mapping.get("default") or [] @@ -371,7 +426,7 @@ def compile_graph(spec: GraphSpec, run_ctx: RunContext) -> StateGraph: handles = {e.sourceHandle for e in outgoing if e.sourceHandle} needs_routing = node.type in ROUTING_TYPES or ( node.type == NodeType.HUMAN and handles & {"approved", "rejected"} - ) + ) or bool(reject_targets(spec, node)) if needs_routing: route, path_map = _make_router(node, spec) diff --git a/backend/app/engine/context.py b/backend/app/engine/context.py index 7cc915ea..26dee96b 100644 --- a/backend/app/engine/context.py +++ b/backend/app/engine/context.py @@ -7,7 +7,7 @@ from typing import Any from langgraph.config import get_stream_writer from app.core.events import EventType -from app.engine.schema import GraphNode, GraphSpec +from app.engine.schema import REJECT_HANDLE, GraphNode, GraphSpec from app.engine.state import GraphState, template_context from app.engine.expressions import render_deep, render_template @@ -34,6 +34,35 @@ class RunContext: agent_limits: dict[str, Any] | None = None extra: dict[str, Any] = field(default_factory=dict) + @property + def prefix(self) -> str: + """子工作流里的节点在这次运行里的 id 前缀（「子工作流节点 id/」，嵌套的一层层接上）；最外层是空串。""" + return str(self.extra.get(PREFIX_KEY) or "") + + def global_id(self, node_id: str) -> str: + """这张图里的节点 id → 这次运行里的 id（事件、工件上记的那个）：最外层原样，子工作流里加前缀。""" + return f"{self.prefix}{node_id}" + + +#: 子工作流里的节点 id 前缀（RunContext.extra 里的键，engine/nodes/multi.run_subgraph 放） +PREFIX_KEY = "node_prefix" +#: 父运行的事件流（RunContext.extra 里的键，同上）。子工作流在父节点里用 ainvoke 跑，LangGraph 不把它的 custom 事件交给 +#: 父图的流（ainvoke 不开 custom 模式，get_stream_writer 拿到的是空的那个）：子工作流里每一步以前在运行的事件里一条都 +#: 没有，时间线看不到、封存也不含它查过的数。所以 run_subgraph 把父图那一刻的写入口交下来，子工作流的节点经它发 +STREAM_KEY = "stream_writer" + + +def stream_writer(run: RunContext) -> Any: + """往这次运行的事件流里写事件的入口：子工作流里是父运行交下来的那个，否则是 LangGraph 这一刻的。不在图执行上下文里 + （单元测试直接调执行器）时为 None。""" + writer = run.extra.get(STREAM_KEY) + if writer is not None: + return writer + try: + return get_stream_writer() + except Exception: # noqa: BLE001 - 不在图执行上下文里 + return None + @dataclass class NodeContext: @@ -51,10 +80,13 @@ class NodeContext: runner 在另一端消费，转成落库的 RunEvent 并广播给前端 —— 画布上的节点高亮、 token 流、工具调用卡片全都来自这里。 + + 子工作流里的节点经父运行交下来的写入口发（stream_writer），节点 id 带上前缀（RunContext.global_id）： + 「子工作流节点 id/节点 id」，嵌套的一层层接上。事件里别的字段（台账条目、口径卡上记的输入）照旧写这张图里的 id， + 读的一方按事件的节点 id 认它在哪一层（api/evidence 的 _scoped）。 """ - try: - writer = get_stream_writer() - except Exception: # noqa: BLE001 - 不在图执行上下文里（比如单元测试） + writer = stream_writer(self.run) + if writer is None: return if str(event_type) == EventType.LLM_END and "unpriced" not in data: from app.providers.catalog import unpriced_call @@ -66,7 +98,7 @@ class NodeContext: { "__agentlab__": True, "type": str(event_type), - "node_id": self.node.id, + "node_id": self.run.global_id(self.node.id), "ts": time.time(), "data": data, } @@ -121,18 +153,36 @@ class NodeRejected(NodeError): 是 NodeError 的子类：节点这一步照旧记 node.failed（时间线、实例进度按它收尾）；但不重试、不被 on_error=continue 接住（engine/compiler._execute：那两项是给故障准备的，驳回被吞掉运行就以 succeeded 结束了）。原话（str）照旧写给人看， - Run.error 和以前一字不差。kind 见 REJECTION_KINDS；decision 是 reject 或 expired_reject（同待办的结论）。""" + Run.error 和以前一字不差。kind 见 REJECTION_KINDS；decision 是 reject 或 expired_reject（同待办的结论）。 + + 节点在画布上接了「驳回」出口（REJECT_HANDLE，engine/schema.reject_targets）时，驳回不让运行结束：compiler 接住它， + 这一步正常结束、产出 exit_payload()、沿驳回那条边往下走（docs/design/orchestration-round/00-总方案.md §3）。stop 为真 + 的除外：审批节点明说了「驳回即终止运行」，那条驳回边照旧走不到（以前就是这样），运行照旧收成被驳回。""" def __init__(self, node_id: str, message: str, *, kind: str, decision: str = "reject", - actor: str | None = None, note: str = "") -> None: + actor: str | None = None, note: str = "", stop: bool = False) -> None: super().__init__(node_id, message) self.kind = kind self.decision = decision self.actor = actor self.note = note + self.stop = stop @classmethod - def of(cls, ctx: Any, message: str, *, kind: str, decision: Any) -> NodeRejected: + def of(cls, ctx: Any, message: str, *, kind: str, decision: Any, stop: bool = False) -> NodeRejected: """按节点上下文和读出来的审批结论（engine/approval.Decision）造一个：谁驳回的取 ctx.actor()。""" return cls(ctx.node.id, message, kind=kind, decision="expired_reject" if decision.expired else "reject", - actor=ctx.actor(), note=decision.note or "") + actor=ctx.actor(), note=decision.note or "", stop=stop) + + def exit_payload(self) -> dict[str, Any]: + """走驳回出口时这一步的产出：{rejected, decision, actor, note, kind, text}，加路由用的 __decision__。 + + text 是原话（「「付款」未确认（被驳回），没有发送」），下游的「通知」「退回修改」写 {{ nodes..text }} 就有一句 + 说得清的话；note 是驳回人写的备注。审批模式另带 approved=false：以前接了驳回边的审批节点产出就是 + {approved: false, note}，下游模板读 approved 的照旧读得到。""" + out: dict[str, Any] = {"rejected": True, "decision": self.decision, "actor": self.actor, "note": self.note, + "kind": self.kind, "text": str(self), "__decision__": REJECT_HANDLE} + if self.kind == "approve": + out["approved"] = False + return out + diff --git a/backend/app/engine/governance.py b/backend/app/engine/governance.py index 4f78c8f2..3f7c0c54 100644 --- a/backend/app/engine/governance.py +++ b/backend/app/engine/governance.py @@ -129,6 +129,7 @@ def lint_for_publish(spec: GraphSpec, *, level: str, checkers: Mapping[str, Any] "（那里会保存取数快照）", code="governed.code_network", node_id=node.id, field="network") _lint_evidence(spec, flag, strict=strict) + _lint_reject_exits(spec, flag) _lint_sql(spec, flag, checkers) _load_extra_lints() for check in EXTRA_LINTS: @@ -158,6 +159,41 @@ def publish_issues(spec: GraphSpec, *, level: str, checkers: Mapping[str, Any] | return [*_not_repeated(validate_graph(spec).issues, gate, spec), *gate] +def _lint_reject_exits(spec: GraphSpec, flag: Callable[..., None]) -> None: + """「驳回」出口那条路要能走到成果节点或者结束（一个不再往下连的节点）（总方案 §3）：走不到的，驳回之后只能一直绕下去， + 运行会在步数上限处中止。各级都报错——这不是受管才有的要求，任何一次驳回都会把运行拖到那里。 + + 按图上的可达性认：从驳回出口连到的节点往下走，碰到成果节点或者没有出边的节点就算走得到。绕回关卡自己（「退回修改」： + 改完再问一次）算走得到：再问一次还能批准、走普通出口。受管级里驳回路径上的出具照常判档、契约照常核对，不另立规矩。""" + from app.engine.schema import reject_targets + + nodes = spec.node_map() + for gate in spec.nodes: + targets = [t for t in reject_targets(spec, gate) if t in nodes] + if not targets or _reaches_an_end(spec, targets): + continue + flag(f"{_who(gate)}的「驳回」出口接到{'、'.join(f'「{nodes[t].title}」' for t in dict.fromkeys(targets))}之后，" + "走不到任何成果节点或结束：驳回之后运行只能一直绕下去，会在步数上限处中止。请让驳回之后的那条路接到成果节点，" + "或者在一个不再往下连的节点结束", + code="reject_exit.no_end", node_id=gate.id, always=True) + + +def _reaches_an_end(spec: GraphSpec, start: list[str]) -> bool: + nodes = spec.node_map() + seen: set[str] = set() + todo = list(start) + while todo: + cur = todo.pop() + if cur in seen or cur not in nodes: + continue + seen.add(cur) + outgoing = [e.target for e in spec.outgoing(cur) if e.target in nodes] + if nodes[cur].type == NodeType.OUTPUT or not outgoing: + return True + todo.extend(outgoing) + return False + + def _lint_sql(spec: GraphSpec, flag: Callable[..., None], checkers: Mapping[str, Any] | None) -> None: """调用工具节点里写死的 SQL 对照数据目录检查（data/sqlcheck.py）。 diff --git a/backend/app/engine/layout.py b/backend/app/engine/layout.py index c079ac99..813e47a0 100644 --- a/backend/app/engine/layout.py +++ b/backend/app/engine/layout.py @@ -25,7 +25,7 @@ from __future__ import annotations from collections import defaultdict, deque from statistics import median -from app.engine.schema import GraphNode, GraphSpec, NodeType, back_edges +from app.engine.schema import REJECT_HANDLE, GraphNode, GraphSpec, NodeType, back_edges, reject_targets #: 画布节点的宽度。前端 NodeCard 的 `width: 238` 是写死的，这里必须跟着它—— #: 排版算出来的走廊是给连线用的，宽度估错了走廊就白留。 @@ -108,11 +108,12 @@ def _ranks( return rank -def _handle_order(node: GraphNode) -> list[str]: +def _handle_order(node: GraphNode, spec: GraphSpec | None = None) -> list[str]: """节点出口在画布上从上到下的顺序。分支的 yes/no 就该是这个顺序。 前端 `sourceHandles()` 决定出口怎么画，这里决定它们怎么排。两份必须一致， - 否则分支标签的上下关系会和连线对不上。 + 否则分支标签的上下关系会和连线对不上。关卡节点接了「驳回」出口的（engine/schema.reject_targets）， + 驳回排在普通出口下面；没接的只有一个出口，排不排都一样。 """ cfg = node.config if node.type == NodeType.BRANCH: @@ -125,7 +126,9 @@ def _handle_order(node: GraphNode) -> list[str]: if node.type == NodeType.LOOP: return ["body", "done"] if node.type == NodeType.HUMAN and cfg.get("mode", "approve") == "approve": - return ["approved", "rejected"] + return ["approved", REJECT_HANDLE] + if spec is not None and reject_targets(spec, node): + return ["out", REJECT_HANDLE] return ["out"] @@ -225,7 +228,7 @@ def _apply_handle_order( outgoing[e.source].append(e) for node in spec.nodes: - handles = _handle_order(node) + handles = _handle_order(node, spec) if len(handles) < 2 or node.id not in nodes: continue outs = outgoing.get(node.id, []) diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index 58428f0f..b334e5cb 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/backend/app/engine/nodes/dual_check.py @@ -1379,8 +1379,8 @@ def _merge(per_input: list[tuple[dict[str, Any], dict[str, Any], dict[str, Any]] def _choices(ctx: NodeContext, response: Any, rows: list[dict[str, Any]]) -> tuple[dict[str, str], str]: """审核人的回复 → ({指标 id: a | b | neither}, 备注)。回复的 value 是 {choices, note}；某个指标没选、选了认不出的 不进 choices：照实记成「没有作出选择」，不替人说「两边都不对」，也不替人认可任何一边（缺口照样记、出具照样降档）。 - 通用的输入卡送来的是一句话，所有指标都按没选记。明确驳回（approved 为否）的按人工驳回让节点失败，和人工审批节点的 - 填写模式一样：这一步没有驳回的出口。""" + 通用的输入卡送来的是一句话，所有指标都按没选记。明确驳回（approved 为否）的抛 NodeRejected，和人工审批节点的 + 填写模式一样：画布上接了「驳回」出口的沿它往下走（engine/compiler），没接的运行收成被驳回。""" decision = read_decision(response) if isinstance(response, dict) and decision.rejected: raise NodeRejected.of(ctx, f"人工驳回：{decision.note or '未说明原因'}", kind="dual_check", decision=decision) diff --git a/backend/app/engine/nodes/human.py b/backend/app/engine/nodes/human.py index 1e20e9b6..b112a39f 100644 --- a/backend/app/engine/nodes/human.py +++ b/backend/app/engine/nodes/human.py @@ -10,6 +10,7 @@ from app.db.base import SessionLocal from app.engine.approval import read_decision from app.engine.context import NodeContext, NodeError, NodeRejected from app.engine.replay import ask, once +from app.engine.schema import reject_targets from app.engine.state import GraphState, merge_usage, message_text, template_context from app.providers import catalog from app.providers.factory import ModelSpec, get_chat_model @@ -51,15 +52,20 @@ async def run_human(state: GraphState, ctx: NodeContext) -> dict[str, Any]: "note": decision.note, "__decision__": "approved" if approved else "rejected", } - if not approved and ctx.cfg("stop_on_reject", False): - raise NodeRejected.of(ctx, f"人工驳回：{result['note'] or '未说明原因'}", kind="approve", decision=decision) + # 驳回即终止：照旧收成被驳回，接没接驳回边都一样（明说了终止，那条边以前就走不到；stop=True，compiler 不接）。 + # 画布上接了「驳回」出口的：同样交给 NodeRejected，由 compiler 接住、这一步正常结束、沿驳回边往下走，产出和别的 + # 关卡同一种形状（engine/schema.reject_targets）。两样都没有的照旧按 __decision__ 路由（没有驳回边就结束在这里） + stop = bool(ctx.cfg("stop_on_reject", False)) + if not approved and (stop or reject_targets(ctx.run.spec, ctx.node)): + raise NodeRejected.of(ctx, f"人工驳回：{result['note'] or '未说明原因'}", kind="approve", decision=decision, + stop=stop) else: # edit / input # 驳回在这两种模式下以前是个摆设：__decision__ 硬编码成 approved， # approved 字段看都不看。界面上按钮在、toast 说"已驳回"，运行却照常 # 往下走——比没有这个按钮更糟。 # - # 这两种模式在图上只有一个 out 出口（没有 rejected 那条边），所以驳回 - # 的正确语义是终止运行，而不是当作通过继续。缺省仍是通过：只传 value + # 这两种模式在图上本来只有一个 out 出口，所以驳回的语义是终止运行，而不是当作通过继续； + # 画布上接了「驳回」出口的，compiler 接住 NodeRejected、沿驳回边往下走。缺省仍是通过：只传 value # 不带 approved 的老调用方行为不变。 # # 只有结构体里明说了"不"才算驳回：这两种模式下裸字符串就是内容本身， diff --git a/backend/app/engine/nodes/tools.py b/backend/app/engine/nodes/tools.py index 129abc99..7c299fcb 100644 --- a/backend/app/engine/nodes/tools.py +++ b/backend/app/engine/nodes/tools.py @@ -302,8 +302,9 @@ async def _run_write(state: GraphState, ctx: NodeContext, name: str, args: dict[ await once(ctx, EventType.HUMAN_RESOLVED, tool=name, approved=decision.approved, note=decision.note, actor=ctx.actor(), payload_hash=confirmed) if not decision.approved: - # 驳回是人的决定，不是故障：运行到这里结束（没有可走的出口），收成 cancelled + end_reason=rejected - # （NodeRejected），说法写成「未确认（被驳回）」，实例页、通知按同一个结论说（apps/resolve.instance_state） + # 驳回是人的决定，不是故障：抛 NodeRejected，说法写成「未确认（被驳回）」，实例页、通知按同一个结论说 + # （apps/resolve.instance_state）。抛在发送之前：画布上接了「驳回」出口的沿它往下走（engine/compiler），什么都 + # 没发；没接的运行到这里结束，收成 cancelled + end_reason=rejected raise NodeRejected.of(ctx, f"「{ctx.node.title}」未确认（被驳回），没有发送" + (f"：{decision.note}" if decision.note else ""), kind="action_confirm", decision=decision) diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index c3acfd8c..27fea92e 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -1039,13 +1039,13 @@ class RunManager: # 切断的半截答案当成完整结论落了库 clipped, truncated = _clip(output) data: dict[str, Any] = {"output": clipped, "usage": usage, - "duration_ms": timing["active_ms"], "timing": timing} + "duration_ms": timing["active_ms"], "timing": timing, **_rejected_via(state)} if truncated: data["output_truncated"] = True event = self._event(run_id, EventType.RUN_FINISHED, data=data) elif terminal is not None: kind, data = terminal - event = self._event(run_id, kind, data={**data, "timing": timing}) + event = self._event(run_id, kind, data={**data, "timing": timing, **_rejected_via(state)}) if run: run.status = status @@ -1346,6 +1346,16 @@ def _trust_event(tool_trust: dict[str, str] | None) -> Any: return "formal" if tool_trust.get(FORMAL_MARK) else snapshot_levels(tool_trust) +def _rejected_via(state: dict[str, Any]) -> dict[str, Any]: + """这次运行途经的驳回出口（engine/compiler._rejected_exit 记在 trail 上的 outcome=rejected）：{rejected_via: [{node_id, + label, kind, decision, actor, note, to}]}，按先后。终态事件带上它，运行照它的路径结束（成功结束也一样）照实说「途经 + 驳回」；没有途经的不带这个键，终态事件和以前一字不差。读的是收尾时的状态：节点出错、被驳回结束的那几种收尾拿不到 + 状态（final_state 为空），不带；子工作流里的驳回出口不在这里（它的 trail 是子工作流自己的状态）。""" + via = [{k: t.get(k) for k in ("node_id", "label", "kind", "decision", "actor", "note", "to")} + for t in state.get("trail") or [] if isinstance(t, dict) and t.get("outcome") == "rejected"] + return {"rejected_via": via} if via else {} + + def rejected_data(exc: NodeRejected, spec: GraphSpec) -> dict[str, Any]: """被驳回的 run.cancelled 带什么：原因（rejected）、在哪一步（节点 id 和图上的名字）、哪种关卡、结论（驳回 / 超时 驳回）、谁、备注。进封存：谁在哪一步驳回、为什么，事后查得到。""" diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py index 8f6fca69..5f37fa06 100644 --- a/backend/app/engine/schema.py +++ b/backend/app/engine/schema.py @@ -144,6 +144,37 @@ class GraphSpec(BaseModel): return [n for n in self.nodes if not self.outgoing(n.id)] +# -------------------------------------------------------------------------- +# 驳回出口（docs/design/orchestration-round/00-总方案.md §3） +# -------------------------------------------------------------------------- + +#: 会被人驳回、能接「驳回」出口的节点：人工节点（审批、补充输入、编辑稿件）、调用工具（工具调用审批、写入前的动作确认）、 +#: 沙箱代码（代码执行审批）、双盲复算（裁决）。和 context.REJECTION_KINDS 的六种关卡一一对得上 +GATE_TYPES = frozenset({NodeType.HUMAN, NodeType.TOOL, NodeType.CODE, NodeType.DUAL_CHECK}) +#: 「驳回」出口的 sourceHandle（同 context.REJECT_HANDLE） +REJECT_HANDLE = "rejected" + + +def approve_mode(spec: GraphSpec, node: GraphNode) -> bool: + """人工节点是不是审批模式（和 NodeContext.cfg 同一个规则：节点上没写取图级 defaults，都没写是审批）。审批模式本来就有 + 「批准」「驳回」两个出口，按 __decision__ 路由；别的关卡只有一个普通出口，接了驳回出口才多一个。""" + if node.type != NodeType.HUMAN: + return False + mode = node.config.get("mode") + return (spec.defaults.get("mode") if mode in (None, "") else mode) in (None, "", "approve") + + +def reject_targets(spec: GraphSpec, node: GraphNode) -> list[str]: + """这个关卡节点「驳回」出口连到的节点（按连线的先后）；不是关卡、没接驳回出口的为空。 + + 接了：驳回（含超时驳回）时这一步正常结束、沿这几条边往下走（engine/compiler）；没接：驳回让运行结束（cancelled + + end_reason=rejected），和以前一字不差。审批模式另有一条老规矩：没设「驳回即终止」时，驳回本来就按 __decision__ 走 + 驳回边（没接就结束在这里、运行成功），那一条照旧。""" + if node.type not in GATE_TYPES: + return [] + return [e.target for e in spec.outgoing(node.id) if e.sourceHandle == REJECT_HANDLE] + + # -------------------------------------------------------------------------- # 校验 # -------------------------------------------------------------------------- @@ -869,6 +900,14 @@ def validate_graph(spec: GraphSpec) -> ValidationResult: result.add(f"连线的起点节点「{edge.source}」不存在", edge_id=edge.id) if edge.target not in nodes: result.add(f"连线的终点节点「{edge.target}」不存在", edge_id=edge.id) + source = nodes.get(edge.source) + if edge.sourceHandle == REJECT_HANDLE and source is not None \ + and source.type not in GATE_TYPES and source.type != NodeType.BRANCH: + # 「驳回」出口只在会被人驳回的节点上有意义：别的节点上这条线不会被路由，每次都会走（条件分支的 case 标识 + # 可以叫 rejected，那是分支自己的出口） + result.add(f"「{source.title}」不会被人驳回，不能接「驳回」出口：只有人工节点（审批、补充输入、编辑稿件）、" + "调用工具（工具调用审批、动作确认）、沙箱代码（代码执行审批）、双盲复算能接。请把这条连线改接到" + "节点的普通出口", node_id=source.id, edge_id=edge.id, code="graph.reject_exit_invalid") if not spec.entry_nodes(): result.add("找不到起始节点：每个节点都有上游连线，工作流中存在环路且没有起点") @@ -911,6 +950,11 @@ def validate_graph(spec: GraphSpec) -> ValidationResult: for code, message, field in human_issues(node): result.add(message, node_id=node.id, field=field, code=code) + if cfg.get("stop_on_reject") and approve_mode(spec, node) and reject_targets(spec, node): + # 驳回即终止优先（以前就是这样）：驳回出口那条线走不到。两样都写着，说明有一样不是作者的本意 + result.add("这个审批设了「驳回即终止运行」，「驳回」出口的连线不会走到：驳回时运行直接结束。" + "要让驳回之后走那条路，请关掉「驳回即终止运行」", level="warning", node_id=node.id, + field="stop_on_reject", code="human.stop_on_reject_exit") elif node.type == NodeType.SUBGRAPH: if not cfg.get("workflow_id"): result.add("「子工作流」节点还没有选择要嵌套的工作流", node_id=node.id, diff --git a/backend/tests/test_app_instance_page.py b/backend/tests/test_app_instance_page.py index ce502afb..ff5c1622 100644 --- a/backend/tests/test_app_instance_page.py +++ b/backend/tests/test_app_instance_page.py @@ -191,11 +191,12 @@ async def test_the_instance_says_which_outcome_it_reached(): await _wait(back, ("succeeded",)) view = await _page(app_id, back, user("qianqi")) # 走了退回分支：运行是「已完成」，流程的结论是「已退回」。这是否定的结局（从「驳回」出口走到的成果节点），状态 - # 标记同「未确认（被驳回）」，不画成成功；时间线上驳回的那一步同样 + # 标记同「未确认（被驳回）」，不画成成功；时间线上驳回的那一步同样，照实写被驳回、转到了哪一步（驳回出口， + # docs/design/orchestration-round/00-总方案.md §3） assert (view["hdr"]["status_code"], view["hdr"]["status_text"]) == ("cancelled", "已退回") assert view["tl"]["status_code"] == "cancelled" review = next(s for s in view["tl"]["steps"] if s["node"] == "review") - assert (review["state"], review["text"]) == ("rejected", "已驳回") + assert (review["state"], review["text"]) == ("rejected", "被驳回，转到「已退回」") # 判定之后没回复的都撤回了：会签人数按判定那一刻的算，不缩成 1 assert view["signs"]["progress"] == {"policy": "all", "voters": 2, "need": 2, "agreed": 0, "rejected": 1} from app.apps.resolve import list_instances diff --git a/backend/tests/test_dual_check_node.py b/backend/tests/test_dual_check_node.py index 3ca9af15..67e08954 100644 --- a/backend/tests/test_dual_check_node.py +++ b/backend/tests/test_dual_check_node.py @@ -534,6 +534,27 @@ async def test_rejecting_the_approval_ends_the_run_as_rejected(monkeypatch, shop assert (ended.data["kind"], ended.data["note"]) == ("dual_check", "口径要重新对") +async def test_rejecting_the_approval_with_a_reject_exit_goes_on_along_it(monkeypatch, shop, client): + """裁决接了「驳回」出口（docs/design/orchestration-round/00-总方案.md §3）：驳回时这一步正常结束、沿驳回边往下走，产出带 + 哪种关卡、结论、备注；运行照它的路径结束。""" + name, _ = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + spec = graph(name, BY_ORDER_DATE) + spec["nodes"].append(node("redo", "output", "重新核对", fields=[{"name": "结论", "value": "{{ nodes.dual.note }}"}])) + spec["edges"].append({"source": "dual", "target": "redo", "sourceHandle": "rejected"}) + run = await start(spec) + approval = await pending(run.id) + res = await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": False, "note": "口径要重新对"}) + assert res.status_code == 200, res.text + done = await wait(run.id, ("succeeded", "failed", "cancelled")) + assert (done.status, done.end_reason) == ("succeeded", None), done.error + assert done.output == {"结论": "口径要重新对"} + evs = await events(run.id) + [finished] = [e for e in evs if e.type == "node.finished" and e.node_id == "dual"] + assert (finished.data["outcome"], finished.data["kind"]) == ("rejected", "dual_check") + assert not [e for e in evs if e.type == "node.finished" and e.node_id == "out"], "普通出口没有走" + + async def test_on_reject_fail_stops_the_run(monkeypatch, shop, client): name, source_id = shop Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) diff --git a/backend/tests/test_process_actions.py b/backend/tests/test_process_actions.py index b0d0e992..c093f98d 100644 --- a/backend/tests/test_process_actions.py +++ b/backend/tests/test_process_actions.py @@ -473,6 +473,30 @@ async def test_a_rejected_confirmation_reads_as_a_decision_not_a_fault(erp, capa assert erp.requests == [] +async def test_a_rejected_confirmation_with_a_reject_exit_sends_nothing_and_goes_on(erp, capabilities): + """动作确认接了「驳回」出口（docs/design/orchestration-round/00-总方案.md §3）：驳回时照旧不发送写入、记「未确认（被驳回）」， + 这一步正常结束、沿驳回边往下走（这里是通知申请人），运行照它的路径结束；实例页写「未确认（被驳回），转到「通知申请人」」。""" + capabilities(await _endpoint(test=True), lookup=False) + graph = pay_graph() + graph["nodes"].append(node("tell", "output", fields=[{"name": "结论", "value": "{{ nodes.pay.text }}"}])) + graph["nodes"][-1]["data"]["label"] = "通知申请人" + graph["edges"].append({"source": "pay", "target": "tell", "sourceHandle": "rejected"}) + run_id = await _formal(graph) + task = await _task(run_id, "action_confirm") + await tasks.respond(task.id, ANN, decision="reject", note="金额不对") + final = await _wait(run_id, ("succeeded", "failed", "cancelled")) + assert (final.status, final.end_reason) == ("succeeded", None), final.error + assert final.output == {"结论": "「pay」未确认（被驳回），没有发送：金额不对"} + assert erp.requests == [], "驳回之后走出口也不发送" + assert await _records(run_id) == [] + [done] = [e for e in await _events(run_id, "node.finished") if e.node_id == "pay"] + assert (done.data["outcome"], done.data["kind"], done.data["decision"]) == ("rejected", "action_confirm", "reject") + assert not [e for e in await _events(run_id, "tool.start") if e.node_id == "pay"] + view = await _instance_view(run_id) + step = next(s for s in view["tl"]["steps"] if s["node"] == "pay") + assert (step["state"], step["text"]) == ("rejected", "未确认（被驳回），转到「通知申请人」") + + async def test_unknown_then_reconciled_reads_as_waiting_then_a_footnote(erp, capabilities): capabilities(await _endpoint(test=True), lookup=True) erp.mode = "timeout_after_write" diff --git a/backend/tests/test_reject_exit.py b/backend/tests/test_reject_exit.py new file mode 100644 index 00000000..c6330d4f --- /dev/null +++ b/backend/tests/test_reject_exit.py @@ -0,0 +1,340 @@ +"""驳回出口（docs/design/orchestration-round/00-总方案.md §3）。 + +审批、补充输入、编辑稿件、工具调用审批、代码执行审批（动作确认、双盲裁决在 test_process_actions.py、test_dual_check_node.py +里，要假的外部系统、口径卡）在画布上多一个可选的「驳回」出口（sourceHandle="rejected"）： +- 接了：驳回（含超时驳回）时这一步正常结束、产出 {rejected, decision, actor, note, kind, text}，运行沿驳回那条边往下走； + node.finished 带 outcome=rejected（哪种关卡、结论、谁、备注、转到哪几步），运行照它的路径结束，终态事件照实带 + rejected_via；批准了照旧走普通出口； +- 没接：行为一字不变（cancelled + end_reason=rejected，test_run_rejected.py 照过）；审批没设驳回即终止的老规矩照旧； +- 「失败重试」「出错时继续」照旧不管驳回； +- 发布检查：驳回出口那条路走不到出口或结束（只能绕回去）的拦下； +- 实例进度写「被驳回，转到 X」。 +""" +from __future__ import annotations + +import asyncio +from datetime import timedelta + +import pytest +from sqlalchemy import select + +from app.db.base import SessionLocal, utcnow +from app.db.models import Run, RunEvent, Task +from app.engine.runner import run_manager + + +@pytest.fixture(autouse=True) +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}} + + +START = node("start", "input", fields=[{"name": "q"}]) +DONE = node("done", "output", "办结", fields=[{"name": "结果", "value": "通过了"}]) +BACK = node("back", "output", "退回申请人", fields=[{"name": "结果", "value": "退回：{{ nodes.gate.note }}"}, + {"name": "谁", "value": "{{ nodes.gate.actor }}"}]) + +#: (关卡, 驳回之后的原话) +GATES = { + "tool": (node("gate", "tool", label="算一下", tool="calculator", args={"expression": "1+1"}, approval="always"), + "工具 calculator 的调用已被驳回：不对"), + "code": (node("gate", "code", label="跑代码", code="print(1)", language="python", approval="always"), + "代码执行已被驳回：不对"), + "approve": (node("gate", "human", label="把关", mode="approve", title="可以吗"), "人工驳回：不对"), + "edit": (node("gate", "human", label="改稿", mode="edit", title="改一下", draft="初稿"), "人工驳回：不对"), + "input": (node("gate", "human", label="补充", mode="input", title="补一下"), "人工驳回：不对"), +} + + +def gated(gate: dict, *, exit_to: dict | None = BACK, normal_handle: str | None = None) -> dict: + """入口 → 关卡 → 办结；关卡的「驳回」出口接到 exit_to。审批模式的普通出口是 approved，别的是 out。""" + normal = normal_handle or ("approved" if gate["data"]["config"].get("mode") == "approve" else "out") + nodes = [START, gate, DONE] + ([exit_to] if exit_to else []) + edges = [{"source": "start", "target": "gate"}, {"source": "gate", "target": "done", "sourceHandle": normal}] + if exit_to: + edges.append({"source": "gate", "target": exit_to["id"], "sourceHandle": "rejected"}) + return {"nodes": nodes, "edges": edges} + + +async def _wait(run_id: str, statuses: tuple[str, ...], timeout: float = 20.0) -> Run: + deadline = asyncio.get_event_loop().time() + timeout + while asyncio.get_event_loop().time() < deadline: + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row is not None and row.status in statuses and not run_manager.is_active(run_id): + return row + await asyncio.sleep(0.05) + raise AssertionError(f"运行 {run_id} 没有进入 {statuses}") + + +async def _events(run_id: str, kind: str | None = None) -> list[RunEvent]: + async with SessionLocal() as session: + query = select(RunEvent).where(RunEvent.run_id == run_id).order_by(RunEvent.seq) + if kind: + query = query.where(RunEvent.type == kind) + return list((await session.execute(query)).scalars()) + + +async def _answered(graph: dict, reply: dict) -> Run: + run = await run_manager.start(graph=graph, input_payload={"q": "x"}) + await _wait(run.id, ("interrupted",)) + await run_manager.wait_idle(run.id) + await run_manager.resume(run.id, reply, actor="张工") + return await _wait(run.id, ("cancelled", "failed", "succeeded")) + + +@pytest.mark.parametrize("kind", list(GATES)) +async def test_a_rejection_takes_the_reject_exit_and_the_run_finishes_along_it(kind): + gate, said = GATES[kind] + final = await _answered(gated(gate), {"approved": False, "note": "不对"}) + # 接了驳回出口：运行没有被驳回结束，照驳回那条路走完 + assert (final.status, final.end_reason, final.error) == ("succeeded", None, None) + assert final.output == {"结果": "退回：不对", "谁": "张工"} + finished = {e.node_id: e for e in await _events(final.id, "node.finished")} + assert "back" in finished and "done" not in finished + assert not [e for e in await _events(final.id, "node.failed") if e.node_id == "gate"], "被驳回走出口不是失败" + data = finished["gate"].data + assert {k: data.get(k) for k in ("outcome", "kind", "decision", "actor", "note")} == { + "outcome": "rejected", "kind": kind, "decision": "reject", "actor": "张工", "note": "不对"} + assert data["to"] == [{"node_id": "back", "label": "退回申请人"}] + assert data["preview"]["__decision__"] == "rejected" and data["preview"]["rejected"] is True + # 这一步的产出：哪种关卡、结论、谁、备注、原话 + async with SessionLocal() as session: + row = await session.get(Run, final.id) + # 运行照路径结束，终态事件照实说途经驳回 + [ended] = await _events(final.id, "run.finished") + assert ended.data["rejected_via"] == [{"node_id": "gate", "label": gate["data"]["label"], "kind": kind, + "decision": "reject", "actor": "张工", "note": "不对", + "to": [{"node_id": "back", "label": "退回申请人"}]}] + assert row.manifest_seq == ended.seq + from app.core.artifact_store import load + + payload = load(data["artifact"]) + assert {k: payload.get(k) for k in ("rejected", "decision", "actor", "note", "kind", "text")} == { + "rejected": True, "decision": "reject", "actor": "张工", "note": "不对", "kind": kind, "text": said} + + +@pytest.mark.parametrize("kind", list(GATES)) +async def test_approving_takes_the_normal_exit(kind): + gate, _ = GATES[kind] + final = await _answered(gated(gate), {"approved": True, "value": "补好了"}) + assert final.status == "succeeded", final.error + assert final.output == {"结果": "通过了"} + finished = {e.node_id: e for e in await _events(final.id, "node.finished")} + assert "done" in finished and "back" not in finished + assert "outcome" not in finished["gate"].data + [ended] = await _events(final.id, "run.finished") + assert "rejected_via" not in ended.data, "没有途经驳回的运行，终态事件和以前一字不差" + + +async def test_a_plain_out_handle_without_a_name_is_the_normal_exit_too(): + """普通出口的连线没写 sourceHandle（老图、助手建的图）：照样是普通出口。""" + gate, _ = GATES["input"] + graph = gated(gate) + graph["edges"][1].pop("sourceHandle") + final = await _answered(graph, {"approved": True, "value": "补好了"}) + assert final.status == "succeeded" and final.output == {"结果": "通过了"} + + +async def test_without_the_exit_nothing_changes(): + """没接驳回出口：照旧被驳回结束（cancelled + end_reason=rejected、node.failed reason=rejected）。""" + gate, said = GATES["edit"] + final = await _answered(gated(gate, exit_to=None), {"approved": False, "note": "不对"}) + assert (final.status, final.end_reason, final.error) == ("cancelled", "rejected", said) + [failed] = [e for e in await _events(final.id, "node.failed") if e.node_id == "gate"] + assert failed.data["reason"] == "rejected" + [cancelled] = await _events(final.id, "run.cancelled") + assert "rejected_via" not in cancelled.data + + +async def test_an_approval_without_stop_on_reject_keeps_its_old_routing(): + """审批没设驳回即终止、也没接驳回边：照旧按 __decision__ 路由，驳回就结束在这里、运行成功，产出和以前一样。""" + gate = node("gate", "human", label="把关", mode="approve", title="可以吗", assign_to="ok") + final = await _answered(gated(gate, exit_to=None), {"approved": False, "note": "不对"}) + assert final.status == "succeeded" and final.output in ({}, None) + finished = {e.node_id: e for e in await _events(final.id, "node.finished")} + assert finished["gate"].data["preview"] == {"approved": False, "note": "不对", "__decision__": "rejected"} + assert "outcome" not in finished["gate"].data and "done" not in finished + + +async def test_stop_on_reject_still_stops_even_with_a_reject_edge(): + """审批明说了「驳回即终止运行」：照旧收成被驳回，驳回边走不到（以前就是这样），画布校验另给一条提示。""" + from app.engine.schema import validate_graph + + gate = node("gate", "human", label="把关", mode="approve", title="可以吗", stop_on_reject=True) + final = await _answered(gated(gate), {"approved": False, "note": "不对"}) + assert (final.status, final.end_reason, final.error) == ("cancelled", "rejected", "人工驳回：不对") + assert not [e for e in await _events(final.id, "node.finished") if e.node_id in ("gate", "back")] + [hint] = [i for i in validate_graph(_spec(gated(gate))).issues if i.code == "human.stop_on_reject_exit"] + assert hint.level == "warning" and hint.node_id == "gate" and hint.field == "stop_on_reject" + + +async def test_an_approval_with_a_reject_edge_keeps_approved_false_and_the_variable(): + """接了驳回边的审批（不管设没设驳回即终止）：产出照旧带 approved=false，「结果存为变量」照旧写 false。""" + gate = node("gate", "human", label="把关", mode="approve", title="可以吗", assign_to="ok") + back = node("back", "output", "退回申请人", fields=[{"name": "批了", "value": "{{ vars.ok }}"}, + {"name": "approved", "value": "{{ nodes.gate.approved }}"}]) + final = await _answered(gated(gate, exit_to=back), {"approved": False, "note": "不对"}) + assert final.status == "succeeded" and final.output == {"批了": "False", "approved": "False"} + [gate_done] = [e for e in await _events(final.id, "node.finished") if e.node_id == "gate"] + assert gate_done.data["outcome"] == "rejected" and gate_done.data["kind"] == "approve" + + +async def test_a_timeout_rejection_takes_the_exit_too(): + from app.process import tasks + + gate = node("gate", "human", label="把关", mode="approve", title="可以吗", due="PT30M", + on_timeout={"action": "reject"}) + run = await run_manager.start(graph=gated(gate), input_payload={"q": "x"}) + await _wait(run.id, ("interrupted",)) + await run_manager.wait_idle(run.id) + ticked = await tasks.tick(utcnow() + timedelta(hours=1)) + async with SessionLocal() as session: + mine = (await session.execute(select(Task.id).where(Task.run_id == run.id))).scalar_one() + assert mine in ticked + final = await _wait(run.id, ("cancelled", "failed", "succeeded")) + assert final.status == "succeeded", final.error + [gate_done] = [e for e in await _events(final.id, "node.finished") if e.node_id == "gate"] + assert (gate_done.data["outcome"], gate_done.data["decision"], gate_done.data["actor"]) == ( + "rejected", "expired_reject", None) + assert final.output["结果"].startswith("退回") + + +async def test_retries_and_continue_on_error_still_leave_rejections_alone(): + """「失败重试」「出错时继续」照旧不管驳回：接了出口的照出口走，没接的照旧被驳回结束（不重试、不吞掉）。""" + gate, _ = GATES["tool"] + patient = {**gate, "data": {**gate["data"], "config": {**gate["data"]["config"], "retries": 2, + "on_error": "continue"}}} + final = await _answered(gated(patient, exit_to=None), {"approved": False, "note": "不对"}) + assert (final.status, final.end_reason) == ("cancelled", "rejected") + assert not [e for e in await _events(final.id, "log") if (e.data or {}).get("code") == "node_retry"] + final = await _answered(gated(patient), {"approved": False, "note": "不对"}) + assert final.status == "succeeded" and final.output["结果"] == "退回：不对" + assert len(await _events(final.id, "human.requested")) == 1 + + +async def test_the_reject_path_can_send_the_work_back_and_ask_again(): + """「退回修改」：驳回出口接回上游，改完再问一次。第二次批准了走普通出口；时间线上同一个关卡两次执行，第一次被驳回。""" + draft = node("draft", "transform", "起草", mode="template", template="稿子") + gate = node("gate", "human", label="把关", mode="approve", title="可以吗") + graph = {"nodes": [START, draft, gate, DONE], + "edges": [{"source": "start", "target": "draft"}, {"source": "draft", "target": "gate"}, + {"source": "gate", "target": "done", "sourceHandle": "approved"}, + {"source": "gate", "target": "draft", "sourceHandle": "rejected"}]} + run = await run_manager.start(graph=graph, 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="张工") + await _wait(run.id, ("interrupted",)) + await run_manager.wait_idle(run.id) + await run_manager.resume(run.id, {"approved": True}, actor="张工") + final = await _wait(run.id, ("cancelled", "failed", "succeeded")) + assert final.status == "succeeded" and final.output == {"结果": "通过了"} + outcomes = [e.data.get("outcome") for e in await _events(run.id, "node.finished") if e.node_id == "gate"] + assert outcomes == ["rejected", None] + [ended] = await _events(run.id, "run.finished") + assert [v["to"] for v in ended.data["rejected_via"]] == [[{"node_id": "draft", "label": "起草"}]] + + +# -------------------------------------------------------------------------- +# 画布校验、发布检查 +# -------------------------------------------------------------------------- + + +def _spec(graph: dict): + from app.engine.schema import GraphSpec + + return GraphSpec.model_validate(graph) + + +def test_a_reject_exit_on_a_node_that_cannot_be_rejected_is_an_error(): + from app.engine.schema import validate_graph + + calc = node("gate", "transform", "整理", mode="template", template="x") + graph = gated(calc) + issues = [i for i in validate_graph(_spec(graph)).issues if i.code == "graph.reject_exit_invalid"] + assert [(i.level, i.node_id) for i in issues] == [("error", "gate")] + assert "驳回" in issues[0].message + assert not [i for i in validate_graph(_spec(gated(GATES["tool"][0]))).issues + if i.code == "graph.reject_exit_invalid"] + + +@pytest.mark.parametrize("level", ["published", "governed"]) +def test_publishing_refuses_a_reject_path_that_never_reaches_an_end(level): + from app.engine.governance import lint_for_publish + + # 驳回之后绕回关卡自己的上游，那里只有一条路又回到关卡：除了再驳回一次，走不到任何出口 + loop_a = node("a", "transform", "甲", mode="template", template="x") + loop_b = node("b", "transform", "乙", mode="template", template="y") + gate = node("gate", "human", label="把关", mode="approve", title="可以吗") + graph = {"nodes": [START, gate, DONE, loop_a, loop_b], + "edges": [{"source": "start", "target": "gate"}, + {"source": "gate", "target": "done", "sourceHandle": "approved"}, + {"source": "gate", "target": "a", "sourceHandle": "rejected"}, + {"source": "a", "target": "b"}, {"source": "b", "target": "a"}]} + issues = [i for i in lint_for_publish(_spec(graph), level=level).issues if i.code == "reject_exit.no_end"] + assert [(i.level, i.node_id) for i in issues] == [("error", "gate")] + assert "「把关」" in issues[0].message and "「甲」" in issues[0].message + # 走得到出口、走得到结束（没有下游的节点）、退回上游再问一次（重新问还能批准）的都放过 + for exit_to, extra_edges in ((BACK, []), (node("note", "transform", "记一笔", mode="template", template="z"), []),): + ok = gated(gate, exit_to=exit_to) + ok["edges"] += extra_edges + assert not [i for i in lint_for_publish(_spec(ok), level=level).issues if i.code == "reject_exit.no_end"] + draft = node("draft", "transform", "起草", mode="template", template="稿子") + again = {"nodes": [START, draft, gate, DONE], + "edges": [{"source": "start", "target": "draft"}, {"source": "draft", "target": "gate"}, + {"source": "gate", "target": "done", "sourceHandle": "approved"}, + {"source": "gate", "target": "draft", "sourceHandle": "rejected"}]} + assert not [i for i in lint_for_publish(_spec(again), level=level).issues if i.code == "reject_exit.no_end"] + + +def test_layout_orders_the_reject_exit_below_the_normal_one(): + from app.engine.layout import _handle_order + + spec = _spec(gated(GATES["tool"][0])) + assert _handle_order(spec.node_map()["gate"], spec) == ["out", "rejected"] + approve = _spec(gated(GATES["approve"][0])) + assert _handle_order(approve.node_map()["gate"], approve) == ["approved", "rejected"] + plain = _spec(gated(GATES["tool"][0], exit_to=None)) + assert _handle_order(plain.node_map()["gate"], plain) == ["out"] + + +# -------------------------------------------------------------------------- +# 应用：实例进度、实例状态 +# -------------------------------------------------------------------------- + + +async def test_the_instance_says_rejected_and_where_it_went(): + from types import SimpleNamespace + + from app.apps.resolve import _timeline, _trails, instance_state + + gate, _ = GATES["tool"] + + async def said(final: Run) -> tuple[dict, dict]: + async with SessionLocal() as session: + run = await session.get(Run, final.id) + steps = {s["node"]: s for s in (await _timeline(session, SimpleNamespace(instance=run)))["steps"]} + return steps, instance_state(run, (await _trails(session, [run]))[run.id]) + + # 驳回之后去通知一声就结束（只有一个成果节点）：实例写谁被驳回、转到了哪一步，是人的决定，不画成成功 + notify = node("back", "transform", "通知申请人", mode="template", template="退回了") + steps, state = await said(await _answered(gated(gate, exit_to=notify), {"approved": False, "note": "不对"})) + assert (steps["gate"]["state"], steps["gate"]["text"]) == ("rejected", "被驳回，转到「通知申请人」") + assert state == {"code": "cancelled", "text": "「算一下」被驳回，转到「通知申请人」"} + # 驳回出口走到另一个成果节点（流程的另一种结局）：照旧写走到的那个成果节点，记成人的决定 + steps, state = await said(await _answered(gated(gate), {"approved": False, "note": "不对"})) + assert (steps["gate"]["state"], steps["gate"]["text"]) == ("rejected", "被驳回，转到「退回申请人」") + assert state == {"code": "cancelled", "text": "退回申请人"} + # 批准了的照旧是办完 + ok = await _answered(gated(gate), {"approved": True}) + async with SessionLocal() as session: + run = await session.get(Run, ok.id) + steps = {s["node"]: s for s in (await _timeline(session, SimpleNamespace(instance=run)))["steps"]} + state = instance_state(run, (await _trails(session, [run]))[run.id]) + assert steps["gate"]["state"] == "done" and state == {"code": "succeeded", "text": "办结"}