新增: 工具工作流补全(diff_files工具 + 技能清单注入 + 工作流进度共享 + human端到端测试 + 知识库MCP工具)

2026-08-08 21:24:53 +08:00
parent 6ba6daf188
commit 9eb2995a74
14 changed files with 807 additions and 93 deletions
@@ -0,0 +1,235 @@
+//! human 节点端到端集成测试 — DagDef → NodeRegistry.build_dag → DagExecutor.run
+//!
+//! df-workflow 不依赖 df-nodes(反向依赖),此处用等价阻塞节点模拟 HumanNode 的
+//! 「发审批请求 → 挂起等待 → 外部 approve → 返回结果」链路,覆盖
+//! DagDef 序列化 → 注册表 → 执行器全链路 + 事件序列断言(R6 send 缺 await /
+//! R7 契约失配的存活土壤)。
+
+use std::time::Duration;
+
+use async_trait::async_trait;
+use serde_json::json;
+
+use df_types::events::{SelectType, WorkflowEvent};
+use df_types::types::NodeStatus;
+use df_workflow::dag_def::DagDef;
+use df_workflow::eventbus::{EventBus, EventSubscriber};
+use df_workflow::executor::DagExecutor;
+use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema};
+use df_workflow::registry::NodeRegistry;
+
+/// 前驱节点:sleep 后返回空输出(仿 executor 单测 SleepNode)。
+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: json!(null),
+            output: json!(null),
+        }
+    }
+
+    fn node_type(&self) -> &str {
+        "sleep"
+    }
+}
+
+/// 模拟 HumanNode 的阻塞审批节点:先 subscribe 再发 HumanApprovalRequest(broadcast 不回放),
+/// select! 等待匹配 execution_id + node_id 的 HumanApprovalResponse,收到则返回 decision。
+struct ApprovalNode;
+
+#[async_trait]
+impl Node for ApprovalNode {
+    async fn execute(&self, ctx: NodeContext) -> NodeResult {
+        let mut rx = ctx.event_bus.subscribe();
+        let _ = ctx
+            .event_bus
+            .send(WorkflowEvent::HumanApprovalRequest {
+                execution_id: ctx.execution_id.clone(),
+                node_id: ctx.node_id.clone(),
+                title: "确认发布".to_string(),
+                description: String::new(),
+                options: vec!["同意".into(), "拒绝".into()],
+                select_type: SelectType::Single,
+            })
+            .await;
+        let timeout_secs = ctx
+            .config
+            .get("timeout_secs")
+            .and_then(|v| v.as_u64())
+            .unwrap_or(2);
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs);
+        loop {
+            tokio::select! {
+                recv = rx.recv() => match recv {
+                    Ok(WorkflowEvent::HumanApprovalResponse {
+                        execution_id, node_id, decision, decisions, ..
+                    }) if execution_id == ctx.execution_id && node_id == ctx.node_id => {
+                        let primary = if decision.is_empty() {
+                            decisions.first().cloned().unwrap_or_default()
+                        } else {
+                            decision
+                        };
+                        return Ok(NodeOutput::from_value(json!({ "decision": primary })));
+                    }
+                    Ok(_) => continue,
+                    Err(_) => return Err(anyhow::anyhow!("事件总线关闭,审批无法完成")),
+                },
+                _ = tokio::time::sleep_until(deadline) => {
+                    return Err(anyhow::anyhow!("人工审批等待超时({timeout_secs}s)"));
+                }
+            }
+        }
+    }
+
+    fn schema(&self) -> NodeSchema {
+        NodeSchema {
+            params: json!(null),
+            output: json!(null),
+        }
+    }
+
+    fn is_blocking(&self) -> bool {
+        true
+    }
+
+    fn node_type(&self) -> &str {
+        "human"
+    }
+}
+
+/// 从事件流中等待一条 HumanApprovalRequest(跳过 NodeStarted/NodeCompleted 等其他事件)。
+async fn wait_approval_request(mut rx: EventSubscriber) -> WorkflowEvent {
+    tokio::time::timeout(Duration::from_millis(1000), async move {
+        loop {
+            if let Ok(ev @ WorkflowEvent::HumanApprovalRequest { .. }) = rx.recv().await {
+                return ev;
+            }
+        }
+    })
+    .await
+    .expect("应收到 HumanApprovalRequest,实际超时")
+}
+
+/// 端到端主链路:前驱 sleep(a) → human(b)。
+/// 执行到 human 后挂起等审批 → 外部 approve(模拟 approve_human_approval IPC) → 返回结果、工作流完成。
+#[tokio::test]
+async fn human_workflow_end_to_end_approval() {
+    let bus = EventBus::new();
+    let rx = bus.subscribe(); // 先 subscribe 再执行(broadcast 不回放)
+    let exec_id = "exec-e2e-approve";
+
+    // DagDef(可序列化定义)→ NodeRegistry.build_dag → 运行时 Dag
+    let mut def = DagDef::new();
+    def.add_node("a", "sleep", json!({}));
+    def.add_node("b", "human", json!({}));
+    def.add_edge("a", "b");
+    let mut registry = NodeRegistry::new();
+    registry.register("sleep", |_| Box::new(SleepNode { sleep_ms: 10 }));
+    registry.register("human", |_| Box::new(ApprovalNode));
+    let dag = registry.build_dag(&def).expect("build_dag 应成功");
+
+    let mut executor = DagExecutor::new(bus.clone(), exec_id.into());
+    let sm = executor.state_machine(); // 共享状态机(spawn 后仍可读)
+    let run_handle = tokio::spawn(async move { executor.run(&dag, json!({})).await });
+
+    // a 层完成、b 层 human subscribe + send Request 后应收到审批请求
+    let request = wait_approval_request(rx).await;
+    match request {
+        WorkflowEvent::HumanApprovalRequest {
+            node_id, title, options, ..
+        } => {
+            assert_eq!(node_id, "b");
+            assert_eq!(title, "确认发布");
+            assert_eq!(options, vec!["同意".to_string(), "拒绝".to_string()]);
+        }
+        other => panic!("期望 HumanApprovalRequest,收到 {:?}", other),
+    }
+
+    // 外部 approve(模拟 approve_human_approval IPC 发送 HumanApprovalResponse 到总线)
+    bus.send(WorkflowEvent::HumanApprovalResponse {
+        execution_id: exec_id.into(),
+        node_id: "b".to_string(),
+        decision: "同意".into(),
+        decisions: vec![],
+        comment: Some("可以发布".into()),
+    })
+    .await;
+
+    let outputs = run_handle
+        .await
+        .expect("run 不应 panic")
+        .expect("run 应成功");
+    assert!(outputs.contains_key("a"), "outputs 应含前驱节点 a");
+    assert_eq!(outputs["b"].data["decision"], json!("同意"));
+    assert_eq!(sm.get(&"a".to_string()), NodeStatus::Completed);
+    assert_eq!(sm.get(&"b".to_string()), NodeStatus::Completed);
+}
+
+/// 取消路径:外部 set_cancelled(模拟 cancel_workflow_node IPC 写共享状态机)
+/// → human select! 轮询 is_cancelled → Err → 状态保持 Cancelled。
+#[tokio::test]
+async fn human_workflow_cancel_via_shared_state_machine() {
+    let bus = EventBus::new();
+    let rx = bus.subscribe();
+    let exec_id = "exec-e2e-cancel";
+
+    let mut def = DagDef::new();
+    def.add_node("h", "human", json!({}));
+    let mut registry = NodeRegistry::new();
+    registry.register("human", |_| Box::new(ApprovalNode));
+    let dag = registry.build_dag(&def).expect("build_dag 应成功");
+
+    let mut executor = DagExecutor::new(bus.clone(), exec_id.into());
+    let sm = executor.state_machine();
+    let run_handle = tokio::spawn(async move { executor.run(&dag, json!({})).await });
+
+    // 等 human 发出审批请求(确认已 subscribe + send + 进入 select! 轮询)
+    let _ = wait_approval_request(rx).await;
+
+    // 外部 cancel:写共享状态机置 Cancelled,humar select! 的 cancel_tick 会读到
+    sm.set_cancelled("h".to_string());
+
+    let result = run_handle.await.unwrap();
+    assert!(result.is_err(), "取消应致 run 返回 Err");
+    let err = result.unwrap_err().to_string();
+    assert!(
+        err.contains("取消") || err.contains("执行失败"),
+        "应返回取消相关错误,实际: {err}"
+    );
+    // 状态保持 Cancelled(未被 set_failed 覆盖为 Failed)—— 取消终态不被失败语义覆盖
+    assert_eq!(sm.get(&"h".to_string()), NodeStatus::Cancelled);
+}
+
+/// 挂起等待路径:收到请求后不 approve → human 挂起直到超时 → run 返 Err。
+/// 证明节点真在等待外部响应(而非空转直接返回)。
+#[tokio::test]
+async fn human_workflow_waits_then_times_out_without_approve() {
+    let bus = EventBus::new();
+    let rx = bus.subscribe();
+    let exec_id = "exec-e2e-timeout";
+
+    let mut def = DagDef::new();
+    def.add_node("h", "human", json!({ "timeout_secs": 1 }));
+    let mut registry = NodeRegistry::new();
+    registry.register("human", |_| Box::new(ApprovalNode));
+    let dag = registry.build_dag(&def).expect("build_dag 应成功");
+
+    let mut executor = DagExecutor::new(bus.clone(), exec_id.into());
+    let run_handle = tokio::spawn(async move { executor.run(&dag, json!({})).await });
+
+    let _ = wait_approval_request(rx).await;
+    let result = run_handle.await.unwrap();
+    assert!(result.is_err(), "无审批响应应超时失败");
+    // executor 包了「节点 h 执行失败」context,用 {:#} 展开整条错误链断言根因
+    let chain = format!("{:#}", result.unwrap_err());
+    assert!(chain.contains("超时"), "应返回超时错误,实际: {chain}");
+}