//! 人工审批节点 — 阻塞工作流,等待人工确认或输入 //! //! 拒绝语义化关键字判定等纯逻辑见 `human_node_helpers`(本文件只保留 //! HumanNode 的 struct + impl,impl 块约束)。 use async_trait::async_trait; use std::time::Duration; use tokio::sync::broadcast; use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema}; use df_types::events::{SelectType, WorkflowEvent}; // 抽离的纯函数/常量(glob 引入,tests `use super::*` 间接可见) #[allow(unused_imports)] use crate::human_node_helpers::{contains_reject, is_reject_decision}; /// 人工审批节点(阻塞节点) pub struct HumanNode; #[async_trait] impl Node for HumanNode { async fn execute(&self, ctx: NodeContext) -> NodeResult { let config = ctx.config.as_object().cloned().unwrap_or_default(); let title = config.get("title") .and_then(|v| v.as_str()) .unwrap_or("请确认"); let description = config.get("description") .and_then(|v| v.as_str()) .unwrap_or(""); let options: Vec = config.get("options") .and_then(|v| v.as_array()) .map(|arr| arr.iter() .filter_map(|v| v.as_str()) .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); // F-260615-01: 解析 select_type(缺省 Single,向后兼容)。非 "multiple" 一律按 Single 处理。 let select_type = match config.get("select_type").and_then(|v| v.as_str()) { Some("multiple") => SelectType::Multiple, _ => SelectType::Single, }; // 1. 先订阅,再发请求 —— broadcast 不回放历史消息, // 若 subscribe 晚于 send,Response 会在 receiver 建位前发出而丢失,节点死等到超时。 let mut rx = ctx.event_bus.subscribe(); // 2. 发送人工审批请求到事件总线 // send 是 async fn(broadcast 同步发但 async 包装),必须 await 否则 Future 不 poll、Request 不进 channel(B-03b-R6) 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.clone(), select_type, }) .await; // 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 { tokio::select! { recv = rx.recv() => match recv { Ok(WorkflowEvent::HumanApprovalResponse { execution_id, node_id, decision, decisions, comment, }) if execution_id == ctx.execution_id && node_id == ctx.node_id => { // F-260615-01: 归一化决策集合(优先用 decisions 数组,空则回退兼容 decision 单值) // select_type=Single → 决策数必须 =1 // select_type=Multiple → 决策数必须 ≥1 // options 空 → 允许自由文本(仅受数量约束); // options 非空 → 每项必须 ∈ options // 兼容旧调用方:decisions 空但 decision 非空时,按 [decision] 单值处理。 let mut picked: Vec = decisions; if picked.is_empty() && !decision.is_empty() { picked.push(decision.clone()); } let count_ok = match select_type { SelectType::Single => picked.len() == 1, SelectType::Multiple => !picked.is_empty(), }; let each_valid = picked.iter().all(|d| !d.is_empty()) && (options.is_empty() || picked.iter().all(|d| options.contains(d))); if count_ok && each_valid { // F-260616-06 阶段2: 拒绝语义化。 // 审批拒绝此前与同意一样返 Ok —— 语义反转(审批被拒却报"成功"), // 下游无法据 failed 触发退回/重做。 // 现:decision 命中拒绝关键字(见 REJECT_KEYWORDS)→ 返 Err // "人工审批被拒绝(用户选择: )",executor Err 分支 set_failed // → 工作流 failed 状态 → 阶段2 推进链可据 failed 触发退回。 // 行为变更:审批拒绝从 Ok → Err,标注(同步通知主代理)。 if contains_reject(&picked) { let primary = picked.first().cloned().unwrap_or_default(); let comment_str = comment.unwrap_or_default(); return Err(anyhow::anyhow!( "人工审批被拒绝(用户选择: {}){}", primary, if comment_str.is_empty() { String::new() } else { format!(";意见: {}", comment_str) } )); } // 输出统一含 decisions 数组;保留 decision 取首项(向后兼容下游消费者) let primary = picked.first().cloned().unwrap_or_default(); return Ok(NodeOutput::from_value(serde_json::json!({ "decision": primary, "decisions": picked, "comment": comment.unwrap_or_default(), }))); } tracing::warn!( node_id = %ctx.node_id, execution_id = %ctx.execution_id, select_type = ?select_type, decision = %decision, decisions = ?picked, options = ?options, "审批决策非法,忽略并继续等待下一条合法 Response(由超时兜底)" ); continue; } 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!("人工审批被取消")); } } } } } fn schema(&self) -> NodeSchema { NodeSchema { params: serde_json::json!({ "type": "object", "properties": { "title": { "type": "string" }, "description": { "type": "string" }, "options": { "type": "array", "items": { "type": "string" } }, "select_type": { "type": "string", "enum": ["single", "multiple"], "default": "single" } }, "required": ["title"] }), output: serde_json::json!({ "type": "object", "properties": { "decision": { "type": "string" }, "decisions": { "type": "array", "items": { "type": "string" } }, "comment": { "type": "string" } } }), } } fn is_blocking(&self) -> bool { true } fn node_type(&self) -> &str { "human" } } #[cfg(test)] mod tests { use super::*; use df_types::types::NodeStatus; use df_workflow::dag::Dag; use df_workflow::eventbus::EventBus; use df_workflow::executor::DagExecutor; 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 走完后的链路)。 /// F-260615-01: 单选调用方仅填 decision;多选调用方填 decisions。 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(), decisions: vec![], comment: comment.map(String::from), }) .await; } /// F-260615-01: 多选响应发送助手(填 decisions 数组,decision 留空) async fn send_response_multi( event_bus: &EventBus, execution_id: &str, node_id: &str, decisions: &[&str], comment: Option<&str>, ) { event_bus .send(WorkflowEvent::HumanApprovalResponse { execution_id: execution_id.to_string(), node_id: node_id.to_string(), decision: String::new(), decisions: decisions.iter().map(|s| s.to_string()).collect(), 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 request_is_emitted_to_bus() { // B-03b-R8: 验证 HumanNode 真发出 HumanApprovalRequest 到事件总线 // (R6 修复前 execute 内 event_bus.send().await 缺 await → Request 未进 channel,此测会超时失败) let bus = EventBus::new(); let mut rx = bus.subscribe(); // subscribe 先于 execute send(broadcast 不回放) 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 }); // execute send Request 后 rx 应收到 let received = tokio::time::timeout(Duration::from_millis(500), rx.recv()).await; assert!( received.is_ok(), "应收到 HumanApprovalRequest(R6 Request 真发出),实际超时" ); match received.unwrap().unwrap() { WorkflowEvent::HumanApprovalRequest { execution_id, node_id, title, .. } => { assert_eq!(execution_id, "exec-1"); assert_eq!(node_id, "node-h"); assert_eq!(title, "请确认"); } other => panic!("期望 HumanApprovalRequest,收到 {:?}", other), } // 发 Response 让 execute 正常返回(防 task 泄漏) send_response(&bus_clone, "exec-1", "node-h", "同意", None).await; let _ = handle.await; } #[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_ignored_then_timeout() { // R-P1-6: options 非空且 decision ∉ options → 不立即 Err(一次手误不应杀死节点), // warn 记录 + continue 续等下一条合法 Response;此处仅发一条非法的,短超时验证 → Err 超时。 let bus = EventBus::new(); let ctx = make_ctx( &bus, "exec-1", "node-h", json!({ "options": ["同意", "拒绝"], "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; send_response(&bus_clone, "exec-1", "node-h", "随便", None).await; let err = handle.await.unwrap().unwrap_err().to_string(); 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_ignored_then_timeout_even_with_empty_options() { // R-P1-6: 空 decision 始终非法,但同样不立即 Err——warn + continue 续等,短超时验证 → Err 超时。 let bus = EventBus::new(); let ctx = make_ctx(&bus, "exec-1", "node-h", json!({ "options": [], "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; 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 后可达,需独立集成测试,此处不强造。 // ===== B-03b-R8: executor 级端到端集成测试 ===== // 上述单测均为 HumanNode.execute 直接调用级,不覆盖 executor 驱动含 human 的 DAG。 // 端到端覆盖 NodeStarted/Completed 事件流转、共享 state_machine、Request 经 executor 驱动真发出、 // outputs 收集,是 R6(send 缺 await)/R7(契约失配)的存活土壤。 /// 简单测试节点:sleep 指定毫秒后返回空输出(仿 executor.rs SleepNode,作 human 的前驱) 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" } } /// B-03b-R8: executor 驱动 a(SleepNode) → b(HumanNode) 两层 DAG 端到端。 /// a 完成 → b 阻塞等审批 → 外部发 Response(模拟 approve_human_approval IPC) → b Ok → 工作流完成。 /// 验证:Request 经 executor 驱动真发出(R6)、outputs 收集两层、共享 state_machine 两节点皆 Completed。 #[tokio::test] async fn end_to_end_human_approval_completes_workflow() { let bus = EventBus::new(); let mut rx = bus.subscribe(); // subscribe 先于 executor send(broadcast 不回放) let exec_id = "exec-e2e-1"; let mut dag = Dag::new(); dag.add_node("a".to_string(), Box::new(SleepNode { sleep_ms: 10 })); dag.add_node("b".to_string(), Box::new(HumanNode)); dag.add_edge("a".to_string(), "b".to_string()); let mut executor = DagExecutor::new(bus.clone(), exec_id.to_string()); let sm = executor.state_machine(); // 共享状态机(spawn 后仍可读) let run_handle = tokio::spawn(async move { executor .run( &dag, json!({ "title": "确认发布", "options": ["同意", "拒绝"], "timeout_secs": 5 }), ) .await }); // 等 a 层完成、b 层 human subscribe + send Request。 // executor 先发 NodeStarted(a)/NodeStarted(b),再发 human 的 Request —— 过滤跳过非 Request 事件 let request = tokio::time::timeout(Duration::from_millis(1000), async { loop { if let Ok(ev @ WorkflowEvent::HumanApprovalRequest { .. }) = rx.recv().await { return ev; } // NodeStarted/NodeCompleted 等其他事件,继续等 } }) .await; assert!( request.is_ok(), "应收到 b 的 HumanApprovalRequest(R6 端到端),实际超时" ); match request.unwrap() { 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), } // 外部 IPC 模拟:审批通过 send_response(&bus, exec_id, "b", "同意", Some("可以发布")).await; let outputs = run_handle.await.unwrap().unwrap(); assert!(outputs.contains_key("a"), "outputs 应含前驱节点 a"); assert_eq!(outputs["b"].data["decision"], json!("同意")); assert_eq!(outputs["b"].data["decisions"], json!(["同意"])); assert_eq!(outputs["b"].data["comment"], json!("可以发布")); assert_eq!(sm.get(&"a".to_string()), NodeStatus::Completed); assert_eq!(sm.get(&"b".to_string()), NodeStatus::Completed); } /// B-03b-R8: executor 驱动 human DAG,外部 set_cancelled(模拟 cancel_workflow_node IPC) /// → human cancel_tick 命中 is_cancelled → Err → executor 跳过 set_failed(Cancelled 已终态,R1) /// → run 返回取消 Err + 状态保持 Cancelled。覆盖取消在真实 HumanNode 上的端到端(R2)。 #[tokio::test] async fn end_to_end_human_approval_cancelled() { let bus = EventBus::new(); let mut rx = bus.subscribe(); // subscribe 先于 executor send let exec_id = "exec-e2e-cancel"; let mut dag = Dag::new(); dag.add_node("h".to_string(), Box::new(HumanNode)); let mut executor = DagExecutor::new(bus.clone(), exec_id.to_string()); let sm = executor.state_machine(); let run_handle = tokio::spawn(async move { executor.run(&dag, json!({ "timeout_secs": 5 })).await }); // 等 human 发出 Request(确认已 subscribe + send + 进入 select! 轮询) let saw_request = tokio::time::timeout(Duration::from_millis(1000), async { while !matches!(rx.recv().await, Ok(WorkflowEvent::HumanApprovalRequest { .. })) {} }) .await .is_ok(); assert!(saw_request, "human 应先发出 HumanApprovalRequest"); // 外部 IPC 模拟:取消节点(写共享 state_machine,human 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(); // executor 把节点 Err 包成 "节点 h 执行失败"(context),原 "人工审批被取消" 在 source chain assert!( err.contains("取消") || err.contains("执行失败"), "应返回取消相关错误,实际: {}", err ); // 状态保持 Cancelled(未被 set_failed 覆盖为 Failed)—— R1 修法端到端验证 assert_eq!(sm.get(&"h".to_string()), NodeStatus::Cancelled); } // ===== F-260615-01: 多选审批覆盖 ===== /// F-260615-01: select_type=multiple + 多 decisions(均∈options) → 返回 decisions 数组 #[tokio::test] async fn multiple_select_returns_decisions_array() { let bus = EventBus::new(); let ctx = make_ctx( &bus, "exec-multi", "node-h", json!({ "options": ["A", "B", "C"], "select_type": "multiple" }), ); 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_multi(&bus_clone, "exec-multi", "node-h", &["A", "C"], Some("多选")).await; let out = handle.await.unwrap().unwrap(); assert_eq!(out.data["decision"], json!("A"), "decision 取首项"); assert_eq!(out.data["decisions"], json!(["A", "C"])); assert_eq!(out.data["comment"], json!("多选")); } /// F-260615-01: select_type=single 缺省 + decisions 多个 → 校验失败(count!=1)忽略后超时 #[tokio::test] async fn single_select_rejects_multiple_decisions_then_timeout() { let bus = EventBus::new(); let ctx = make_ctx( &bus, "exec-single-multi", "node-h", json!({ "options": ["A", "B"], "timeout_secs": 1 }), // 缺省 single ); let bus_clone = bus.clone(); let handle = tokio::spawn(async move { HumanNode.execute(ctx).await }); tokio::time::sleep(Duration::from_millis(50)).await; // single 模式却发多个 → 非法 send_response_multi(&bus_clone, "exec-single-multi", "node-h", &["A", "B"], None).await; let err = handle.await.unwrap().unwrap_err().to_string(); assert!(err.contains("超时"), "single 下多 decisions 应被忽略后超时, 实际: {}", err); } /// F-260615-01: select_type=multiple 但 decisions 含 ∉ options 的项 → 非法忽略后超时 #[tokio::test] async fn multiple_select_invalid_option_ignored_then_timeout() { let bus = EventBus::new(); let ctx = make_ctx( &bus, "exec-multi-bad", "node-h", json!({ "options": ["A", "B"], "select_type": "multiple", "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; // 一项合法一项不合法 → 整批非法 send_response_multi(&bus_clone, "exec-multi-bad", "node-h", &["A", "X"], None).await; let err = handle.await.unwrap().unwrap_err().to_string(); assert!(err.contains("超时"), "含非法 option 应被忽略后超时, 实际: {}", err); } /// F-260615-01: 兼容旧调用方 —— 不填 select_type(缺省 single) + 只填 decision 单值,应正常通过 /// (即所有未改造的现有 Request 均按 single 解析,零改动) #[tokio::test] async fn default_single_with_legacy_decision_single_value() { let bus = EventBus::new(); // 注意:不传 select_type,验证缺省=Single let ctx = make_ctx(&bus, "exec-leg", "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; // 旧链路只填 decision,decisions 为空 → HumanNode 兼容回退 send_response(&bus_clone, "exec-leg", "node-h", "同意", None).await; let out = handle.await.unwrap().unwrap(); assert_eq!(out.data["decision"], json!("同意")); assert_eq!(out.data["decisions"], json!(["同意"]), "兼容回退后 decisions 应含单值"); } // ===== F-260616-06 阶段2: 审批拒绝语义化(行为变更: 拒绝从 Ok → Err) ===== /// F-260616-06: 默认 options `["同意","拒绝"]` 下选「拒绝」→ Err(不再 Ok)。 /// 阶段2 推进链依赖工作流 failed 触发退回,故拒绝必须让节点返 Err → executor set_failed。 #[tokio::test] async fn reject_decision_returns_error() { let bus = EventBus::new(); let ctx = make_ctx( &bus, "exec-reject", "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-reject", "node-h", "拒绝", Some("方案不行")).await; let err = handle.await.unwrap().unwrap_err().to_string(); assert!( err.contains("拒绝") && err.contains("人工审批"), "拒绝 decision 应返 Err「人工审批被拒绝...」, 实际: {}", err ); assert!( err.contains("方案不行"), "拒绝 Err 应携带 comment, 实际: {}", err ); } /// F-260616-06: 同一 options 下选「同意」→ Ok(通过路径不回归)。 #[tokio::test] async fn approve_decision_still_ok() { let bus = EventBus::new(); let ctx = make_ctx( &bus, "exec-approve", "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-approve", "node-h", "同意", None).await; let out = handle.await.unwrap().unwrap(); assert_eq!(out.data["decision"], json!("同意")); } /// F-260616-06: 英文 reject 关键字同样识别为拒绝 → Err(归一化大小写/空白)。 /// 多关键字覆盖走 reject_keyword_detection_normalized 纯单元测试,此处仅验证端到端一条。 #[tokio::test] async fn english_reject_keyword_returns_error() { let bus = EventBus::new(); let ctx = make_ctx( &bus, "exec-en", "node-h", json!({ "options": ["approve", "reject"], "timeout_secs": 5 }), ); 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-en", "node-h", "reject", None).await; let err = handle.await.unwrap().unwrap_err().to_string(); assert!( err.contains("拒绝"), "英文 reject 关键字应返 Err, 实际: {}", err ); } /// F-260616-06: 多选场景,picked 含一项拒绝 → 整单拒绝 → Err /// (选了「驳回」即驳回,即便同时选了「同意」)。 #[tokio::test] async fn multiple_select_with_one_reject_returns_error() { let bus = EventBus::new(); let ctx = make_ctx( &bus, "exec-multi-reject", "node-h", json!({ "options": ["同意", "驳回", "备注"], "select_type": "multiple" }), ); 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_multi( &bus_clone, "exec-multi-reject", "node-h", &["同意", "驳回"], Some("驳回其中一项"), ) .await; let err = handle.await.unwrap().unwrap_err().to_string(); assert!(err.contains("拒绝"), "多选含拒绝项应返 Err, 实际: {}", err); } /// F-260616-06: options 空的自由文本场景 —— 明确拒绝词("拒绝")仍返 Err, /// 其余自由文本(非拒绝词)仍按通过处理(向后兼容,不阻断自由反馈)。 #[tokio::test] async fn empty_options_free_text_reject_keyword_still_errors() { let bus = EventBus::new(); let ctx = make_ctx(&bus, "exec-free-reject", "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-free-reject", "node-h", "拒绝", None).await; let err = handle.await.unwrap().unwrap_err().to_string(); assert!(err.contains("拒绝"), "自由文本明确为拒绝词仍应 Err, 实际: {}", err); } /// F-260616-06: options 空的自由文本场景 —— 非拒绝词自由文本仍返 Ok(不误伤自由反馈)。 /// (empty_options_allows_free_text 已覆盖 "改成先做B方案" → Ok,此处补一条非拒绝中文短句。) #[tokio::test] async fn empty_options_non_reject_free_text_still_ok() { let bus = EventBus::new(); let ctx = make_ctx(&bus, "exec-free-ok", "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-free-ok", "node-h", "再讨论一下", None).await; let out = handle.await.unwrap().unwrap(); assert_eq!(out.data["decision"], json!("再讨论一下")); } /// F-260616-06 单元: 关键字判定函数归一化(去空白+小写)与边界。 #[test] fn reject_keyword_detection_normalized() { assert!(is_reject_decision("拒绝")); assert!(is_reject_decision(" 拒绝 ")); assert!(is_reject_decision("Reject")); assert!(is_reject_decision(" decline ")); assert!(is_reject_decision("驳回")); assert!(is_reject_decision("退回")); assert!(!is_reject_decision("同意")); assert!(!is_reject_decision("approve")); assert!(!is_reject_decision("再讨论")); assert!(!is_reject_decision("")); // 不是关键字开头,是整词匹配 —— "拒绝啦" 不应误判 assert!(!is_reject_decision("拒绝啦")); assert!(contains_reject(&["同意".into()]) == false); assert!(contains_reject(&["同意".into(), "拒绝".into()]) == true); } }