From 845809486a1d29cd6754b2f5981acfcf67283c28 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 10:14:27 +0800 Subject: [PATCH] =?UTF-8?q?=E9=A9=B3=E5=9B=9E=E4=B8=8D=E5=86=8D=E8=AE=B0?= =?UTF-8?q?=E6=88=90=E5=A4=B1=E8=B4=A5=EF=BC=88=E5=90=8E=E7=AB=AF=EF=BC=89?= =?UTF-8?q?=EF=BC=9A=E4=BA=BA=E7=9A=84=E9=A9=B3=E5=9B=9E=E6=94=B6=E6=88=90?= =?UTF-8?q?=20cancelled=20+=20end=5Freason=3Drejected=EF=BC=8C=E6=B6=88?= =?UTF-8?q?=E8=B4=B9=E6=96=B9=E9=83=BD=E8=AF=BB=E8=BF=99=E4=B8=80=E4=B8=AA?= =?UTF-8?q?=E5=AD=97=E6=AE=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：动作确认、工具 / 代码审批、审批驳回即终止、填写 / 编辑被驳回、双盲裁决的驳回以前抛普通 NodeError，运行记成 failed。 通知推「实例失败」，定时推「运行失败」，发布单把驳回过的试运行说成「试运行失败了」并挡住发布，记录页还给「继续运行」 （接着跑只会读到同一个驳回答复、再停一次）。人的决定被说成故障，每个消费方各写一份映射，已经漏了四处（05-已知遗留 (a)、总方案 X9）。 做了什么： - engine/context 新增 NodeRejected(NodeError)，带 node_id、kind、decision（reject / expired_reject）、actor、note；六处驳回改抛它， 原话一字不变。待办超时驳回的答复带 expired，Decision.expired 据此记成 expired_reject。 - runner._drive 在通用 except 之前单独接住：status=cancelled、end_reason=rejected、error 写原话、error_node 写那一步， run.cancelled 带 reason/node_id/label/kind/decision/actor/note；关审批、回收沙箱、关流、封存走取消的那一套。手动停止 （执行中取消、放弃断点上的运行）写 end_reason=stopped，run.cancelled 带 reason=stopped。 - 继续运行：被驳回的 409 code=run_rejected「被驳回的运行不能接着跑：改好内容后重新发起」；存量 failed+驳回（失败那一步上 最近一条处理完的待办是驳回，和实例页同一份判断 resolve.legacy_rejected）也拒。RunOut 带 end_reason。 - apps/resolve.instance_state：cancelled+rejected 和存量 failed+驳回说同一句（「「付款」未确认（被驳回）」「「x」被驳回」）。 - notify_run：状态那一句取自 instance_state，被驳回的走「实例结束」、ok=False，走到「已退回」成果的不再写「已完成」； 手动停止照旧不推。顺带修 04-对外集成与触达 B1：发起人是认证身份（run.origin.actor_verified）时 users 传 [started_by]， 个人渠道订阅了实例完成 / 失败终于收得到；自报署名不算。 - 定时 webhook：cancelled+rejected 按「运行结束」推「未出具（被驳回）」、写驳回的那一步，不写备注和原话、不画成功色； 手动停止仍不推。 - 发布单：试运行被驳回不算 plan.trial_failed，记 warning plan.trial_rejected「最近一次试运行在「x」被驳回、没有走完」（带白话）。 - (h) 重启时排队中的运行改成 failed 时补发 run.failed（同一句原因），时间线里有了失败原因这一行；不写结束时间、不封存，同以前。 - (f) tasks.tick / actions.tick 交回这一拍处理过的 id 列表（调度循环只记条数），测试改成只断言自己造的待办；先加了预置 一条无关到期待办的 fixture，旧的按条数断言当场失败（3 == 1），改完通过。 已封存的老运行不改写。测试：新增 test_run_rejected（五种关卡参数化、超时驳回、继续运行拒绝含存量、手动停止、补发 run.failed、 instance_state 新老形状同一句）、test_notify_run（驳回走实例结束、手动停止不推、已退回不写已完成、个人渠道 B1 两条）； webhook、发布单、动作确认、双盲、审批的老断言跟上。去掉 runner 里接 NodeRejected 的分支回跑，新测试 8 条失败。 --- backend/app/api/runs.py | 11 +- backend/app/apps/resolve.py | 48 ++++-- backend/app/blueprints/publish_plan.py | 20 +++ backend/app/engine/approval.py | 4 + backend/app/engine/context.py | 29 ++++ backend/app/engine/nodes/dual_check.py | 4 +- backend/app/engine/nodes/human.py | 6 +- backend/app/engine/nodes/tools.py | 19 ++- backend/app/engine/runner.py | 71 ++++++-- backend/app/notify/service.py | 26 ++- backend/app/plain.py | 3 + backend/app/process/actions.py | 9 +- backend/app/process/tasks.py | 14 +- backend/app/scheduler/loop.py | 4 +- backend/app/scheduler/webhook.py | 41 ++++- backend/tests/test_app_instance_page.py | 2 +- backend/tests/test_approval.py | 13 +- backend/tests/test_blueprint_trial.py | 2 +- backend/tests/test_dual_check_node.py | 10 +- backend/tests/test_notify_run.py | 199 ++++++++++++++++++++++ backend/tests/test_process_actions.py | 20 ++- backend/tests/test_process_tasks.py | 37 ++-- backend/tests/test_publish_plan.py | 24 +++ backend/tests/test_records.py | 2 +- backend/tests/test_run_rejected.py | 214 ++++++++++++++++++++++++ backend/tests/test_scheduler_webhook.py | 24 ++- 26 files changed, 768 insertions(+), 88 deletions(-) create mode 100644 backend/tests/test_notify_run.py create mode 100644 backend/tests/test_run_rejected.py diff --git a/backend/app/api/runs.py b/backend/app/api/runs.py index 3a35364e..2832e340 100644 --- a/backend/app/api/runs.py +++ b/backend/app/api/runs.py @@ -56,6 +56,9 @@ class RunOut(BaseModel): input: dict[str, Any] output: dict[str, Any] error: str | None + #: 结束在 cancelled 的原因（docs/design/next-round/00-总方案.md X9）：rejected 人工驳回、stopped 手动停止。 + #: 其余终态、加这一列之前的运行为空。界面、通知、定时推送都读这一个字段，不再解析报错原文 + end_reason: str | None = None usage: dict[str, Any] run_class: str | None = "exploratory" version: int | None = None @@ -289,11 +292,13 @@ async def cancel_run(run_id: str, x_actor: str | None = Header(default=None)) -> run = await session.get(Run, run_id) if run is None: raise HTTPException(404, "运行记录不存在") - raise HTTPException(409, _cannot_cancel(run.status)) + raise HTTPException(409, _cannot_cancel(run.status, run.end_reason)) return {"ok": True, "status": outcome} -def _cannot_cancel(status: str) -> str: +def _cannot_cancel(status: str, end_reason: str | None = None) -> str: + if status == "cancelled" and end_reason == "rejected": + return "这次运行已被驳回结束，没有要停止的。" return { "succeeded": "这次运行已完成，没有要停止的。", "failed": "这次运行已失败结束，没有要停止的；如需继续，请点「继续运行」。", @@ -396,7 +401,7 @@ async def continue_run( except KeyError as e: raise _missing(e) from e except ContinueRefused as e: - # 正式运行带图（formal_graph_locked）、写入结果未知还没核实（action_unresolved）：带机器码 + # 正式运行带图（formal_graph_locked）、写入结果未知还没核实（action_unresolved）、被驳回了的（run_rejected）：带机器码 raise CodedHTTPException(409, str(e), e.code) from e except ValueError as e: raise HTTPException(409, str(e)) from e diff --git a/backend/app/apps/resolve.py b/backend/app/apps/resolve.py index 97564ea6..0a23a758 100644 --- a/backend/app/apps/resolve.py +++ b/backend/app/apps/resolve.py @@ -51,6 +51,8 @@ NEEDS_UPGRADE = "app_needs_upgrade" STATUS_TEXT = {"queued": "排队中", "running": "运行中", "interrupted": "等待处理", "succeeded": "已完成", "failed": "失败", "cancelled": "已取消"} +#: 被驳回、又说不出是哪一步时（没有节点可指）的说法 +REJECTED_TEXT = "被驳回" TIER_TEXT = {"formal": "完整出具", "degraded": "降档出具", "withheld": "不予出具"} NO_TIER = "未声明出具契约" #: 蓝图试运行（docs/design/normal-user-apps.md N4）：印章的来源、说明，趋势只有一个点时的说明，预览里入口按钮为什么点不了 @@ -1277,8 +1279,9 @@ async def _trails(session: AsyncSession, runs: list[Run]) -> dict[str, _Trail]: for row in (await session.execute(select(ActionRecord).where(ActionRecord.run_id.in_(ids)) .order_by(ActionRecord.created_at, ActionRecord.attempt))).scalars(): out[row.run_id].actions.append(row) - # 节点事件只有结束了的运行要看（走到了哪个成果节点、停在哪个节点上） - wanted = [r.id for r in runs if r.status in ("succeeded", "failed")] + # 节点事件只有结束了的运行要看（走到了哪个成果节点、停在哪个节点上）。取消的只有被驳回的要看（停在哪一步） + wanted = [r.id for r in runs if r.status in ("succeeded", "failed") + or (r.status == "cancelled" and r.end_reason == "rejected")] if wanted: for run_id, etype, node_id in (await session.execute( select(RunEvent.run_id, RunEvent.type, RunEvent.node_id).where( @@ -1325,6 +1328,26 @@ def rejected_outputs(graph: Any) -> dict[str, set[str]]: return out +def _rejection_text(trail: _Trail, node_id: str, title: str) -> str | None: + """停在被人驳回的待办上时怎么说：动作确认写「「付款」未确认（被驳回）」，别的写「「x」被驳回」；不是驳回的为 None。""" + kind, decision = _decision(trail, node_id) + if kind == "action_confirm" and decision == "reject": + return f"「{title}」未确认（被驳回）" + if decision in _REJECTIONS: + return f"「{title}」被驳回" + return None + + +async def legacy_rejected(session: AsyncSession, run: Run) -> bool: + """加 end_reason 之前被人驳回的运行记的是 failed（封存不改写）：失败的那一步上最近一条处理完的待办是驳回。和实例页 + 认存量驳回（instance_state 的 failed 分支）是同一份判断；runner 据此不许它接着跑。""" + if run.status != "failed": + return False + trail = (await _trails(session, [run]))[run.id] + failed = [nid for etype, nid in trail.events if etype == "node.failed"] + return bool(failed) and _decision(trail, failed[-1])[1] in _REJECTIONS + + def instance_state(run: Run, trail: _Trail) -> dict[str, str]: """实例给人看的状态 {code, text}：code 是界面画状态标记用的状态（运行的那几种，加 held），text 说这件事办到哪了、 结论是什么，不只是运行状态（P1）。 @@ -1334,10 +1357,12 @@ def instance_state(run: Run, trail: _Trail) -> dict[str, str]: 否定的那种结局——这次运行里人工节点驳回了、从它的「驳回」出口一路走到的成果节点（rejected_outputs）——code 记 cancelled（同「未确认（被驳回）」，是人的决定，不画成成功），文字照旧写成果节点的标题； - 停着：有已发出、结果未知的动作，写「等待核实写入结果」；否则按在等的待办说（等待审批、等待确认发送……）； - - 失败：停在一个被人驳回的待办上的——动作确认被驳回（「付款」未确认（被驳回））、审批设了驳回即终止、补充输入被 - 驳回——是人的决定，不是故障，code 记 cancelled；核实为未生效、选了放弃的同理。外部系统拒绝之类照写失败。 + - 被驳回（cancelled + end_reason=rejected，engine/context.NodeRejected）：动作确认被驳回写「「付款」未确认（被驳回）」， + 审批设了驳回即终止、补充输入被驳回等写「「x」被驳回」，code 记 cancelled；手动停止的写「已取消」； + - 失败：加 end_reason 之前被人驳回的运行记的是 failed（封存不改写），停在被驳回的待办上的照上一条说，和新形状说法 + 一样；核实为未生效、选了放弃的同理，code 记 cancelled。外部系统拒绝之类照写失败。 - 运行本身的状态不改（驳回动作确认在引擎里就是节点失败、运行结束）：这里只是把它说清楚。""" + 通知（notify/service.notify_run）的文字也取自这里：实例页和推送说同一句话。""" from app.apps.tree import node_map nodes = node_map(run.graph) @@ -1360,15 +1385,18 @@ def instance_state(run: Run, trail: _Trail) -> dict[str, str]: negative = any(_decision(trail, gate)[1] in _REJECTIONS for gate in gates) return {"code": "cancelled" if negative else "succeeded", "text": title(reached[0])} 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"] + node_id = run.error_node_id or (failed[-1] if failed else None) + said = _rejection_text(trail, node_id, title(node_id)) if node_id else None + # 待办没留下（老的审批、没开出待办）时按驳回的那一步说，不退回「已取消」：这是人的决定，不是有人停了它 + return {"code": "cancelled", "text": said or (f"「{title(node_id)}」被驳回" if node_id else REJECTED_TEXT)} if run.status == "failed": failed = [nid for etype, nid in trail.events if etype == "node.failed"] if failed: node_id = failed[-1] - kind, decision = _decision(trail, node_id) - if kind == "action_confirm" and decision == "reject": - return {"code": "cancelled", "text": f"「{title(node_id)}」未确认（被驳回）"} - if decision in _REJECTIONS: - return {"code": "cancelled", "text": f"「{title(node_id)}」被驳回"} + if said := _rejection_text(trail, node_id, title(node_id)): + return {"code": "cancelled", "text": said} record = next((a for a in reversed(trail.actions) if a.node_id == node_id), None) if record is not None and record.state == "failed": if (record.resolution or {}).get("outcome") == "abandon": diff --git a/backend/app/blueprints/publish_plan.py b/backend/app/blueprints/publish_plan.py index 3bbcecf1..cdc1749b 100644 --- a/backend/app/blueprints/publish_plan.py +++ b/backend/app/blueprints/publish_plan.py @@ -70,6 +70,8 @@ NOT_INSTANTIATED = "plan.not_instantiated" GRAPH_CHANGED = "plan.graph_changed" UNTRIALED = "plan.untrialed" TRIAL_FAILED = "plan.trial_failed" +#: 最近一次试运行在某一步被人驳回、没有走完（cancelled + end_reason=rejected）：人的决定，不是故障，不挡发布、照实说 +TRIAL_REJECTED = "plan.trial_rejected" TRIAL_DIFFERS = "plan.trial_differs" #: 蓝图检查：写回的结果不明时没人核实（blueprints/ops.blueprint_issues） RECONCILER_UNBOUND = "bp.reconciler_unbound" @@ -363,6 +365,12 @@ async def _preflight(session: AsyncSession, bp: Blueprint, target: _Target, *, l if is_trial_of(trial, bp) and trial.status == "failed": why = (trial.error or "").strip().splitlines()[0][:200] if (trial.error or "").strip() else "" out.append(_problem(TRIAL_FAILED, f"最近一次试运行失败了{f'：{why}' if why else ''}", source="plan")) + elif is_trial_of(trial, bp) and trial.status == "cancelled" and trial.end_reason == "rejected": + # 搭建者在试运行里驳回了写入、审批：那是他的决定，不是图出了错，不说成「试运行失败了」、也不挡发布；只照实说 + # 没走完——驳回之后的步骤没用真数看过 + where = _node_title(trial.graph, trial.error_node_id) + out.append(_problem(TRIAL_REJECTED, f"最近一次试运行{f'在「{where}」' if where else ''}被驳回、没有走完", + "warning", source="plan")) elif not is_trial_of(trial, bp) or trial.status != "succeeded": out.append(_problem(UNTRIALED, "还没有成功的试运行：发布之前没用真数看过结果", "warning", source="plan")) elif graph_hash(trial.graph or {}) != target.digest: @@ -370,6 +378,17 @@ async def _preflight(session: AsyncSession, bp: Blueprint, target: _Target, *, l return [_with_who(p) for p in out] +def _node_title(graph: Any, node_id: str | None) -> str | None: + """节点在图上的名字（没有名字的给 id）；没有节点时为 None。""" + if not node_id: + return None + for node in (graph or {}).get("nodes") or []: + if isinstance(node, dict) and node.get("id") == node_id: + data = node.get("data") if isinstance(node.get("data"), dict) else {} + return str(data.get("label") or node_id) + return node_id + + def _with_who(issue: dict[str, Any]) -> dict[str, Any]: """问题带上白话（没登记白话的照旧只有技术说法）和归谁处理（you / assistant / admin，摘要按它分三栏）。""" from app.plain import who_of, with_plain @@ -428,6 +447,7 @@ def _summary(bp: Blueprint, target: _Target, *, level: str, first_run: bool, opt if current and current > target.version else "" codes = {p.get("code") for p in problems} tried = "；发布的内容和最近一次试运行不同（试运行之后又改过），建议再试一次" if TRIAL_DIFFERS in codes else \ + "；最近一次试运行被驳回、没有走完" if TRIAL_REJECTED in codes else \ "；还没有用真数试运行过" if UNTRIALED in codes else "" if target.published_as == level: say("publish_workflow", f"「{name}」v{target.version} 已经按{LEVEL_TEXT[level]}级别发布过，不重复发布{tried}") diff --git a/backend/app/engine/approval.py b/backend/app/engine/approval.py index cb8beb21..5da191b8 100644 --- a/backend/app/engine/approval.py +++ b/backend/app/engine/approval.py @@ -39,6 +39,9 @@ class Decision: #: 审批卡上点的是「始终允许」：批准这次，并且这个工具以后改由门控把关（tools/trust.py）。 #: 只认明确的 True，而且只跟着批准走 always: bool = False + #: 这次驳回是待办超时、按「超时驳回」替人做的（process/tasks.py 的 tick），不是有人点了驳回。只跟着驳回走， + #: 节点据此在 NodeRejected 上记 decision=expired_reject + expired: bool = False def _word(value: Any) -> str: @@ -71,6 +74,7 @@ def read_decision(response: Any) -> Decision: code=code if isinstance(code, str) and code else None, value=response.get("value"), always=verdict is True and response.get("always") is True, + expired=verdict is False and response.get("expired") is True, ) verdict = _verdict(response) # 不是一个是/否词的字符串（"拒绝，这个表是生产库"）既算不上批准，本身又 diff --git a/backend/app/engine/context.py b/backend/app/engine/context.py index ad197844..9e4654a3 100644 --- a/backend/app/engine/context.py +++ b/backend/app/engine/context.py @@ -106,3 +106,32 @@ class NodeError(RuntimeError): super().__init__(message) self.node_id = node_id self.retryable = retryable + + +#: 驳回发生在哪种关卡上（NodeRejected.kind）：人工节点的三种模式、写入前的动作确认、代码执行审批、工具调用审批、双盲裁决 +REJECTION_KINDS = ("approve", "input", "edit", "action_confirm", "code", "tool", "dual_check") + + +class NodeRejected(NodeError): + """人的决定让运行到这里结束：驳回、超时驳回（docs/design/next-round/05-已知遗留.md (a)）。 + + 这不是故障。以前六处驳回都抛普通的 NodeError，运行记成 failed：通知推「实例失败」，定时推「运行失败」，发布单把 + 驳回过的试运行说成「试运行失败了」，记录页还给「继续运行」——接着跑只会读到同一个驳回答复、再停一次。现在 runner + 单独接住它，收成 cancelled + end_reason=rejected，run.cancelled 带上在哪一步、谁、结论和备注，消费方都读这一个字段。 + + 是 NodeError 的子类：节点失败的那一套（node.failed、on_error=continue）照旧认它；原话（str）照旧写给人看， + Run.error 和以前一字不差。kind 见 REJECTION_KINDS；decision 是 reject 或 expired_reject（同待办的结论）。""" + + def __init__(self, node_id: str, message: str, *, kind: str, decision: str = "reject", + actor: str | None = None, note: str = "") -> None: + super().__init__(node_id, message) + self.kind = kind + self.decision = decision + self.actor = actor + self.note = note + + @classmethod + def of(cls, ctx: Any, message: str, *, kind: str, decision: Any) -> 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 "") diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index 0152179c..d50cdcc0 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/backend/app/engine/nodes/dual_check.py @@ -70,7 +70,7 @@ from app.db.base import SessionLocal from app.db.models import DataSource from app.engine import deadline, recompute from app.engine.approval import read_decision -from app.engine.context import NodeContext, NodeError +from app.engine.context import NodeContext, NodeError, NodeRejected from app.engine.dual_check_facts import preview, preview_joint, record_choices, upload_dates from app.engine.evidence import build_catalog, is_number, next_exec from app.engine.expressions import ExpressionError, _cell_call, _chain_root, cell_value, column_kind, numeric_text, \ @@ -1396,7 +1396,7 @@ def _choices(ctx: NodeContext, response: Any, rows: list[dict[str, Any]]) -> tup 填写模式一样：这一步没有驳回的出口。""" decision = read_decision(response) if isinstance(response, dict) and decision.rejected: - raise NodeError(ctx.node.id, f"人工驳回：{decision.note or '未说明原因'}") + raise NodeRejected.of(ctx, f"人工驳回：{decision.note or '未说明原因'}", kind="dual_check", decision=decision) value = decision.value if isinstance(decision.value, dict) else response if isinstance(response, dict) else {} raw = value.get("choices") if isinstance(value.get("choices"), dict) else {} choices = {r["metric_id"]: raw[r["metric_id"]] for r in rows if raw.get(r["metric_id"]) in CHOICES} diff --git a/backend/app/engine/nodes/human.py b/backend/app/engine/nodes/human.py index 0c266bec..1e20e9b6 100644 --- a/backend/app/engine/nodes/human.py +++ b/backend/app/engine/nodes/human.py @@ -8,7 +8,7 @@ from typing import Any from app.core.events import EventType from app.db.base import SessionLocal from app.engine.approval import read_decision -from app.engine.context import NodeContext, NodeError +from app.engine.context import NodeContext, NodeError, NodeRejected from app.engine.replay import ask, once from app.engine.state import GraphState, merge_usage, message_text, template_context from app.providers import catalog @@ -52,7 +52,7 @@ async def run_human(state: GraphState, ctx: NodeContext) -> dict[str, Any]: "__decision__": "approved" if approved else "rejected", } if not approved and ctx.cfg("stop_on_reject", False): - raise NodeError(ctx.node.id, f"人工驳回：{result['note'] or '未说明原因'}") + raise NodeRejected.of(ctx, f"人工驳回：{result['note'] or '未说明原因'}", kind="approve", decision=decision) else: # edit / input # 驳回在这两种模式下以前是个摆设：__decision__ 硬编码成 approved， # approved 字段看都不看。界面上按钮在、toast 说"已驳回"，运行却照常 @@ -65,7 +65,7 @@ async def run_human(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 只有结构体里明说了"不"才算驳回：这两种模式下裸字符串就是内容本身， # 用户在输入框里回一句"no"是在交稿，不是在驳回。 if isinstance(response, dict) and decision.rejected: - raise NodeError(ctx.node.id, f"人工驳回：{decision.note or '未说明原因'}") + raise NodeRejected.of(ctx, f"人工驳回：{decision.note or '未说明原因'}", kind=mode, decision=decision) note = decision.note if isinstance(response, dict) else "" # 裸字符串是内容，不是备注 result = {"value": decision.value, "note": note, "__decision__": "approved"} diff --git a/backend/app/engine/nodes/tools.py b/backend/app/engine/nodes/tools.py index 1b26352d..e3591d35 100644 --- a/backend/app/engine/nodes/tools.py +++ b/backend/app/engine/nodes/tools.py @@ -11,7 +11,7 @@ from langgraph.func import task from app.core.artifact_store import canonical_json, content_hash from app.engine.approval import read_decision -from app.engine.context import NodeContext, NodeError +from app.engine.context import NodeContext, NodeError, NodeRejected from app.engine.evidence import ledger_enabled, next_exec, query_entry_fields, schema_ledger_entry from app.engine.replay import ask, once from app.engine.state import GraphState @@ -127,8 +127,8 @@ async def run_tool(state: GraphState, ctx: NodeContext) -> dict[str, Any]: await once(ctx, EventType.HUMAN_RESOLVED, tool=name, approved=decision.approved, note=decision.note, actor=ctx.actor(), **({"always": True} if always else {})) if not decision.approved: - raise NodeError(ctx.node.id, f"工具 {name} 的调用已被驳回" - + (f"：{decision.note}" if decision.note else "")) + raise NodeRejected.of(ctx, f"工具 {name} 的调用已被驳回" + (f"：{decision.note}" if decision.note else ""), + kind="tool", decision=decision) limit = limit_of(tool, name, args) ctx.emit(EventType.TOOL_START, tool=name, args=args, **limit_fields(limit)) @@ -299,10 +299,11 @@ 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: - # 驳回是人的决定，不是故障：运行到这里结束（状态记失败，没有可走的出口），说法写成「未确认（被驳回）」， - # 实例页按待办的结论另说（apps/resolve.instance_states） - raise NodeError(ctx.node.id, f"「{ctx.node.title}」未确认（被驳回），没有发送" - + (f"：{decision.note}" if decision.note else "")) + # 驳回是人的决定，不是故障：运行到这里结束（没有可走的出口），收成 cancelled + end_reason=rejected + # （NodeRejected），说法写成「未确认（被驳回）」，实例页、通知按同一个结论说（apps/resolve.instance_state） + raise NodeRejected.of(ctx, f"「{ctx.node.title}」未确认（被驳回），没有发送" + + (f"：{decision.note}" if decision.note else ""), + kind="action_confirm", decision=decision) if confirmed != prep.payload_hash: # 动作确认不接受改参数：要改在上游的表单里改。确认之后上游的数据变了，看过的就不是要发的，拒发 raise NodeError(ctx.node.id, f"「{name}」确认时看到的内容和现在要发送的不一致（内容指纹对不上），" @@ -380,8 +381,8 @@ async def run_code(state: GraphState, ctx: NodeContext) -> dict[str, Any]: await once(ctx, EventType.HUMAN_RESOLVED, approved=decision.approved, note=decision.note, actor=ctx.actor()) if not decision.approved: - raise NodeError(ctx.node.id, "代码执行已被驳回" - + (f"：{decision.note}" if decision.note else "")) + raise NodeRejected.of(ctx, "代码执行已被驳回" + (f"：{decision.note}" if decision.note else ""), + kind="code", decision=decision) # 隔离档位：strict 要硬件级（microVM），fast 要低延迟（Seatbelt/bwrap）， # 留空跟随整机默认。要的那档不可用时会安静退回默认，不阻断运行。 diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index ef22bd1e..7fcd0a62 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -23,7 +23,7 @@ from app.db.base import SessionLocal from app.db.models import Approval, Run, RunEvent, Workflow from app.engine.approval import read_decision from app.engine.compiler import compile_graph, initial_state -from app.engine.context import NodeError, RunContext +from app.engine.context import NodeError, NodeRejected, RunContext from app.engine.deadline import segment from app.engine.replay import PROTOCOL, PROTOCOL_KEY, protocol_of from app.engine.schema import GraphSpec, NodeType, loop_steps, topology_of, validate_graph @@ -42,9 +42,18 @@ _EPHEMERAL = {EventType.LLM_TOKEN, EventType.LLM_THINKING_DELTA} RUN_OPTIONS_KEY = "run_options" -#: 接着跑被拒的机器码：正式运行带图（F2），有结果未知、还没核实的写入 +#: 接着跑被拒的机器码：正式运行带图（F2），有结果未知、还没核实的写入，被人驳回了的 FORMAL_GRAPH_LOCKED = "formal_graph_locked" ACTION_UNRESOLVED = "action_unresolved" +RUN_REJECTED = "run_rejected" + +#: 运行为什么结束在 cancelled（Run.end_reason，docs/design/next-round/00-总方案.md X9） +END_REJECTED = "rejected" +END_STOPPED = "stopped" +#: 被驳回的运行不能接着跑：从断点重跑会读到同一个驳回答复、再停在同一处 +REJECTED_CONTINUE = "被驳回的运行不能接着跑：改好内容后重新发起" +#: 服务重启时还在排队的运行改成失败时写的话（run.error 和补发的 run.failed 是同一句） +QUEUED_AT_RESTART = "服务重启时这次运行仍在排队，没有可恢复的进度，请重新发起运行" class ContinueRefused(ValueError): @@ -94,6 +103,9 @@ class RunManager: stranded = list((await session.execute( select(Run.id, Run.last_seq).where(Run.status == "running") )).all()) + queued = list((await session.execute( + select(Run.id, Run.last_seq).where(Run.status == "queued") + )).all()) await session.execute( update(Run) .where(Run.status == "running") @@ -102,7 +114,7 @@ class RunManager: await session.execute( update(Run) .where(Run.status == "queued") - .values(status="failed", error="服务重启时这次运行仍在排队，没有可恢复的进度，请重新发起运行") + .values(status="failed", error=QUEUED_AT_RESTART) ) await session.commit() # 进程被强杀时连 server_shutdown 日志都来不及发，事件流停在半路。补上这一条， @@ -113,6 +125,11 @@ class RunManager: "level": "warn", "code": "server_shutdown", "message": "服务上次未正常关闭，这次运行已中断；可从断点继续运行", }) + # 排队中的改成了失败，也补一条 run.failed：以前只改状态，时间线里没有失败原因这一行（05-已知遗留 (h)）。 + # 不写结束时间、不封存：它从没开始执行，同以前（定时通知按「没有开始时间」认出它，scheduler/webhook.py） + for run_id, last_seq in queued: + bus.set_seq(run_id, last_seq or 0) + await self._emit(run_id, EventType.RUN_FAILED, data={"error": QUEUED_AT_RESTART}) async def shutdown(self) -> None: self._closing = True @@ -505,8 +522,13 @@ class RunManager: Approval.run_id == run_id, Approval.status.in_(["pending", "answered"]) ).limit(1) )).first() is not None + if run.status == "cancelled" and run.end_reason == END_REJECTED: + raise ContinueRefused(REJECTED_CONTINUE, code=RUN_REJECTED) if not (run.status == "failed" or (run.status == "interrupted" and not has_pending)): raise ValueError(_cannot_continue(run.status, has_pending)) + if run.status == "failed" and await _rejected_before(session, run): + # 加 end_reason 之前被驳回的运行记的是 failed（封存不改写）：接着跑会读到同一个驳回答复、再停一次 + raise ContinueRefused(REJECTED_CONTINUE, code=RUN_REJECTED) if graph is not None and run.run_class == "formal": raise ContinueRefused( "正式运行固定在发布的版本上，不能修改配置后继续运行。如需修改，请在画布上修改并发布新版本后重新发起", @@ -613,7 +635,7 @@ class RunManager: # 或者另一个取消先到了，这里就什么都不做 claimed = await session.execute( update(Run).where(Run.id == run_id, Run.status == "interrupted") - .values(status="cancelled", finished_at=now) + .values(status="cancelled", end_reason=END_STOPPED, finished_at=now) ) if claimed.rowcount != 1: await session.rollback() @@ -631,7 +653,7 @@ class RunManager: bus.set_seq(run_id, await _seq_floor(session, run_id, run.last_seq)) # 和状态同一个事务提交，理由同 _finalize event = self._event(run_id, EventType.RUN_CANCELLED, data={ - "timing": timing, "actor": actor, + "reason": END_STOPPED, "timing": timing, "actor": actor, "message": f"放弃了这次运行，{closed} 条待审批一并关闭" if closed else "放弃了这次挂起的运行", }) @@ -684,6 +706,8 @@ class RunManager: #: 终态事件（run.failed / run.cancelled）等 _finalize 算完耗时再发，timing 才带得上 terminal: tuple[EventType, dict[str, Any]] | None = None error_node: str | None = None + #: 收成 cancelled 的原因（Run.end_reason）：rejected 人的驳回，stopped 手动停止 + end_reason: str | None = None #: 结果未知、挂起等核实的动作（ActionPending），收尾之后发 action.unknown、安排核实 pending_action: str | None = None @@ -796,7 +820,9 @@ class RunManager: else: status = "cancelled" error = "用户取消" - terminal = (EventType.RUN_CANCELLED, {"actor": self._stopped_by.pop(run_id, None)}) + end_reason = END_STOPPED + terminal = (EventType.RUN_CANCELLED, {"reason": END_STOPPED, + "actor": self._stopped_by.pop(run_id, None)}) raise except ActionPending as e: # 写外部系统的结果未知（app/process/actions.py）：不在节点里等，运行收成挂起、不封存，等核实之后 @@ -829,6 +855,12 @@ class RunManager: "（AGENTLAB_MAX_GRAPH_STEPS）" ) terminal = (EventType.RUN_FAILED, {"error": error}) + except NodeRejected as e: + # 人的决定（驳回、超时驳回），不是故障：收成 cancelled，同手动停止走一样的收尾（关掉并行支路上的审批、 + # 回收沙箱、关流、封存、不许继续运行）。原话照旧记在 error 上，在哪一步、谁、结论、备注进 run.cancelled + status, end_reason = "cancelled", END_REJECTED + error, error_node = str(e), e.node_id + terminal = (EventType.RUN_CANCELLED, rejected_data(e, spec)) except Exception as e: # noqa: BLE001 status = "failed" error, terminal = _failure(e, spec) @@ -840,7 +872,7 @@ class RunManager: self._finalizing.add(run_id) try: await self._finalize(run_id, status, error, final_state, elapsed, - terminal=terminal, error_node=error_node) + terminal=terminal, error_node=error_node, end_reason=end_reason) finally: # 收尾被打断（关停时的取消、库连接出错）也得把自己摘掉。以前写在 # _finalize 后面，一被打断就漏掉，留下的任务让下一次 shutdown 的 @@ -944,6 +976,7 @@ class RunManager: *, terminal: tuple[EventType, dict[str, Any]] | None = None, error_node: str | None = None, + end_reason: str | None = None, ) -> None: usage = dict(state.get("usage") or {}) output = dict(state.get("output") or {}) @@ -1000,6 +1033,7 @@ class RunManager: run.output = output run.usage = usage run.error_node_id = error_node + run.end_reason = end_reason if status != "interrupted": run.finished_at = now if status in ("failed", "cancelled"): @@ -1024,11 +1058,12 @@ class RunManager: # 关停时挂起的也要让在线的客户端收尾：进程马上就没了，不说一声， # 界面会一直以为它还在跑 await bus.close(run_id) - if app_id and status in ("succeeded", "failed"): - # 应用实例结束、失败：按应用订阅推送（app/notify/service.py）。放到后台，接收端慢不拖着收尾 + if app_id and (status in ("succeeded", "failed") or end_reason == END_REJECTED): + # 应用实例结束、失败：按应用订阅推送（app/notify/service.py）。放到后台，接收端慢不拖着收尾。被驳回的走 + # 「实例结束」（next-round X6）：发起人要知道结果，订阅「实例失败」的人盯的是故障；手动停止的不推，同以前 from app.notify.service import notify_run, spawn - spawn(notify_run(run_id, "run.finished" if status == "succeeded" else "run.failed")) + spawn(notify_run(run_id, "run.failed" if status == "failed" else "run.finished")) async def _seal(self, run_id: str) -> None: """对到此为止的全部事件算清单哈希，记下封到了哪一条。""" @@ -1269,6 +1304,22 @@ 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_data(exc: NodeRejected, spec: GraphSpec) -> dict[str, Any]: + """被驳回的 run.cancelled 带什么：原因（rejected）、在哪一步（节点 id 和图上的名字）、哪种关卡、结论（驳回 / 超时 + 驳回）、谁、备注。进封存：谁在哪一步驳回、为什么，事后查得到。""" + node = spec.node_map().get(exc.node_id) + return {"reason": END_REJECTED, "node_id": exc.node_id, "label": node.title if node else exc.node_id, + "kind": exc.kind, "decision": exc.decision, "actor": exc.actor, "note": exc.note} + + +async def _rejected_before(session: Any, run: Run) -> bool: + """加 end_reason 之前被人驳回、记成 failed 的运行：失败的那一步上最近一条处理完的待办是驳回（和实例页认驳回的是 + 同一份判断，apps/resolve.legacy_rejected）。""" + from app.apps.resolve import legacy_rejected + + return await legacy_rejected(session, run) + + def _failure(exc: BaseException, spec: GraphSpec) -> tuple[str, tuple[EventType, dict[str, Any]]]: """失败的首行报错（给人看）和 run.failed 的负载（带上定位和原始异常）。""" if isinstance(exc, NodeError): diff --git a/backend/app/notify/service.py b/backend/app/notify/service.py index a836574a..33b8ff78 100644 --- a/backend/app/notify/service.py +++ b/backend/app/notify/service.py @@ -201,22 +201,36 @@ async def notify_task(task_id: str, event: str, *, users: list[str]) -> int: async def notify_run(run_id: str, event: str) -> int: - """应用实例结束、失败（run.finished / run.failed）。只推应用发起的运行。""" + """应用实例结束、失败（run.finished / run.failed）。只推应用发起的运行。 + + 状态那一句取自实例页的说法（apps/resolve.instance_state），两边说同一句话：被驳回的（cancelled + end_reason=rejected） + 走「实例结束」、写「「付款」未确认（被驳回）」；走到「已退回」成果的写「已退回」，不再一律写「已完成」。ok 只在真的 + 办成了（code 是 succeeded）时为真。 + + 发起人是认证过的身份（run.origin.actor_verified）时，他自己订阅了这件事的个人渠道也收得到（「我发起的实例完成 / 失败」， + 04-对外集成与触达 B1）。以前这里给 _channels 的 users 是空的，个人渠道勾了这两项永远收不到。自报的署名不算：关登录 + 时谁都能填别人的名字。""" + from app.apps.resolve import _trails, instance_state + async with SessionLocal() as session: run = await session.get(Run, run_id) if run is None or not run.app_id: return 0 - channels = await _channels(session, event, run, []) + verified = bool((run.origin or {}).get("actor_verified")) and bool(run.started_by) + channels = await _channels(session, event, run, [run.started_by] if verified else []) if not channels: return 0 base = await _base_url(session) where = await _where(session, run) + state = instance_state(run, (await _trails(session, [run]))[run.id]) finished = event == "run.finished" - lines = [("状态", "已完成" if finished else "失败"), ("实例", f"#{run_id[:6]}")] - if not finished and run.error_node_id: + ok = state["code"] == "succeeded" + lines = [("状态", state["text"]), ("实例", f"#{run_id[:6]}")] + if not finished and run.error_node_id and state["code"] == "failed": lines.append(("失败节点", f"「{_label(run.graph, run.error_node_id)}」")) - message = {"event": event, "status": run.status, "ok": finished, - "title": f"{where or '应用'}的实例{'已完成' if finished else '失败'}", "lines": lines, + head = ("已完成" if ok else "已结束") if finished else ("已结束" if state["code"] == "cancelled" else "失败") + message = {"event": event, "status": run.status, "ok": ok, + "title": f"{where or '应用'}的实例{head}", "lines": lines, "links": [("查看实例", f"{base}/apps/{run.app_id}/i/{run_id}")] if base else [], "subject": {"run_id": run_id, "app_id": run.app_id}} return await _push(channels, message, subject_kind="run", subject_id=run_id, event=event, run_id=run_id, diff --git a/backend/app/plain.py b/backend/app/plain.py index 617afe92..6c38d3a4 100644 --- a/backend/app/plain.py +++ b/backend/app/plain.py @@ -181,6 +181,9 @@ PLAIN: dict[str, Plain] = { "plan.trial_failed": Plain( "最近一次试运行失败了", "用真数运行出了错，这样发布出去，每次出数都会是同一个错。", _Y, "回到试运行看出错的原因（可以交给助手修），修好后再用真数试一次。"), + "plan.trial_rejected": Plain( + "最近一次试运行被驳回、没有走完", "那是你（或审批人）的决定，不是出错；驳回之后的步骤没用真数看过。", _Y, + "要看完整结果，再试运行一次并批准；确认不需要的话可以直接发布。"), "plan.untrialed": Plain( "还没用真数试运行过", "没看过真数就发布，可能不是你要的结果。", _Y, "先试运行一次，看过结果再发布。"), "plan.trial_differs": Plain( diff --git a/backend/app/process/actions.py b/backend/app/process/actions.py index e75a269e..87ab7fca 100644 --- a/backend/app/process/actions.py +++ b/backend/app/process/actions.py @@ -572,17 +572,18 @@ async def _continue(run_id: str, actor: str | None) -> Any: return None -async def tick(now: datetime | None = None) -> int: - """调度循环每一拍：对登记了 lookup 的结果未知动作带退避地查询。返回查了几条。""" +async def tick(now: datetime | None = None) -> list[str]: + """调度循环每一拍：对登记了 lookup 的结果未知动作带退避地查询。交回这一拍查过的动作记录 id（同 tasks.tick）。""" now = now or utcnow() async with SessionLocal() as session: due = list((await session.execute(select(ActionRecord.id, ActionRecord.next_lookup_at).where( ActionRecord.state == "unknown", ActionRecord.next_lookup_at.is_not(None), ActionRecord.next_lookup_at <= now))).all()) - done = 0 + done: list[str] = [] for action_id, claimed_at in due: try: - done += await _lookup(action_id, claimed_at, now) + if await _lookup(action_id, claimed_at, now): + done.append(action_id) except Exception: # noqa: BLE001 - 一条出错不拖住这一拍的其他动作 logger.exception("动作 %s 的结果查询失败，下一拍再试", action_id) return done diff --git a/backend/app/process/tasks.py b/backend/app/process/tasks.py index 9f1dd240..6ec167af 100644 --- a/backend/app/process/tasks.py +++ b/backend/app/process/tasks.py @@ -761,8 +761,9 @@ async def _notify(task_id: str, event: str, users: list[str]) -> int: # -------------------------------------------------------------------------- -async def tick(now: datetime | None = None) -> int: - """调度循环每一拍：提醒、升级、超时驳回（永不自动批准）。返回处理的条数。试运行的待办不处理（只显示时限）。""" +async def tick(now: datetime | None = None) -> list[str]: + """调度循环每一拍：提醒、升级、超时驳回（永不自动批准）。交回这一拍处理过的待办 id（生效可查：办了哪几条， + 不只是几条；测试按自己造的那条断言，不受同一个库里别的待办影响）。试运行的待办不处理（只显示时限）。""" now = now or utcnow() action = func.json_extract(Task.on_timeout, "$.action") trial = select(Run.id).where(Run.origin_kind == TRIAL_ORIGIN) @@ -773,10 +774,11 @@ async def tick(now: datetime | None = None) -> int: and_(Task.due_at.is_not(None), Task.due_at <= now, action == "reject"), and_(Task.due_at.is_not(None), Task.due_at <= now, action == "escalate", Task.escalation_level == 0), )).order_by(Task.due_at))).scalars()) - done = 0 + done: list[str] = [] for task_id in due: try: - done += await _on_timeout(task_id, now) + if await _on_timeout(task_id, now): + done.append(task_id) except Exception: # noqa: BLE001 - 一条出错不拖住这一拍的其他待办 logger.exception("待办 %s 的超时处理失败，下一拍再试", task_id) return done @@ -838,7 +840,9 @@ async def _on_timeout(task_id: str, now: datetime) -> int: from app.engine.runner import run_manager try: - await run_manager.resume(run_id, {"approved": False, "note": EXPIRED_NOTE}, approval_id=approval_id) + # expired：节点据此把这次驳回记成超时驳回（engine/approval.Decision.expired → NodeRejected.decision） + await run_manager.resume(run_id, {"approved": False, "note": EXPIRED_NOTE, "expired": True}, + approval_id=approval_id) except (ValueError, KeyError): logger.warning("待办 %s 超时驳回时运行已无法恢复", task_id) async with SessionLocal() as session: diff --git a/backend/app/scheduler/loop.py b/backend/app/scheduler/loop.py index 243f5f97..ee6373c2 100644 --- a/backend/app/scheduler/loop.py +++ b/backend/app/scheduler/loop.py @@ -130,7 +130,9 @@ async def process_due(now: datetime | None = None) -> None: for name, step in (("待办的时限处理", tasks.tick), ("写入结果的查询", actions.tick), ("没有提交的上传文件清理", files_store.cleanup_tick)): try: - await step(now) + done = await step(now) + if isinstance(done, list) and done: + logger.info("%s：这一拍处理了 %d 条", name, len(done)) except Exception: # noqa: BLE001 logger.exception("%s这一拍出错，下一拍照常进行", name) diff --git a/backend/app/scheduler/webhook.py b/backend/app/scheduler/webhook.py index 4a19e1c6..2062db3c 100644 --- a/backend/app/scheduler/webhook.py +++ b/backend/app/scheduler/webhook.py @@ -5,7 +5,8 @@ 再导出，调用方和测试不用改；_transport、RETRY_DELAYS 仍是这个模块自己的（测试照旧替换它们），send 每次调用时读。 通知里只有任务名、状态、出具档位、缺口条数、运行链接、封存哈希（界面上叫清单哈希）前 12 位和起止时间，失败时 -另写失败节点的名字和下一步（不许诺能继续运行）；运行成功结束后和同一任务上一次成功的运行的对比（auto_compare）只写 +另写失败节点的名字和下一步（不许诺能继续运行）；被人驳回的那一期按「运行结束」推「未出具（被驳回）」、写驳回的那一步， +手动停止的不推；运行成功结束后和同一任务上一次成功的运行的对比（auto_compare）只写 各类指标的个数和对比页链接。不放报告里的具体数字，也不放失败时的原始报错——报错里可能带着查询到的数据，Webhook 推去的群不一定有权看。 """ @@ -71,6 +72,11 @@ NEXT_AS_USUAL = "下一次触发会照常发起" #: 从没开始执行、任务又不在排期上（停用了）：原因刚写过，记录页上也没有更多可看的，下一步是重新发起。「立即运行」 #: 是定时任务页上那个按钮的名字，停用的任务也能点 RERUN_ACTION = "如需重新发起，请在 AgentLab 的「定时任务」中点击「立即运行」" +#: 被人驳回的那一期（cancelled + end_reason=rejected，next-round X9）：是人的决定，不是故障。按「运行结束」推（订阅 +#: 「运行失败」的人盯的是故障），标题、状态照实写被驳回；驳回的备注不写（可能带着数），驳回的是哪一步写节点名 +REJECTED_HEADLINE = "未出具（被驳回）" +REJECTED_STATUS = "被驳回" +REJECTED_ACTION = "驳回的说明请在 AgentLab 的「记录」中查看" # -------------------------------------------------------------------------- @@ -110,12 +116,18 @@ class Notice: #: 挂起的原因是写外部系统的结果未知（等核实，核实后自动接着跑），不是服务重启。只在 suspended 时有意义 action_pending: bool = False error_node: str | None = None + #: 运行结束在 cancelled 的原因（Run.end_reason）：rejected 时按「未出具（被驳回）」说，error_node 是驳回的那一步 + end_reason: str | None = None next_fire: bool = False compare: dict[str, Any] | None = None compare_url: str | None = None fmt: str | None = None signed: bool = False + @property + def rejected(self) -> bool: + return self.status == "cancelled" and self.end_reason == "rejected" + @property def never_started(self) -> bool: """失败的运行从没开始执行（没有开始时间）：服务重启时还在排队、被改成失败的就是这种，没有断点可续。""" @@ -142,6 +154,8 @@ def message(notice: Notice) -> tuple[str, list[tuple[str, str]], str | None]: headline = "未能发起运行" elif notice.suspended: headline = "运行已挂起" + elif notice.rejected: + headline = REJECTED_HEADLINE else: headline = _HEADLINE.get(notice.event, notice.event) title = f"定时任务「{notice.schedule_name}」{headline}" @@ -163,6 +177,10 @@ def message(notice: Notice) -> tuple[str, list[tuple[str, str]], str | None]: lines += [("原因", ACTION_PENDING_REASON), ("处理", ACTION_PENDING_ACTION)] elif notice.suspended: lines += [("原因", SUSPENDED_REASON), ("处理", SUSPENDED_ACTION)] + elif notice.rejected: + if notice.error_node: + lines.append(("驳回节点", f"「{notice.error_node}」")) + lines.append(("处理", f"{REJECTED_ACTION}；{NEXT_AS_USUAL}" if notice.next_fire else REJECTED_ACTION)) elif notice.status == "failed": # 不写原始报错（可能带着查询到的数据），写失败节点的名字和下一步 if notice.error_node: @@ -219,6 +237,8 @@ def _status_label(notice: Notice) -> str: if notice.pending_approvals: return "等待审批" return "已挂起，等待核实" if notice.action_pending else "已挂起，可继续运行" + if notice.rejected: + return REJECTED_STATUS return _STATUS_LABEL.get(notice.status or "", notice.status or "") @@ -239,7 +259,9 @@ def _links(notice: Notice) -> list[tuple[str, str]]: def _message(notice: Notice) -> core.Message: title, lines, _ = message(notice) - return core.Message(title=title, lines=lines, links=_links(notice), ok=notice.event in ("finished", "test")) + # 被驳回的那一期按 finished 推，但没出具：不画成成功的颜色 + return core.Message(title=title, lines=lines, links=_links(notice), + ok=notice.event in ("finished", "test") and not notice.rejected) def _text(notice: Notice) -> str: @@ -332,10 +354,17 @@ async def _record(schedule_id: str, fire_id: str | None, run_id: str | None, eve #: 运行到了这些状态才可能要推送 _WATCHED = ("succeeded", "failed", "cancelled", "interrupted") -#: 结束的运行 → 事件。cancelled 是有人手动停的，不推。interrupted 分停在审批和挂起两种，见 _state_of +#: 结束的运行 → 事件。cancelled 是有人手动停的，不推；被驳回的（end_reason=rejected）按 finished 推，见 _event_of。 +#: interrupted 分停在审批和挂起两种，见 _state_of _EVENT_OF = {"succeeded": "finished", "failed": "failed", "cancelled": None} +def _event_of(run: Run) -> str | None: + if run.status == "cancelled" and run.end_reason == "rejected": + return "finished" + return _EVENT_OF[run.status] + + @dataclass class _Job: fmt: str @@ -469,7 +498,7 @@ async def _state_of(session: Any, fire: ScheduleFire, run: Run | None, now: date since = run.finished_at or run.updated_at or fire.created_at if not sealed and (now - service.as_utc(since)).total_seconds() < SEAL_WAIT_S: return _Seen(None) - return _Seen(run.status, _EVENT_OF[run.status], sealed=sealed) + return _Seen(run.status, _event_of(run), sealed=sealed) if run.status != "interrupted": return _Seen(None) counts = dict((await session.execute( @@ -516,7 +545,9 @@ async def _notice(session: Any, schedule: Schedule, fire: ScheduleFire, run: Run pending_approvals=pending or None, suspended=run is not None and run.status == "interrupted" and not pending, action_pending=seen.action_pending, - error_node=_failed_node(run) if run is not None and run.status == "failed" else None, + error_node=_failed_node(run) if run is not None and (run.status == "failed" or run.end_reason == "rejected") + else None, + end_reason=run.end_reason if run is not None else None, next_fire=bool(schedule.enabled and schedule.next_run_at is not None), ) diff --git a/backend/tests/test_app_instance_page.py b/backend/tests/test_app_instance_page.py index 764273a0..ce502afb 100644 --- a/backend/tests/test_app_instance_page.py +++ b/backend/tests/test_app_instance_page.py @@ -258,7 +258,7 @@ async def test_after_escalation_the_rule_is_one_director_not_zero_of_three(): run_id = await _start(app_id, "报销审批·钱七") task = await _task(run_id) await tasks.respond(task.id, user(a), decision="approve") - assert await tasks.tick(utcnow() + timedelta(hours=2)) >= 1 + assert task.id in await tasks.tick(utcnow() + timedelta(hours=2)) detail = await tasks.detail(task.id, user(boss)) p = detail["progress"] # 升级加派的那一份不算进会签人数：还是 1/2，另写升级给了谁、他一人说了算 diff --git a/backend/tests/test_approval.py b/backend/tests/test_approval.py index f3d005a6..0f45888c 100644 --- a/backend/tests/test_approval.py +++ b/backend/tests/test_approval.py @@ -138,9 +138,10 @@ async def test_tool_node_takes_a_refusal_as_a_refusal(): run_id = await _start(_calc_graph()) await _wait(run_id, ("interrupted",)) await run_manager.resume(run_id, "拒绝") - run = await _wait(run_id, ("failed", "succeeded")) + run = await _wait(run_id, ("failed", "succeeded", "cancelled")) - assert run.status == "failed" and "驳回" in (run.error or "") + # 驳回是人的决定，运行收成 cancelled + end_reason=rejected（05-已知遗留 (a)） + assert (run.status, run.end_reason) == ("cancelled", "rejected") and "驳回" in (run.error or "") assert not [e for e in await _events(run_id) if e.type == "tool.start"], "被拒绝的工具执行了" @@ -208,8 +209,8 @@ async def test_a_refused_write_leaves_the_data_alone(writable): run_id = await _start(_query_graph(tool, "WITH k AS (SELECT 1) DELETE FROM t")) await _wait(run_id, ("interrupted",)) await run_manager.resume(run_id, "拒绝") - run = await _wait(run_id, ("failed", "succeeded")) - assert run.status == "failed" + run = await _wait(run_id, ("failed", "succeeded", "cancelled")) + assert (run.status, run.end_reason) == ("cancelled", "rejected") assert count() == 5 @@ -392,5 +393,5 @@ async def test_parallel_rejection_with_content_is_still_a_rejection(): approval_id=pending["h1"].id) await run_manager.resume(run_id, {"approved": False, "note": "不行", "value": "乙"}, approval_id=pending["h2"].id) - run = await _wait(run_id, ("failed", "succeeded")) - assert run.status == "failed" and "驳回" in (run.error or ""), run.error + run = await _wait(run_id, ("failed", "succeeded", "cancelled")) + assert (run.status, run.end_reason) == ("cancelled", "rejected") and "驳回" in (run.error or ""), run.error diff --git a/backend/tests/test_blueprint_trial.py b/backend/tests/test_blueprint_trial.py index d5ad468b..957a8549 100644 --- a/backend/tests/test_blueprint_trial.py +++ b/backend/tests/test_blueprint_trial.py @@ -157,7 +157,7 @@ async def test_process_trial_assigns_every_step_to_the_trial_runner_and_never_es assert task.due_at is not None and task.on_timeout["action"] == "escalate" # 时限到了：调度不升级试运行的待办（同一个库里别的测试的待办照常处理，只看这一条） - await tasks.tick(utcnow() + timedelta(hours=2)) + assert task.id not in await tasks.tick(utcnow() + timedelta(hours=2)) async with SessionLocal() as session: task = await session.get(Task, task.id) count = len(list((await session.execute(select(TaskAssignment).where( diff --git a/backend/tests/test_dual_check_node.py b/backend/tests/test_dual_check_node.py index 436cd9bc..3ca9af15 100644 --- a/backend/tests/test_dual_check_node.py +++ b/backend/tests/test_dual_check_node.py @@ -519,15 +519,19 @@ async def test_a_missing_choice_is_recorded_as_no_choice(monkeypatch, shop, clie assert not logs(evs, "dual_check_catalog") and await notes_of(source_id) is None -async def test_rejecting_the_approval_fails_the_node(monkeypatch, shop, client): +async def test_rejecting_the_approval_ends_the_run_as_rejected(monkeypatch, shop, client): + """明确驳回裁决：人的决定，不是故障（05-已知遗留 (a)）。运行收成 cancelled + end_reason=rejected，原话照旧。""" name, _ = shop Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) run = await start(graph(name, BY_ORDER_DATE)) 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")) - assert done.status == "failed" and "人工驳回" in (done.error or "") and "口径要重新对" in done.error + done = await wait(run.id, ("succeeded", "failed", "cancelled")) + assert (done.status, done.end_reason) == ("cancelled", "rejected"), done.error + assert "人工驳回" in (done.error or "") and "口径要重新对" in done.error + [ended] = [e for e in await events(run.id) if e.type == "run.cancelled"] + assert (ended.data["kind"], ended.data["note"]) == ("dual_check", "口径要重新对") async def test_on_reject_fail_stops_the_run(monkeypatch, shop, client): diff --git a/backend/tests/test_notify_run.py b/backend/tests/test_notify_run.py new file mode 100644 index 00000000..a57e32ff --- /dev/null +++ b/backend/tests/test_notify_run.py @@ -0,0 +1,199 @@ +"""应用实例结束的通知（notify/service.notify_run）：说法取自实例页，个人渠道按「我发起的实例」推。 + +- 被驳回的实例（cancelled + end_reason=rejected）走「实例结束」，状态写实例页那一句（「「review」被驳回」），不推 + 「实例失败」（05-已知遗留 (a)、总方案 X6）； +- 走到「已退回」那种结局的运行，实例页写成果节点的标题、记成人的决定，推送以前却写「已完成」：现在两边一致； +- 个人渠道勾了「实例完成 / 失败」以前永远收不到（_channels 拿到的 users 是空的）：发起人是认证过的身份时，他自己的 + 个人渠道收得到；自报的署名不算（04-对外集成与触达 B1）。 +""" +from __future__ import annotations + +import asyncio +import json +import uuid + +import httpx +import pytest +from sqlalchemy import select + +from app.core.crypto import encrypt +from app.db.base import SessionLocal +from app.db.models import App, AppVersion, NotifyChannel, NotifyDelivery, Run, User +from app.engine.runner import run_manager +from app.notify import core, service + +URL = "https://hooks.example.com/agentlab/run-3b8e1f" + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await service.drain() + await run_manager.shutdown() + + +@pytest.fixture +def received(monkeypatch): + got: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + got.append(request) + return httpx.Response(200, json={"ok": True}) + + monkeypatch.setattr(core, "_transport", httpx.MockTransport(handler)) + monkeypatch.setattr(core, "RETRY_DELAYS", (0, 0, 0)) + # 个人渠道发送前要解析域名、核对不是内网：测试里不连外网 + async def public(url: str) -> None: + return None + + monkeypatch.setattr(service, "personal_url_problem", public) + return got + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def flow(**human): + nodes = [node("start", "input", fields=[{"name": "q"}]), + node("review", "human", mode="approve", title="审一下", **human), + node("ok", "output", fields=[{"name": "结论", "value": "通过"}]), + node("no", "output", fields=[{"name": "结论", "value": "退回"}])] + edges = [{"source": "start", "target": "review"}, + {"source": "review", "target": "ok", "sourceHandle": "approved"}, + {"source": "review", "target": "no", "sourceHandle": "rejected"}] + return {"nodes": nodes, "edges": edges} + + +async def _app(events: list[str]) -> tuple[str, str]: + async with SessionLocal() as session: + channel = NotifyChannel(name=f"群-{uuid.uuid4().hex[:6]}", format="generic", url=encrypt(URL), events=[]) + session.add(channel) + app = App(name=f"报销-{uuid.uuid4().hex[:6]}", published_version=1) + session.add(app) + await session.flush() + session.add(AppVersion(app_id=app.id, version=1, spec_hash="h" * 64, bindings={}, + spec={"notify": [{"channel": channel.id, "events": events}]})) + await session.commit() + return app.id, channel.id + + +async def _wait(run_id: str, statuses: tuple[str, ...]) -> Run: + for _ in range(400): + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row is not None and row.status in statuses: + return row + await asyncio.sleep(0.05) + raise AssertionError(f"运行 {run_id} 没有进入 {statuses}") + + +async def _decide(run_id: str, approved: bool) -> Run: + """经待办服务回复（同界面上点同意 / 驳回）：实例页认结局靠待办的结论。""" + from app.api.deps import Principal + from app.db.models import Task + from app.process import tasks + + await _wait(run_id, ("interrupted",)) + await run_manager.wait_idle(run_id) + for _ in range(100): + async with SessionLocal() as session: + task = (await session.execute(select(Task).where(Task.run_id == run_id, Task.status == "pending")))\ + .scalars().first() + if task is not None: + break + await asyncio.sleep(0.05) + await tasks.respond(task.id, Principal(username=None, role="admin", mode="off"), + decision="approve" if approved else "reject", note="金额 45678 不对") + final = await _wait(run_id, ("succeeded", "failed", "cancelled")) + await run_manager.wait_idle(run_id) + await service.drain() + return final + + +async def _deliveries(channel_id: str) -> list[tuple[str, str]]: + async with SessionLocal() as session: + rows = (await session.execute(select(NotifyDelivery).where(NotifyDelivery.channel_id == channel_id))).scalars() + return [(r.event, r.status) for r in rows] + + +async def test_a_rejected_instance_is_pushed_as_finished_in_the_instance_page_words(received): + app_id, channel_id = await _app(["run.finished", "run.failed"]) + run = await run_manager.start(graph=flow(stop_on_reject=True), input_payload={"q": "x"}, app_id=app_id, + app_version=1) + final = await _decide(run.id, approved=False) + assert (final.status, final.end_reason) == ("cancelled", "rejected") + assert await _deliveries(channel_id) == [("run.finished", "ok")], "驳回不进「实例失败」" + [request] = received + body = json.loads(request.content) + assert body["event"] == "run.finished" and body["title"].endswith("的实例已结束") + assert "状态：「review」被驳回" in body["text"] and "失败" not in body["text"] + assert "45678" not in request.content.decode(), "驳回的备注不进消息（可能带着数）" + + +async def test_a_manual_stop_is_not_pushed(received): + app_id, channel_id = await _app(["run.finished", "run.failed"]) + run = await run_manager.start(graph=flow(), input_payload={"q": "x"}, app_id=app_id, app_version=1) + await _wait(run.id, ("interrupted",)) + await run_manager.wait_idle(run.id) + assert await run_manager.cancel(run.id) == "cancelled" + await service.drain() + assert await _deliveries(channel_id) == [] and received == [] + + +async def test_reaching_the_returned_outcome_no_longer_says_completed(received): + app_id, channel_id = await _app(["run.finished"]) + run = await run_manager.start(graph=flow(), input_payload={"q": "x"}, app_id=app_id, app_version=1) + final = await _decide(run.id, approved=False) + assert final.status == "succeeded" and final.output.get("结论") == "退回" + [request] = received + body = json.loads(request.content) + # 实例页写走到的成果节点（no），记成人的决定：推送也这么写，不再说「已完成」 + assert "状态：no" in body["text"] and "已完成" not in body["title"] and body["title"].endswith("的实例已结束") + + +async def _personal(username: str, events: list[str]) -> str: + async with SessionLocal() as session: + user = (await session.execute(select(User).where(User.username == username))).scalar_one_or_none() + if user is None: + user = User(username=username, role="user") + session.add(user) + await session.flush() + channel = NotifyChannel(name="我的", scope="user", owner_id=user.id, format="generic", url=encrypt(URL), + events=events) + session.add(channel) + await session.commit() + return channel.id + + +async def _finished_instance(app_id: str, *, started_by: str, verified: bool) -> str: + async with SessionLocal() as session: + run = Run(workflow_name="报销", status="succeeded", graph=flow(), input={}, app_id=app_id, app_version=1, + started_by=started_by, origin_kind="app_entry", + origin={"kind": "app_entry", "actor": started_by, "actor_verified": verified}) + session.add(run) + await session.commit() + return run.id + + +async def test_the_initiator_s_personal_channel_receives_my_instance_finished(received): + """B1：个人渠道订阅了 run.finished，发起人自己发起的实例结束了收得到；应用自己没订阅也一样。""" + app_id, _ = await _app([]) + name = f"uma{uuid.uuid4().hex[:6]}" + mine = await _personal(name, ["run.finished"]) + run_id = await _finished_instance(app_id, started_by=name, verified=True) + assert await service.notify_run(run_id, "run.finished") == 1 + assert await _deliveries(mine) == [("run.finished", "ok")] + body = json.loads(received[0].content) + assert body["title"].endswith("的实例已完成") and "状态：已完成" in body["text"] + + +async def test_a_self_declared_initiator_does_not_reach_a_personal_channel(received): + """关登录时署名是自己填的，谁都能填别人的名字：不按它推给那个人的个人渠道。""" + app_id, _ = await _app([]) + name = f"vic{uuid.uuid4().hex[:6]}" + mine = await _personal(name, ["run.finished"]) + run_id = await _finished_instance(app_id, started_by=name, verified=False) + assert await service.notify_run(run_id, "run.finished") == 0 + assert await _deliveries(mine) == [] and received == [] diff --git a/backend/tests/test_process_actions.py b/backend/tests/test_process_actions.py index f11ae634..b0d0e992 100644 --- a/backend/tests/test_process_actions.py +++ b/backend/tests/test_process_actions.py @@ -173,8 +173,8 @@ async def test_unknown_result_with_lookup_is_reconciled_automatically_and_sent_o await run_manager.continue_failed(run_id) assert refused.value.code == "action_unresolved" - assert await actions.tick(utcnow()) == 0 # 还没到第一次查询的时候 - assert await actions.tick(utcnow() + timedelta(minutes=2)) == 1 + assert rec.id not in await actions.tick(utcnow()) # 还没到第一次查询的时候 + assert rec.id in await actions.tick(utcnow() + timedelta(minutes=2)) final = await _wait(run_id, ("succeeded", "failed")) assert final.status == "succeeded", final.error assert len(erp.writes()) == 1 and len(erp.lookups) == 1 @@ -410,7 +410,7 @@ async def test_a_changed_params_schema_does_not_fail_a_write_that_already_went_o with pytest.raises(actions.ActionPending): await actions.prepare(run_id=run_id, node_id="pay", exec_no=1, capability=caps[PAY], args=args) - assert await actions.tick(utcnow() + timedelta(minutes=2)) == 1 + assert rec.id in await actions.tick(utcnow() + timedelta(minutes=2)) final = await _wait(run_id, ("succeeded", "failed")) assert final.status == "succeeded", final.error assert len(erp.writes()) == 1 @@ -457,9 +457,15 @@ async def test_a_rejected_confirmation_reads_as_a_decision_not_a_fault(erp, capa run_id = await _formal(pay_graph()) task = await _task(run_id, "action_confirm") await tasks.respond(task.id, ANN, decision="reject", note="金额不对") - final = await _wait(run_id, ("succeeded", "failed")) - # 引擎里驳回就是节点失败、运行结束（没有可走的出口）；说法写成未确认（被驳回），实例页按人的决定说 - assert final.status == "failed" and final.error.startswith("「pay」未确认（被驳回），没有发送：金额不对") + final = await _wait(run_id, ("succeeded", "failed", "cancelled")) + # 驳回是人的决定，不是故障（05-已知遗留 (a)）：运行收成 cancelled + end_reason=rejected，原话照旧写未确认（被驳回）， + # run.cancelled 带上在哪一步、谁、备注；实例页按人的决定说 + assert (final.status, final.end_reason) == ("cancelled", "rejected") + assert final.error.startswith("「pay」未确认（被驳回），没有发送：金额不对") + [ended] = await _events(run_id, "run.cancelled") + assert {k: ended.data[k] for k in ("reason", "node_id", "kind", "decision", "note")} == { + "reason": "rejected", "node_id": "pay", "kind": "action_confirm", "decision": "reject", "note": "金额不对"} + assert not await _events(run_id, "run.failed") view = await _instance_view(run_id) assert (view["hdr"]["status_code"], view["hdr"]["status_text"]) == ("cancelled", "「pay」未确认（被驳回）") step = next(s for s in view["tl"]["steps"] if s["node"] == "pay") @@ -481,7 +487,7 @@ async def test_unknown_then_reconciled_reads_as_waiting_then_a_footnote(erp, cap [row] = view["act"]["records"] assert row["state"] == "unknown" and "error" not in row and "ReadTimeout" in row["note"] - assert await actions.tick(utcnow() + timedelta(minutes=2)) == 1 + assert row["id"] in await actions.tick(utcnow() + timedelta(minutes=2)) final = await _wait(run_id, ("succeeded", "failed")) assert final.status == "succeeded", final.error view = await _instance_view(run_id) diff --git a/backend/tests/test_process_tasks.py b/backend/tests/test_process_tasks.py index 3bb85db9..b1f875fc 100644 --- a/backend/tests/test_process_tasks.py +++ b/backend/tests/test_process_tasks.py @@ -271,7 +271,24 @@ async def test_login_on_only_assignees_can_respond(): # -------------------------------------------------------------------------- -async def test_escalation_adds_the_director_whose_answer_decides(): +@pytest.fixture +async def stray_due_task(): + """别的测试留下的、早就到期、可升级的待办（-n 8 下同一个库里常有）。tick 扫全库，按条数断言的测试会被它带偏： + 这里先放一条，测试只断言自己造的那条在不在 tick 交回的 id 里（05-已知遗留 (f)）。""" + async with SessionLocal() as session: + run = Run(workflow_name="别的测试留下的", status="interrupted", graph={}, input={}) + session.add(run) + await session.flush() + task = Task(run_id=run.id, node_id="x", kind="approve", title="残留", due_at=utcnow() - timedelta(days=1), + on_timeout={"action": "escalate", "to_slot": "nobody"}) + session.add(task) + await session.commit() + yield task.id + await session.execute(delete(Run).where(Run.id == run.id)) + await session.commit() + + +async def test_escalation_adds_the_director_whose_answer_decides(stray_due_task): a, b, g, boss = f"a{_u()}", f"b{_u()}", f"finance{_u()}", f"boss{_u()}" await _people(a, b, group=g) run_id = await _app_run( @@ -280,9 +297,9 @@ async def test_escalation_adds_the_director_whose_answer_decides(): {"finance": [{"kind": "group", "name": g}], "director": [{"kind": "user", "name": boss}]}) task = await _task(run_id) assert task.due_at is not None - assert await tasks.tick(utcnow()) == 0 # 还没到时限 - assert await tasks.tick(utcnow() + timedelta(hours=2)) == 1 - assert await tasks.tick(utcnow() + timedelta(hours=3)) == 0 # 只升一次 + assert task.id not in await tasks.tick(utcnow()) # 还没到时限 + assert task.id in await tasks.tick(utcnow() + timedelta(hours=2)) + assert task.id not in await tasks.tick(utcnow() + timedelta(hours=3)) # 只升一次 detail = await tasks.detail(task.id, off(None)) assert detail["escalation_level"] == 1 and detail["status"] == "pending" assert ("user", boss, "escalation") in [(x["principal_kind"], x["principal"], x["source"]) @@ -293,13 +310,13 @@ async def test_escalation_adds_the_director_whose_answer_decides(): assert (await _wait(run_id, ("succeeded", "failed"))).output.get("结论") == "通过" -async def test_timeout_reject_never_approves(): +async def test_timeout_reject_never_approves(stray_due_task): run = await run_manager.start( graph=approval_flow(due="PT30M", on_timeout={"action": "reject"}), input_payload={"q": "x"}) await _wait(run.id, ("interrupted",)) await run_manager.wait_idle(run.id) task = await _task(run.id) - assert await tasks.tick(utcnow() + timedelta(hours=1)) == 1 + assert task.id in await tasks.tick(utcnow() + timedelta(hours=1)) final = await _wait(run.id, ("succeeded", "failed")) assert final.output.get("结论") == "退回" assert len(await _events(run.id, "task.expired")) == 1 @@ -308,16 +325,16 @@ async def test_timeout_reject_never_approves(): assert row.status == "resolved" and row.outcome["decision"] == "expired_reject" -async def test_remind_repeats_every_interval(): +async def test_remind_repeats_every_interval(stray_due_task): run = await run_manager.start( graph=approval_flow(due="PT1H", on_timeout={"action": "remind", "every": "PT1H"}), input_payload={"q": "x"}) await _wait(run.id, ("interrupted",)) await run_manager.wait_idle(run.id) task = await _task(run.id) later = utcnow() + timedelta(hours=1, minutes=1) - assert await tasks.tick(later) == 1 - assert await tasks.tick(later) == 0 - assert await tasks.tick(later + timedelta(hours=1)) == 1 + assert task.id in await tasks.tick(later) + assert task.id not in await tasks.tick(later) + assert task.id in await tasks.tick(later + timedelta(hours=1)) assert len(await _events(run.id, "task.reminded")) == 2 async with SessionLocal() as session: assert (await session.get(Task, task.id)).status == "pending" diff --git a/backend/tests/test_publish_plan.py b/backend/tests/test_publish_plan.py index e38b8d97..00afa3ea 100644 --- a/backend/tests/test_publish_plan.py +++ b/backend/tests/test_publish_plan.py @@ -186,6 +186,30 @@ async def test_a_failed_trial_blocks_the_plan(client, tmp_path): assert "plan.untrialed" not in codes +async def test_a_rejected_trial_is_a_warning_not_a_failure(client, tmp_path): + """最近一次试运行在某一步被驳回（cancelled + end_reason=rejected）：那是人的决定，不说成「试运行失败了」、不挡发布， + 照实提醒没走完（05-已知遗留 (a)）。""" + from sqlalchemy import update + + bp = await _board_from_run(client, tmp_path) + async with SessionLocal() as session: + trial = await session.get(Run, bp["last_trial_run_id"]) + node_id = trial.graph["nodes"][-1]["id"] + label = trial.graph["nodes"][-1]["data"]["label"] + await session.execute(update(Run).where(Run.id == trial.id).values( + status="cancelled", end_reason="rejected", error_node_id=node_id, error="人工驳回：不对")) + await session.commit() + plan = (await client.post(f"/api/blueprints/{bp['id']}/publish-plan", json={"level": "published"})).json() + codes = {p["code"]: p for p in plan["problems"]} + assert "plan.trial_failed" not in codes and "plan.untrialed" not in codes + rejected = codes["plan.trial_rejected"] + assert rejected["level"] == "warning" and rejected["message"] == f"最近一次试运行在「{label}」被驳回、没有走完" + assert rejected["plain"]["title"] == "最近一次试运行被驳回、没有走完" + assert plan["ok"] is True, [p["code"] for p in plan["problems"] if p["level"] == "error"] + summary = {x["step"]: x["text"] for x in plan["summary"]} + assert "被驳回、没有走完" in summary["publish_workflow"] + + async def test_a_process_with_an_unbound_step_stops_at_preflight(client): wf_id = await s.published_workflow(s.expense_graph(), f"报销-{uuid.uuid4().hex[:4]}", description="报销审批") bp = (await client.post("/api/blueprints", json={"source": {"kind": "workflow_version", "id": wf_id}, diff --git a/backend/tests/test_records.py b/backend/tests/test_records.py index 88f435aa..c051c126 100644 --- a/backend/tests/test_records.py +++ b/backend/tests/test_records.py @@ -461,7 +461,7 @@ async def test_unknown_result_is_found_by_the_lookup_and_not_written_again(engin if (await session.get(ActionRecord, rec.id)).next_lookup_at is not None: break await asyncio.sleep(0.05) - assert await actions.tick(utcnow() + timedelta(minutes=5)) == 1 + assert rec.id in await actions.tick(utcnow() + timedelta(minutes=5)) done = await wait(run.id, ("succeeded", "failed")) assert done.status == "succeeded", done.error async with SessionLocal() as session: diff --git a/backend/tests/test_run_rejected.py b/backend/tests/test_run_rejected.py new file mode 100644 index 00000000..520d2d4f --- /dev/null +++ b/backend/tests/test_run_rejected.py @@ -0,0 +1,214 @@ +"""驳回 ≠ 失败（docs/design/next-round/05-已知遗留.md (a)、(h)；总方案 X9）。 + +人的驳回（工具调用审批、代码执行审批、审批设了驳回即终止、填写 / 编辑被驳回、动作确认、双盲裁决）以前在引擎里抛普通的 +NodeError，运行记成 failed：通知推「实例失败」，定时推「运行失败」，记录页给「继续运行」——接着跑只会读到同一个驳回答复、 +再停一次。现在统一抛 NodeRejected，runner 收成 cancelled + end_reason=rejected，run.cancelled 带上在哪一步、哪种关卡、 +结论、谁、备注；原话（Run.error）一字不变。手动停止记 end_reason=stopped。被驳回的不能接着跑，存量的 failed+驳回也不能。 + +动作确认、双盲裁决的驳回在 test_process_actions.py、test_dual_check_node.py 里（要假的外部系统、口径卡）。 +""" +from __future__ import annotations + +import asyncio +import uuid +from datetime import timedelta + +import pytest +from httpx import ASGITransport, AsyncClient +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() + + +@pytest.fixture +async def client(): + from app.main import app + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +def node(nid, ntype, label=None, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}} + + +def chain(*nodes): + return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +START = node("start", "input", fields=[{"name": "q"}]) +OUT = node("out", "output", fields=[{"name": "结果", "value": "到了"}]) + +#: (关卡, 那一步, 驳回之后 Run.error 的开头) +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="可以吗", stop_on_reject=True), + "人工驳回：不对"), + "edit": (node("gate", "human", label="改稿", mode="edit", title="改一下", draft="初稿"), "人工驳回：不对"), + "input": (node("gate", "human", label="补充", mode="input", title="补一下"), "人工驳回：不对"), +} + + +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: + 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 _rejected_run(kind: str, *, note: str = "不对", **reply) -> str: + gate, _ = GATES[kind] + 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": note, **reply}, actor="张工") + return run.id + + +@pytest.mark.parametrize("kind", list(GATES)) +async def test_a_rejection_ends_the_run_as_cancelled_with_who_and_why(kind): + run_id = await _rejected_run(kind) + final = await _wait(run_id, ("cancelled", "failed", "succeeded")) + assert (final.status, final.end_reason) == ("cancelled", "rejected"), final.error + # 原话不变：给人看的那一句照旧 + assert final.error == GATES[kind][1] and final.error_node_id == "gate" and final.finished_at is not None + assert not await _events(run_id, "run.failed"), "驳回不是失败" + [cancelled] = await _events(run_id, "run.cancelled") + data = {k: v for k, v in cancelled.data.items() if k != "timing"} + assert data == {"reason": "rejected", "node_id": "gate", "label": GATES[kind][0]["data"]["label"], "kind": kind, + "decision": "reject", "actor": "张工", "note": "不对"} + # 收尾和取消一样：封存了，在线的流收到终态 + async with SessionLocal() as session: + row = await session.get(Run, run_id) + assert row.manifest_hash and row.manifest_seq == cancelled.seq + + +async def test_a_timeout_rejection_is_recorded_as_expired_reject(): + from app.process import tasks + + gate = node("gate", "human", label="把关", mode="approve", title="可以吗", stop_on_reject=True, due="PT30M", + on_timeout={"action": "reject"}) + 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) + 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, final.end_reason) == ("cancelled", "rejected") + [cancelled] = await _events(run.id, "run.cancelled") + assert cancelled.data["decision"] == "expired_reject" and cancelled.data["actor"] is None + + +async def test_the_api_carries_end_reason_and_refuses_to_continue_a_rejected_run(client): + run_id = await _rejected_run("approve") + await _wait(run_id, ("cancelled",)) + body = (await client.get(f"/api/runs/{run_id}")).json() + assert (body["status"], body["end_reason"]) == ("cancelled", "rejected") + r = await client.post(f"/api/runs/{run_id}/continue", json={}) + assert r.status_code == 409 and r.json()["code"] == "run_rejected" + assert r.json()["detail"] == "被驳回的运行不能接着跑：改好内容后重新发起" + stop = await client.post(f"/api/runs/{run_id}/cancel") + assert stop.status_code == 409 and "被驳回" in stop.json()["detail"] + + +async def test_a_legacy_failed_rejection_cannot_be_continued_either(client): + """加 end_reason 之前被驳回的运行记的是 failed（封存不改写）：失败的那一步上最近一条处理完的待办是驳回，也不许接着跑。 + 别的失败照旧能接着跑（这里断点不在，说的是没有断点，不是被驳回）。""" + async with SessionLocal() as session: + old = Run(workflow_name="老的驳回", status="failed", graph=chain(START, GATES["approve"][0], OUT), input={}, + error="人工驳回：不对", error_node_id="gate") + plain = Run(workflow_name="老的失败", status="failed", graph=chain(START, OUT), input={}, error="出错了") + session.add_all([old, plain]) + await session.flush() + session.add_all([ + RunEvent(run_id=old.id, seq=1, type="node.failed", node_id="gate", ts=1.0, data={"error": "人工驳回：不对"}), + Task(run_id=old.id, node_id="gate", kind="approve", title="可以吗", status="resolved", + outcome={"decision": "reject"}), + RunEvent(run_id=plain.id, seq=1, type="node.failed", node_id="out", ts=1.0, data={"error": "出错了"}), + ]) + await session.commit() + old_id, plain_id = old.id, plain.id + r = await client.post(f"/api/runs/{old_id}/continue", json={}) + assert r.status_code == 409 and r.json()["code"] == "run_rejected", r.text + other = await client.post(f"/api/runs/{plain_id}/continue", json={}) + assert other.status_code == 409 and other.json().get("code") != "run_rejected" + + +async def test_instance_state_says_the_same_for_the_new_shape_and_the_legacy_one(): + from app.apps.resolve import _trails, instance_state + + run_id = await _rejected_run("approve") + await _wait(run_id, ("cancelled",)) + async with SessionLocal() as session: + new = await session.get(Run, run_id) + legacy = Run(workflow_name="老的驳回", status="failed", graph=new.graph, input={}, error=new.error, + error_node_id="gate") + session.add(legacy) + await session.flush() + session.add_all([ + RunEvent(run_id=legacy.id, seq=1, type="node.failed", node_id="gate", ts=1.0, data={}), + Task(run_id=legacy.id, node_id="gate", kind="approve", title="可以吗", status="resolved", + outcome={"decision": "reject"}), + ]) + await session.commit() + trails = await _trails(session, [new, legacy]) + said = {"code": "cancelled", "text": "「把关」被驳回"} + assert instance_state(new, trails[new.id]) == said + assert instance_state(legacy, trails[legacy.id]) == said + + +async def test_stopping_is_recorded_as_stopped(): + gate, _ = GATES["approve"] + 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) + assert await run_manager.cancel(run.id, actor="李工") == "cancelled" + final = await _wait(run.id, ("cancelled",)) + assert final.end_reason == "stopped" + [cancelled] = await _events(run.id, "run.cancelled") + assert cancelled.data["reason"] == "stopped" and cancelled.data["actor"] == "李工" + + +async def test_a_run_still_queued_at_restart_gets_its_run_failed_event(): + """(h)：重启时排队中的运行改成 failed，以前只改状态，时间线里没有失败原因这一行。""" + from app.core.bus import bus + + async with SessionLocal() as session: + run = Run(workflow_name=f"排队-{uuid.uuid4().hex[:6]}", status="queued", graph={}, input={}, last_seq=2) + session.add(run) + await session.commit() + run_id = run.id + await run_manager.shutdown() + bus._seq.pop(run_id, None) + await run_manager.setup() + async with SessionLocal() as session: + row = await session.get(Run, run_id) + 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} diff --git a/backend/tests/test_scheduler_webhook.py b/backend/tests/test_scheduler_webhook.py index 635a59ab..0b6207f4 100644 --- a/backend/tests/test_scheduler_webhook.py +++ b/backend/tests/test_scheduler_webhook.py @@ -426,11 +426,33 @@ async def test_without_webhook_old_fires_are_not_pushed_later(recorder): async def test_cancelled_run_is_not_pushed(recorder): - _, fire_id, _ = await make(run_status="cancelled") + _, fire_id, run_id = await make(run_status="cancelled") + await set_run(run_id, end_reason="stopped") await webhook.notify_due() assert recorder.requests == [] and (await fire_row(fire_id)).notified_status == "cancelled" +async def test_a_rejected_run_is_pushed_as_not_issued(recorder): + """被人驳回的那一期（cancelled + end_reason=rejected）：按「运行结束」推「未出具（被驳回）」，写驳回的那一步，不写驳回的 + 备注和原话（可能带着数）；订阅「运行失败」的不收（05-已知遗留 (a)）。""" + _, fire_id, run_id = await make(run_status="cancelled", output={}, fmt="wecom", url=WECOM) + await set_run(run_id, end_reason="rejected", error_node_id="pay", error="「付款」未确认（被驳回），没有发送：金额 0042 不对", + graph={"nodes": [{"id": "pay", "type": "tool", "data": {"label": "付款", "config": {}}}], + "edges": []}) + assert await webhook.notify_due() == 1 + content = json.loads(recorder.requests[0].content)["markdown"]["content"] + assert "定时任务「门店周报」未出具（被驳回）" in content and '状态：被驳回' in content + assert "驳回节点：「付款」" in content and "失败" not in content + assert "0042" not in content and "金额" not in content + assert (await fire_row(fire_id)).notified_status == "cancelled" + [delivery] = await deliveries(fire_id) + assert delivery.event == "finished" + + _, only_failed, other = await make(run_status="cancelled", output={}, events=("failed",)) + await set_run(other, end_reason="rejected") + assert await webhook.notify_due() == 0 and (await fire_row(only_failed)).notified_status == "cancelled" + + async def test_waits_for_the_seal_before_pushing(recorder): _, fire_id, run_id = await make(sealed=False) assert await webhook.notify_due() == 0 and (await fire_row(fire_id)).notified_status is None