sz_rust_workflow/scheduling/
candidate_resolver.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2024-2026 SZ-Rust Team
3//
4use std::sync::Arc;
5
6use async_trait::async_trait;
7
8use crate::definition::CandidateStrategy;
9use crate::error::{WorkflowError, WorkflowErrorCode, WorkflowResult};
10use crate::guard::GuardEvaluator;
11use sz_rust_capability::CapabilityRegistry;
12
13/// 候选人解析 trait，对齐 design 2.2.2.6。
14#[async_trait]
15pub trait CandidateResolver: Send + Sync + 'static {
16    async fn resolve(
17        &self,
18        strategy: &CandidateStrategy,
19        context: &serde_json::Value,
20    ) -> WorkflowResult<Vec<String>>;
21}
22
23/// 默认候选人解析器。
24pub struct DefaultCandidateResolver {
25    guard_evaluator: Arc<dyn GuardEvaluator>,
26    capability_registry: Arc<CapabilityRegistry>,
27}
28
29impl DefaultCandidateResolver {
30    pub fn new(
31        guard_evaluator: Arc<dyn GuardEvaluator>,
32        capability_registry: Arc<CapabilityRegistry>,
33    ) -> Self {
34        Self {
35            guard_evaluator,
36            capability_registry,
37        }
38    }
39}
40
41#[async_trait]
42impl CandidateResolver for DefaultCandidateResolver {
43    async fn resolve(
44        &self,
45        strategy: &CandidateStrategy,
46        context: &serde_json::Value,
47    ) -> WorkflowResult<Vec<String>> {
48        let candidates = match strategy {
49            CandidateStrategy::Static { users, roles } => {
50                let mut result: Vec<String> = users.clone();
51                let _ = roles;
52                result.sort();
53                result.dedup();
54                result
55            }
56            CandidateStrategy::Dynamic { expr } => {
57                let val = self.guard_evaluator.evaluate(expr, context).await?;
58                if let serde_json::Value::Array(arr) =
59                    serde_json::to_value(val).unwrap_or(serde_json::Value::Null)
60                {
61                    arr.into_iter()
62                        .filter_map(|v| v.as_str().map(|s| s.to_string()))
63                        .collect()
64                } else {
65                    return Err(WorkflowError::new(
66                        WorkflowErrorCode::CandidateFormatError,
67                        "Dynamic 表达式求值结果非数组",
68                    ));
69                }
70            }
71            CandidateStrategy::Capability {
72                capability_name,
73                args,
74            } => {
75                let result = self
76                    .capability_registry
77                    .call(capability_name, args.clone())
78                    .await
79                    .map_err(|e| {
80                        WorkflowError::new(
81                            WorkflowErrorCode::CandidateFormatError,
82                            format!("能力调用失败：{e}"),
83                        )
84                    })?;
85                match result {
86                    serde_json::Value::Array(arr) => arr
87                        .into_iter()
88                        .filter_map(|v| v.as_str().map(|s| s.to_string()))
89                        .collect(),
90                    _ => {
91                        return Err(WorkflowError::with_field(
92                            WorkflowErrorCode::CandidateFormatError,
93                            "能力返回值非数组",
94                            "capability",
95                            capability_name,
96                        ))
97                    }
98                }
99            }
100        };
101        if candidates.is_empty() {
102            return Err(WorkflowError::new(
103                WorkflowErrorCode::NoCandidates,
104                "候选人为空集合",
105            ));
106        }
107        Ok(candidates)
108    }
109}
110
111#[cfg(test)]
112mod tests {
113    use super::*;
114    use crate::guard::DefaultGuardEvaluator;
115
116    #[tokio::test]
117    async fn static_strategy() {
118        let resolver = DefaultCandidateResolver::new(
119            Arc::new(DefaultGuardEvaluator::default()),
120            Arc::new(CapabilityRegistry::new()),
121        );
122        let strategy = CandidateStrategy::Static {
123            users: vec!["u1".into(), "u2".into()],
124            roles: vec![],
125        };
126        let result = resolver
127            .resolve(&strategy, &serde_json::json!({}))
128            .await
129            .unwrap();
130        assert_eq!(result, vec!["u1", "u2"]);
131    }
132
133    #[tokio::test]
134    async fn static_strategy_empty() {
135        let resolver = DefaultCandidateResolver::new(
136            Arc::new(DefaultGuardEvaluator::default()),
137            Arc::new(CapabilityRegistry::new()),
138        );
139        let strategy = CandidateStrategy::Static {
140            users: vec![],
141            roles: vec![],
142        };
143        let result = resolver.resolve(&strategy, &serde_json::json!({})).await;
144        assert!(result.is_err());
145        assert_eq!(result.unwrap_err().code, WorkflowErrorCode::NoCandidates);
146    }
147
148    #[tokio::test]
149    async fn static_strategy_dedup_and_sort() {
150        let resolver = DefaultCandidateResolver::new(
151            Arc::new(DefaultGuardEvaluator::default()),
152            Arc::new(CapabilityRegistry::new()),
153        );
154        let strategy = CandidateStrategy::Static {
155            users: vec!["b".into(), "a".into(), "b".into()],
156            roles: vec![],
157        };
158        let result = resolver
159            .resolve(&strategy, &serde_json::json!({}))
160            .await
161            .unwrap();
162        assert_eq!(result, vec!["a", "b"]);
163    }
164
165    #[tokio::test]
166    async fn dynamic_strategy_non_array_error() {
167        let resolver = DefaultCandidateResolver::new(
168            Arc::new(DefaultGuardEvaluator::default()),
169            Arc::new(CapabilityRegistry::new()),
170        );
171        let strategy = CandidateStrategy::Dynamic {
172            expr: "true".into(),
173        };
174        let result = resolver.resolve(&strategy, &serde_json::json!({})).await;
175        assert!(result.is_err());
176        assert_eq!(
177            result.unwrap_err().code,
178            WorkflowErrorCode::CandidateFormatError
179        );
180    }
181
182    #[tokio::test]
183    async fn capability_strategy_not_found_error() {
184        let resolver = DefaultCandidateResolver::new(
185            Arc::new(DefaultGuardEvaluator::default()),
186            Arc::new(CapabilityRegistry::new()),
187        );
188        let strategy = CandidateStrategy::Capability {
189            capability_name: "nonexistent".into(),
190            args: serde_json::json!({}),
191        };
192        let result = resolver.resolve(&strategy, &serde_json::json!({})).await;
193        assert!(result.is_err());
194        assert_eq!(
195            result.unwrap_err().code,
196            WorkflowErrorCode::CandidateFormatError
197        );
198    }
199}
sz_rust_workflow/scheduling/candidate_resolver.rs

sz_rust_workflow/scheduling/
candidate_resolver.rs
