DevFlow/crates/df-nodes/src/human_node.rs

//! 人工审批节点 — 阻塞工作流，等待人工确认或输入
//!
//! 拒绝语义化关键字判定等纯逻辑见 `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};

/// 默认审批超时(秒)。
/// 1800s = 30min,合理审批窗口。原 3600s(1h)过长,用户忘关致任务挂 1h。
const DEFAULT_TIMEOUT_SECS: u64 = 1800;

/// 人工审批节点（阻塞节点）
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()]);

        // 默认 1800s(30min):合理审批窗口。原 3600s(1h)过长,用户忘关致任务挂 1h。
        let timeout_secs = config.get("timeout_secs")
            .and_then(|v| v.as_u64())
            .unwrap_or(DEFAULT_TIMEOUT_SECS);

        // 解析 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-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);
        // FR-P4：取消检测由 500ms interval 轮询改为 Notify 被动唤醒（set_cancelled 触发）。
        // 先建 notified future 再进 loop，避免 notify 早于 await 的竞态窗口；
        // 每轮 select! 后重建（Notified 单次消费），配合 is_cancelled 复查兜底。

        loop {
            let cancel_wait = ctx.node_status.cancel_notified();
            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 => {
                        // 归一化决策集合(优先用 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 {
                            // 拒绝语义化。
                            //   审批拒绝此前与同意一样返 Ok —— 语义反转(审批被拒却报"成功"),
                            //   下游无法据 failed 触发退回/重做。
                            //   现:decision 命中拒绝关键字(见 REJECT_KEYWORDS)→ 返 Err
                            //   "人工审批被拒绝(用户选择: <decision>)",executor Err 分支 set_failed
                            //   → 工作流 failed 状态 → 推进链可据 failed 触发退回。
                            // 行为变更:审批拒绝从 Ok → Err,标注(同步通知主代理)。
                            if contains_reject(&picked) {
                                let primary = picked.first().cloned().unwrap_or_default();
                                let comment_str = comment.unwrap_or_default();
                                // 意见后缀:空 comment 不拼接,非空才追加(避免空括号)
                                let suffix = if comment_str.is_empty() {
                                    String::new()
                                } else {
                                    format!("；意见: {}", comment_str)
                                };
                                return Err(anyhow::anyhow!(
                                    "人工审批被拒绝(用户选择: {}){}",
                                    primary,
                                    suffix,
                                ));
                            }
                            // 输出统一含 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_wait => {
                    // FR-P4：Notify 唤醒后复查 is_cancelled（Notify 无状态，
                    // 可能因其他节点取消而误唤醒，须确认本节点确被取消）。
                    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.into(),
            event_bus: event_bus.clone(),
            node_status: StateMachine::new(),
        }
    }

    /// 发一条审批响应到事件总线（模拟前端 approve_human_approval IPC 走完后的链路）。
    /// 单选调用方仅填 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.into(),
                node_id: node_id.to_string(),
                decision: decision.to_string(),
                decisions: vec![],
                comment: comment.map(String::from),
            })
            .await;
    }

    /// 多选响应发送助手(填 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.into(),
                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-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() {
        // 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() {
        // 空 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-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-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.into());
        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-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.into());
        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);
    }

    // ===== 多选审批覆盖 =====

    /// 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!("多选"));
    }

    /// 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);
    }

    /// 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);
    }

    /// 兼容旧调用方 —— 不填 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 应含单值");
    }

    // ===== 审批拒绝语义化(行为变更: 拒绝从 Ok → Err) =====

    /// 默认 options `["同意","拒绝"]` 下选「拒绝」→ Err（不再 Ok）。
    /// 推进链依赖工作流 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
        );
    }

    /// 同一 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!("同意"));
    }

    /// 英文 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
        );
    }

    /// 多选场景,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);
    }

    /// 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);
    }

    /// 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!("再讨论一下"));
    }

    /// 单元: 关键字判定函数归一化(去空白+小写)与边界。
    #[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);
    }

    /// 默认审批超时 1800s(30min)。
    /// 回归保护:防止有人无意改回 3600s(1h 过长,用户忘关致任务挂 1h)。
    #[test]
    fn default_timeout_is_1800_secs() {
        assert_eq!(DEFAULT_TIMEOUT_SECS, 1800, "默认审批超时应为 1800s(30min)");
    }
}