新增: 初始化 DevFlow 项目仓库

Tauri 2 + Vue 3 + Vite 6 桌面应用，Rust workspace 含 13 个 crate
(df-ai / df-storage / df-workflow / df-core / df-execute 等)。
核心能力：AI 聊天 agentic 循环(工具调用+人工审批)、工作流引擎、
任务/想法/项目/阶段管理、可追溯性，及配套前端组件。

crates/df-workflow/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "df-workflow"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+df-core = { path = "../df-core" }
+serde = { workspace = true }
+serde_json = { workspace = true }
+tokio = { workspace = true }
+async-trait = { workspace = true }
+futures = "0.3"
+thiserror = { workspace = true }
+anyhow = { workspace = true }
+tracing = { workspace = true }

crates/df-workflow/src/conditions.rs

@@ -0,0 +1,33 @@
+//! 条件表达式引擎 — 用于 DAG 边上的条件判断
+//!
+//! TODO: 实现完整的条件表达式解析与求值
+
+use serde_json::Value;
+
+/// 条件表达式求值器
+pub struct ConditionEngine;
+
+impl ConditionEngine {
+    /// 求值条件表达式
+    ///
+    /// TODO: 实现完整的表达式解析，当前仅支持简单的 JSON Path 比较
+    pub fn evaluate(expr: &str, _context: &Value) -> anyhow::Result<bool> {
+        // 骨架：简单的 true/false 字面量
+        let trimmed = expr.trim();
+        if trimmed == "true" {
+            return Ok(true);
+        }
+        if trimmed == "false" {
+            return Ok(false);
+        }
+
+        // TODO: 实现如下语法：
+        // - "$.status == 'completed'"  — JSON Path 比较
+        // - "$.count > 10"             — 数值比较
+        // - "$.tags contains 'ai'"     — 包含检查
+        // - "and/or/not"               — 逻辑组合
+
+        tracing::warn!("条件表达式引擎尚未完整实现，表达式: {}", expr);
+        Ok(true)
+    }
+}

crates/df-workflow/src/dag.rs

@@ -0,0 +1,141 @@
+//! DAG（有向无环图）定义与拓扑排序
+
+use std::collections::{HashMap, HashSet, VecDeque};
+
+use df_core::error::Result;
+use df_core::types::NodeId;
+
+use crate::node::Node;
+
+/// 图的边：从 source 到 target
+#[derive(Debug, Clone)]
+pub struct Edge {
+    pub source: NodeId,
+    pub target: NodeId,
+    /// 边的条件表达式（可选）
+    pub condition: Option<String>,
+}
+
+/// DAG 结构
+pub struct Dag {
+    /// 节点集合
+    pub nodes: HashMap<NodeId, Box<dyn Node>>,
+    /// 边集合
+    pub edges: Vec<Edge>,
+}
+
+impl Dag {
+    /// 创建空 DAG
+    pub fn new() -> Self {
+        Self {
+            nodes: HashMap::new(),
+            edges: Vec::new(),
+        }
+    }
+
+    /// 添加节点
+    pub fn add_node(&mut self, id: NodeId, node: Box<dyn Node>) {
+        self.nodes.insert(id, node);
+    }
+
+    /// 添加边
+    pub fn add_edge(&mut self, source: NodeId, target: NodeId) {
+        self.edges.push(Edge {
+            source,
+            target,
+            condition: None,
+        });
+    }
+
+    /// 添加带条件的边
+    pub fn add_edge_with_condition(
+        &mut self,
+        source: NodeId,
+        target: NodeId,
+        condition: String,
+    ) {
+        self.edges.push(Edge {
+            source,
+            target,
+            condition: Some(condition),
+        });
+    }
+
+    /// 获取指定节点的所有上游节点 ID
+    pub fn predecessors(&self, node_id: &NodeId) -> Vec<NodeId> {
+        self.edges
+            .iter()
+            .filter(|e| &e.target == node_id)
+            .map(|e| e.source.clone())
+            .collect()
+    }
+
+    /// 获取指定节点的所有下游节点 ID
+    pub fn successors(&self, node_id: &NodeId) -> Vec<NodeId> {
+        self.edges
+            .iter()
+            .filter(|e| &e.source == node_id)
+            .map(|e| e.target.clone())
+            .collect()
+    }
+
+    /// 拓扑排序 — 返回按执行顺序排列的节点 ID 层级
+    ///
+    /// 返回 Vec<Vec<NodeId>>，每层内的节点可以并行执行
+    pub fn topological_layers(&self) -> Result<Vec<Vec<NodeId>>> {
+        let node_ids: HashSet<NodeId> = self.nodes.keys().cloned().collect();
+        let mut in_degree: HashMap<NodeId, usize> = HashMap::new();
+
+        // 初始化入度
+        for id in &node_ids {
+            in_degree.insert(id.clone(), 0);
+        }
+        for edge in &self.edges {
+            if node_ids.contains(&edge.source) && node_ids.contains(&edge.target) {
+                *in_degree.entry(edge.target.clone()).or_insert(0) += 1;
+            }
+        }
+
+        // BFS 分层
+        let mut layers = Vec::new();
+        let mut queue: VecDeque<NodeId> = in_degree
+            .iter()
+            .filter(|(_, &deg)| deg == 0)
+            .map(|(id, _)| id.clone())
+            .collect();
+
+        let mut processed = 0usize;
+        while !queue.is_empty() {
+            let mut layer: Vec<NodeId> = Vec::new();
+            let layer_size = queue.len();
+            for _ in 0..layer_size {
+                if let Some(id) = queue.pop_front() {
+                    layer.push(id.clone());
+                    for succ in self.successors(&id) {
+                        let deg = in_degree.get_mut(&succ).unwrap();
+                        *deg -= 1;
+                        if *deg == 0 {
+                            queue.push_back(succ);
+                        }
+                    }
+                    processed += 1;
+                }
+            }
+            layers.push(layer);
+        }
+
+        if processed != node_ids.len() {
+            return Err(df_core::error::Error::Workflow(
+                "DAG 中存在环，无法进行拓扑排序".to_string(),
+            ));
+        }
+
+        Ok(layers)
+    }
+}
+
+impl Default for Dag {
+    fn default() -> Self {
+        Self::new()
+    }
+}

