af_workflow/
registry.rs

1//! Node-type registry. Port of `platform/registry.py`.
2//!
3//! Maps a node type string (`transform.state_append`) to a factory that builds
4//! a [`StepNode`] from its config. Products extend the registry with their own
5//! business node types without touching this crate.
6
7use std::collections::{BTreeMap, HashMap};
8
9use serde_json::Value;
10
11use crate::node::StepNode;
12use crate::{CapabilityManifest, CapabilityPin, GuardKind};
13
14/// Node construction or registry failure.
15#[derive(Debug, thiserror::Error)]
16pub enum NodeError {
17    /// Unknown node type '' (not registered).
18    #[error("unknown node type '{0}' (not registered)")]
19    UnknownType(String),
20    /// Node '`node_type`' has invalid config: `reason`.
21    #[error("node '{node_type}' has invalid config: {reason}")]
22    InvalidConfig {
23        /// Node type being built.
24        node_type: String,
25        /// What is wrong with the config.
26        reason: String,
27    },
28    /// Invalid capability manifest.
29    #[error("invalid capability manifest: {0}")]
30    InvalidCapability(String),
31    /// Capability '' is already registered.
32    #[error("capability '{0}' is already registered")]
33    DuplicateCapability(String),
34}
35
36/// Builds a step node from its (raw, unresolved) config value.
37pub type StepFactory = fn(config: &Value) -> Result<Box<dyn StepNode>, NodeError>;
38
39/// Config field type in a node schema.
40#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
41#[serde(rename_all = "lowercase")]
42pub enum FieldType {
43    /// String.
44    String,
45    /// Number.
46    Number,
47    /// Boolean.
48    Bool,
49    /// Array.
50    Array,
51    /// Object.
52    Object,
53    /// Any JSON value.
54    Any,
55}
56
57/// One config field.
58#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
59pub struct FieldSpec {
60    /// Field name.
61    pub key: String,
62    /// Expected type.
63    pub ty: FieldType,
64    /// Whether it must be present.
65    pub required: bool,
66}
67
68impl NodeSchema {
69    /// JSON Schema projection used by forms and the same validator rule.
70    pub fn to_json_schema(&self) -> Value {
71        let mut properties = serde_json::Map::new();
72        let mut required = Vec::new();
73        for field in &self.fields {
74            let schema = match field.ty {
75                FieldType::String => serde_json::json!({"type":"string"}),
76                FieldType::Number => serde_json::json!({"type":"number"}),
77                FieldType::Bool => serde_json::json!({"type":"boolean"}),
78                FieldType::Array => serde_json::json!({"type":"array"}),
79                FieldType::Object => serde_json::json!({"type":"object"}),
80                FieldType::Any => serde_json::json!({}),
81            };
82            properties.insert(field.key.clone(), schema);
83            if field.required {
84                required.push(field.key.clone());
85            }
86        }
87        serde_json::json!({"type":"object","properties":properties,"required":required,"additionalProperties":false})
88    }
89}
90
91impl FieldSpec {
92    /// A required field.
93    pub fn required(key: impl Into<String>, ty: FieldType) -> Self {
94        Self {
95            key: key.into(),
96            ty,
97            required: true,
98        }
99    }
100
101    /// An optional field.
102    pub fn optional(key: impl Into<String>, ty: FieldType) -> Self {
103        Self {
104            key: key.into(),
105            ty,
106            required: false,
107        }
108    }
109}
110
111/// Config schema of a node type.
112#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
113pub struct NodeSchema {
114    /// Fields in declaration order.
115    pub fields: Vec<FieldSpec>,
116}
117
118/// Registry of step-node factories. Ingress types are tracked by name only —
119/// the runner owns their execution.
120#[derive(Debug, Clone)]
121pub struct NodeRegistry {
122    steps: HashMap<String, StepFactory>,
123    ingress: std::collections::HashSet<String>,
124    fan_out: std::collections::HashSet<String>,
125    side_effect_guards: std::collections::HashSet<String>,
126    guard_kinds: HashMap<String, GuardKind>,
127    capabilities: BTreeMap<CapabilityPin, CapabilityManifest>,
128    capability_steps: BTreeMap<CapabilityPin, StepFactory>,
129    capability_schemas: BTreeMap<CapabilityPin, NodeSchema>,
130    capability_guards: BTreeMap<CapabilityPin, GuardKind>,
131    capability_fan_out: std::collections::BTreeSet<CapabilityPin>,
132    schemas: HashMap<String, NodeSchema>,
133}
134
135impl NodeRegistry {
136    /// Empty registry — no node types known.
137    pub fn empty() -> Self {
138        Self {
139            steps: HashMap::new(),
140            ingress: Default::default(),
141            fan_out: Default::default(),
142            side_effect_guards: Default::default(),
143            guard_kinds: Default::default(),
144            capabilities: Default::default(),
145            capability_steps: Default::default(),
146            capability_schemas: Default::default(),
147            capability_guards: Default::default(),
148            capability_fan_out: Default::default(),
149            schemas: HashMap::new(),
150        }
151    }
152
153    /// Registry pre-loaded with the generic (non-business) node types.
154    pub fn with_builtins() -> Self {
155        let mut r = Self::empty();
156        crate::builtins::register_builtins(&mut r);
157        r
158    }
159
160    /// Register a step-node factory under `node_type`.
161    pub fn register_step(&mut self, node_type: impl Into<String>, factory: StepFactory) {
162        self.steps.insert(node_type.into(), factory);
163    }
164
165    /// Mark a node type as an ingress source.
166    pub fn register_ingress(&mut self, node_type: impl Into<String>) {
167        self.ingress.insert(node_type.into());
168    }
169
170    /// Mark a node type as fan-out-capable (its `process` may return `FanOut`).
171    /// The validator (R4') bans these upstream of `execute.*`.
172    pub fn register_fan_out(&mut self, node_type: impl Into<String>) {
173        self.fan_out.insert(node_type.into());
174    }
175
176    /// Whether the node type may return `FanOut`.
177    pub fn is_fan_out_capable(&self, node_type: &str) -> bool {
178        self.fan_out.contains(node_type)
179    }
180
181    /// Declare a product node as an authorization guard for `execute.*` side effects.
182    pub fn register_side_effect_guard(&mut self, node_type: impl Into<String>) {
183        self.register_guard(node_type, GuardKind::Authorization);
184    }
185
186    /// Declare the role a guard plays on an action's dominating path.
187    pub fn register_guard(&mut self, node_type: impl Into<String>, kind: GuardKind) {
188        let node_type = node_type.into();
189        self.side_effect_guards.insert(node_type.clone());
190        self.guard_kinds.insert(node_type, kind);
191    }
192
193    /// Guard role of a node type, if registered as a guard.
194    pub fn guard_kind(&self, node_type: &str) -> Option<GuardKind> {
195        self.guard_kinds.get(node_type).copied()
196    }
197
198    /// Whether the node type is a registered guard.
199    pub fn is_side_effect_guard(&self, node_type: &str) -> bool {
200        self.side_effect_guards.contains(node_type)
201    }
202
203    /// Whether the node type is an ingress source.
204    pub fn is_ingress(&self, node_type: &str) -> bool {
205        self.ingress.contains(node_type) || node_type.starts_with("ingress.")
206    }
207
208    /// Whether the node type can be compiled as a step.
209    pub fn is_step(&self, node_type: &str) -> bool {
210        self.steps.contains_key(node_type)
211            || self
212                .capabilities
213                .values()
214                .any(|manifest| manifest.id == node_type)
215    }
216
217    /// Build a step node instance from a spec node's type + config.
218    pub fn build_step(
219        &self,
220        node_type: &str,
221        config: &Value,
222    ) -> Result<Box<dyn StepNode>, NodeError> {
223        let factory = self
224            .steps
225            .get(node_type)
226            .ok_or_else(|| NodeError::UnknownType(node_type.to_string()))?;
227        factory(config)
228    }
229
230    /// Registered step node types.
231    pub fn known_step_types(&self) -> impl Iterator<Item = &str> {
232        self.steps.keys().map(|s| s.as_str())
233    }
234
235    /// Attach a config schema to a node type.
236    pub fn register_schema(&mut self, node_type: impl Into<String>, schema: NodeSchema) {
237        self.schemas.insert(node_type.into(), schema);
238    }
239
240    /// Config schema of a node type.
241    pub fn schema(&self, node_type: &str) -> Option<&NodeSchema> {
242        self.schemas.get(node_type)
243    }
244
245    /// Stable node configuration schemas for authoring catalogs.
246    pub fn authoring_schemas(&self) -> Vec<(String, Value)> {
247        let mut entries = self
248            .schemas
249            .iter()
250            .map(|(id, schema)| (id.clone(), schema.to_json_schema()))
251            .collect::<Vec<_>>();
252        entries.sort_by(|left, right| left.0.cmp(&right.0));
253        entries
254    }
255
256    /// Register the immutable contract for a trigger, expression, guard or action.
257    pub fn register_capability(&mut self, manifest: CapabilityManifest) -> Result<(), NodeError> {
258        manifest
259            .validate()
260            .map_err(|error| NodeError::InvalidCapability(error.to_string()))?;
261        let pin = CapabilityPin {
262            id: manifest.id.clone(),
263            contract_version: manifest.contract_version.clone(),
264            content_digest: manifest.content_digest.clone(),
265        };
266        if self.capabilities.contains_key(&pin) {
267            return Err(NodeError::DuplicateCapability(manifest.id));
268        }
269        self.capabilities.insert(pin, manifest);
270        Ok(())
271    }
272
273    /// Bind executable behavior and optional authoring/guard metadata to one immutable capability pin.
274    pub fn register_capability_implementation(
275        &mut self,
276        pin: CapabilityPin,
277        factory: StepFactory,
278        schema: Option<NodeSchema>,
279        guard: Option<GuardKind>,
280        fan_out: bool,
281    ) -> Result<(), NodeError> {
282        if !self.capabilities.contains_key(&pin) {
283            return Err(NodeError::InvalidCapability(format!(
284                "capability '{}' implementation has no registered manifest at {} ({})",
285                pin.id, pin.contract_version, pin.content_digest
286            )));
287        }
288        if self.capability_steps.contains_key(&pin) {
289            return Err(NodeError::DuplicateCapability(pin.id));
290        }
291        self.capability_steps.insert(pin.clone(), factory);
292        if let Some(schema) = schema {
293            self.capability_schemas.insert(pin.clone(), schema);
294        }
295        if let Some(guard) = guard {
296            self.capability_guards.insert(pin.clone(), guard);
297        }
298        if fan_out {
299            self.capability_fan_out.insert(pin);
300        }
301        Ok(())
302    }
303
304    /// Registered manifest by capability id when exactly one version is selected.
305    pub fn capability(&self, id: &str) -> Option<&CapabilityManifest> {
306        let mut matches = self
307            .capabilities
308            .values()
309            .filter(|manifest| manifest.id == id);
310        let manifest = matches.next()?;
311        matches.next().is_none().then_some(manifest)
312    }
313
314    /// Registered manifest by its immutable pin.
315    pub fn capability_by_pin(&self, pin: &CapabilityPin) -> Option<&CapabilityManifest> {
316        self.capabilities.get(pin)
317    }
318
319    /// Clone this registry with only the exact capability versions pinned by one revision.
320    pub fn for_capability_pins(&self, pins: &[CapabilityPin]) -> Result<Self, NodeError> {
321        let mut selected = self.clone();
322        selected.capabilities.clear();
323        selected
324            .fan_out
325            .retain(|node_type| !pins.iter().any(|pin| pin.id == *node_type));
326        for pin in pins {
327            let manifest = self.capability_by_pin(pin).ok_or_else(|| {
328                NodeError::InvalidCapability(format!(
329                    "capability '{}' is unavailable at {} ({})",
330                    pin.id, pin.contract_version, pin.content_digest
331                ))
332            })?;
333            selected.capabilities.insert(pin.clone(), manifest.clone());
334            let versions = self
335                .capabilities
336                .keys()
337                .filter(|candidate| candidate.id == pin.id)
338                .count();
339            if let Some(factory) = self.capability_steps.get(pin) {
340                selected.steps.insert(pin.id.clone(), *factory);
341            } else if versions > 1 && self.steps.contains_key(&pin.id) {
342                return Err(NodeError::InvalidCapability(format!(
343                    "capability '{}' has multiple versions but no executable implementation for {} ({})",
344                    pin.id, pin.contract_version, pin.content_digest
345                )));
346            }
347            if let Some(schema) = self.capability_schemas.get(pin) {
348                selected.schemas.insert(pin.id.clone(), schema.clone());
349            } else if versions > 1 && self.schemas.contains_key(&pin.id) {
350                return Err(NodeError::InvalidCapability(format!(
351                    "capability '{}' has multiple versions but no authoring schema for {} ({})",
352                    pin.id, pin.contract_version, pin.content_digest
353                )));
354            }
355            if let Some(kind) = self.capability_guards.get(pin).copied() {
356                selected.side_effect_guards.insert(pin.id.clone());
357                selected.guard_kinds.insert(pin.id.clone(), kind);
358            } else if manifest.kind == crate::CapabilityKind::Guard
359                && (versions > 1 || !self.guard_kinds.contains_key(&pin.id))
360            {
361                return Err(NodeError::InvalidCapability(format!(
362                    "guard capability '{}' has no guard implementation for {} ({})",
363                    pin.id, pin.contract_version, pin.content_digest
364                )));
365            }
366            if self.capability_fan_out.contains(pin) {
367                selected.fan_out.insert(pin.id.clone());
368            }
369        }
370        Ok(selected)
371    }
372
373    /// Every registered manifest.
374    pub fn capability_manifests(&self) -> impl Iterator<Item = &CapabilityManifest> {
375        self.capabilities.values()
376    }
377}
378
379#[cfg(test)]
380mod tests {
381    use super::*;
382    use crate::{CapabilityKind, Effect, IdempotencyMode};
383
384    struct Noop;
385
386    #[async_trait::async_trait]
387    impl StepNode for Noop {
388        async fn process(
389            &self,
390            event: &crate::Event,
391            _ctx: &crate::WorkflowContext,
392        ) -> crate::StepResult {
393            crate::StepResult::Pass(event.clone())
394        }
395    }
396
397    fn noop(_: &Value) -> Result<Box<dyn StepNode>, NodeError> {
398        Ok(Box::new(Noop))
399    }
400
401    #[test]
402    fn capability_versions_are_indexed_and_selected_by_full_pin() {
403        let manifest = |version: &str, digest: &str| {
404            CapabilityManifest::action(
405                "action.versioned",
406                version,
407                digest,
408                Effect::ExternalWrite,
409                IdempotencyMode::Native,
410                true,
411            )
412        };
413        let v1 = manifest("1", "digest-v1");
414        let v2 = manifest("2", "digest-v2");
415        let pin = CapabilityPin {
416            id: v1.id.clone(),
417            contract_version: v1.contract_version.clone(),
418            content_digest: v1.content_digest.clone(),
419        };
420        let mut registry = NodeRegistry::empty();
421        registry.register_capability(v1.clone()).unwrap();
422        registry.register_capability(v2).unwrap();
423
424        assert!(registry.capability("action.versioned").is_none());
425        assert_eq!(registry.capability_by_pin(&pin), Some(&v1));
426        let selected = registry.for_capability_pins(&[pin]).unwrap();
427        assert_eq!(selected.capability("action.versioned"), Some(&v1));
428    }
429
430    #[test]
431    fn versioned_capability_selects_exact_factory_schema_and_guard() {
432        let manifest = |version: &str, digest: &str| {
433            let mut manifest = CapabilityManifest::action(
434                "guard.versioned",
435                version,
436                digest,
437                Effect::Pure,
438                IdempotencyMode::Native,
439                true,
440            );
441            manifest.kind = CapabilityKind::Guard;
442            manifest
443        };
444        let v1 = manifest("1", "digest-v1");
445        let v2 = manifest("2", "digest-v2");
446        let pin = CapabilityPin {
447            id: v1.id.clone(),
448            contract_version: v1.contract_version.clone(),
449            content_digest: v1.content_digest.clone(),
450        };
451        let schema = NodeSchema {
452            fields: vec![FieldSpec::required("approved", FieldType::Bool)],
453        };
454        let mut registry = NodeRegistry::empty();
455        registry.register_capability(v1).unwrap();
456        registry.register_capability(v2).unwrap();
457        registry
458            .register_capability_implementation(
459                pin.clone(),
460                noop,
461                Some(schema.clone()),
462                Some(GuardKind::Authorization),
463                false,
464            )
465            .unwrap();
466
467        let selected = registry.for_capability_pins(&[pin]).unwrap();
468        assert_eq!(selected.schema("guard.versioned"), Some(&schema));
469        assert_eq!(
470            selected.guard_kind("guard.versioned"),
471            Some(GuardKind::Authorization)
472        );
473        assert!(selected.build_step("guard.versioned", &Value::Null).is_ok());
474    }
475
476    #[test]
477    fn versioned_capability_selects_fan_out_by_full_pin() {
478        let manifest = |version: &str, digest: &str| {
479            CapabilityManifest::action(
480                "transform.versioned",
481                version,
482                digest,
483                Effect::Pure,
484                IdempotencyMode::Native,
485                true,
486            )
487        };
488        let v1 = manifest("1", "digest-v1");
489        let v2 = manifest("2", "digest-v2");
490        let pin = |manifest: &CapabilityManifest| CapabilityPin {
491            id: manifest.id.clone(),
492            contract_version: manifest.contract_version.clone(),
493            content_digest: manifest.content_digest.clone(),
494        };
495        let mut registry = NodeRegistry::empty();
496        registry.register_capability(v1.clone()).unwrap();
497        registry.register_capability(v2.clone()).unwrap();
498        registry
499            .register_capability_implementation(pin(&v1), noop, None, None, true)
500            .unwrap();
501        registry
502            .register_capability_implementation(pin(&v2), noop, None, None, false)
503            .unwrap();
504
505        assert!(registry
506            .for_capability_pins(&[pin(&v1)])
507            .unwrap()
508            .is_fan_out_capable("transform.versioned"));
509        assert!(!registry
510            .for_capability_pins(&[pin(&v2)])
511            .unwrap()
512            .is_fan_out_capable("transform.versioned"));
513    }
514}
af_workflow/registry.rs

af_workflow/
registry.rs
