# DAG 引擎详解 > 来源：基于 `crates/df-workflow/src/` 实际代码核对编写，2026-06-14 > 关联模块文档：[df-workflow-工作流引擎-2026-06-12.md](./df-workflow-工作流引擎-2026-06-12.md)（接口级参考） > 代码版本：Sprint 22 收尾（StateMachine Arc 共享 + 审批取消闭环已落地） --- ## 一、什么是 DAG DAG = **Directed Acyclic Graph（有向无环图）**。DevFlow 用它来编排工作流——把一个复杂任务拆成多个节点，按依赖关系连接，引擎自动算出执行顺序。 ``` ┌─────────┐ │ 检查环境 │ ← 第 1 层（无依赖，先执行） └────┬────┘ │ ┌────┴────┐ │ 运行测试 │ ← 第 2 层（依赖「检查环境」完成） └────┬────┘ │ ┌────┴────┐ │ 构建产物 │ ← 第 3 层（依赖「运行测试」完成） └─────────┘ ``` 也可以有分支和并行： ``` ┌──────────┐ │ 代码扫描 │ └──┬───┬───┘ ← 第 1 层 │ │ ┌────┘ └────┐ ▼ ▼ ┌──────┐ ┌──────────┐ │ 单元测试 │ │ AI 代码审查 │ ← 第 2 层（两个并行执行） └──┬───┘ └────┬─────┘ │ │ └──────┬───────┘ ▼ ┌────────────┐ │ 汇总报告 │ ← 第 3 层（等上面两个都完成） └────────────┘ ``` --- ## 二、核心组件（6 个文件，各司其职） ``` 用户定义工作流（JSON） │ ▼ ┌──────────┐ 序列化/反序列化 ┌──────────┐ │ DagDef │ ←──────────────────→ │ 数据库持久化 │ │ (dag_def) │ └──────────┘ └─────┬─────┘ │ NodeRegistry.build_dag() ▼ ┌──────────┐ 拓扑排序分层 ┌──────────────┐ │ Dag │ ──────────────────→ │ Vec> │ │ (dag) │ │ 按层组织 │ └──────────┘ └──────┬───────┘ │ ┌──────┴───────┐ ▼ │ ┌────────────┐ │ │ DagExecutor │ ←─── StateMachine │ (executor) │ (状态转换校验) └──────┬─────┘ │ ┌──────┴─────┐ │ EventBus │ → 事件广播到前端 │ (eventbus) │ └────────────┘ ``` ### 2.1 DagDef（定义层）— 工作流的"蓝图" > 源文件：`dag_def.rs` ```rust pub struct DagDef { pub nodes: HashMap, // 节点定义 pub edges: Vec, // 依赖关系 } pub struct NodeDef { pub id: String, // 如 "check-env" pub node_type: String, // 如 "script"、"ai"、"human" pub config: serde_json::Value, // 节点参数（命令、prompt 等） } pub struct EdgeDef { pub source: String, // 如 "check-env" pub target: String, // 如 "run-tests" pub condition: Option, // 可选条件（如 "true"） } ``` 这是**可序列化**的，存到 SQLite 的 `workflow_defs.dag` 字段（JSON），也可以从模板加载。 ### 2.2 Node（节点抽象）— 所有节点的统一接口 > 源文件：`node.rs` ```rust #[async_trait] pub trait Node: Send + Sync { async fn execute(&self, ctx: NodeContext) -> NodeResult; fn schema(&self) -> NodeSchema; fn is_blocking(&self) -> bool { false } // 阻塞节点（如人工审批） fn node_type(&self) -> &str; } ``` 每个节点执行时拿到一个 `NodeContext`： ```rust pub struct NodeContext { pub node_id: String, pub inputs: HashMap, // 上游节点的输出 pub config: serde_json::Value, // 节点配置 pub execution_id: String, // 本次执行 ID pub event_bus: EventBus, // 事件总线（发事件给前端） pub node_status: StateMachine, // 共享状态机（检查是否被取消） } ``` **关键设计**：`node_status` 是 `StateMachine` 的 clone，但内部是 `Arc>`，所以 clone 共享同一底层数据——外部 IPC 调 `set_cancelled` 写入后，运行中的节点能立即读到。 ### 2.3 Dag（运行时图）+ 拓扑排序 > 源文件：`dag.rs` `DagDef` 是数据蓝图，`Dag` 是装入真实节点实例的运行时图。核心方法是 `topological_layers()`： ``` 输入：节点 + 边的依赖关系 输出：Vec> —— 分层的节点 ID 算法：Kahn 算法（BFS 分层） 1. 算每个节点的入度（有多少上游） 2. 入度=0 的节点入队（第一层） 3. 取出一层节点，将它们的下游入度 -1 4. 入度归零的下游入队（下一层） 5. 重复直到所有节点处理完 6. 如果处理数 ≠ 节点总数 → 有环，报错 ``` 复杂度 O(V+E)：一次性遍历边构建邻接索引（出边表）与入度表，BFS 分层只走索引查询（每条边仅被访问一次）。 这就是引擎的核心智能：**用户只需画依赖关系（谁依赖谁），引擎自动算出执行顺序和并行机会**。 ### 2.4 DagExecutor（执行器）— 三阶段调度 > 源文件：`executor.rs` ```rust pub async fn run(&mut self, dag: &Dag, config: Value) -> Result> ``` 对每一层执行**三阶段模式**： ``` ┌─ 阶段一：准备 ──────────────────────────────────────┐ │ 遍历层内每个节点： │ │ ① emit NodeStarted 事件 │ │ ② state_machine.set_running（Pending→Running） │ │ ③ 收集上游输出作为 inputs │ │ ④ 构建 NodeContext │ │ ⑤ 创建 async future（不立即执行） │ └──────────────────────────────────────────────────────┘ │ ▼ ┌─ 阶段二：并发执行 ───────────────────────────────────┐ │ futures::future::join_all(node_futures).await │ │ │ │ 同层所有节点真正并发跑（tokio 异步运行时） │ │ 等待全部完成（或某个失败） │ └──────────────────────────────────────────────────────┘ │ ▼ ┌─ 阶段三：收尾 ───────────────────────────────────────┐ │ 遍历结果： │ │ ✅ 成功 → set_completed + emit NodeCompleted + 存输出 │ │ ❌ 失败 → set_failed + emit NodeFailed + 记录错误 │ │ 🚫 取消 → 跳过 set_failed（Cancelled 是终态） │ │ │ │ 任一失败 → return Err，后续层不执行 │ │ 全部成功 → 进入下一层 │ └──────────────────────────────────────────────────────┘ ``` **为什么三阶段不合并？** 避免 Rust 的借用冲突——阶段一需要 `&self`（读状态机），阶段二的 future 捕获节点引用独立运行，阶段三回来再更新 `&mut self`。 **入边索引优化**：执行前预建 `adjacency_in: HashMap>`（O(E) 一次构建），避免内层循环中调用 `dag.predecessors()`（每次 O(E) 全表扫描，整体退化 O(V·E)）。 ### 2.5 StateMachine（状态机）— 防止非法状态转换 > 源文件：`state.rs` ``` 合法转换图： Pending ──→ Running ──→ Completed ✅ 正常完成 │ └──→ Failed ❌ 执行失败 任何节点 ←── set_cancelled 🚫 外部取消（不走转换校验） ``` ```rust fn is_legal(from: &NodeStatus, to: &NodeStatus) -> bool { matches!((from, to), (Pending, Running) // 启动 | (Running, Completed) // 完成 | (Running, Failed) // 失败 ) } ``` 非法转换直接 `bail!`（如 Pending → Completed 跳过执行），防止状态被意外覆盖。 **共享语义**是关键： ```rust pub struct StateMachine { states: Arc>>, } ``` `clone()` 只是 Arc 引用计数 +1，所有副本共享同一 HashMap。这让取消机制可以跨边界工作： ``` 前端点「取消审批」 │ ▼ cancel_workflow_node IPC │ ▼ AppState.workflow_state_registry[execution_id].set_cancelled(node_id) │ （写入共享 HashMap） ▼ HumanNode.execute 内部 ctx.node_status.is_cancelled() → true │ （读到写入，因为是同一份 HashMap） ▼ 返回 Err("人工审批被取消") │ ▼ Executor 检测到 is_cancelled → 跳过 set_failed（避免 Cancelled→Failed 非法转换） ``` ### 2.6 EventBus（事件总线）— 对外广播 > 源文件：`eventbus.rs` ```rust pub struct EventBus { sender: broadcast::Sender, // tokio broadcast } ``` 容量 256，所有事件通过 `broadcast` 广播。Tauri IPC 层订阅后转发给前端： ``` Executor 发事件 → EventBus broadcast → IPC 层 listen → app.emit("workflow-event") → 前端实时展示 ``` 事件类型：`NodeStarted` / `NodeCompleted` / `NodeFailed` / `WorkflowCompleted` / `HumanApprovalRequest` / `HumanApprovalResponse` ### 2.7 ConditionEngine（条件分支）— 当前最简实现 > 源文件：`conditions.rs` ```rust pub fn evaluate(expr: &str, _context: &Value) -> Result { if expr.trim() == "true" { return Ok(true); } if expr.trim() == "false" { return Ok(false); } // 其他一律 false（保守拒绝，不静默放行） Ok(false) } ``` 边上的 `condition` 字段目前只认 `"true"` / `"false"` 字面量。升级为真表达式求值是待办（T-260614-11）。 ### 2.8 NodeRegistry（节点注册表）— 工厂模式创建节点 > 源文件：`registry.rs` ```rust pub struct NodeRegistry { factories: HashMap, } ``` 核心方法 `build_dag(def: &DagDef) -> Result`：遍历 DagDef 的节点定义，按 `node_type` 查工厂函数创建真实节点实例，组装成运行时 Dag。 **不实现 Default trait**：原 Default 注册了一个会 panic 的 script 工厂（`unimplemented!`），违反项目铁律「无 panic」。所有调用方必须显式 `new()` + 手动 `register` 真实节点。 --- ## 三、一次完整的执行流程 以 ProjectDetail.vue 中的 3 节点工作流为例： ``` 用户点「运行工作流」 │ ▼ IPC: run_workflow(dag_json) │ ▼ ① NodeRegistry.build_dag(dag_def) → 从 JSON 创建 3 个真实节点实例 + 边 → 运行时 Dag │ ▼ ② Dag.topological_layers() → 算出分层：[["check-env"], ["run-tests"], ["build"]] │ ▼ ③ DagExecutor.new(event_bus, execution_id).run(dag, config) │ ├─ 第 1 层 ["check-env"] │ ├─ set_running("check-env") │ ├─ ScriptNode.execute(ctx) → tokio::process::Command("cmd /C node -v") │ └─ set_completed("check-env") + emit NodeCompleted │ ├─ 第 2 层 ["run-tests"] │ ├─ inputs = { "check-env": 上一步输出 } │ ├─ set_running("run-tests") │ ├─ ScriptNode.execute(ctx) → cmd /C npm test │ └─ set_completed("run-tests") │ └─ 第 3 层 ["build"] ├─ set_running("build") ├─ ScriptNode.execute(ctx) → cmd /C npm run build └─ set_completed("build") + emit WorkflowCompleted │ ▼ ④ 返回 HashMap<节点ID, 输出> + 前端实时看到日志 ``` --- ## 四、现有节点类型 | 节点 | 实现状态 | 能力 | |------|---------|------| | **ScriptNode** | ✅ 真实可用 | 执行 Shell 命令（Windows: `cmd /C`），非零退出码判断为失败 | | **AiNode** | ✅ 真实可用 | 调用 LLM（OpenAI/Anthropic），config 驱动 provider（base_url/api_key/model/prompt），支持上游输入优先 | | **HumanNode** | ✅ 审批闭环 | 阻塞等待人工审批：subscribe→send→select!（响应/超时/取消），execution_id+node_id 双键过滤 | | DockerNode | ❌ 未实现 | 设计意图：Docker 容器内执行命令（隔离构建/测试环境） | | GitNode | ❌ 未实现 | 设计意图：Git 操作（commit/push/merge/分支管理） | | HTTPNode | ❌ 未实现 | 设计意图：发起 HTTP 请求（调外部 API/Webhook） | | NotifyNode | ❌ 未实现 | 设计意图：多渠道通知（桌面/飞书/钉钉/Webhook） | | SubflowNode | ❌ 未实现 | 设计意图：嵌套子工作流（复杂流程拆分复用） | --- ## 五、当前局限 | 局限 | 说明 | 对应待办 | |------|------|---------| | 条件引擎只有 true/false | 无法做 `$.status == 'ok'` 这种真条件分支 | T-260614-11 | | 无断点续跑 | 执行中崩溃后无法从失败节点恢复 | Phase 4 | | 无暂停/恢复 | 只能取消，不能暂停后继续 | Phase 4 | | 无 DAG 可视化编辑器 | 用户只能写 JSON 定义 | Phase 5 | | 持久化仅存定义 | 执行状态不落盘（纯内存），重启丢失 | 待需求驱动 | | 缺 human 节点端到端测试 | 单测绿但前端无 human DAG 入口，审批闭环未端到端验证 | B-03b-R8（P0） |