From 22964a2e203b226baa203c47f49c5ea5624498b9 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 14:27:08 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D:=20=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E6=B5=81=E5=AE=A1=E6=89=B9=E9=97=AD=E7=8E=AF=EF=BC=88execution?= =?UTF-8?q?=5Fid=20=E4=B8=8B=E6=B2=89=20+=20=E5=85=B1=E4=BA=AB=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E6=9C=BA=20+=20HumanNode=20=E7=9C=9F=E5=AE=A1?= =?UTF-8?q?=E6=89=B9=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - B-06: DagExecutor 接收 execution_id 下沉到 NodeContext，消除 dummy-execution-id 硬编码 - B-07: NodeContext.node_status 共享 self.state_machine.clone()（is_cancelled 可工作） - B-03a: HumanNode execute 重写为 subscribe→send→select! 循环（响应/超时/取消 + execution_id+node_id 双键 + Lagged 容忍），删假返回"同意" - eventbus.rs 删 try_recv_human_approval 死代码 - 新增 human_node 7 单测（正常/双键过滤/超时/非法决策/自由文本） 照 B-03-人工审批响应机制.md 设计。cargo check 0 error，df-nodes 14 + df-workflow 11 test 全过 --- crates/df-nodes/src/human_node.rs | 265 ++++++++++++++++++++++++++--- crates/df-workflow/src/eventbus.rs | 9 +- crates/df-workflow/src/executor.rs | 17 +- src-tauri/src/commands/workflow.rs | 2 +- 4 files changed, 254 insertions(+), 39 deletions(-) diff --git a/crates/df-nodes/src/human_node.rs b/crates/df-nodes/src/human_node.rs index ed378cd..c2fa11e 100644 --- a/crates/df-nodes/src/human_node.rs +++ b/crates/df-nodes/src/human_node.rs @@ -2,7 +2,7 @@ use async_trait::async_trait; use std::time::Duration; -use tokio::time; +use tokio::sync::broadcast; use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema}; use df_core::events::WorkflowEvent; @@ -21,44 +21,77 @@ impl Node for HumanNode { .and_then(|v| v.as_str()) .unwrap_or(""); - let options = config.get("options") + let options: Vec = config.get("options") .and_then(|v| v.as_array()) .map(|arr| arr.iter() .filter_map(|v| v.as_str()) - .collect::>()) - .unwrap_or_else(|| vec!["同意", "拒绝"]); + .map(String::from) + .collect()) + .unwrap_or_else(|| vec!["同意".into(), "拒绝".into()]); - // 发送人工审批请求到事件总线 + let timeout_secs = config.get("timeout_secs") + .and_then(|v| v.as_u64()) + .unwrap_or(3600); + + // 1. 先订阅，再发请求 —— broadcast 不回放历史消息， + // 若 subscribe 晚于 send，Response 会在 receiver 建位前发出而丢失，节点死等到超时。 + let mut rx = ctx.event_bus.subscribe(); + + // 2. 发送人工审批请求到事件总线 let _ = ctx.event_bus.send(WorkflowEvent::HumanApprovalRequest { execution_id: ctx.execution_id.clone(), node_id: ctx.node_id.clone(), title: title.to_string(), description: description.to_string(), - options: options.iter().map(|s| s.to_string()).collect(), + options: options.clone(), }); - // 等待用户响应（最多等待 1 小时） - let timeout = Duration::from_secs(3600); - let start_time = std::time::Instant::now(); + // 3. select! 循环：Response / 超时 / 取消 + let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs); + let mut cancel_tick = tokio::time::interval(Duration::from_millis(500)); + cancel_tick.tick().await; // 丢弃首个立即触发的 tick loop { - // 检查是否超时 - if start_time.elapsed() > timeout { - return Err(anyhow::anyhow!("人工审批超时")); + tokio::select! { + recv = rx.recv() => match recv { + Ok(WorkflowEvent::HumanApprovalResponse { + execution_id, node_id, decision, comment, + }) if execution_id == ctx.execution_id && node_id == ctx.node_id => { + // decision 合法性校验： + // options 空 → 允许自由文本决策； + // options 非空 → 强制 decision ∈ options，否则报错而非静默放行。 + // 空 decision 始终非法。 + if !decision.is_empty() && (options.is_empty() || options.contains(&decision)) { + return Ok(NodeOutput::from_value(serde_json::json!({ + "decision": decision, + "comment": comment.unwrap_or_default(), + }))); + } + return Err(anyhow::anyhow!("审批决策非法: {}", decision)); + } + Ok(_) => continue, // 其他节点/类型的事件，忽略 + Err(broadcast::error::RecvError::Lagged(n)) => { + tracing::warn!( + "HumanNode {} 漏收 {} 条事件（可能错过自身响应，继续等待）", + ctx.node_id, n + ); + // 容量 256 + 审批低频，漏自身 Response 概率极低； + // 丢弃则可能误判超时，故 continue。 + continue; + } + Err(broadcast::error::RecvError::Closed) => { + return Err(anyhow::anyhow!("事件总线关闭，审批无法完成")); + } + }, + _ = tokio::time::sleep_until(deadline) => { + return Err(anyhow::anyhow!("人工审批超时({}s)", timeout_secs)); + } + _ = cancel_tick.tick() => { + if ctx.node_status.is_cancelled(&ctx.node_id) { + return Err(anyhow::anyhow!("人工审批被取消")); + } + } } - - // 检查节点是否被取消 - if ctx.node_status.is_cancelled(&ctx.node_id) { - return Err(anyhow::anyhow!("人工审批被取消")); - } - - // TODO: 检查审批响应（需要前端实现） - // 目前直接返回同意 - let output = serde_json::json!({ - "decision": "同意", - "comment": "", - }); - return Ok(NodeOutput::from_value(output)); } } @@ -94,3 +127,185 @@ impl Node for HumanNode { "human" } } + +#[cfg(test)] +mod tests { + use super::*; + use df_workflow::eventbus::EventBus; + use df_workflow::state::StateMachine; + use serde_json::json; + use std::collections::HashMap; + use std::time::Duration; + + /// 构造 NodeContext：自定义 execution_id / node_id / config，共享同一 EventBus。 + fn make_ctx( + event_bus: &EventBus, + execution_id: &str, + node_id: &str, + config: serde_json::Value, + ) -> NodeContext { + NodeContext { + node_id: node_id.to_string(), + inputs: HashMap::new(), + config, + execution_id: execution_id.to_string(), + event_bus: event_bus.clone(), + node_status: StateMachine::new(), + } + } + + /// 发一条审批响应到事件总线（模拟前端 approve_human_approval IPC 走完后的链路）。 + async fn send_response( + event_bus: &EventBus, + execution_id: &str, + node_id: &str, + decision: &str, + comment: Option<&str>, + ) { + event_bus + .send(WorkflowEvent::HumanApprovalResponse { + execution_id: execution_id.to_string(), + node_id: node_id.to_string(), + decision: decision.to_string(), + comment: comment.map(String::from), + }) + .await; + } + + #[tokio::test] + async fn normal_approval_returns_decision() { + // 正常审批：发匹配(exec_id+node_id) Response → 返回该 decision + let bus = EventBus::new(); + let ctx = make_ctx(&bus, "exec-1", "node-h", json!({})); + let bus_clone = bus.clone(); + + let handle = tokio::spawn(async move { + HumanNode.execute(ctx).await + }); + + // 给 execute 一点时间 subscribe + send Request + tokio::time::sleep(Duration::from_millis(50)).await; + send_response(&bus_clone, "exec-1", "node-h", "同意", Some("好的")).await; + + let out = handle.await.unwrap().unwrap(); + assert_eq!(out.data["decision"], json!("同意")); + assert_eq!(out.data["comment"], json!("好的")); + } + + #[tokio::test] + async fn mismatched_execution_id_filtered_then_timeout() { + // execution_id 不匹配：Response 被过滤，短超时验证 → Err 超时 + let bus = EventBus::new(); + let ctx = make_ctx(&bus, "exec-1", "node-h", json!({ "timeout_secs": 1 })); + let bus_clone = bus.clone(); + + let handle = tokio::spawn(async move { + HumanNode.execute(ctx).await + }); + + tokio::time::sleep(Duration::from_millis(50)).await; + // 错配的 execution_id + send_response(&bus_clone, "exec-OTHER", "node-h", "同意", None).await; + + let err = handle.await.unwrap().unwrap_err().to_string(); + assert!(err.contains("超时"), "不匹配 exec_id 应被过滤后超时, 实际: {}", err); + } + + #[tokio::test] + async fn mismatched_node_id_filtered_then_timeout() { + // node_id 不匹配：Response 被过滤，短超时验证 → Err 超时 + let bus = EventBus::new(); + let ctx = make_ctx(&bus, "exec-1", "node-h", json!({ "timeout_secs": 1 })); + let bus_clone = bus.clone(); + + let handle = tokio::spawn(async move { + HumanNode.execute(ctx).await + }); + + tokio::time::sleep(Duration::from_millis(50)).await; + // 错配的 node_id + send_response(&bus_clone, "exec-1", "node-OTHER", "同意", None).await; + + let err = handle.await.unwrap().unwrap_err().to_string(); + assert!(err.contains("超时"), "不匹配 node_id 应被过滤后超时, 实际: {}", err); + } + + #[tokio::test] + async fn timeout_when_no_response() { + // 无 Response → sleep_until(deadline) 触发超时 + let bus = EventBus::new(); + let ctx = make_ctx(&bus, "exec-1", "node-h", json!({ "timeout_secs": 1 })); + + let err = HumanNode.execute(ctx).await.unwrap_err().to_string(); + assert!(err.contains("超时"), "应报超时, 实际: {}", err); + assert!(err.contains("1s"), "超时消息应带秒数, 实际: {}", err); + } + + #[tokio::test] + async fn invalid_decision_rejected() { + // options 非空，decision ∉ options → Err 决策非法 + let bus = EventBus::new(); + let ctx = make_ctx( + &bus, + "exec-1", + "node-h", + json!({ "options": ["同意", "拒绝"] }), + ); + let bus_clone = bus.clone(); + + let handle = tokio::spawn(async move { + HumanNode.execute(ctx).await + }); + + tokio::time::sleep(Duration::from_millis(50)).await; + send_response(&bus_clone, "exec-1", "node-h", "随便", None).await; + + let err = handle.await.unwrap().unwrap_err().to_string(); + assert!(err.contains("非法"), "非法 decision 应报错, 实际: {}", err); + assert!(err.contains("随便"), "错误信息应含 decision 值, 实际: {}", err); + } + + #[tokio::test] + async fn empty_options_allows_free_text() { + // options 空 → 允许任意自由文本 decision（不校验） + let bus = EventBus::new(); + let ctx = make_ctx(&bus, "exec-1", "node-h", json!({ "options": [] })); + let bus_clone = bus.clone(); + + let handle = tokio::spawn(async move { + HumanNode.execute(ctx).await + }); + + tokio::time::sleep(Duration::from_millis(50)).await; + send_response(&bus_clone, "exec-1", "node-h", "改成先做B方案", None).await; + + let out = handle.await.unwrap().unwrap(); + assert_eq!( + out.data["decision"], + json!("改成先做B方案"), + "options 空应允许自由文本 decision" + ); + } + + #[tokio::test] + async fn empty_decision_rejected_even_with_empty_options() { + // 空 decision 始终非法（即使 options 空也不允许空 decision） + let bus = EventBus::new(); + let ctx = make_ctx(&bus, "exec-1", "node-h", json!({ "options": [] })); + let bus_clone = bus.clone(); + + let handle = tokio::spawn(async move { + HumanNode.execute(ctx).await + }); + + tokio::time::sleep(Duration::from_millis(50)).await; + send_response(&bus_clone, "exec-1", "node-h", "", None).await; + + let err = handle.await.unwrap().unwrap_err().to_string(); + assert!(err.contains("非法"), "空 decision 应报错, 实际: {}", err); + } + + // Closed 分支覆盖：execute 持有的 ctx 含 EventBus clone（即一个 sender）， + // 在 execute 运行期间通道无法 Closed（至少该 sender 存活）。 + // 该分支仅在所有 sender drop 后可达，需独立集成测试，此处不强造。 +} diff --git a/crates/df-workflow/src/eventbus.rs b/crates/df-workflow/src/eventbus.rs index d708591..704efe9 100644 --- a/crates/df-workflow/src/eventbus.rs +++ b/crates/df-workflow/src/eventbus.rs @@ -2,7 +2,7 @@ use tokio::sync::broadcast; use tokio::sync::broadcast::error::SendError; -use df_core::events::{WorkflowEvent, HumanApprovalResponse}; +use df_core::events::WorkflowEvent; /// 事件总线 #[derive(Debug)] @@ -44,13 +44,6 @@ impl EventBus { pub fn emit_human_approval_request(&self, event: WorkflowEvent) -> Result> { self.sender.send(event) } - - /// 尝试获取人工审批响应（简化版本，实际需要实现状态存储） - pub fn try_recv_human_approval(&self, _execution_id: &str, _node_id: &str) -> Option { - // TODO: 实现审批响应的存储和检索 - // 目前返回 None，需要配合前端实现 - None - } } impl Default for EventBus { diff --git a/crates/df-workflow/src/executor.rs b/crates/df-workflow/src/executor.rs index 8e05f19..acaed97 100644 --- a/crates/df-workflow/src/executor.rs +++ b/crates/df-workflow/src/executor.rs @@ -16,14 +16,20 @@ pub struct DagExecutor { event_bus: EventBus, /// 节点状态机 state_machine: StateMachine, + /// 工作流执行 ID（由调用方传入，下沉到每个 NodeContext） + execution_id: String, } impl DagExecutor { /// 创建执行器 - pub fn new(event_bus: EventBus) -> Self { + /// + /// `execution_id` 为本次工作流执行的唯一标识，会下沉到每个节点的 NodeContext， + /// 用于节点内的事件关联、审计追踪等。 + pub fn new(event_bus: EventBus, execution_id: String) -> Self { Self { event_bus, state_machine: StateMachine::new(), + execution_id, } } @@ -73,9 +79,10 @@ impl DagExecutor { node_id: node_id.clone(), inputs, config: initial_config.clone(), - execution_id: "dummy-execution-id".to_string(), // TODO: 应该从 workflow_execution 获取 + execution_id: self.execution_id.clone(), event_bus: self.event_bus.clone(), - node_status: StateMachine::new(), + // 共享执行器状态机，使节点（如 HumanNode）能读取真实状态而非空状态机 + node_status: self.state_machine.clone(), }; // 仅捕获节点共享引用与所有权上下文，避免与 self 借用冲突 @@ -196,7 +203,7 @@ mod tests { 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()); + 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(); @@ -219,7 +226,7 @@ mod tests { 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()); + let mut executor = DagExecutor::new(EventBus::new(), "test-exec".to_string()); let err = executor .run(&dag, serde_json::Value::Null) .await diff --git a/src-tauri/src/commands/workflow.rs b/src-tauri/src/commands/workflow.rs index 0874bcf..63194c7 100644 --- a/src-tauri/src/commands/workflow.rs +++ b/src-tauri/src/commands/workflow.rs @@ -101,7 +101,7 @@ pub async fn run_workflow( let db = state.db.clone(); let exec_id = execution_id.clone(); tauri::async_runtime::spawn(async move { - let mut executor = DagExecutor::new(event_bus.clone()); + let mut executor = DagExecutor::new(event_bus.clone(), exec_id.clone()); let result = executor.run(&runtime_dag, config).await; let (status, error) = match &result {