crates/df-workflow/src/dag_def.rs

@@ -0,0 +1,104 @@
+//! DAG 定义 — 可序列化/反序列化，用于持久化和模板
+
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+use crate::dag::Dag;
+
+/// DAG 定义 — 可序列化/反序列化，用于持久化和模板
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct DagDef {
+    pub nodes: HashMap<String, NodeDef>,
+    pub edges: Vec<EdgeDef>,
+}
+
+/// 节点定义
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct NodeDef {
+    pub id: String,
+    pub node_type: String,
+    pub config: serde_json::Value,
+    #[serde(default)]
+    pub label: Option<String>,
+}
+
+/// 边定义
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct EdgeDef {
+    pub source: String,
+    pub target: String,
+    #[serde(default)]
+    pub condition: Option<String>,
+}
+
+impl DagDef {
+    /// 创建空的 DAG 定义
+    pub fn new() -> Self {
+        Self {
+            nodes: HashMap::new(),
+            edges: Vec::new(),
+        }
+    }
+
+    /// 添加节点定义
+    pub fn add_node(&mut self, id: impl Into<String>, node_type: impl Into<String>, config: serde_json::Value) {
+        let id = id.into();
+        self.nodes.insert(id.clone(), NodeDef {
+            id,
+            node_type: node_type.into(),
+            config,
+            label: None,
+        });
+    }
+
+    /// 添加边定义
+    pub fn add_edge(&mut self, source: impl Into<String>, target: impl Into<String>) {
+        self.edges.push(EdgeDef {
+            source: source.into(),
+            target: target.into(),
+            condition: None,
+        });
+    }
+
+    /// 添加带条件的边定义
+    pub fn add_edge_with_condition(
+        &mut self,
+        source: impl Into<String>,
+        target: impl Into<String>,
+        condition: impl Into<String>,
+    ) {
+        self.edges.push(EdgeDef {
+            source: source.into(),
+            target: target.into(),
+            condition: Some(condition.into()),
+        });
+    }
+
+    /// 从已有的运行时 Dag 提取定义（忽略节点实例，仅保留元数据）
+    ///
+    /// 注意：此方法只能提取边的定义，节点配置无法从 trait object 反推
+    pub fn from_dag_edges(dag: &Dag) -> Self {
+        let edges: Vec<EdgeDef> = dag.edges.iter().map(|e| EdgeDef {
+            source: e.source.clone(),
+            target: e.target.clone(),
+            condition: e.condition.clone(),
+        }).collect();
+
+        let nodes: HashMap<String, NodeDef> = dag.nodes.iter().map(|(id, node)| {
+            (id.clone(), NodeDef {
+                id: id.clone(),
+                node_type: node.node_type().to_string(),
+                config: serde_json::Value::Null,
+                label: None,
+            })
+        }).collect();
+
+        Self { nodes, edges }
+    }
+}
+
+impl Default for DagDef {
+    fn default() -> Self {
+        Self::new()
+    }
+}

