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` 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 /// Workflow spec identifier.
24 pub spec_id: String,
25 /// Semantic version string.
26 pub version: String,
27
28 /// Human-readable description.
29 #[serde(default)]
30 pub description: String,
31
32 /// Alternate ids this spec answers to.
33 #[serde(default, skip_serializing_if = "Vec::is_empty")]
34 pub aliases: Vec<String>,
35
36 /// JSON-Schema for per-instance configuration, if the spec is templated.
37 #[serde(default, skip_serializing_if = "Option::is_none")]
38 pub instance_config_schema: Option<serde_json::Value>,
39
40 /// Presentation metadata, keyed by node id. Opaque to the engine.
41 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
42 pub display: BTreeMap<String, serde_json::Value>,
43
44 /// Branches; the first is the root.
45 pub branches: Vec<Branch>,
46}
47
48/// One DAG within a spec. The root branch is conventionally `__root__`.
49#[derive(Debug, Clone, Serialize, Deserialize)]
50pub struct Branch {
51 /// Workflow branch this record belongs to.
52 pub branch_id: String,
53 /// Nodes in this graph.
54 pub nodes: Vec<Node>,
55 /// Directed edges between nodes.
56 #[serde(default)]
57 pub edges: Vec<Edge>,
58}
59
60/// A typed node expression.
61#[derive(Debug, Clone, Serialize, Deserialize)]
62pub struct Node {
63 /// Stable identifier of this record.
64 pub id: String,
65 /// Namespaced node type, e.g. `transform.state_append`. The wire key is
66 /// `type`; we rename to avoid the Rust keyword.
67 #[serde(rename = "type")]
68 pub node_type: String,
69 /// Expression parameters. Shape depends on `node_type`; validated by the
70 /// executor against the node-type registry, not here.
71 #[serde(default)]
72 pub config: serde_json::Value,
73}
74
75impl Node {
76 /// The category half of the node type (`transform` in `transform.foo`).
77 pub fn category(&self) -> &str {
78 self.node_type.split('.').next().unwrap_or(&self.node_type)
79 }
80
81 /// The action half (`foo` in `transform.foo`), or `""` if untyped.
82 pub fn action(&self) -> &str {
83 self.node_type.split_once('.').map(|(_, a)| a).unwrap_or("")
84 }
85}
86
87/// A directed wire from one node's output to another's input.
88#[derive(Debug, Clone, Serialize, Deserialize)]
89pub struct Edge {
90 /// Source node id.
91 pub source: String,
92 /// Target node id.
93 pub target: String,
94}
95
96/// A spec is structurally invalid.
97#[derive(Debug, thiserror::Error)]
98pub enum SpecError {
99 /// Spec parse error.
100 #[error("spec parse error: {0}")]
101 Parse(#[from] serde_json::Error),
102 /// Invalid spec '`spec_id`': `reason`.
103 #[error("invalid spec '{spec_id}': {reason}")]
104 Invalid {
105 /// Spec id.
106 spec_id: String,
107 /// What is wrong.
108 reason: String,
109 },
110}
111
112impl Spec {
113 /// Parse a spec from JSON.
114 pub fn from_json(s: &str) -> Result<Self, SpecError> {
115 Ok(serde_json::from_str(s)?)
116 }
117
118 /// Structural validation that does not require the node-type registry:
119 /// every edge endpoint must reference a node id that exists in its branch.
120 /// (Cycle detection and node-type resolution belong to the executor phase.)
121 pub fn validate_structure(&self) -> Result<(), SpecError> {
122 for branch in &self.branches {
123 let ids: std::collections::HashSet<&str> =
124 branch.nodes.iter().map(|n| n.id.as_str()).collect();
125 for edge in &branch.edges {
126 for (role, endpoint) in [("source", &edge.source), ("target", &edge.target)] {
127 if !ids.contains(endpoint.as_str()) {
128 return Err(SpecError::Invalid {
129 spec_id: self.spec_id.clone(),
130 reason: format!(
131 "branch '{}' edge {} '{}' references unknown node",
132 branch.branch_id, role, endpoint
133 ),
134 });
135 }
136 }
137 }
138 }
139 Ok(())
140 }
141}
142
143#[cfg(test)]
144mod tests {
145 use super::*;
146
147 #[test]
148 fn node_type_split() {
149 let n = Node {
150 id: "x".into(),
151 node_type: "transform.state_append".into(),
152 config: serde_json::json!({}),
153 };
154 assert_eq!(n.category(), "transform");
155 assert_eq!(n.action(), "state_append");
156 }
157
158 #[test]
159 fn validate_catches_dangling_edge() {
160 let spec = Spec {
161 spec_id: "t".into(),
162 version: "1.0".into(),
163 description: String::new(),
164 aliases: vec![],
165 instance_config_schema: None,
166 display: Default::default(),
167 branches: vec![Branch {
168 branch_id: "__root__".into(),
169 nodes: vec![Node {
170 id: "a".into(),
171 node_type: "ingress.cron".into(),
172 config: serde_json::json!({}),
173 }],
174 edges: vec![Edge {
175 source: "a".into(),
176 target: "ghost".into(),
177 }],
178 }],
179 };
180 assert!(spec.validate_structure().is_err());
181 }
182}