diff --git a/crates/df-workflow/src/executor.rs b/crates/df-workflow/src/executor.rs
index 2f2c928..217d9b3 100644
--- a/crates/df-workflow/src/executor.rs
+++ b/crates/df-workflow/src/executor.rs
@@ -206,308 +206,5 @@ impl DagExecutor {
 }
 
 #[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::node::{Node, NodeResult, NodeSchema};
-    use async_trait::async_trait;
-    use std::time::Duration;
-
-    /// 测试节点：sleep 指定毫秒后返回空输出
-    struct SleepNode {
-        sleep_ms: u64,
-    }
-
-    #[async_trait]
-    impl Node for SleepNode {
-        async fn execute(&self, _ctx: NodeContext) -> NodeResult {
-            tokio::time::sleep(Duration::from_millis(self.sleep_ms)).await;
-            Ok(NodeOutput::empty())
-        }
-
-        fn schema(&self) -> NodeSchema {
-            NodeSchema {
-                params: serde_json::Value::Null,
-                output: serde_json::Value::Null,
-            }
-        }
-
-        fn node_type(&self) -> &str {
-            "sleep"
-        }
-    }
-
-    /// 测试节点：直接返回错误
-    struct FailNode;
-
-    /// 测试节点：把 ctx.config 的指定 key 字符串原样回显到 output（验证节点级 config 下沉）
-    struct EchoConfigNode {
-        key: String,
-    }
-
-    #[async_trait]
-    impl Node for EchoConfigNode {
-        async fn execute(&self, ctx: NodeContext) -> NodeResult {
-            let v = ctx
-                .config
-                .get(&self.key)
-                .and_then(|v| v.as_str())
-                .unwrap_or("")
-                .to_string();
-            Ok(NodeOutput::from_value(serde_json::json!({ "echo": v })))
-        }
-
-        fn schema(&self) -> NodeSchema {
-            NodeSchema {
-                params: serde_json::Value::Null,
-                output: serde_json::Value::Null,
-            }
-        }
-
-        fn node_type(&self) -> &str {
-            "echo_config"
-        }
-    }
-
-    #[async_trait]
-    impl Node for FailNode {
-        async fn execute(&self, _ctx: NodeContext) -> NodeResult {
-            Err(anyhow::anyhow!("故意失败"))
-        }
-
-        fn schema(&self) -> NodeSchema {
-            NodeSchema {
-                params: serde_json::Value::Null,
-                output: serde_json::Value::Null,
-            }
-        }
-
-        fn node_type(&self) -> &str {
-            "fail"
-        }
-    }
-
-    /// ④-1:节点级 config 下沉验证。
-    /// DagDef 节点 config={foo:"bar"} + 全局 config={foo:"GLOBAL"} → NodeContext.config.foo=="bar"
-    /// （节点级覆盖全局级）。同时验证缺节点 config 的节点回退全局 config（零行为破坏现有调用方）。
-    #[tokio::test]
-    async fn node_config_overrides_global_in_node_context() {
-        use crate::registry::NodeRegistry;
-
-        // 构造 DagDef：n1 节点写 config={foo:"bar"}；n2 节点 config 空（验证回退全局）
-        let mut def = crate::dag_def::DagDef::new();
-        def.add_node("n1".to_string(), "echo".to_string(), serde_json::json!({ "foo": "bar" }));
-        def.add_node("n2".to_string(), "echo".to_string(), serde_json::json!({}));
-
-        // 注册 echo 工厂 → EchoConfigNode（读 config.foo）
-        let mut registry = NodeRegistry::new();
-        registry.register("echo", |_cfg| {
-            Box::new(EchoConfigNode { key: "foo".to_string() })
-        });
-
-        let dag = registry.build_dag(&def).expect("build_dag");
-
-        let mut executor = DagExecutor::new(EventBus::new(), "test-nodecfg".to_string());
-        // 全局 config 写 foo=GLOBAL（验证被 n1 节点级覆盖，n2 回退用全局）
-        let outputs = executor
-            .run(&dag, serde_json::json!({ "foo": "GLOBAL" }))
-            .await
-            .expect("run");
-
-        // n1:节点级 foo="bar" 覆盖全局 "GLOBAL"
-        assert_eq!(
-            outputs["n1"].data["echo"],
-            serde_json::json!("bar"),
-            "节点级 config 应覆盖全局"
-        );
-        // n2:节点 config 空 → 回退全局 foo="GLOBAL"（零行为破坏现有调用方）
-        assert_eq!(
-            outputs["n2"].data["echo"],
-            serde_json::json!("GLOBAL"),
-            "空节点 config 应回退全局"
-        );
-    }
-
-    #[tokio::test]
-    async fn test_same_layer_runs_in_parallel() {
-        // 两个无依赖节点位于同一层，各 sleep 100ms
-        let mut dag = Dag::new();
-        dag.add_node("a".to_string(), Box::new(SleepNode { sleep_ms: 100 }));
-        dag.add_node("b".to_string(), Box::new(SleepNode { sleep_ms: 100 }));
-
-        let mut executor = DagExecutor::new(EventBus::new(), "test-exec".to_string());
-        let start = std::time::Instant::now();
-        let outputs = executor.run(&dag, serde_json::Value::Null).await.unwrap();
-        let elapsed = start.elapsed();
-
-        assert_eq!(outputs.len(), 2);
-        // 串行需要约 200ms，并行应明显小于 180ms
-        assert!(
-            elapsed < Duration::from_millis(180),
-            "同层节点应并行执行，实际耗时 {:?}",
-            elapsed
-        );
-    }
-
-    #[tokio::test]
-    async fn test_layer_failure_aborts_following_layers() {
-        // a(失败) 与 b(成功) 同层，c 依赖 a，失败后 c 不应执行
-        let mut dag = Dag::new();
-        dag.add_node("a".to_string(), Box::new(FailNode));
-        dag.add_node("b".to_string(), Box::new(SleepNode { sleep_ms: 10 }));
-        dag.add_node("c".to_string(), Box::new(SleepNode { sleep_ms: 10 }));
-        dag.add_edge("a".to_string(), "c".to_string());
-
-        let mut executor = DagExecutor::new(EventBus::new(), "test-exec".to_string());
-        let err = executor
-            .run(&dag, serde_json::Value::Null)
-            .await
-            .unwrap_err();
-        assert!(err.to_string().contains("节点 a 执行失败"));
-
-        // 同层成功节点状态正常更新，下游节点保持 Pending
-        use df_types::types::NodeStatus;
-        assert_eq!(executor.state_machine.get(&"a".to_string()), NodeStatus::Failed);
-        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_types::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
-        );
-    }
-
-    /// 测试节点:执行中通过共享 node_status 自取消(模拟外部 IPC set_cancelled),返回 Ok
-    /// 模拟 HumanNode select! 收合法 HumanApprovalResponse 后 return Ok,但 join_all 让出窗口内
-    /// cancel_workflow_node IPC 把节点改 Cancelled 的 TOCTOU 场景。
-    struct CancelSelfThenOkNode;
-
-    #[async_trait]
-    impl Node for CancelSelfThenOkNode {
-        async fn execute(&self, ctx: NodeContext) -> NodeResult {
-            ctx.node_status.set_cancelled(ctx.node_id.clone());
-            Ok(NodeOutput::empty())
-        }
-
-        fn schema(&self) -> NodeSchema {
-            NodeSchema {
-                params: serde_json::Value::Null,
-                output: serde_json::Value::Null,
-            }
-        }
-
-        fn node_type(&self) -> &str {
-            "cancel_self_then_ok"
-        }
-    }
-
-    /// R-P1-3:Ok 后取消的 TOCTOU 防护。节点返回 Ok 但执行期间已被 set_cancelled,
-    /// executor Ok 分支必须跳过 set_completed(transition Cancelled→Completed 非法会 bail),
-    /// 状态保持 Cancelled,run 返回 Ok(已批准审批不应误报失败)。
-    #[tokio::test]
-    async fn test_cancelled_node_skips_set_completed() {
-        use df_types::types::NodeStatus;
-        let mut dag = Dag::new();
-        dag.add_node("y".to_string(), Box::new(CancelSelfThenOkNode));
-
-        let mut executor = DagExecutor::new(EventBus::new(), "test-cancel-ok".to_string());
-        let result = executor.run(&dag, serde_json::Value::Null).await;
-
-        // run 返回 Ok —— Ok 节点不应因 TOCTOU 取消被误判为失败
-        assert!(
-            result.is_ok(),
-            "Ok 后取消应短路 set_completed 而非 bail,实际 err: {:?}",
-            result.err()
-        );
-        // 状态保持 Cancelled(未被 set_completed 覆盖为 Completed,也不 bail)—— R-P1-3 修法核心验证
-        assert_eq!(
-            executor.state_machine.get(&"y".to_string()),
-            NodeStatus::Cancelled
-        );
-    }
-
-    /// 复核-新③:取消节点(Err 路径)emit 的应是 NodeCancelled(非 NodeFailed),
-    /// 前端按 type 区分「取消」与「失败」。订阅事件总线收集所有事件断言。
-    #[tokio::test]
-    async fn test_cancelled_node_emits_node_cancelled_event() {
-        use df_types::events::WorkflowEvent;
-        use df_types::types::NodeStatus;
-
-        let mut dag = Dag::new();
-        dag.add_node("c".to_string(), Box::new(CancelSelfNode));
-
-        let bus = EventBus::new();
-        let mut rx = bus.subscribe();
-        let mut executor = DagExecutor::new(bus, "test-cancel-event".to_string());
-        let _ = executor.run(&dag, serde_json::Value::Null).await;
-
-        // 收集所有事件(run 已结束,事件总线无新事件)
-        let mut events: Vec<WorkflowEvent> = Vec::new();
-        while let Ok(ev) = rx.try_recv() {
-            events.push(ev);
-        }
-
-        // 必含 NodeCancelled { node_id: "c" }
-        let cancelled = events.iter().any(|e| matches!(
-            e,
-            WorkflowEvent::NodeCancelled { node_id } if node_id == "c"
-        ));
-        assert!(cancelled, "取消节点应 emit NodeCancelled, 实际事件: {:?}", events);
-
-        // 不应含 node "c" 的 NodeFailed(失败事件)—— 语义双标修复的核心验证
-        let failed = events.iter().any(|e| matches!(
-            e,
-            WorkflowEvent::NodeFailed { node_id, .. } if node_id == "c"
-        ));
-        assert!(!failed, "取消节点不应 emit NodeFailed, 实际事件: {:?}", events);
-
-        // 状态保持 Cancelled
-        assert_eq!(
-            executor.state_machine.get(&"c".to_string()),
-            NodeStatus::Cancelled
-        );
-    }
-}
+#[path = "executor_helpers.rs"]
+mod tests;
diff --git a/crates/df-workflow/src/executor_helpers.rs b/crates/df-workflow/src/executor_helpers.rs
new file mode 100644
index 0000000..c4e92b6
--- /dev/null
+++ b/crates/df-workflow/src/executor_helpers.rs
@@ -0,0 +1,314 @@
+//! DagExecutor 测试套件 — 从 executor.rs 抽离(纯搬迁,零行为变更)
+//!
+//! 抽离说明:executor.rs 原 513 行中 tests mod 占 305 行(60%),主体 run 仅 157 行。
+//! 独立成文件降低阅读 run 主体的滚动噪音,测试逻辑、断言、节点 mock 全部原样搬迁。
+//! 通过 `#[path = "executor_helpers.rs"]` 内联回 executor.rs 的 #[cfg(test)] mod,
+//! 不进 lib.rs、不改 pub 路径、不影响外部调用方。
+//!
+//! SW-01 TOCTOU 相关测试(test_cancelled_node_skips_set_failed /
+//! test_cancelled_node_skips_set_completed / test_cancelled_node_emits_node_cancelled_event)
+//! 原样保留,锁定 executor.rs:136/138-153 的 Ok/Err 分支 is_cancelled 短路行为。
+
+use super::*;
+use crate::node::{Node, NodeResult, NodeSchema};
+use async_trait::async_trait;
+use std::time::Duration;
+
+/// 测试节点：sleep 指定毫秒后返回空输出
+struct SleepNode {
+    sleep_ms: u64,
+}
+
+#[async_trait]
+impl Node for SleepNode {
+    async fn execute(&self, _ctx: NodeContext) -> NodeResult {
+        tokio::time::sleep(Duration::from_millis(self.sleep_ms)).await;
+        Ok(NodeOutput::empty())
+    }
+
+    fn schema(&self) -> NodeSchema {
+        NodeSchema {
+            params: serde_json::Value::Null,
+            output: serde_json::Value::Null,
+        }
+    }
+
+    fn node_type(&self) -> &str {
+        "sleep"
+    }
+}
+
+/// 测试节点：直接返回错误
+struct FailNode;
+
+/// 测试节点：把 ctx.config 的指定 key 字符串原样回显到 output（验证节点级 config 下沉）
+struct EchoConfigNode {
+    key: String,
+}
+
+#[async_trait]
+impl Node for EchoConfigNode {
+    async fn execute(&self, ctx: NodeContext) -> NodeResult {
+        let v = ctx
+            .config
+            .get(&self.key)
+            .and_then(|v| v.as_str())
+            .unwrap_or("")
+            .to_string();
+        Ok(NodeOutput::from_value(serde_json::json!({ "echo": v })))
+    }
+
+    fn schema(&self) -> NodeSchema {
+        NodeSchema {
+            params: serde_json::Value::Null,
+            output: serde_json::Value::Null,
+        }
+    }
+
+    fn node_type(&self) -> &str {
+        "echo_config"
+    }
+}
+
+#[async_trait]
+impl Node for FailNode {
+    async fn execute(&self, _ctx: NodeContext) -> NodeResult {
+        Err(anyhow::anyhow!("故意失败"))
+    }
+
+    fn schema(&self) -> NodeSchema {
+        NodeSchema {
+            params: serde_json::Value::Null,
+            output: serde_json::Value::Null,
+        }
+    }
+
+    fn node_type(&self) -> &str {
+        "fail"
+    }
+}
+
+/// ④-1:节点级 config 下沉验证。
+/// DagDef 节点 config={foo:"bar"} + 全局 config={foo:"GLOBAL"} → NodeContext.config.foo=="bar"
+/// （节点级覆盖全局级）。同时验证缺节点 config 的节点回退全局 config（零行为破坏现有调用方）。
+#[tokio::test]
+async fn node_config_overrides_global_in_node_context() {
+    use crate::registry::NodeRegistry;
+
+    // 构造 DagDef：n1 节点写 config={foo:"bar"}；n2 节点 config 空（验证回退全局）
+    let mut def = crate::dag_def::DagDef::new();
+    def.add_node("n1".to_string(), "echo".to_string(), serde_json::json!({ "foo": "bar" }));
+    def.add_node("n2".to_string(), "echo".to_string(), serde_json::json!({}));
+
+    // 注册 echo 工厂 → EchoConfigNode（读 config.foo）
+    let mut registry = NodeRegistry::new();
+    registry.register("echo", |_cfg| {
+        Box::new(EchoConfigNode { key: "foo".to_string() })
+    });
+
+    let dag = registry.build_dag(&def).expect("build_dag");
+
+    let mut executor = DagExecutor::new(EventBus::new(), "test-nodecfg".to_string());
+    // 全局 config 写 foo=GLOBAL（验证被 n1 节点级覆盖，n2 回退用全局）
+    let outputs = executor
+        .run(&dag, serde_json::json!({ "foo": "GLOBAL" }))
+        .await
+        .expect("run");
+
+    // n1:节点级 foo="bar" 覆盖全局 "GLOBAL"
+    assert_eq!(
+        outputs["n1"].data["echo"],
+        serde_json::json!("bar"),
+        "节点级 config 应覆盖全局"
+    );
+    // n2:节点 config 空 → 回退全局 foo="GLOBAL"（零行为破坏现有调用方）
+    assert_eq!(
+        outputs["n2"].data["echo"],
+        serde_json::json!("GLOBAL"),
+        "空节点 config 应回退全局"
+    );
+}
+
+#[tokio::test]
+async fn test_same_layer_runs_in_parallel() {
+    // 两个无依赖节点位于同一层，各 sleep 100ms
+    let mut dag = Dag::new();
+    dag.add_node("a".to_string(), Box::new(SleepNode { sleep_ms: 100 }));
+    dag.add_node("b".to_string(), Box::new(SleepNode { sleep_ms: 100 }));
+
+    let mut executor = DagExecutor::new(EventBus::new(), "test-exec".to_string());
+    let start = std::time::Instant::now();
+    let outputs = executor.run(&dag, serde_json::Value::Null).await.unwrap();
+    let elapsed = start.elapsed();
+
+    assert_eq!(outputs.len(), 2);
+    // 串行需要约 200ms，并行应明显小于 180ms
+    assert!(
+        elapsed < Duration::from_millis(180),
+        "同层节点应并行执行，实际耗时 {:?}",
+        elapsed
+    );
+}
+
+#[tokio::test]
+async fn test_layer_failure_aborts_following_layers() {
+    // a(失败) 与 b(成功) 同层，c 依赖 a，失败后 c 不应执行
+    let mut dag = Dag::new();
+    dag.add_node("a".to_string(), Box::new(FailNode));
+    dag.add_node("b".to_string(), Box::new(SleepNode { sleep_ms: 10 }));
+    dag.add_node("c".to_string(), Box::new(SleepNode { sleep_ms: 10 }));
+    dag.add_edge("a".to_string(), "c".to_string());
+
+    let mut executor = DagExecutor::new(EventBus::new(), "test-exec".to_string());
+    let err = executor
+        .run(&dag, serde_json::Value::Null)
+        .await
+        .unwrap_err();
+    assert!(err.to_string().contains("节点 a 执行失败"));
+
+    // 同层成功节点状态正常更新，下游节点保持 Pending
+    use df_types::types::NodeStatus;
+    assert_eq!(executor.state_machine.get(&"a".to_string()), NodeStatus::Failed);
+    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_types::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
+    );
+}
+
+/// 测试节点:执行中通过共享 node_status 自取消(模拟外部 IPC set_cancelled),返回 Ok
+/// 模拟 HumanNode select! 收合法 HumanApprovalResponse 后 return Ok,但 join_all 让出窗口内
+/// cancel_workflow_node IPC 把节点改 Cancelled 的 TOCTOU 场景。
+struct CancelSelfThenOkNode;
+
+#[async_trait]
+impl Node for CancelSelfThenOkNode {
+    async fn execute(&self, ctx: NodeContext) -> NodeResult {
+        ctx.node_status.set_cancelled(ctx.node_id.clone());
+        Ok(NodeOutput::empty())
+    }
+
+    fn schema(&self) -> NodeSchema {
+        NodeSchema {
+            params: serde_json::Value::Null,
+            output: serde_json::Value::Null,
+        }
+    }
+
+    fn node_type(&self) -> &str {
+        "cancel_self_then_ok"
+    }
+}
+
+/// R-P1-3:Ok 后取消的 TOCTOU 防护。节点返回 Ok 但执行期间已被 set_cancelled,
+/// executor Ok 分支必须跳过 set_completed(transition Cancelled→Completed 非法会 bail),
+/// 状态保持 Cancelled,run 返回 Ok(已批准审批不应误报失败)。
+#[tokio::test]
+async fn test_cancelled_node_skips_set_completed() {
+    use df_types::types::NodeStatus;
+    let mut dag = Dag::new();
+    dag.add_node("y".to_string(), Box::new(CancelSelfThenOkNode));
+
+    let mut executor = DagExecutor::new(EventBus::new(), "test-cancel-ok".to_string());
+    let result = executor.run(&dag, serde_json::Value::Null).await;
+
+    // run 返回 Ok —— Ok 节点不应因 TOCTOU 取消被误判为失败
+    assert!(
+        result.is_ok(),
+        "Ok 后取消应短路 set_completed 而非 bail,实际 err: {:?}",
+        result.err()
+    );
+    // 状态保持 Cancelled(未被 set_completed 覆盖为 Completed,也不 bail)—— R-P1-3 修法核心验证
+    assert_eq!(
+        executor.state_machine.get(&"y".to_string()),
+        NodeStatus::Cancelled
+    );
+}
+
+/// 复核-新③:取消节点(Err 路径)emit 的应是 NodeCancelled(非 NodeFailed),
+/// 前端按 type 区分「取消」与「失败」。订阅事件总线收集所有事件断言。
+#[tokio::test]
+async fn test_cancelled_node_emits_node_cancelled_event() {
+    use df_types::events::WorkflowEvent;
+    use df_types::types::NodeStatus;
+
+    let mut dag = Dag::new();
+    dag.add_node("c".to_string(), Box::new(CancelSelfNode));
+
+    let bus = EventBus::new();
+    let mut rx = bus.subscribe();
+    let mut executor = DagExecutor::new(bus, "test-cancel-event".to_string());
+    let _ = executor.run(&dag, serde_json::Value::Null).await;
+
+    // 收集所有事件(run 已结束,事件总线无新事件)
+    let mut events: Vec<WorkflowEvent> = Vec::new();
+    while let Ok(ev) = rx.try_recv() {
+        events.push(ev);
+    }
+
+    // 必含 NodeCancelled { node_id: "c" }
+    let cancelled = events.iter().any(|e| matches!(
+        e,
+        WorkflowEvent::NodeCancelled { node_id } if node_id == "c"
+    ));
+    assert!(cancelled, "取消节点应 emit NodeCancelled, 实际事件: {:?}", events);
+
+    // 不应含 node "c" 的 NodeFailed(失败事件)—— 语义双标修复的核心验证
+    let failed = events.iter().any(|e| matches!(
+        e,
+        WorkflowEvent::NodeFailed { node_id, .. } if node_id == "c"
+    ));
+    assert!(!failed, "取消节点不应 emit NodeFailed, 实际事件: {:?}", events);
+
+    // 状态保持 Cancelled
+    assert_eq!(
+        executor.state_machine.get(&"c".to_string()),
+        NodeStatus::Cancelled
+    );
+}