Vakyartha documentation

1use std::collections::{BTreeMap, VecDeque};
2
3use crate::types::FlowDef;
4
5#[derive(Debug, thiserror::Error)]
6pub enum ParseError {
7 #[error("toml error: {0}")]
8 Toml(#[from] toml::de::Error),
9 #[error("flow has no nodes")]
10 Empty,
11 #[error("duplicate node id '{0}'")]
12 Duplicate(String),
13 #[error("node '{0}' has unknown type '{1}' (agent|bash|approval|merge)")]
14 UnknownType(String, String),
15 #[error("node '{0}' of type '{1}' is missing required field {2}")]
16 MissingField(String, String, &'static str),
17 #[error("node '{0}' depends on unknown node '{1}'")]
18 UnknownDep(String, String),
19 #[error("node '{0}' depends on itself")]
20 SelfDep(String),
21 #[error("cycle detected involving: {0}")]
22 Cycle(String),
23}
24
25pub const NODE_TYPES: [&str; 4] = ["agent", "bash", "approval", "merge"];
26
27#[derive(serde::Deserialize)]
28struct FlowFile {
29 flow: FlowHeader,
30 #[serde(default)]
31 nodes: Vec<crate::types::NodeDef>,
32}
33
34#[derive(serde::Deserialize)]
35struct FlowHeader {
36 name: String,
37 #[serde(default)]
38 description: String,
39}
40
41pub fn parse_flow(toml_str: &str) -> Result<FlowDef, ParseError> {
42 let file: FlowFile = toml::from_str(toml_str)?;
43 let mut flow = FlowDef {
44 name: file.flow.name,
45 description: file.flow.description,
46 nodes: file.nodes,
47 };
48 validate(&mut flow)?;
49 Ok(flow)
50}
51
52pub fn validate(flow: &mut FlowDef) -> Result<(), ParseError> {
53 if flow.nodes.is_empty() {
54 return Err(ParseError::Empty);
55 }
56 let mut ids = BTreeMap::new();
57 for n in &flow.nodes {
58 if ids.contains_key(&n.id) {
59 return Err(ParseError::Duplicate(n.id.clone()));
60 }
61 if !NODE_TYPES.contains(&n.r#type.as_str()) {
62 return Err(ParseError::UnknownType(n.id.clone(), n.r#type.clone()));
63 }
64 match n.r#type.as_str() {
65 "agent" if n.prompt.is_none() => {
66 return Err(ParseError::MissingField(
67 n.id.clone(),
68 "agent".into(),
69 "prompt",
70 ));
71 }
72 "bash" if n.command.is_none() => {
73 return Err(ParseError::MissingField(
74 n.id.clone(),
75 "bash".into(),
76 "command",
77 ));
78 }
79 "approval" if n.message.is_none() => {
80 return Err(ParseError::MissingField(
81 n.id.clone(),
82 "approval".into(),
83 "message",
84 ));
85 }
86 _ => {}
87 }
88 ids.insert(n.id.clone(), ());
89 }
90 for n in &flow.nodes {
91 for dep in &n.deps {
92 if dep == &n.id {
93 return Err(ParseError::SelfDep(n.id.clone()));
94 }
95 if !ids.contains_key(dep) {
96 return Err(ParseError::UnknownDep(n.id.clone(), dep.clone()));
97 }
98 }
99 }
100
101 // Template references imply dependencies.
102 let mut resolved: Vec<crate::types::NodeDef> = Vec::with_capacity(flow.nodes.len());
103 for n in &flow.nodes {
104 let mut node = n.clone();
105 for (_, reference) in template_refs(&node) {
106 if !ids.contains_key(&reference) {
107 return Err(ParseError::UnknownDep(node.id.clone(), reference));
108 }
109 if reference != node.id && !node.deps.contains(&reference) {
110 node.deps.push(reference);
111 }
112 }
113 resolved.push(node);
114 }
115 flow.nodes = resolved;
116
117 // Kahn's algorithm: cycle detection + layer assignment.
118 let layers = layers(flow)?;
119 let _ = layers;
120 Ok(())
121}
122
123/// Extracts `{{identifier}}` references from a node's prompt or command.
124fn template_refs(node: &crate::types::NodeDef) -> Vec<(String, String)> {
125 let sources = [
126 ("prompt", node.prompt.as_deref().unwrap_or_default()),
127 ("command", node.command.as_deref().unwrap_or_default()),
128 ];
129 let mut out = Vec::new();
130 for (field, text) in sources {
131 let bytes = text.as_bytes();
132 let mut i = 0;
133 while i + 3 < bytes.len() {
134 if &text[i..i + 2] == "{{"
135 && let Some(end_offset) = text[i + 2..].find("}}")
136 {
137 let inner = text[i + 2..i + 2 + end_offset].trim();
138 if !inner.is_empty()
139 && inner
140 .chars()
141 .all(|c| c.is_alphanumeric() || c == '_' || c == '-')
142 {
143 out.push((field.to_string(), inner.to_string()));
144 }
145 i += 2 + end_offset + 2;
146 } else {
147 i += 1;
148 }
149 }
150 }
151 out
152}
153
154/// Topological layers: nodes in the same layer have no mutual dependencies
155/// and may run concurrently.
156pub fn layers(flow: &FlowDef) -> Result<Vec<Vec<String>>, ParseError> {
157 let mut indegree: BTreeMap<&str, usize> = BTreeMap::new();
158 let mut dependents: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
159 for n in &flow.nodes {
160 indegree.entry(n.id.as_str()).or_insert(0);
161 for dep in &n.deps {
162 *indegree.entry(n.id.as_str()).or_insert(0) += 1;
163 dependents
164 .entry(dep.as_str())
165 .or_default()
166 .push(n.id.as_str());
167 }
168 }
169
170 let mut queue: VecDeque<&str> = indegree
171 .iter()
172 .filter(|(_, d)| **d == 0)
173 .map(|(id, _)| *id)
174 .collect();
175 let mut layers: Vec<Vec<String>> = Vec::new();
176 let mut placed = 0usize;
177
178 while !queue.is_empty() {
179 let mut next_queue = VecDeque::new();
180 let mut layer = Vec::new();
181 for id in queue.drain(..) {
182 layer.push(id.to_string());
183 placed += 1;
184 if let Some(deps) = dependents.get(id) {
185 for d in deps {
186 if let Some(e) = indegree.get_mut(*d) {
187 *e -= 1;
188 if *e == 0 {
189 next_queue.push_back(*d);
190 }
191 }
192 }
193 }
194 }
195 layers.push(layer);
196 queue = next_queue;
197 }
198
199 if placed != flow.nodes.len() {
200 let stuck: Vec<String> = indegree
201 .iter()
202 .filter(|(_, d)| **d > 0)
203 .map(|(id, _)| id.to_string())
204 .collect();
205 return Err(ParseError::Cycle(stuck.join(", ")));
206 }
207 Ok(layers)
208}
209

1use std::collections::{BTreeMap, VecDeque};
2
3use crate::types::FlowDef;
4
5#[derive(Debug, thiserror::Error)]
6pub enum ParseError {
7 #[error("toml error: {0}")]
8 Toml(#[from] toml::de::Error),
9 #[error("flow has no nodes")]
10 Empty,
11 #[error("duplicate node id '{0}'")]
12 Duplicate(String),
13 #[error("node '{0}' has unknown type '{1}' (agent|bash|approval|merge)")]
14 UnknownType(String, String),
15 #[error("node '{0}' of type '{1}' is missing required field {2}")]
16 MissingField(String, String, &'static str),
17 #[error("node '{0}' depends on unknown node '{1}'")]
18 UnknownDep(String, String),
19 #[error("node '{0}' depends on itself")]
20 SelfDep(String),
21 #[error("cycle detected involving: {0}")]
22 Cycle(String),
23}
24
25pub const NODE_TYPES: [&str; 4] = ["agent", "bash", "approval", "merge"];
26
27#[derive(serde::Deserialize)]
28struct FlowFile {
29 flow: FlowHeader,
30 #[serde(default)]
31 nodes: Vec<crate::types::NodeDef>,
32}
33
34#[derive(serde::Deserialize)]
35struct FlowHeader {
36 name: String,
37 #[serde(default)]
38 description: String,
39}
40
41pub fn parse_flow(toml_str: &str) -> Result<FlowDef, ParseError> {
42 let file: FlowFile = toml::from_str(toml_str)?;
43 let mut flow = FlowDef {
44 name: file.flow.name,
45 description: file.flow.description,
46 nodes: file.nodes,
47 };
48 validate(&mut flow)?;
49 Ok(flow)
50}
51
52pub fn validate(flow: &mut FlowDef) -> Result<(), ParseError> {
53 if flow.nodes.is_empty() {
54 return Err(ParseError::Empty);
55 }
56 let mut ids = BTreeMap::new();
57 for n in &flow.nodes {
58 if ids.contains_key(&n.id) {
59 return Err(ParseError::Duplicate(n.id.clone()));
60 }
61 if !NODE_TYPES.contains(&n.r#type.as_str()) {
62 return Err(ParseError::UnknownType(n.id.clone(), n.r#type.clone()));
63 }
64 match n.r#type.as_str() {
65 "agent" if n.prompt.is_none() => {
66 return Err(ParseError::MissingField(
67 n.id.clone(),
68 "agent".into(),
69 "prompt",
70 ));
71 }
72 "bash" if n.command.is_none() => {
73 return Err(ParseError::MissingField(
74 n.id.clone(),
75 "bash".into(),
76 "command",
77 ));
78 }
79 "approval" if n.message.is_none() => {
80 return Err(ParseError::MissingField(
81 n.id.clone(),
82 "approval".into(),
83 "message",
84 ));
85 }
86 _ => {}
87 }
88 ids.insert(n.id.clone(), ());
89 }
90 for n in &flow.nodes {
91 for dep in &n.deps {
92 if dep == &n.id {
93 return Err(ParseError::SelfDep(n.id.clone()));
94 }
95 if !ids.contains_key(dep) {
96 return Err(ParseError::UnknownDep(n.id.clone(), dep.clone()));
97 }
98 }
99 }
100
101 // Template references imply dependencies.
102 let mut resolved: Vec<crate::types::NodeDef> = Vec::with_capacity(flow.nodes.len());
103 for n in &flow.nodes {
104 let mut node = n.clone();
105 for (_, reference) in template_refs(&node) {
106 if !ids.contains_key(&reference) {
107 return Err(ParseError::UnknownDep(node.id.clone(), reference));
108 }
109 if reference != node.id && !node.deps.contains(&reference) {
110 node.deps.push(reference);
111 }
112 }
113 resolved.push(node);
114 }
115 flow.nodes = resolved;
116
117 // Kahn's algorithm: cycle detection + layer assignment.
118 let layers = layers(flow)?;
119 let _ = layers;
120 Ok(())
121}
122
123/// Extracts `{{identifier}}` references from a node's prompt or command.
124fn template_refs(node: &crate::types::NodeDef) -> Vec<(String, String)> {
125 let sources = [
126 ("prompt", node.prompt.as_deref().unwrap_or_default()),
127 ("command", node.command.as_deref().unwrap_or_default()),
128 ];
129 let mut out = Vec::new();
130 for (field, text) in sources {
131 let bytes = text.as_bytes();
132 let mut i = 0;
133 while i + 3 < bytes.len() {
134 if &text[i..i + 2] == "{{"
135 && let Some(end_offset) = text[i + 2..].find("}}")
136 {
137 let inner = text[i + 2..i + 2 + end_offset].trim();
138 if !inner.is_empty()
139 && inner
140 .chars()
141 .all(|c| c.is_alphanumeric() || c == '_' || c == '-')
142 {
143 out.push((field.to_string(), inner.to_string()));
144 }
145 i += 2 + end_offset + 2;
146 } else {
147 i += 1;
148 }
149 }
150 }
151 out
152}
153
154/// Topological layers: nodes in the same layer have no mutual dependencies
155/// and may run concurrently.
156pub fn layers(flow: &FlowDef) -> Result<Vec<Vec<String>>, ParseError> {
157 let mut indegree: BTreeMap<&str, usize> = BTreeMap::new();
158 let mut dependents: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
159 for n in &flow.nodes {
160 indegree.entry(n.id.as_str()).or_insert(0);
161 for dep in &n.deps {
162 *indegree.entry(n.id.as_str()).or_insert(0) += 1;
163 dependents
164 .entry(dep.as_str())
165 .or_default()
166 .push(n.id.as_str());
167 }
168 }
169
170 let mut queue: VecDeque<&str> = indegree
171 .iter()
172 .filter(|(_, d)| **d == 0)
173 .map(|(id, _)| *id)
174 .collect();
175 let mut layers: Vec<Vec<String>> = Vec::new();
176 let mut placed = 0usize;
177
178 while !queue.is_empty() {
179 let mut next_queue = VecDeque::new();
180 let mut layer = Vec::new();
181 for id in queue.drain(..) {
182 layer.push(id.to_string());
183 placed += 1;
184 if let Some(deps) = dependents.get(id) {
185 for d in deps {
186 if let Some(e) = indegree.get_mut(*d) {
187 *e -= 1;
188 if *e == 0 {
189 next_queue.push_back(*d);
190 }
191 }
192 }
193 }
194 }
195 layers.push(layer);
196 queue = next_queue;
197 }
198
199 if placed != flow.nodes.len() {
200 let stuck: Vec<String> = indegree
201 .iter()
202 .filter(|(_, d)| **d > 0)
203 .map(|(id, _)| id.to_string())
204 .collect();
205 return Err(ParseError::Cycle(stuck.join(", ")));
206 }
207 Ok(layers)
208}
209

Indexing the workspace…