驳回不再记成失败（后端）：人的驳回收成 cancelled + end_reason=rejected，消费方都读这一个字段

为什么：动作确认、工具 / 代码审批、审批驳回即终止、填写 / 编辑被驳回、双盲裁决的驳回以前抛普通 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

@@ -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

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":

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}")

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)
     # 不是一个是/否词的字符串（"拒绝，这个表是生产库"）既算不上批准，本身又

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 "")

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}

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"}
 

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），
     # 留空跟随整机默认。要的那档不可用时会安静退回默认，不阻断运行。

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):

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,

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(

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

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:

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)
 

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),
     )
 

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，另写升级给了谁、他一人说了算

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

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(

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):

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 == []

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)

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"

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},

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:

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}

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 '状态：<font color="warning">被驳回</font>' 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