新增: AI Chat多项增强(审批去重/编辑重发/导出/实体引用/会话置顶搜索)+任务推进链df-nodes落地

crates/df-workflow/src/dag.rs

@@ -22,6 +22,12 @@ pub struct Dag {
     pub nodes: HashMap<NodeId, Box<dyn Node>>,
     /// 边集合
     pub edges: Vec<Edge>,
+    /// 节点级配置（DagDef.nodes[id].config 提取自 build_dag，run 时与全局 config deep_merge）
+    ///
+    /// 语义：节点级覆盖全局级。DagExecutor 构造 NodeContext 时执行
+    /// `deep_merge(initial_config, node_configs[id])`，节点定义里写的同名 key 优先于
+    /// run_workflow 传入的全局 config。
+    pub node_configs: HashMap<NodeId, serde_json::Value>,
 }
 
 impl Dag {
@@ -30,6 +36,7 @@ impl Dag {
         Self {
             nodes: HashMap::new(),
             edges: Vec::new(),
+            node_configs: HashMap::new(),
         }
     }
 
@@ -38,6 +45,19 @@ impl Dag {
         self.nodes.insert(id, node);
     }
 
+    /// 添加节点（附带节点级配置）
+    ///
+    /// 节点级配置在 DagExecutor 构造 NodeContext 时覆盖全局 config（deep_merge）。
+    pub fn add_node_with_config(
+        &mut self,
+        id: NodeId,
+        node: Box<dyn Node>,
+        config: serde_json::Value,
+    ) {
+        self.nodes.insert(id.clone(), node);
+        self.node_configs.insert(id, config);
+    }
+
     /// 添加边
     pub fn add_edge(&mut self, source: NodeId, target: NodeId) {
         self.edges.push(Edge {
@@ -135,3 +155,85 @@ impl Default for Dag {
         Self::new()
     }
 }
+
+/// 深度合并两个 JSON 值：节点级 `node` 覆盖全局级 `global`。
+///
+/// 规则：
+/// - 两端均为 Object：递归合并（同 key 时节点级覆盖全局级，新增 key 各自保留）。
+/// - 非两端 Object：节点级直接覆盖全局级（包括 `Null`，符合 JSON 显式空语义）。
+/// - 节点级为 `Null` 的语义是「显式置空」，按非 Object 规则覆盖（直接返回 Null）；
+///   若调用方希望「跳过节点配置」，应在构造 Dag 时不写入该 key（node_configs 缺失即用全局）。
+///
+/// 用途：DagExecutor 构造 NodeContext 时 `deep_merge(initial_config, node_config)`，
+/// 让节点定义里写的 NodeDef.config 优先于 run_workflow 传入的全局 config。
+pub fn deep_merge(global: &serde_json::Value, node: &serde_json::Value) -> serde_json::Value {
+    use serde_json::Value;
+    match (global, node) {
+        (Value::Object(g), Value::Object(n)) => {
+            let mut merged = g.clone();
+            for (k, nv) in n {
+                // 同 key 递归（支持嵌套对象局部覆盖）；不同类型时递归退化为「节点级直接覆盖」
+                let gv = merged.get(k).cloned();
+                let merged_v = match gv {
+                    Some(gv) => deep_merge(&gv, nv),
+                    None => nv.clone(),
+                };
+                merged.insert(k.clone(), merged_v);
+            }
+            Value::Object(merged)
+        }
+        // 非两端 Object：节点级覆盖全局级（含 Null、标量、数组）
+        _ => node.clone(),
+    }
+}
+
+#[cfg(test)]
+mod deep_merge_tests {
+    use super::deep_merge;
+    use serde_json::json;
+
+    #[test]
+    fn node_overrides_global_scalar() {
+        assert_eq!(deep_merge(&json!({"a": 1}), &json!({"a": 2})), json!({"a": 2}));
+    }
+
+    #[test]
+    fn node_adds_new_key() {
+        assert_eq!(
+            deep_merge(&json!({"a": 1}), &json!({"b": 2})),
+            json!({"a": 1, "b": 2})
+        );
+    }
+
+    #[test]
+    fn nested_object_recursive_merge() {
+        assert_eq!(
+            deep_merge(&json!({"o": {"x": 1, "y": 2}}), &json!({"o": {"y": 9}})),
+            json!({"o": {"x": 1, "y": 9}})
+        );
+    }
+
+    #[test]
+    fn empty_node_config_is_noop() {
+        // 核心保现有调用方行为：节点 config 为空对象 → 结果等同全局（零行为破坏）
+        let global = json!({"title": "确认", "options": ["同意"]});
+        assert_eq!(deep_merge(&global, &json!({})), global);
+    }
+
+    #[test]
+    fn missing_node_config_treated_as_noop() {
+        // DagExecutor 对缺失 key 的节点不调 deep_merge（直接用 global），此处覆盖语义对照：
+        // 若强行把缺失视作 Null，应覆盖（显式空语义），与「缺失」不同（缺失走 if let Some 分支）。
+        let global = json!({"a": 1});
+        assert_eq!(deep_merge(&global, &json!(null)), json!(null));
+    }
+
+    #[test]
+    fn array_replaces_not_concat() {
+        // 非两端 Object（数组）→ 节点级直接覆盖，不拼接
+        assert_eq!(
+            deep_merge(&json!({"arr": [1, 2]}), &json!({"arr": [3]})),
+            json!({"arr": [3]})
+        );
+    }
+}

crates/df-workflow/src/executor.rs

@@ -99,7 +99,13 @@ impl DagExecutor {
                 let ctx = NodeContext {
                     node_id: node_id.clone(),
                     inputs,
-                    config: initial_config.clone(),
+                    // 节点级覆盖全局级：deep_merge(initial_config, node_configs[id])。
+                    // 节点 NodeDef.config 未定义（node_configs 缺该 key）→ 直接用全局 initial_config，
+                    // 与旧行为一致（零行为破坏现有 HumanNode/AiNode 等读全局 config 的调用方）。
+                    config: match dag.node_configs.get(node_id) {
+                        Some(node_cfg) => crate::dag::deep_merge(&initial_config, node_cfg),
+                        None => initial_config.clone(),
+                    },
                     execution_id: self.execution_id.clone(),
                     event_bus: self.event_bus.clone(),
                     // 共享执行器状态机，使节点（如 HumanNode）能读取真实状态而非空状态机
@@ -142,16 +148,24 @@ impl DagExecutor {
                         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) {
+                        if self.state_machine.is_cancelled(&node_id) {
+                            // emit NodeCancelled(非 NodeFailed):取消语义有别于失败,前端按 type 归「取消」非「失败」
+                            self.event_bus
+                                .send(WorkflowEvent::NodeCancelled {
+                                    node_id: node_id.clone(),
+                                })
+                                .await;
+                        } else {
                             self.state_machine.set_failed(node_id.clone())?;
+                            self.event_bus
+                                .send(WorkflowEvent::NodeFailed {
+                                    node_id: node_id.clone(),
+                                    error: error_msg,
+                                })
+                                .await;
                         }
-                        self.event_bus
-                            .send(WorkflowEvent::NodeFailed {
-                                node_id: node_id.clone(),
-                                error: error_msg,
-                            })
-                            .await;
-                        // 同层多个失败时只报告第一个
+                        // 取消与失败一致:节点返回 Err 即中止后续层(run 返回 Err);
+                        // 仅事件类型区分,工作流结果语义保持不变(零行为破坏)。
                         if first_err.is_none() {
                             first_err = Some(e.context(format!("节点 {} 执行失败", node_id)));
                         }
@@ -210,6 +224,35 @@ mod tests {
     /// 测试节点：直接返回错误
     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 {
@@ -228,6 +271,47 @@ mod tests {
         }
     }
 
+    /// ④-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
@@ -369,4 +453,46 @@ mod tests {
             NodeStatus::Cancelled
         );
     }
+
+    /// 复核-新③:取消节点(Err 路径)emit 的应是 NodeCancelled(非 NodeFailed),
+    /// 前端按 type 区分「取消」与「失败」。订阅事件总线收集所有事件断言。
+    #[tokio::test]
+    async fn test_cancelled_node_emits_node_cancelled_event() {
+        use df_core::events::WorkflowEvent;
+        use df_core::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
+        );
+    }
 }

crates/df-workflow/src/registry.rs

@@ -46,7 +46,10 @@ impl NodeRegistry {
         let mut node_ids: std::collections::HashSet<&String> = std::collections::HashSet::new();
         for (id, node_def) in &def.nodes {
             let node = self.create(&node_def.node_type, &node_def.config)?;
-            dag.add_node(id.clone(), node);
+            // 节点级 config 下沉到 Dag.node_configs：DagExecutor.run 构造 NodeContext 时
+            // deep_merge(initial_config, node_configs[id])（节点级覆盖全局级），解决此前
+            // initial_config.clone() 直接当 ctx.config、忽略 NodeDef.config 的问题。
+            dag.add_node_with_config(id.clone(), node, node_def.config.clone());
             node_ids.insert(id);
         }