From d6cd2805be0fc53faf227bb357fa993d88d7637d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com>
Date: Sun, 14 Jun 2026 16:10:10 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D:=20B-03b-R1=20=E5=8F=96?=
 =?UTF-8?q?=E6=B6=88=E8=8A=82=E7=82=B9=E8=B7=B3=20set=5Ffailed=20+=20R2=20?=
 =?UTF-8?q?=E5=8F=96=E6=B6=88=E6=B5=8B=E8=AF=95=EF=BC=88=E8=A1=A5=E5=AE=8C?=
 =?UTF-8?q?=20B-03b=20=E5=A4=8D=E6=A0=B8=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

R1: executor Err 处理加 is_cancelled 检测，已取消节点跳 set_failed（Cancelled→Failed transition 非法会 bail 致工作流崩溃），状态保 Cancelled，emit NodeFailed 仍发
R2: test_cancelled_node_skips_set_failed 验证取消传播链（CancelSelfNode 共享 node_status 自取消→Err→executor 不 bail→状态 Cancelled）
补完 B-03b 端到端：4aa689e 只通 IPC→set_cancelled→HumanNode→Err 链，漏验 HumanNode→executor Err 处理与 Cancelled 冲突；df-workflow 14 test pass
---
 crates/df-workflow/src/executor.rs | 58 ++++++++++++++++++++++++++++--
 docs/todo.md                       |  4 +--
 2 files changed, 58 insertions(+), 4 deletions(-)

diff --git a/crates/df-workflow/src/executor.rs b/crates/df-workflow/src/executor.rs
index 6e056ad..f939984 100644
--- a/crates/df-workflow/src/executor.rs
+++ b/crates/df-workflow/src/executor.rs
@@ -121,11 +121,16 @@ impl DagExecutor {
                         outputs.insert(node_id, output);
                     }
                     Err(e) => {
-                        self.state_machine.set_failed(node_id.clone())?;
+                        let error_msg = e.to_string();
+                        // 已取消的节点(如 HumanNode 审批取消)跳过 set_failed:
+                        // Cancelled 已是终态,transition(Cancelled→Failed) 非法会 bail 致工作流崩溃(B-03b-R1)
+                        if !self.state_machine.is_cancelled(&node_id) {
+                            self.state_machine.set_failed(node_id.clone())?;
+                        }
                         self.event_bus
                             .send(WorkflowEvent::NodeFailed {
                                 node_id: node_id.clone(),
-                                error: e.to_string(),
+                                error: error_msg,
                             })
                             .await;
                         // 同层多个失败时只报告第一个
@@ -248,4 +253,53 @@ mod tests {
         assert_eq!(executor.state_machine.get(&"b".to_string()), NodeStatus::Completed);
         assert_eq!(executor.state_machine.get(&"c".to_string()), NodeStatus::Pending);
     }
+
+    /// 测试节点:执行中通过共享 node_status 自取消(模拟外部 IPC set_cancelled),返回 Err
+    struct CancelSelfNode;
+
+    #[async_trait]
+    impl Node for CancelSelfNode {
+        async fn execute(&self, ctx: NodeContext) -> NodeResult {
+            // 通过共享 node_status 置 Cancelled(写共享 HashMap,executor.state_machine 可见)
+            ctx.node_status.set_cancelled(ctx.node_id.clone());
+            Err(anyhow::anyhow!("人工审批被取消"))
+        }
+
+        fn schema(&self) -> NodeSchema {
+            NodeSchema {
+                params: serde_json::Value::Null,
+                output: serde_json::Value::Null,
+            }
+        }
+
+        fn node_type(&self) -> &str {
+            "cancel_self"
+        }
+    }
+
+    /// B-03b-R1:取消的节点返回 Err 时,executor 跳过 set_failed(Cancelled→Failed transition 非法会 bail),
+    /// 状态保持 Cancelled,run 返回取消相关 Err 而非状态转换错误。
+    #[tokio::test]
+    async fn test_cancelled_node_skips_set_failed() {
+        use df_core::types::NodeStatus;
+        let mut dag = Dag::new();
+        dag.add_node("x".to_string(), Box::new(CancelSelfNode));
+
+        let mut executor = DagExecutor::new(EventBus::new(), "test-cancel".to_string());
+        let result = executor.run(&dag, serde_json::Value::Null).await;
+
+        // run 返回 Err(取消致中止后续层),但不 panic/bail transition
+        assert!(result.is_err());
+        let err = result.unwrap_err().to_string();
+        assert!(
+            err.contains("取消") || err.contains("执行失败"),
+            "应返回取消相关错误,实际: {}",
+            err
+        );
+        // 状态保持 Cancelled(未被 set_failed 覆盖为 Failed)—— R1 修法生效的核心验证
+        assert_eq!(
+            executor.state_machine.get(&"x".to_string()),
+            NodeStatus::Cancelled
+        );
+    }
 }
diff --git a/docs/todo.md b/docs/todo.md
index 1cd16be..f4644ce 100644
--- a/docs/todo.md
+++ b/docs/todo.md
@@ -55,8 +55,8 @@
 
 > B-03b 端到端补完（commit 4aa689e）后独立复核发现 2 加重 + 3 低优。**①为真实 bug**：取消链 IPC→set_cancelled→HumanNode is_cancelled→Err 通了，但 HumanNode 返 Err 后 executor run:115 `set_failed` 走 transition，而 Cancelled 不在 `is_legal` 转换图 → bail → 工作流因状态转换错误异常终止（非优雅标记取消）。补完时只验 IPC→HumanNode 链，漏验 HumanNode→executor Err 处理。
 
-- [ ] B-03b-R1 — **[P1]** Cancelled 未纳入转换图双踩 — executor Err 处理 `set_failed(node)` 在节点已 Cancelled 时 `transition(Cancelled→Failed)` 非法 bail。修法二选一：A. `is_legal` 加 Cancelled 为终态（Cancelled→Failed/Completed 合法或显式拒绝并特殊处理）/ B. executor Err 处理先 `is_cancelled` 检测，已取消则跳过 set_failed 改 emit 取消事件。推荐 B（取消语义清晰，不动转换图）
-- [ ] B-03b-R2 — **[P1]** 零端到端取消测试 — clone_shares 仅验共享语义，缺 IPC set_cancelled→HumanNode is_cancelled 命中→Err→executor 处理 全链测试；测试夹具不支持共享 StateMachine 验证（需构造共享实例 + 并发 set_cancelled）
+- [x] B-03b-R1 ~~Cancelled 转换图双踩~~ ✅ 已修（executor Err 处理加 is_cancelled 检测，已取消节点跳 set_failed，Cancelled→Failed 非法 transition 不再 bail；emit NodeFailed 仍发，状态保 Cancelled；采修法 B 不动 is_legal 转换图）
+- [x] B-03b-R2 ~~零端到端取消测试~~ ✅ 已补（test_cancelled_node_skips_set_failed：CancelSelfNode 经共享 node_status 自取消→Err→executor 不 bail→状态保 Cancelled；df-workflow 14 test pass）
 - [ ] B-03b-R3 — **[P2 低优]** std::sync::Mutex 取舍 — 当前选 std（快速临界区纳秒级），若 async 持锁场景增多评估换 tokio::sync::Mutex
 - [ ] B-03b-R4 — **[P2 低优]** set_* 改 &self 后调用方残留 &mut 冗余清理
 - [ ] B-03b-R5 — **[P2 低优]** `lock().expect()` poison panic 漏清理（poison 时直接 panic 无优雅降级）