crates/df-workflow/src/eventbus.rs

@@ -0,0 +1,68 @@
+//! 事件总线 — 基于 tokio::sync::broadcast 的发布/订阅
+
+use tokio::sync::broadcast;
+use tokio::sync::broadcast::error::SendError;
+use df_core::events::{WorkflowEvent, HumanApprovalResponse};
+
+/// 事件总线
+#[derive(Debug)]
+pub struct EventBus {
+    sender: broadcast::Sender<WorkflowEvent>,
+}
+
+/// 事件订阅者
+pub type EventSubscriber = broadcast::Receiver<WorkflowEvent>;
+
+/// 默认事件通道容量
+const DEFAULT_CAPACITY: usize = 256;
+
+impl EventBus {
+    /// 创建事件总线
+    pub fn new() -> Self {
+        let (sender, _) = broadcast::channel(DEFAULT_CAPACITY);
+        Self { sender }
+    }
+
+    /// 创建指定容量的事件总线
+    pub fn with_capacity(capacity: usize) -> Self {
+        let (sender, _) = broadcast::channel(capacity);
+        Self { sender }
+    }
+
+    /// 发送事件
+    pub async fn send(&self, event: WorkflowEvent) {
+        // broadcast::send 是同步的，忽略接收者已关闭的错误
+        let _ = self.sender.send(event);
+    }
+
+    /// 订阅事件
+    pub fn subscribe(&self) -> EventSubscriber {
+        self.sender.subscribe()
+    }
+
+    /// 发送人工审批请求
+    pub fn emit_human_approval_request(&self, event: WorkflowEvent) -> Result<usize, SendError<WorkflowEvent>> {
+        self.sender.send(event)
+    }
+
+    /// 尝试获取人工审批响应（简化版本，实际需要实现状态存储）
+    pub fn try_recv_human_approval(&self, _execution_id: &str, _node_id: &str) -> Option<HumanApprovalResponse> {
+        // TODO: 实现审批响应的存储和检索
+        // 目前返回 None，需要配合前端实现
+        None
+    }
+}
+
+impl Default for EventBus {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Clone for EventBus {
+    fn clone(&self) -> Self {
+        Self {
+            sender: self.sender.clone(),
+        }
+    }
+}

crates/df-workflow/src/executor.rs

@@ -0,0 +1,235 @@
+//! DAG 执行器 — 按拓扑层级调度执行
+
+use std::collections::HashMap;
+
+use df_core::events::WorkflowEvent;
+use df_core::types::NodeId;
+
+use crate::dag::Dag;
+use crate::eventbus::EventBus;
+use crate::node::{NodeContext, NodeOutput};
+use crate::state::StateMachine;
+
+/// DAG 执行器
+pub struct DagExecutor {
+    /// 事件总线
+    event_bus: EventBus,
+    /// 节点状态机
+    state_machine: StateMachine,
+}
+
+impl DagExecutor {
+    /// 创建执行器
+    pub fn new(event_bus: EventBus) -> Self {
+        Self {
+            event_bus,
+            state_machine: StateMachine::new(),
+        }
+    }
+
+    /// 执行 DAG 工作流
+    ///
+    /// 同一拓扑层内的节点并发执行，层与层之间串行；
+    /// 本层全部节点完成后统一更新状态，任一失败则中止后续层。
+    pub async fn run(
+        &mut self,
+        dag: &Dag,
+        initial_config: serde_json::Value,
+    ) -> anyhow::Result<HashMap<NodeId, NodeOutput>> {
+        let layers = dag.topological_layers()?;
+        let mut outputs: HashMap<NodeId, NodeOutput> = HashMap::new();
+        let start = std::time::Instant::now();
+
+        tracing::info!("DAG 执行开始，共 {} 层", layers.len());
+
+        for (layer_idx, layer) in layers.iter().enumerate() {
+            tracing::info!("执行第 {} 层，共 {} 个节点", layer_idx, layer.len());
+
+            // 阶段一：逐节点发 NodeStarted、置为运行中，并构建执行 future
+            let mut node_futures = Vec::with_capacity(layer.len());
+            for node_id in layer {
+                let node = dag.nodes.get(node_id).ok_or_else(|| {
+                    anyhow::anyhow!("节点 {} 不存在于 DAG 中", node_id)
+                })?;
+
+                // 发送 NodeStarted 事件
+                self.event_bus
+                    .send(WorkflowEvent::NodeStarted {
+                        node_id: node_id.clone(),
+                    })
+                    .await;
+
+                self.state_machine.set_running(node_id.clone())?;
+
+                // 构建节点上下文
+                let mut inputs = HashMap::new();
+                for pred_id in dag.predecessors(node_id) {
+                    if let Some(out) = outputs.get(&pred_id) {
+                        inputs.insert(pred_id, out.clone());
+                    }
+                }
+
+                let ctx = NodeContext {
+                    node_id: node_id.clone(),
+                    inputs,
+                    config: initial_config.clone(),
+                    execution_id: "dummy-execution-id".to_string(), // TODO: 应该从 workflow_execution 获取
+                    event_bus: self.event_bus.clone(),
+                    node_status: StateMachine::new(),
+                };
+
+                // 仅捕获节点共享引用与所有权上下文，避免与 self 借用冲突
+                let id = node_id.clone();
+                node_futures.push(async move {
+                    let node_start = std::time::Instant::now();
+                    let result = node.execute(ctx).await;
+                    (id, result, node_start.elapsed().as_millis() as u64)
+                });
+            }
+
+            // 阶段二：同层节点并发执行，等待全部完成
+            let results = futures::future::join_all(node_futures).await;
+
+            // 阶段三：统一更新状态并发送事件，任一失败则整体返回 Err
+            let mut first_err: Option<anyhow::Error> = None;
+            for (node_id, result, duration) in results {
+                match result {
+                    Ok(output) => {
+                        self.state_machine.set_completed(node_id.clone())?;
+                        self.event_bus
+                            .send(WorkflowEvent::NodeCompleted {
+                                node_id: node_id.clone(),
+                                duration_ms: duration,
+                            })
+                            .await;
+                        outputs.insert(node_id, output);
+                    }
+                    Err(e) => {
+                        self.state_machine.set_failed(node_id.clone())?;
+                        self.event_bus
+                            .send(WorkflowEvent::NodeFailed {
+                                node_id: node_id.clone(),
+                                error: e.to_string(),
+                            })
+                            .await;
+                        // 同层多个失败时只报告第一个
+                        if first_err.is_none() {
+                            first_err = Some(e.context(format!("节点 {} 执行失败", node_id)));
+                        }
+                    }
+                }
+            }
+
+            if let Some(e) = first_err {
+                return Err(e);
+            }
+        }
+
+        let total = start.elapsed().as_millis() as u64;
+        self.event_bus
+            .send(WorkflowEvent::WorkflowCompleted {
+                total_duration_ms: total,
+            })
+            .await;
+
+        tracing::info!("DAG 执行完成，耗时 {}ms", total);
+        Ok(outputs)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::node::{Node, NodeResult, NodeSchema};
+    use async_trait::async_trait;
+    use std::time::Duration;
+
+    /// 测试节点：sleep 指定毫秒后返回空输出
+    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"
+        }
+    }
+
+    /// 测试节点：直接返回错误
+    struct FailNode;
+
+    #[async_trait]
+    impl Node for FailNode {
+        async fn execute(&self, _ctx: NodeContext) -> NodeResult {
+            Err(anyhow::anyhow!("故意失败"))
+        }
+
+        fn schema(&self) -> NodeSchema {
+            NodeSchema {
+                params: serde_json::Value::Null,
+                output: serde_json::Value::Null,
+            }
+        }
+
+        fn node_type(&self) -> &str {
+            "fail"
+        }
+    }
+
+    #[tokio::test]
+    async fn test_same_layer_runs_in_parallel() {
+        // 两个无依赖节点位于同一层，各 sleep 100ms
+        let mut dag = Dag::new();
+        dag.add_node("a".to_string(), Box::new(SleepNode { sleep_ms: 100 }));
+        dag.add_node("b".to_string(), Box::new(SleepNode { sleep_ms: 100 }));
+
+        let mut executor = DagExecutor::new(EventBus::new());
+        let start = std::time::Instant::now();
+        let outputs = executor.run(&dag, serde_json::Value::Null).await.unwrap();
+        let elapsed = start.elapsed();
+
+        assert_eq!(outputs.len(), 2);
+        // 串行需要约 200ms，并行应明显小于 180ms
+        assert!(
+            elapsed < Duration::from_millis(180),
+            "同层节点应并行执行，实际耗时 {:?}",
+            elapsed
+        );
+    }
+
+    #[tokio::test]
+    async fn test_layer_failure_aborts_following_layers() {
+        // a(失败) 与 b(成功) 同层，c 依赖 a，失败后 c 不应执行
+        let mut dag = Dag::new();
+        dag.add_node("a".to_string(), Box::new(FailNode));
+        dag.add_node("b".to_string(), Box::new(SleepNode { sleep_ms: 10 }));
+        dag.add_node("c".to_string(), Box::new(SleepNode { sleep_ms: 10 }));
+        dag.add_edge("a".to_string(), "c".to_string());
+
+        let mut executor = DagExecutor::new(EventBus::new());
+        let err = executor
+            .run(&dag, serde_json::Value::Null)
+            .await
+            .unwrap_err();
+        assert!(err.to_string().contains("节点 a 执行失败"));
+
+        // 同层成功节点状态正常更新，下游节点保持 Pending
+        use df_core::types::NodeStatus;
+        assert_eq!(executor.state_machine.get(&"a".to_string()), NodeStatus::Failed);
+        assert_eq!(executor.state_machine.get(&"b".to_string()), NodeStatus::Completed);
+        assert_eq!(executor.state_machine.get(&"c".to_string()), NodeStatus::Pending);
+    }
+}

