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

//! SubflowNode — 嵌套子工作流节点
//!
//! 加载一个子模板(DagDef)并在当前执行上下文中递归执行。
//! 用于复用通用流程(如"代码审查"可作为任意主流程的子步骤)。

use async_trait::async_trait;
use df_workflow::dag_def::DagDef;
use df_workflow::node::{Node, NodeContext, NodeOutput, NodeResult, NodeSchema};
use serde::{Deserialize, Serialize};

/// SubflowNode 配置
#[derive(Debug, Clone, Serialize, Deserialize)]
struct SubflowConfig {
    /// 子工作流 DAG 定义(内联 JSON)
    dag: serde_json::Value,
    /// 嵌套深度限制(防无限递归,默认 MAX_DEPTH)
    #[serde(default)]
    max_depth: Option<u32>,
}

/// 最大递归深度(安全阀)
const MAX_DEPTH: u32 = 10;

/// SubflowNode — 嵌套子工作流
pub struct SubflowNode;

#[async_trait]
impl Node for SubflowNode {
    async fn execute(&self, ctx: NodeContext) -> NodeResult {
        tracing::info!("SubflowNode 执行: {}", ctx.node_id);

        let config: SubflowConfig = serde_json::from_value(ctx.config.clone())
            .map_err(|e| anyhow::anyhow!("SubflowNode 配置解析失败: {}", e))?;

        let max_depth = config.max_depth.unwrap_or(MAX_DEPTH).min(MAX_DEPTH);

        // 反序列化子 DAG
        let sub_dag: DagDef = serde_json::from_value(config.dag)
            .map_err(|e| anyhow::anyhow!("子工作流 DAG 解析失败: {}", e))?;

        if sub_dag.nodes.is_empty() {
            anyhow::bail!("子工作流节点不能为空");
        }

        if max_depth == 0 {
            anyhow::bail!("SubflowNode 超过最大嵌套深度");
        }

        tracing::info!(
            node_count = sub_dag.nodes.len(),
            edge_count = sub_dag.edges.len(),
            max_depth,
            "SubflowNode 加载子工作流"
        );

        // 返回子 DAG 的 JSON 快照供 DagExecutor 消费
        Ok(NodeOutput::from_value(serde_json::json!({
            "subflow": true,
            "node_count": sub_dag.nodes.len(),
            "edge_count": sub_dag.edges.len(),
            "max_depth": max_depth,
            "dag": sub_dag,
        })))
    }

    fn schema(&self) -> NodeSchema {
        NodeSchema {
            params: serde_json::json!({
                "type": "object",
                "properties": {
                    "dag": {
                        "type": "object",
                        "description": "子工作流 DAG 定义(nodes + edges)",
                        "properties": {
                            "nodes": { "type": "object" },
                            "edges": { "type": "array" }
                        }
                    },
                    "max_depth": {
                        "type": "integer",
                        "description": "最大嵌套深度(默认10)",
                        "default": 10
                    }
                },
                "required": ["dag"]
            }),
            output: serde_json::json!({}),
        }
    }

    fn node_type(&self) -> &'static str {
        "subflow"
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use df_workflow::eventbus::EventBus;
    use df_workflow::state::StateMachine;

    fn make_ctx(config: serde_json::Value) -> NodeContext {
        NodeContext {
            node_id: "test_subflow".into(),
            inputs: Default::default(),
            config,
            execution_id: "exec_1".into(),
            event_bus: EventBus::new(),
            node_status: StateMachine::new(),
        }
    }

    #[tokio::test]
    async fn sub_01_valid_dag_returns_metadata() {
        let node = SubflowNode;
        let ctx = make_ctx(serde_json::json!({
            "dag": {
                "nodes": {
                    "child": {
                        "id": "child",
                        "node_type": "script",
                        "config": {}
                    }
                },
                "edges": []
            }
        }));
        let result = node.execute(ctx).await.unwrap();
        assert_eq!(result.data["subflow"], true);
        assert_eq!(result.data["node_count"], 1);
    }

    #[tokio::test]
    async fn sub_02_empty_dag_rejected() {
        let node = SubflowNode;
        let ctx = make_ctx(serde_json::json!({
            "dag": { "nodes": {}, "edges": [] }
        }));
        let result = node.execute(ctx).await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("不能为空"));
    }

    #[tokio::test]
    async fn sub_03_max_depth_zero_guard() {
        let node = SubflowNode;
        let ctx = make_ctx(serde_json::json!({
            "dag": {
                "nodes": { "c": { "id": "c", "node_type": "script", "config": {} } },
                "edges": []
            },
            "max_depth": 0
        }));
        let result = node.execute(ctx).await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("最大嵌套深度"));
    }

    #[tokio::test]
    async fn sub_04_missing_dag_field_errors() {
        let node = SubflowNode;
        let ctx = make_ctx(serde_json::json!({}));
        let result = node.execute(ctx).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn sub_05_schema_has_required_fields() {
        let node = SubflowNode;
        let schema = node.schema();
        let params = &schema.params;
        assert!(params["properties"]["dag"].is_object());
        assert!(params["required"].as_array().unwrap().contains(&serde_json::json!("dag")));
    }
}