绝尘
a2330de2a8
- 新建 df-nodes/human_node_helpers.rs(32行): REJECT_KEYWORDS + is_reject_decision + contains_reject(B-260615-05拒绝语义保留) - human_node.rs 850→831: HumanNode struct+impl+tests保留 + use helpers(impl块约束主体内联) - lib.rs: mod human_node_helpers 主代兜底(独立 -p df-nodes): cargo 0 + test 82(B-260615-05 reject全绿) strategy: 核心库, 纯函数+常量抽离, impl保留; human_node主体内联execute大(impl约束, 可拆有限) git add指定(df-nodes/*)
832 lines
35 KiB
Rust
832 lines
35 KiB
Rust
//! 人工审批节点 — 阻塞工作流，等待人工确认或输入
|
||
//!
|
||
//! 拒绝语义化关键字判定等纯逻辑见 `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, REJECT_KEYWORDS};
|
||
|
||
/// 人工审批节点（阻塞节点）
|
||
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<String> = 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<String> = 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
|
||
// "人工审批被拒绝(用户选择: <decision>)",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 修复前 :41 send 缺 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);
|
||
}
|
||
}
|