crates/df-workflow/src/lib.rs

@@ -0,0 +1,10 @@
+//! df-workflow: 工作流引擎 — DAG 定义、节点抽象、执行器、状态机、事件总线
+
+pub mod conditions;
+pub mod dag;
+pub mod dag_def;
+pub mod eventbus;
+pub mod executor;
+pub mod node;
+pub mod registry;
+pub mod state;

crates/df-workflow/src/node.rs

@@ -0,0 +1,83 @@
+//! 节点抽象：Node trait、NodeContext、NodeOutput
+
+use async_trait::async_trait;
+use serde::{Deserialize, Serialize};
+
+use df_core::types::NodeId;
+
+use super::eventbus::EventBus;
+use super::state::StateMachine;
+
+/// 节点输入/输出上下文
+#[derive(Debug, Clone)]
+pub struct NodeContext {
+    /// 节点 ID
+    pub node_id: NodeId,
+    /// 上游节点输出（key 为上游节点 ID）
+    pub inputs: std::collections::HashMap<String, NodeOutput>,
+    /// 节点配置参数
+    pub config: serde_json::Value,
+    /// 工作流执行 ID
+    pub execution_id: String,
+    /// 事件总线
+    pub event_bus: EventBus,
+    /// 节点状态机（用于检查取消状态）
+    pub node_status: StateMachine,
+}
+
+/// 节点执行输出
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct NodeOutput {
+    /// 输出数据
+    pub data: serde_json::Value,
+    /// 输出元数据
+    pub metadata: std::collections::HashMap<String, String>,
+}
+
+impl NodeOutput {
+    /// 创建空输出
+    pub fn empty() -> Self {
+        Self {
+            data: serde_json::Value::Null,
+            metadata: std::collections::HashMap::new(),
+        }
+    }
+
+    /// 从 JSON 值创建输出
+    pub fn from_value(data: serde_json::Value) -> Self {
+        Self {
+            data,
+            metadata: std::collections::HashMap::new(),
+        }
+    }
+}
+
+/// 节点执行结果
+pub type NodeResult = anyhow::Result<NodeOutput>;
+
+/// 节点参数 Schema（JSON Schema 格式）
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct NodeSchema {
+    /// 参数 JSON Schema
+    pub params: serde_json::Value,
+    /// 输出 JSON Schema
+    pub output: serde_json::Value,
+}
+
+/// 节点 trait — 所有工作流节点必须实现
+#[async_trait]
+pub trait Node: Send + Sync {
+    /// 执行节点逻辑
+    async fn execute(&self, ctx: NodeContext) -> NodeResult;
+
+    /// 返回节点的参数 Schema
+    fn schema(&self) -> NodeSchema;
+
+    /// 该节点是否为阻塞节点（阻塞工作流，等待外部输入）
+    fn is_blocking(&self) -> bool {
+        false
+    }
+
+    /// 节点类型名称
+    fn node_type(&self) -> &str;
+}

