af_workflow/
spec.rs

1//! Workflow spec data model.
2//!
3//! A workflow is a set of **branches**, each a DAG of **nodes** wired by
4//! **edges**. A node is a typed expression: its `node_type` is a namespaced
5//! verb (`ingress.cron`, `transform.state_append`, `decision.consensus_voting`,
6//! `guard.permission`, `execute.publish`, `sink.log`, …) and
7//! its `config` is the expression's parameters. Composing these typed
8//! expressions into a DAG is the whole programming model — the same shape as the
9//! Python `agent_core/workflow/v2` JSON specs, so existing builtin specs
10//! deserialize unchanged.
11//!
12//! This module is the *data model* only. The executor (scheduling branches,
13//! resolving edge order, dispatching node types, threading run state) is the
14//! next porting phase and will live in a sibling `runtime` module.
15
16use std::collections::BTreeMap;
17
18use serde::{Deserialize, Serialize};
19
20/// A complete workflow specification.
21#[derive(Debug, Clone, Serialize, Deserialize)]
22pub struct Spec {
23    pub spec_id: String,
24    pub version: String,
25
26    #[serde(default)]
27    pub description: String,
28
29    /// Alternate ids this spec answers to.
30    #[serde(default, skip_serializing_if = "Vec::is_empty")]
31    pub aliases: Vec<String>,
32
33    /// JSON-Schema for per-instance configuration, if the spec is templated.
34    #[serde(default, skip_serializing_if = "Option::is_none")]
35    pub instance_config_schema: Option<serde_json::Value>,
36
37    /// Presentation metadata, keyed by node id. Opaque to the engine.
38    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
39    pub display: BTreeMap<String, serde_json::Value>,
40
41    pub branches: Vec<Branch>,
42}
43
44/// One DAG within a spec. The root branch is conventionally `__root__`.
45#[derive(Debug, Clone, Serialize, Deserialize)]
46pub struct Branch {
47    pub branch_id: String,
48    pub nodes: Vec<Node>,
49    #[serde(default)]
50    pub edges: Vec<Edge>,
51}
52
53/// A typed node expression.
54#[derive(Debug, Clone, Serialize, Deserialize)]
55pub struct Node {
56    pub id: String,
57    /// Namespaced node type, e.g. `transform.state_append`. The wire key is
58    /// `type`; we rename to avoid the Rust keyword.
59    #[serde(rename = "type")]
60    pub node_type: String,
61    /// Expression parameters. Shape depends on `node_type`; validated by the
62    /// executor against the node-type registry, not here.
63    #[serde(default)]
64    pub config: serde_json::Value,
65}
66
67impl Node {
68    /// The category half of the node type (`transform` in `transform.foo`).
69    pub fn category(&self) -> &str {
70        self.node_type.split('.').next().unwrap_or(&self.node_type)
71    }
72
73    /// The action half (`foo` in `transform.foo`), or `""` if untyped.
74    pub fn action(&self) -> &str {
75        self.node_type.split_once('.').map(|(_, a)| a).unwrap_or("")
76    }
77}
78
79/// A directed wire from one node's output to another's input.
80#[derive(Debug, Clone, Serialize, Deserialize)]
81pub struct Edge {
82    pub source: String,
83    pub target: String,
84}
85
86#[derive(Debug, thiserror::Error)]
87pub enum SpecError {
88    #[error("spec parse error: {0}")]
89    Parse(#[from] serde_json::Error),
90    #[error("invalid spec '{spec_id}': {reason}")]
91    Invalid { spec_id: String, reason: String },
92}
93
94impl Spec {
95    /// Parse a spec from JSON.
96    pub fn from_json(s: &str) -> Result<Self, SpecError> {
97        Ok(serde_json::from_str(s)?)
98    }
99
100    /// Structural validation that does not require the node-type registry:
101    /// every edge endpoint must reference a node id that exists in its branch.
102    /// (Cycle detection and node-type resolution belong to the executor phase.)
103    pub fn validate_structure(&self) -> Result<(), SpecError> {
104        for branch in &self.branches {
105            let ids: std::collections::HashSet<&str> =
106                branch.nodes.iter().map(|n| n.id.as_str()).collect();
107            for edge in &branch.edges {
108                for (role, endpoint) in [("source", &edge.source), ("target", &edge.target)] {
109                    if !ids.contains(endpoint.as_str()) {
110                        return Err(SpecError::Invalid {
111                            spec_id: self.spec_id.clone(),
112                            reason: format!(
113                                "branch '{}' edge {} '{}' references unknown node",
114                                branch.branch_id, role, endpoint
115                            ),
116                        });
117                    }
118                }
119            }
120        }
121        Ok(())
122    }
123}
124
125#[cfg(test)]
126mod tests {
127    use super::*;
128
129    #[test]
130    fn node_type_split() {
131        let n = Node {
132            id: "x".into(),
133            node_type: "transform.state_append".into(),
134            config: serde_json::json!({}),
135        };
136        assert_eq!(n.category(), "transform");
137        assert_eq!(n.action(), "state_append");
138    }
139
140    #[test]
141    fn validate_catches_dangling_edge() {
142        let spec = Spec {
143            spec_id: "t".into(),
144            version: "1.0".into(),
145            description: String::new(),
146            aliases: vec![],
147            instance_config_schema: None,
148            display: Default::default(),
149            branches: vec![Branch {
150                branch_id: "__root__".into(),
151                nodes: vec![Node {
152                    id: "a".into(),
153                    node_type: "ingress.cron".into(),
154                    config: serde_json::json!({}),
155                }],
156                edges: vec![Edge {
157                    source: "a".into(),
158                    target: "ghost".into(),
159                }],
160            }],
161        };
162        assert!(spec.validate_structure().is_err());
163    }
164}
af_workflow/spec.rs

af_workflow/
spec.rs
