sz_rust_workflow/definition/
validator.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2024-2026 SZ-Rust Team
3//
4use std::collections::{HashMap, HashSet, VecDeque};
5use std::sync::Arc;
6
7use async_trait::async_trait;
8use serde::{Deserialize, Serialize};
9
10use crate::error::{WorkflowErrorCode, WorkflowResult};
11
12use super::models::FlowDefinition;
13use super::node::{NodeConfig, NodeType};
14
15/// 插件可用性检查 trait（解耦 AddonLoader 具体类型）。
16#[async_trait]
17pub trait PluginChecker: Send + Sync + 'static {
18    /// 检查插件是否已启用。
19    async fn is_plugin_enabled(&self, plugin_name: &str) -> bool;
20}
21
22/// Noop 实现：所有插件均视为可用（测试用）。
23pub struct NoopPluginChecker;
24
25#[async_trait]
26impl PluginChecker for NoopPluginChecker {
27    async fn is_plugin_enabled(&self, _plugin_name: &str) -> bool {
28        true
29    }
30}
31
32/// 为 `sz_rust_addons_loader::AddonLoader` 实现 `PluginChecker`。
33#[async_trait]
34impl PluginChecker for sz_rust_addons_loader::AddonLoader {
35    async fn is_plugin_enabled(&self, plugin_name: &str) -> bool {
36        self.is_enabled(plugin_name).unwrap_or(false)
37    }
38}
39
40/// 校验问题严重级别。
41#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
42#[serde(rename_all = "lowercase")]
43pub enum IssueSeverity {
44    Error,
45    Warning,
46}
47
48/// 校验问题。
49#[derive(Debug, Clone, Serialize)]
50pub struct ValidationIssue {
51    pub code: WorkflowErrorCode,
52    pub severity: IssueSeverity,
53    pub message: String,
54    #[serde(skip_serializing_if = "Option::is_none")]
55    pub node_id: Option<String>,
56}
57
58impl ValidationIssue {
59    fn error(code: WorkflowErrorCode, message: impl Into<String>) -> Self {
60        Self {
61            code,
62            severity: IssueSeverity::Error,
63            message: message.into(),
64            node_id: None,
65        }
66    }
67
68    fn error_at(
69        code: WorkflowErrorCode,
70        message: impl Into<String>,
71        node_id: impl Into<String>,
72    ) -> Self {
73        Self {
74            code,
75            severity: IssueSeverity::Error,
76            message: message.into(),
77            node_id: Some(node_id.into()),
78        }
79    }
80
81    fn warning(
82        code: WorkflowErrorCode,
83        message: impl Into<String>,
84        node_id: impl Into<String>,
85    ) -> Self {
86        Self {
87            code,
88            severity: IssueSeverity::Warning,
89            message: message.into(),
90            node_id: Some(node_id.into()),
91        }
92    }
93}
94
95/// 流程定义校验器。
96pub struct DefinitionValidator {
97    plugin_checker: Arc<dyn PluginChecker>,
98}
99
100impl DefinitionValidator {
101    pub fn new(plugin_checker: Arc<dyn PluginChecker>) -> Self {
102        Self { plugin_checker }
103    }
104
105    /// 使用 noop checker 构造（不校验插件可用性）。
106    pub fn new_noop() -> Self {
107        Self::new(Arc::new(NoopPluginChecker))
108    }
109
110    /// 执行全部校验，返回所有 issue（Error 与 Warning）。
111    ///
112    /// 调用方根据是否存在 `IssueSeverity::Error` 决定是否接受定义。
113    pub async fn validate(&self, def: &FlowDefinition) -> WorkflowResult<Vec<ValidationIssue>> {
114        let mut issues = Vec::new();
115        issues.extend(self.validate_structure(def));
116        issues.extend(self.validate_reachability(def));
117        issues.extend(self.validate_termination(def));
118        issues.extend(self.validate_plugin_refs(def).await?);
119        Ok(issues)
120    }
121
122    /// 2.5 结构完整性校验。
123    fn validate_structure(&self, def: &FlowDefinition) -> Vec<ValidationIssue> {
124        let mut issues = Vec::new();
125        let mut id_set: HashSet<&str> = HashSet::new();
126
127        for n in &def.nodes {
128            if !id_set.insert(n.node_id.as_str()) {
129                issues.push(ValidationIssue::error_at(
130                    WorkflowErrorCode::StructureIncomplete,
131                    format!("node_id 重复：{}", n.node_id),
132                    &n.node_id,
133                ));
134            }
135        }
136
137        let has_start = def.nodes.iter().any(|n| n.node_type == NodeType::Start);
138        let has_end = def.nodes.iter().any(|n| n.node_type == NodeType::End);
139        if !has_start {
140            issues.push(ValidationIssue::error(
141                WorkflowErrorCode::StructureIncomplete,
142                "缺少 start 节点",
143            ));
144        }
145        if !has_end {
146            issues.push(ValidationIssue::error(
147                WorkflowErrorCode::StructureIncomplete,
148                "缺少 end 节点",
149            ));
150        }
151
152        if !id_set.contains(def.start_node.as_str()) {
153            issues.push(ValidationIssue::error(
154                WorkflowErrorCode::StructureIncomplete,
155                format!("start_node 引用不存在的节点：{}", def.start_node),
156            ));
157        }
158
159        issues
160    }
161
162    /// 2.6 可达性校验（BFS，识别不可达节点，Warning 级）。
163    fn validate_reachability(&self, def: &FlowDefinition) -> Vec<ValidationIssue> {
164        let mut issues = Vec::new();
165        let node_map: HashMap<&str, &super::node::Node> =
166            def.nodes.iter().map(|n| (n.node_id.as_str(), n)).collect();
167
168        let mut reachable: HashSet<&str> = HashSet::new();
169        let mut queue: VecDeque<&str> = VecDeque::new();
170        if node_map.contains_key(def.start_node.as_str()) {
171            queue.push_back(def.start_node.as_str());
172            reachable.insert(def.start_node.as_str());
173        }
174        while let Some(cur) = queue.pop_front() {
175            if let Some(node) = node_map.get(cur) {
176                for succ in node.successors() {
177                    if node_map.contains_key(succ) && reachable.insert(succ) {
178                        queue.push_back(succ);
179                    }
180                }
181            }
182        }
183
184        for n in &def.nodes {
185            if !reachable.contains(n.node_id.as_str()) {
186                issues.push(ValidationIssue::warning(
187                    WorkflowErrorCode::UnreachableNode,
188                    format!("节点不可达：{}", n.node_id),
189                    &n.node_id,
190                ));
191            }
192        }
193        issues
194    }
195
196    /// 2.6 终止性校验（所有可达节点须能到达某 end 节点）。
197    fn validate_termination(&self, def: &FlowDefinition) -> Vec<ValidationIssue> {
198        let mut issues = Vec::new();
199        let node_map: HashMap<&str, &super::node::Node> =
200            def.nodes.iter().map(|n| (n.node_id.as_str(), n)).collect();
201        let end_nodes: HashSet<&str> = def
202            .nodes
203            .iter()
204            .filter(|n| n.node_type == NodeType::End)
205            .map(|n| n.node_id.as_str())
206            .collect();
207
208        for n in &def.nodes {
209            if end_nodes.contains(n.node_id.as_str()) {
210                continue;
211            }
212            if !self.can_reach_end(n.node_id.as_str(), &node_map, &end_nodes) {
213                issues.push(ValidationIssue::error_at(
214                    WorkflowErrorCode::CannotTerminate,
215                    format!("节点无法到达 end 节点：{}", n.node_id),
216                    &n.node_id,
217                ));
218            }
219        }
220        issues
221    }
222
223    fn can_reach_end(
224        &self,
225        start: &str,
226        node_map: &HashMap<&str, &super::node::Node>,
227        end_nodes: &HashSet<&str>,
228    ) -> bool {
229        let mut visited: HashSet<&str> = HashSet::new();
230        let mut queue: VecDeque<&str> = VecDeque::new();
231        queue.push_back(start);
232        visited.insert(start);
233        while let Some(cur) = queue.pop_front() {
234            if end_nodes.contains(cur) {
235                return true;
236            }
237            if let Some(node) = node_map.get(cur) {
238                for succ in node.successors() {
239                    if visited.insert(succ) {
240                        queue.push_back(succ);
241                    }
242                }
243            }
244        }
245        false
246    }
247
248    /// 2.7 插件节点引用校验。
249    async fn validate_plugin_refs(
250        &self,
251        def: &FlowDefinition,
252    ) -> WorkflowResult<Vec<ValidationIssue>> {
253        let mut issues = Vec::new();
254        for n in &def.nodes {
255            if let NodeConfig::Plugin {
256                capability_name,
257                capability_version_range,
258                ..
259            } = &n.config
260            {
261                let parts: Vec<&str> = capability_name.splitn(2, '.').collect();
262                if parts.len() != 2 || parts[0].is_empty() || parts[1].is_empty() {
263                    issues.push(ValidationIssue::error_at(
264                        WorkflowErrorCode::PluginUnavailable,
265                        format!("capability_name 不符合 {{plugin}}.{{capability}} 命名规范：{capability_name}"),
266                        &n.node_id,
267                    ));
268                    continue;
269                }
270                let plugin_name = parts[0];
271                if !self.plugin_checker.is_plugin_enabled(plugin_name).await {
272                    issues.push(ValidationIssue::error_at(
273                        WorkflowErrorCode::PluginUnavailable,
274                        format!("插件未启用：{plugin_name}"),
275                        &n.node_id,
276                    ));
277                }
278                if let Err(e) = semver::VersionReq::parse(capability_version_range) {
279                    issues.push(ValidationIssue::error_at(
280                        WorkflowErrorCode::PluginUnavailable,
281                        format!("非法版本范围：{capability_version_range} ({e})"),
282                        &n.node_id,
283                    ));
284                }
285            }
286        }
287        Ok(issues)
288    }
289}
290
291#[cfg(test)]
292mod tests {
293    use super::super::models::{DefinitionFormat, FlowDefinition};
294    use super::*;
295
296    use super::super::parser::DefinitionParser;
297
298    fn parse(yaml: &str) -> FlowDefinition {
299        DefinitionParser::new()
300            .parse(yaml, DefinitionFormat::Yaml)
301            .unwrap()
302    }
303
304    const VALID_DEF: &str = r#"
305flow_key: leave_req
306version: "1.0.0"
307name: 请假
308nodes:
309  - node_id: start
310    node_type: start
311    kind: start
312    next: end
313  - node_id: end
314    node_type: end
315    kind: end
316start_node: start
317"#;
318
319    #[tokio::test]
320    async fn valid_definition_no_errors() {
321        let def = parse(VALID_DEF);
322        let v = DefinitionValidator::new_noop();
323        let issues = v.validate(&def).await.unwrap();
324        let errors: Vec<_> = issues
325            .iter()
326            .filter(|i| i.severity == IssueSeverity::Error)
327            .collect();
328        assert!(errors.is_empty(), "不应有 Error 级 issue: {:?}", issues);
329    }
330
331    #[tokio::test]
332    async fn missing_start_node() {
333        let yaml = r#"
334flow_key: leave_req
335version: "1.0.0"
336name: 请假
337nodes:
338  - node_id: end
339    node_type: end
340    kind: end
341start_node: end
342"#;
343        let def = parse(yaml);
344        let v = DefinitionValidator::new_noop();
345        let issues = v.validate(&def).await.unwrap();
346        assert!(issues.iter().any(
347            |i| i.code == WorkflowErrorCode::StructureIncomplete && i.message.contains("start")
348        ));
349    }
350
351    #[tokio::test]
352    async fn duplicate_node_id() {
353        let yaml = r#"
354flow_key: leave_req
355version: "1.0.0"
356name: 请假
357nodes:
358  - node_id: start
359    node_type: start
360    kind: start
361    next: end
362  - node_id: start
363    node_type: start
364    kind: start
365    next: end
366  - node_id: end
367    node_type: end
368    kind: end
369start_node: start
370"#;
371        let def = parse(yaml);
372        let v = DefinitionValidator::new_noop();
373        let issues = v.validate(&def).await.unwrap();
374        assert!(issues.iter().any(
375            |i| i.code == WorkflowErrorCode::StructureIncomplete && i.message.contains("重复")
376        ));
377    }
378
379    #[tokio::test]
380    async fn unreachable_node_warning() {
381        let yaml = r#"
382flow_key: leave_req
383version: "1.0.0"
384name: 请假
385nodes:
386  - node_id: start
387    node_type: start
388    kind: start
389    next: end
390  - node_id: orphan
391    node_type: approval
392    kind: approval
393    approval_strategy: and_sign
394    candidate_strategy:
395      type: static
396      users: ["u1"]
397    next: end
398  - node_id: end
399    node_type: end
400    kind: end
401start_node: start
402"#;
403        let def = parse(yaml);
404        let v = DefinitionValidator::new_noop();
405        let issues = v.validate(&def).await.unwrap();
406        assert!(issues
407            .iter()
408            .any(|i| i.code == WorkflowErrorCode::UnreachableNode
409                && i.severity == IssueSeverity::Warning));
410    }
411
412    #[tokio::test]
413    async fn cannot_terminate() {
414        let yaml = r#"
415flow_key: leave_req
416version: "1.0.0"
417name: 请假
418nodes:
419  - node_id: start
420    node_type: start
421    kind: start
422    next: loop
423  - node_id: loop
424    node_type: approval
425    kind: approval
426    approval_strategy: and_sign
427    candidate_strategy:
428      type: static
429      users: ["u1"]
430    next: loop
431start_node: start
432"#;
433        let def = parse(yaml);
434        let v = DefinitionValidator::new_noop();
435        let issues = v.validate(&def).await.unwrap();
436        assert!(issues
437            .iter()
438            .any(|i| i.code == WorkflowErrorCode::CannotTerminate));
439    }
440
441    #[tokio::test]
442    async fn plugin_bad_capability_name() {
443        let yaml = r#"
444flow_key: leave_req
445version: "1.0.0"
446name: 请假
447nodes:
448  - node_id: start
449    node_type: start
450    kind: start
451    next: p1
452  - node_id: p1
453    node_type: plugin
454    kind: plugin
455    capability_name: invalid_no_dot
456    capability_version_range: "*"
457    fault_strategy: fail
458    next: end
459  - node_id: end
460    node_type: end
461    kind: end
462start_node: start
463"#;
464        let def = parse(yaml);
465        let v = DefinitionValidator::new_noop();
466        let issues = v.validate(&def).await.unwrap();
467        assert!(issues
468            .iter()
469            .any(|i| i.code == WorkflowErrorCode::PluginUnavailable
470                && i.message.contains("命名规范")));
471    }
472
473    #[tokio::test]
474    async fn plugin_bad_version_range() {
475        let yaml = r#"
476flow_key: leave_req
477version: "1.0.0"
478name: 请假
479nodes:
480  - node_id: start
481    node_type: start
482    kind: start
483    next: p1
484  - node_id: p1
485    node_type: plugin
486    kind: plugin
487    capability_name: crm.search
488    capability_version_range: "not_a_valid_range!!!"
489    fault_strategy: fail
490    next: end
491  - node_id: end
492    node_type: end
493    kind: end
494start_node: start
495"#;
496        let def = parse(yaml);
497        let v = DefinitionValidator::new_noop();
498        let issues = v.validate(&def).await.unwrap();
499        assert!(issues
500            .iter()
501            .any(|i| i.code == WorkflowErrorCode::PluginUnavailable
502                && i.message.contains("版本范围")));
503    }
504
505    struct DisabledChecker;
506    #[async_trait]
507    impl PluginChecker for DisabledChecker {
508        async fn is_plugin_enabled(&self, _: &str) -> bool {
509            false
510        }
511    }
512
513    #[tokio::test]
514    async fn plugin_not_enabled() {
515        let yaml = r#"
516flow_key: leave_req
517version: "1.0.0"
518name: 请假
519nodes:
520  - node_id: start
521    node_type: start
522    kind: start
523    next: p1
524  - node_id: p1
525    node_type: plugin
526    kind: plugin
527    capability_name: crm.search
528    capability_version_range: "^1.0"
529    fault_strategy: fail
530    next: end
531  - node_id: end
532    node_type: end
533    kind: end
534start_node: start
535"#;
536        let def = parse(yaml);
537        let v = DefinitionValidator::new(Arc::new(DisabledChecker));
538        let issues = v.validate(&def).await.unwrap();
539        assert!(issues.iter().any(
540            |i| i.code == WorkflowErrorCode::PluginUnavailable && i.message.contains("未启用")
541        ));
542    }
543}
sz_rust_workflow/definition/validator.rs

sz_rust_workflow/definition/
validator.rs