crates/df-workflow/src/registry.rs

@@ -0,0 +1,87 @@
+//! 节点注册表 — 根据 node_type 字符串创建节点实例
+
+use std::collections::HashMap;
+
+use crate::dag::Dag;
+use crate::dag_def::DagDef;
+use crate::node::Node;
+
+/// 节点工厂函数类型
+type NodeFactory = Box<dyn Fn(&serde_json::Value) -> Box<dyn Node> + Send + Sync>;
+
+/// 节点注册表 — 根据 node_type 字符串创建节点实例
+pub struct NodeRegistry {
+    factories: HashMap<String, NodeFactory>,
+}
+
+impl NodeRegistry {
+    /// 创建空的注册表
+    pub fn new() -> Self {
+        Self {
+            factories: HashMap::new(),
+        }
+    }
+
+    /// 注册一个节点工厂
+    pub fn register<F>(&mut self, type_name: &str, factory: F)
+    where
+        F: Fn(&serde_json::Value) -> Box<dyn Node> + 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<Box<dyn Node>> {
+        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<Dag> {
+        let mut dag = Dag::new();
+
+        // 创建所有节点实例
+        for (id, node_def) in &def.nodes {
+            let node = self.create(&node_def.node_type, &node_def.config)?;
+            dag.add_node(id.clone(), node);
+        }
+
+        // 添加所有边
+        for edge_def in &def.edges {
+            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)
+    }
+
+    /// 检查是否已注册指定类型
+    pub fn is_registered(&self, type_name: &str) -> bool {
+        self.factories.contains_key(type_name)
+    }
+
+    /// 列出所有已注册的节点类型
+    pub fn registered_types(&self) -> Vec<&str> {
+        self.factories.keys().map(|s| s.as_str()).collect()
+    }
+}
+
+impl Default for NodeRegistry {
+    fn default() -> Self {
+        let mut registry = Self::new();
+        registry.register("script", |_config| {
+            // ScriptNode 的工厂 — 需要 df-nodes 依赖后才可用
+            // 这里返回一个占位实现，实际项目中由 df-nodes crate 注册
+            unimplemented!("ScriptNode 需要通过 df-nodes 注册")
+        });
+        registry
+    }
+}

crates/df-workflow/src/state.rs

@@ -0,0 +1,147 @@
+//! 节点状态机 — 管理节点的状态转换
+//!
+//! 合法转换：Pending → Running，Running → Completed / Failed，
+//! 非法转换返回错误，避免状态被随意覆盖。
+
+use std::collections::HashMap;
+
+use df_core::types::{NodeId, NodeStatus};
+
+/// 节点状态机
+#[derive(Debug, Clone)]
+pub struct StateMachine {
+    /// 节点状态映射
+    states: HashMap<NodeId, NodeStatus>,
+}
+
+impl StateMachine {
+    /// 创建状态机
+    pub fn new() -> Self {
+        Self {
+            states: HashMap::new(),
+        }
+    }
+
+    /// 获取节点状态
+    pub fn get(&self, node_id: &NodeId) -> NodeStatus {
+        self.states
+            .get(node_id)
+            .cloned()
+            .unwrap_or(NodeStatus::Pending)
+    }
+
+    /// 判断状态转换是否合法
+    fn is_legal(from: &NodeStatus, to: &NodeStatus) -> bool {
+        matches!(
+            (from, to),
+            (NodeStatus::Pending, NodeStatus::Running)
+                | (NodeStatus::Running, NodeStatus::Completed)
+                | (NodeStatus::Running, NodeStatus::Failed)
+        )
+    }
+
+    /// 状态转换 — 校验合法性后更新，非法转换返回错误
+    pub fn transition(&mut self, node_id: NodeId, target: NodeStatus) -> anyhow::Result<()> {
+        let current = self.get(&node_id);
+        if !Self::is_legal(&current, &target) {
+            anyhow::bail!(
+                "节点 {} 状态转换非法：{} -> {}",
+                node_id,
+                current.as_str(),
+                target.as_str()
+            );
+        }
+        self.states.insert(node_id, target);
+        Ok(())
+    }
+
+    /// 设置节点为运行中（仅允许 Pending -> Running）
+    pub fn set_running(&mut self, node_id: NodeId) -> anyhow::Result<()> {
+        self.transition(node_id, NodeStatus::Running)
+    }
+
+    /// 设置节点为已完成（仅允许 Running -> Completed）
+    pub fn set_completed(&mut self, node_id: NodeId) -> anyhow::Result<()> {
+        self.transition(node_id, NodeStatus::Completed)
+    }
+
+    /// 设置节点为失败（仅允许 Running -> Failed）
+    pub fn set_failed(&mut self, node_id: NodeId) -> anyhow::Result<()> {
+        self.transition(node_id, NodeStatus::Failed)
+    }
+
+    /// 设置节点为等待中（暂未纳入转换校验）
+    pub fn set_waiting(&mut self, node_id: NodeId) {
+        self.states.insert(node_id, NodeStatus::Waiting);
+    }
+
+    /// 设置节点为已跳过（暂未纳入转换校验）
+    pub fn set_skipped(&mut self, node_id: NodeId) {
+        self.states.insert(node_id, NodeStatus::Skipped);
+    }
+
+    /// 获取所有状态快照
+    pub fn snapshot(&self) -> &HashMap<NodeId, NodeStatus> {
+        &self.states
+    }
+
+    /// 检查节点是否被取消
+    pub fn is_cancelled(&self, node_id: &NodeId) -> bool {
+        matches!(self.get(node_id), NodeStatus::Cancelled)
+    }
+}
+
+impl Default for StateMachine {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_legal_transitions() {
+        let mut sm = StateMachine::new();
+        let id = "node-a".to_string();
+
+        // Pending -> Running -> Completed 全程合法
+        assert!(sm.set_running(id.clone()).is_ok());
+        assert_eq!(sm.get(&id), NodeStatus::Running);
+        assert!(sm.set_completed(id.clone()).is_ok());
+        assert_eq!(sm.get(&id), NodeStatus::Completed);
+
+        // Running -> Failed 合法
+        let id2 = "node-b".to_string();
+        sm.set_running(id2.clone()).unwrap();
+        assert!(sm.set_failed(id2.clone()).is_ok());
+        assert_eq!(sm.get(&id2), NodeStatus::Failed);
+    }
+
+    #[test]
+    fn test_illegal_transitions_rejected() {
+        let mut sm = StateMachine::new();
+        let id = "node-a".to_string();
+
+        // Pending -> Completed 非法
+        let err = sm.set_completed(id.clone()).unwrap_err();
+        assert!(err.to_string().contains("状态转换非法"));
+        assert!(err.to_string().contains("node-a"));
+        // 状态保持不变
+        assert_eq!(sm.get(&id), NodeStatus::Pending);
+
+        // Pending -> Failed 非法
+        assert!(sm.set_failed(id.clone()).is_err());
+
+        // Completed 为终态，不允许再转 Running
+        sm.set_running(id.clone()).unwrap();
+        sm.set_completed(id.clone()).unwrap();
+        assert!(sm.set_running(id.clone()).is_err());
+
+        // Running -> Running 重复置位非法
+        let id2 = "node-b".to_string();
+        sm.set_running(id2.clone()).unwrap();
+        assert!(sm.set_running(id2.clone()).is_err());
+    }
+}