绝尘
8d18918e39
经代码核验发现 9 份设计文档的状态标注严重滞后(标'待实施'但 实际已完整落地),本次批量同步: 已落地(核验确认): - 局部编辑工具:三层防御+三模式完整,仅文本不支持二进制 - 密钥迁移健壮性:空 key 不覆盖+即时迁移补密钥+阻断保存 - AST 符号解析:符号读取工具已注册+基线测试守护 - 查询能力补全:任务/项目/灵感均多维动态查询 - 条件表达式引擎:手写求值器+JSON Path+执行器集成+前端入口 - 工作流脚本边界:命令白名单/黑名单+危险关键词告警 - 消息拆分存储:消息表+全量迁移+读写全部切换 - 消息级溯源:消息 ID+四场景溯源+切读全部完成 部分落地: - 全局事件总线:基建+20 余个发射点就位,消费者未接(空转) 归档不实施: - 规格契约自检:核心价值已被求助协议+自审闸门覆盖,过度设计
257 lines
17 KiB
Markdown
257 lines
17 KiB
Markdown
# 条件表达式引擎设计（R-PD-3 / T-260614-11）
|
||
|
||
> 来源：全局代码 review `docs/05-代码审查/全局代码review-2026-06-15.md` §🔴 P1 需设计 R-PD-3 + 架构洞察第 4 条
|
||
> 性质：✅ **已落地**(2026-06-28 核验:ConditionEngine 手写求值器 + JSON Path/嵌套/数组索引/数值比较 + executor feature flag 集成 + 前端边条件编辑入口 + 中英文翻译)
|
||
> 关联：todo `T-260614-11 条件表达式引擎升级`、R-P2-13（set_skipped/set_waiting 已删）
|
||
|
||
---
|
||
|
||
## 一、现状
|
||
|
||
### 1.1 ConditionEngine 从未被接线
|
||
|
||
`crates/df-workflow/src/conditions.rs:8` 定义了 `ConditionEngine::evaluate(expr, context) -> Result<bool>`，全仓 grep 确认：**除自身定义与单元测试外，零调用**。executor 从不调它，build_dag 只把 `EdgeDef.condition` 原样写入 runtime `Edge.condition`（`registry.rs:62-69`），写入后无人消费。
|
||
|
||
### 1.2 executor 不区分条件边，无条件灌入前驱输出
|
||
|
||
`executor.rs:62-97` 构建 `adjacency_in: target → Vec<source>` 时丢掉 `edge.condition`，只保留 source/target；随后 inputs 收集处（`executor.rs:91-97`）：
|
||
|
||
```rust
|
||
let mut inputs = HashMap::new();
|
||
if let Some(preds) = adjacency_in.get(node_id) {
|
||
for pred_id in preds {
|
||
if let Some(out) = outputs.get(pred_id) {
|
||
inputs.insert(pred_id.clone(), out.clone());
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
逐前驱无条件灌入。条件边与普通边行为完全相同。
|
||
|
||
### 1.3 topological_layers 把条件边计入入度
|
||
|
||
`dag.rs:81-90` 单次遍历边构建入度表，**不检查 `edge.condition`**。条件边 target 的入度照常 +1，BFS 分层照常把它放入某层。
|
||
|
||
### 1.4 复现：`add_edge_with_condition("a","c","false")` 实际 c 永远执行
|
||
|
||
构造 a → c（condition="false"）的 DAG：
|
||
|
||
- `topological_layers`：a 入度 0，c 入度 1；分层为 `[[a],[c]]`
|
||
- 第 0 层执行 a，输出存入 `outputs["a"]`
|
||
- 第 1 层执行 c：`adjacency_in["c"] = ["a"]`，`outputs["a"]` 存在 → `inputs = {"a": <a 的输出>}`，c 照常 `execute`
|
||
- `ConditionEngine::evaluate("false", ...)` 本应返 `Ok(false)`，但**从无调用点**
|
||
|
||
即条件分支这一 DAG 核心能力整体失效，且对用户静默——前端编了条件边，运行结果与无条件等价，没有任何报错或告警。
|
||
|
||
---
|
||
|
||
## 二、根因
|
||
|
||
三个缺口叠加：
|
||
|
||
| 缺口 | 位置 | 表现 |
|
||
|------|------|------|
|
||
| ① expressions 仅 true/false 字面量 | `conditions.rs:16-33` | TODO 列了 JSON Path / 比较 / contains / and-or-not，全未实现；只认 `"true"`/`"false"` 两字面量 |
|
||
| ② executor 无 evaluate 调用 | `executor.rs:91-97` | inputs 收集不读 `edge.condition`，条件边与普通边行为相同 |
|
||
| ③ topological_layers 计入条件边入度 | `dag.rs:81-90` | 条件边 target 的入度照常 +1，BFS 照常分层调度 |
|
||
|
||
review 第 49 行给的修复方向（"executor inputs 收集处用 ConditionEngine.evaluate 过滤；topological_layers 前过滤无效边或执行时按条件短路 target 为 Skipped"）指向 ②③，本文档补全 ①（表达式能力）与 ③ 短路后的终态机制（**set_skipped 已删**，需设计替代，见 §五）。
|
||
|
||
---
|
||
|
||
## 三、表达式引擎方案
|
||
|
||
### 3.1 现状能力
|
||
|
||
`ConditionEngine::evaluate` 当前支持：
|
||
|
||
- `"true"` / `"false"` 字面量（区分大小写，先 `trim()` 去首尾空白）
|
||
- 其余一律 `Ok(false)` + warn（保守拒绝，B-260614-02 已修，默认 true→false）
|
||
|
||
### 3.2 支持范围（目标语法）
|
||
|
||
工作流条件边的实际诉求是"据前驱节点输出决定下游是否执行"。最小可用集：
|
||
|
||
| 语法 | 示例 | 说明 |
|
||
|------|------|------|
|
||
| 字面量 | `true` / `false` | 已有，保留 |
|
||
| 前驱输出访问 | `pred.output.status == 'completed'` | `pred` 为前驱节点 id，`output` 为其 `NodeOutput` 序列化后字段；多前驱时需指定哪个前驱 |
|
||
| 比较 | `==` `!=` `>` `>=` `<` `<=` | 字符串等值 + 数值大小 |
|
||
| 包含 | `pred.output.tags contains 'ai'` | 数组包含 / 字符串子串 |
|
||
| 逻辑组合 | `and` `or` `not` | 括号分组 |
|
||
| 真值判断 | `pred.output.flag` | 布尔字段直接判真（无比较运算符） |
|
||
|
||
`context: &Value` 入参签名已就位（当前 `_context` 未用）。扩展时把当前节点所有前驱输出按 `{ "<pred_id>": <NodeOutput serde> }` 拼成 context 传入即可。
|
||
|
||
### 3.3 引第三方 expr 库 vs 手写最小求值器
|
||
|
||
| 方案 | 优点 | 缺点 |
|
||
|------|------|------|
|
||
| **第三方 `evalexpr`** | 成熟、支持算术/逻辑/函数/变量；API 简单（`eval_with_context`）；MIT；crates.io 下载量稳定 | 新增依赖（df-workflow 当前 0 expr 库，workspace 也无）；语义需对齐（其变量访问语法 `$var` vs 我们要的 `pred.output.x` 点路径）；引入超出条件边需求的算术/函数能力，扩大攻击面（前端可构造任意表达式） |
|
||
| **第三方 `jsonpath_lib` + 自写比较** | JSON Path 标准成熟，路径表达力强 | 仍需自写比较/逻辑层；两套语法拼装复杂度高于纯手写 |
|
||
| **手写最小递归下降求值器** | 零新依赖；语法完全自定（直接支持 `pred.output.x`）；能力边界可控（拒绝算术/函数，只留比较+逻辑+包含）；~150 行可覆盖 §3.2 全部语法 | 自负维护（但语法面小，测试可固化）|
|
||
|
||
**取舍（推荐）：手写最小求值器**。
|
||
|
||
理由：
|
||
1. 条件边诉求面窄（比较 + 逻辑 + 包含 + 前驱输出访问），不需要通用表达式语言的算术/函数能力。
|
||
2. 前端可构造任意条件表达式（run_workflow IPC 接 DagDef），手写小语法面比引通用 expr 库的攻击面更可控——通用 expr 库默认支持函数调用/算术，需额外配置禁用。
|
||
3. df-workflow 当前是零外部表达式依赖的薄 crate，引入 `evalexpr` 对一个"条件分支"单一能力偏重。
|
||
4. 若后续诉求扩张（如需要正则/数学函数），再评估切换第三方库，届时手写求值器的测试可作迁移回归基准。
|
||
|
||
> **决策点 A（需用户确认）**：表达式引擎走手写最小求值器，还是引 `evalexpr`？本文档默认推荐手写。若用户倾向引库，§五的接线方案不变，仅 §三的"实现"段替换。
|
||
|
||
---
|
||
|
||
## 四、接线方案
|
||
|
||
review 给了两种短路粒度，本文档详析：
|
||
|
||
### 4.1 方案一：数据流过滤（Phase1）
|
||
|
||
**改动点**：executor inputs 收集处（`executor.rs:91-97`）。
|
||
|
||
```rust
|
||
// 伪码
|
||
let mut inputs = HashMap::new();
|
||
if let Some(preds_with_cond) = adjacency_in.get(node_id) {
|
||
for (pred_id, cond_opt) in preds_with_cond {
|
||
if let Some(out) = outputs.get(pred_id) {
|
||
// 边有条件 → 求值；条件 false 则不灌入此条边的数据
|
||
if let Some(cond) = cond_opt {
|
||
let ctx = json!({ pred_id: out }); // 单前驱上下文
|
||
match ConditionEngine::evaluate(cond, &ctx) {
|
||
Ok(true) => { inputs.insert(pred_id.clone(), out.clone()); }
|
||
Ok(false) => { /* 跳过此边，不灌入 */ }
|
||
Err(e) => { /* 求值失败兜底，见 §六 */ }
|
||
}
|
||
} else {
|
||
inputs.insert(pred_id.clone(), out.clone());
|
||
}
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
需配套：`adjacency_in` 的 value 从 `Vec<NodeId>` 改为 `Vec<(NodeId, Option<String>)>`，携带 `edge.condition`（构建处 `executor.rs:62-68` 同步改）。
|
||
|
||
**语义**：target **照常执行**，只是某些前驱的数据不灌入。适合"多前驱汇聚、按条件选择部分输入"的场景。
|
||
|
||
**局限**：target 仍被执行。若用户意图是"a 条件不满足时 c 整个不跑"（单条件边、target 唯一前驱），数据流过滤做不到——target 会以空 inputs 执行，语义错位。
|
||
|
||
### 4.2 方案二：调度短路（Phase2）
|
||
|
||
**改动点**：层调度处（`executor.rs:70-119` 的 for 循环）。
|
||
|
||
执行某层前，对层内每个 target 检查：若**所有入边**条件求值均为 false（或其唯一条件边为 false），则 target 标记"条件跳过"终态、不进入 `node_futures`、不发 NodeStarted。
|
||
|
||
**关键约束：set_skipped 已删**。R-P2-13 删了 `set_waiting/set_skipped`（全仓零调用，误导状态机认知），保留 `set_cancelled` 作"唯一受控旁路"。`NodeStatus::Skipped` 枚举值仍在（`types.rs:243`，`as_str` 能输出 `"skipped"`），但**无 setter**。调度短路需要一个"条件不满足、target 不执行"的终态，必须解决这个缺口。
|
||
|
||
### 4.3 两种短路粒度对比
|
||
|
||
| 维度 | 方案一 数据流过滤 | 方案二 调度短路 |
|
||
|------|------------------|----------------|
|
||
| target 是否执行 | 执行（部分输入被过滤） | 不执行（直接跳过） |
|
||
| 适用场景 | 多前驱汇聚、按条件选输入 | 单条件边、条件不满足则 target 整个不跑 |
|
||
| 用户意图匹配 | 部分 | 完整（用户编条件边的典型意图） |
|
||
| 终态机制 | 不需要（target 走 Completed） | **需要新终态**（set_skipped 已删，见 §五） |
|
||
| 改动面 | inputs 收集 + adjacency_in 携带 condition | 层调度 + 终态机制 + 事件（NodeSkipped?） |
|
||
| 风险 | 低（数据流层面，不影响调度） | 中（动调度循环 + 状态机） |
|
||
|
||
---
|
||
|
||
## 五、推荐分阶段
|
||
|
||
**Phase1（数据流过滤，先打通）**：
|
||
|
||
- 范围：§四.1，仅 inputs 收集处接线 ConditionEngine。
|
||
- 效果：多前驱汇聚场景立即可用；单条件边场景 target 仍执行（空 inputs），需文档标注"Phase1 已知局限"。
|
||
- 改动面小、零终态机制冲突、可独立 ship。
|
||
|
||
**Phase2（调度短路，补完整语义）**：
|
||
|
||
- 范围：§四.2，层调度处短路 target。
|
||
- **前置：解决 set_skipped 删除后的终态机制**——见下。
|
||
|
||
### 5.1 set_skipped 删除后的短路机制设计（Phase2 前置）
|
||
|
||
R-P2-13 删 `set_skipped/set_waiting` 时，"条件跳过"这一用例尚未接线（ConditionEngine 从未调用，无消费方），删除合理。现在 Phase2 要用"条件跳过"终态，三个选项：
|
||
|
||
| 选项 | 做法 | 取舍 |
|
||
|------|------|------|
|
||
| **A. 复活 set_skipped 旁路** | `state.rs` 加回 `set_skipped`，与 `set_cancelled` 同型（不经 transition 校验，直接置 `NodeStatus::Skipped`），注释说明"条件跳过专用，区别于 set_cancelled 的人工取消语义" | 最直接；但与 R-P2-13 删除动机（"全仓零调用、误导状态机认知"）冲突——需明确这是新用例落地后的复活，非反复 |
|
||
| **B. 复用 set_cancelled** | 条件短路也走 `set_cancelled`，target 终态为 Cancelled | 语义污染：Cancelled 现专指"人工审批取消"，条件跳过混入会让 `is_cancelled` 判断与前端"取消"语义混乱。**不推荐** |
|
||
| **C. 走 transition 合法转换** | 扩 `is_legal` 加 `(Running, Skipped)`，调度短路前先 `set_running` 再 `transition(Skipped)` | 走正门最干净，但需 target 先进 Running 再转 Skipped（两步），且 NodeStarted 已发→语义噪声（节点"启动后立即跳过"）。或扩 `(Pending, Skipped)` 直接转换，但破坏"Pending 必经 Running"的不变量 |
|
||
|
||
**推荐 A**：复活 `set_skipped` 旁路，注释明确区分两种"非正常终态"语义：
|
||
|
||
- `set_cancelled`：人工审批取消（外部 IPC 触发，节点可能已 Running）
|
||
- `set_skipped`：条件分支跳过（调度层求值条件为 false，target 从未进入 Running）
|
||
|
||
两者均不经 transition 校验（条件短路时 target 在 Pending 态，`Pending→Skipped` 走 transition 会被 `is_legal` 拒，与 `set_cancelled` 同理需旁路）。
|
||
|
||
> **决策点 B（需用户确认）**：Phase2 的条件跳过终态走选项 A（复活 set_skipped 旁路）？本文档默认推荐 A。若用户倾向 C（扩 transition 合法转换），需同步评估 NodeStarted/NodeSkipped 事件序列与"Pending 必经 Running"不变量的取舍。
|
||
|
||
### 5.2 Phase2 配套事件
|
||
|
||
`df-core/events::WorkflowEvent` 当前有 NodeStarted/NodeCompleted/NodeFailed。Phase2 条件短路需补 `NodeSkipped { node_id, reason }`（reason = 哪条边的条件为 false + 表达式），前端可据 reason 渲染"跳过原因"，闭环"对用户非静默"（R-PD-3 原诉求）。
|
||
|
||
---
|
||
|
||
## 六、改动面 + 风险
|
||
|
||
### 6.1 改动面（行号基于当前 HEAD）
|
||
|
||
| Phase | 文件:行 | 改动 |
|
||
|-------|---------|------|
|
||
| 1 | `crates/df-workflow/src/conditions.rs:16-33` | evaluate 扩展为手写最小求值器（§三.3）|
|
||
| 1 | `crates/df-workflow/src/executor.rs:62-68` | `adjacency_in` value 改 `Vec<(NodeId, Option<String>)>`，携带 condition |
|
||
| 1 | `crates/df-workflow/src/executor.rs:91-97` | inputs 收集处调 `ConditionEngine.evaluate`，false 不灌入 |
|
||
| 2 | `crates/df-workflow/src/executor.rs:70-119` | 层调度前求值各 target 入边条件，全 false 则短路 |
|
||
| 2 | `crates/df-workflow/src/state.rs` | 复活 `set_skipped` 旁路（选项 A）|
|
||
| 2 | `crates/df-core/src/events.rs` | 加 `WorkflowEvent::NodeSkipped` |
|
||
| 2 | `crates/df-workflow/src/executor.rs` | 短路时发 NodeSkipped + set_skipped，不进 node_futures |
|
||
|
||
### 6.2 风险
|
||
|
||
**R1：多前驱 context 拼装**。§四.1 伪码用 `json!({ pred_id: out })` 单前驱上下文。多前驱汇聚时，条件表达式需访问哪个前驱？两种设计：
|
||
- (a) 表达式内显式写前驱 id：`a.output.status == 'ok'`——context 拼成所有前驱 `{ "a":..., "b":... }`，求值器按 `a.output.x` 路径取值
|
||
- (b) target 的所有入边条件独立求值，各用单前驱上下文——不支持"跨前驱联合判断"
|
||
|
||
推荐 (a)，context 拼全前驱，求值器路径访问。`pred.output.xxx` 在 review 第 49 行已示意。
|
||
|
||
**R2：条件求值失败的兜底**。求值出错（语法错 / 路径不存在 / 类型不匹配）时返 `Err`，executor 怎么处理？
|
||
|
||
| 选项 | 语义 | 取舍 |
|
||
|------|------|------|
|
||
| **默认 true** | 求值失败 = 放行 | 与 ConditionEngine 当前的"未识别默认 false"（B-260614-02）相反，破坏保守拒绝原则。**不推荐** |
|
||
| **默认 false** | 求值失败 = 拒绝（条件边不灌入 / target 跳过）| 与 B-260614-02 的保守拒绝一致；但用户表达式写错时 target 静默不跑，需配套告警 |
|
||
| **报错中止工作流** | 求值失败 = 工作流 Failed | 最显式，但单个条件边语法错炸整条工作流，可能过激 |
|
||
|
||
**推荐：默认 false + warn 日志 + Phase2 的 NodeSkipped.reason 透出表达式**。条件边本质是"用户声明的过滤规则"，规则写错应保守拒绝（不执行）而非放行，与现有保守拒绝原则对齐；非静默靠 warn + reason 闭环。区别于 B-260614-02 的"引擎未实现"（那是 TODO 完全未做），这里是"用户表达式语法错"——两者都走 false，但 warn 文案区分。
|
||
|
||
> **决策点 C（需用户确认）**：条件求值失败兜底走"默认 false + warn"（推荐）还是"报错中止工作流"？
|
||
|
||
**R3：topological_layers 计入条件边入度的交互**。Phase1 数据流过滤不动 topological_layers，条件边仍计入入度、target 仍分层调度——这与"条件边语义"不冲突（Phase1 只过滤数据，不拦调度）。Phase2 调度短路有两种实现路径：
|
||
- (a) topological_layers 内部按条件过滤无效边（需把前驱输出传进 topological_layers，但分层时前驱尚未执行，条件无法求值）——**不可行**，条件依赖运行时输出
|
||
- (b) 分层照常（含条件边入度），执行时层调度前求值条件短路 target——**可行**，条件求值发生在前驱已完成、target 将执行的边界
|
||
|
||
推荐 (b)。topological_layers 保持纯结构（不掺运行时），条件求值在 executor 调度边界。
|
||
|
||
**R4：循环依赖风险**。条件边若构成 target 的所有入边均条件 false，target 永不执行。这是用户 DAG 的逻辑，引擎照常短路即可，不需特殊处理（与"用户写了死代码节点"同类）。
|
||
|
||
**R5：与 R-PD-2（ScriptNode 任意 shell）的边界**。条件表达式本身不经 shell，纯内存求值，无 R-PD-2 的 shell 注入面。但若 Phase2 引第三方 expr 库（§三.3 若用户改选 evalexpr），其函数调用能力需配置禁用（前端可构造任意表达式）。
|
||
|
||
---
|
||
|
||
## 七、待用户确认的决策点
|
||
|
||
| # | 决策 | 推荐 | 备选 |
|
||
|---|------|------|------|
|
||
| A | 表达式引擎实现 | 手写最小求值器（零依赖、语法可控）| 引 `evalexpr`（成熟、能力全、攻击面大）|
|
||
| B | Phase2 条件跳过终态机制 | 复活 `set_skipped` 旁路（与 `set_cancelled` 同型，语义区分）| 扩 `is_legal` 走 transition 正门（破坏 Pending→Running 不变量）|
|
||
| C | 条件求值失败兜底 | 默认 false + warn + NodeSkipped.reason 透出 | 报错中止整条工作流 |
|
||
|
||
确认后即可按 §五分阶段推进：Phase1（数据流过滤）独立可 ship，Phase2（调度短路 + 终态机制）依赖决策点 B。
|