//! 节点注册表 — 根据 node_type 字符串创建节点实例 use std::collections::HashMap; use crate::dag::Dag; use crate::dag_def::DagDef; use crate::node::Node; /// 节点工厂函数类型 type NodeFactory = Box Box + Send + Sync>; /// 节点注册表 — 根据 node_type 字符串创建节点实例 pub struct NodeRegistry { factories: HashMap, } impl NodeRegistry { /// 创建空的注册表 pub fn new() -> Self { Self { factories: HashMap::new(), } } /// 注册一个节点工厂 pub fn register(&mut self, type_name: &str, factory: F) where F: Fn(&serde_json::Value) -> Box + Send + Sync + 'static, { self.factories.insert(type_name.to_string(), Box::new(factory)); } /// 根据 node_type 创建节点实例 pub fn create(&self, type_name: &str, config: &serde_json::Value) -> anyhow::Result> { self.factories .get(type_name) .ok_or_else(|| anyhow::anyhow!("未注册的节点类型: {}", type_name)) .map(|factory| factory(config)) } /// 从 DagDef 构建完整的运行时 Dag pub fn build_dag(&self, def: &DagDef) -> anyhow::Result { let mut dag = Dag::new(); // 创建所有节点实例,同时收集节点 ID 集合用于边校验 let mut node_ids: std::collections::HashSet<&String> = std::collections::HashSet::new(); for (id, node_def) in &def.nodes { let node = self.create(&node_def.node_type, &node_def.config)?; // 节点级 config 下沉到 Dag.node_configs:DagExecutor.run 构造 NodeContext 时 // deep_merge(initial_config, node_configs[id])(节点级覆盖全局级),解决此前 // initial_config.clone() 直接当 ctx.config、忽略 NodeDef.config 的问题。 dag.add_node_with_config(id.clone(), node, node_def.config.clone()); node_ids.insert(id); } // 添加所有边 — 校验两端节点均存在,野边早返回 for edge_def in &def.edges { if !node_ids.contains(&edge_def.source) || !node_ids.contains(&edge_def.target) { anyhow::bail!( "边 source/target 节点不存在: {} -> {}", edge_def.source, edge_def.target ); } match &edge_def.condition { Some(cond) => dag.add_edge_with_condition( edge_def.source.clone(), edge_def.target.clone(), cond.clone(), ), None => dag.add_edge(edge_def.source.clone(), edge_def.target.clone()), } } Ok(dag) } } // 不实现 Default:原 Default 注册了一个会 panic 的 script 工厂(unimplemented!), // 违反项目铁律「无 panic——所有占位代码返回空/默认值」。所有调用方应显式 new() + // 手动 register 真实节点(如 state.rs::build_registry 的做法)。