Vakyartha documentation

1//! Run-graph snapshots (docs/design/10-flows.md): typed projection over a
2//! flow-run ledger — zero rendering opinions, safe to serve as JSON or
3//! drive any UI strip.
4
5use crate::types::{FlowState, NodeStatus};
6use serde::Serialize;
7
8#[derive(Debug, Clone, Serialize)]
9pub struct GraphNode {
10 pub id: String,
11 /// Topological layer index (1-based) derived from the frozen
12 /// definition; 0 when the definition no longer parses.
13 pub layer: usize,
14 pub status: &'static str,
15 pub output_bytes: usize,
16}
17
18#[derive(Debug, Clone, Serialize)]
19pub struct RunGraph {
20 pub run_id: String,
21 pub flow_name: String,
22 pub layers_total: usize,
23 pub nodes: Vec<GraphNode>,
24 pub completed: usize,
25 pub failed: usize,
26 pub skipped: usize,
27 pub pending_or_running: usize,
28 pub generated_at: String,
29}
30
31fn status_str(s: NodeStatus) -> &'static str {
32 match s {
33 NodeStatus::Pending => "pending",
34 NodeStatus::Running => "running",
35 NodeStatus::Completed => "completed",
36 NodeStatus::Failed => "failed",
37 NodeStatus::Skipped => "skipped",
38 }
39}
40
41/// Layer assignment mirrors `parse::layers`: repeated passes peel off
42/// ready nodes until exhaustion (cycle-safe).
43fn compute_layers(
44 nodes: &[crate::types::NodeDef],
45) -> (std::collections::HashMap<String, usize>, usize) {
46 let mut map = std::collections::HashMap::new();
47 let mut remaining: Vec<&crate::types::NodeDef> = nodes.iter().collect();
48 let mut layer = 0usize;
49 while !remaining.is_empty() {
50 layer += 1;
51 // Ready = every dep already placed in an EARLIER pass.
52 let placed_now: Vec<String> = Vec::new();
53 let _ = placed_now;
54 let mut progressed = false;
55 let newly: Vec<String> = remaining
56 .iter()
57 .filter(|n| {
58 n.deps
59 .iter()
60 .all(|d| map.contains_key(d) || !remaining.iter().any(|r| &r.id == d))
61 })
62 .map(|n| n.id.clone())
63 .collect();
64 if !newly.is_empty() {
65 for id in &newly {
66 map.insert(id.clone(), layer);
67 }
68 progressed = true;
69 remaining.retain(|n| !newly.contains(&n.id));
70 }
71 if !progressed {
72 // Cycle or dangling deps: pin the rest to the next layer so the
73 // snapshot still renders instead of diverging.
74 for n in remaining {
75 map.insert(n.id.clone(), layer);
76 }
77 break;
78 }
79 }
80 (map, layer)
81}
82
83pub fn graph_snapshot(state: &FlowState) -> RunGraph {
84 let (layer_map, layers_total) = match crate::parse_flow(&state.definition_toml) {
85 Ok(flow) => {
86 let (m, l) = compute_layers(&flow.nodes);
87 (m, l)
88 }
89 Err(_) => (std::collections::HashMap::new(), 0),
90 };
91
92 let mut nodes: Vec<GraphNode> = Vec::new();
93 let (mut completed, mut failed, mut skipped, mut pending) = (0usize, 0, 0, 0);
94 // Deterministic order: by (layer, id).
95 let mut entries: Vec<(&String, &crate::types::NodeResult)> = state.nodes.iter().collect();
96 entries.sort_by(|a, b| {
97 let la = layer_map.get(a.0).copied().unwrap_or(0);
98 let lb = layer_map.get(b.0).copied().unwrap_or(0);
99 la.cmp(&lb).then(a.0.cmp(b.0))
100 });
101 for (id, res) in entries {
102 let status = status_str(res.status);
103 match res.status {
104 NodeStatus::Completed => completed += 1,
105 NodeStatus::Failed => failed += 1,
106 NodeStatus::Skipped => skipped += 1,
107 _ => pending += 1,
108 }
109 nodes.push(GraphNode {
110 id: id.clone(),
111 layer: layer_map.get(id).copied().unwrap_or(0),
112 status,
113 output_bytes: res.output.len(),
114 });
115 }
116
117 RunGraph {
118 run_id: state.run_id.clone(),
119 flow_name: state.flow_name.clone(),
120 layers_total,
121 nodes,
122 completed,
123 failed,
124 skipped,
125 pending_or_running: pending,
126 generated_at: chrono::Utc::now().to_rfc3339(),
127 }
128}
129
130#[cfg(test)]
131#[allow(clippy::unwrap_used, clippy::expect_used)]
132mod tests {
133 use super::*;
134 use crate::types::{FlowState, NodeResult};
135
136 const DEF: &str = "[flow]\nname=\"g\"\ndescription=\"\"\n\n[[nodes]]\nid=\"a\"\ntype=\"bash\"\ncommand=\"echo a\"\n\n[[nodes]]\nid=\"b\"\ntype=\"bash\"\ncommand=\"echo b\"\ndeps=[\"a\"]\n";
137
138 fn state_with(nodes: &[(&str, NodeStatus)]) -> FlowState {
139 let mut st = FlowState {
140 run_id: "r".into(),
141 flow_name: "g".into(),
142 definition_toml: DEF.into(),
143 started_at: chrono::Utc::now(),
144 outcome: None,
145 nodes: Default::default(),
146 };
147 for (id, status) in nodes {
148 st.nodes.insert(
149 (*id).into(),
150 NodeResult {
151 status: *status,
152 output: "x".into(),
153 },
154 );
155 }
156 st
157 }
158
159 #[test]
160 fn snapshot_maps_layers_statuses_and_counts() {
161 let st = state_with(&[("a", NodeStatus::Completed), ("b", NodeStatus::Running)]);
162 let g = graph_snapshot(&st);
163 assert_eq!(g.layers_total, 2);
164 assert_eq!(g.completed, 1);
165 assert_eq!(g.pending_or_running, 1);
166 assert_eq!(g.nodes[0].id, "a");
167 assert_eq!(g.nodes[0].layer, 1);
168 assert_eq!(g.nodes[1].layer, 2);
169 assert_eq!(g.nodes[1].status, "running");
170 }
171
172 #[test]
173 fn unparseable_definition_still_snapshots_with_layer_zero() {
174 let mut st = state_with(&[("a", NodeStatus::Failed)]);
175 st.definition_toml = "not [ valid toml".into();
176 let g = graph_snapshot(&st);
177 assert_eq!(g.layers_total, 0);
178 assert_eq!(g.nodes[0].layer, 0);
179 assert_eq!(g.failed, 1);
180 }
181
182 #[test]
183 fn deterministic_ordering_across_calls() {
184 let st = state_with(&[("b", NodeStatus::Completed), ("a", NodeStatus::Skipped)]);
185 let g1 = graph_snapshot(&st);
186 let g2 = graph_snapshot(&st);
187 assert_eq!(
188 serde_json::to_string(&g1.nodes).unwrap(),
189 serde_json::to_string(&g2.nodes).unwrap()
190 );
191 }
192}
193

1//! Run-graph snapshots (docs/design/10-flows.md): typed projection over a
2//! flow-run ledger — zero rendering opinions, safe to serve as JSON or
3//! drive any UI strip.
4
5use crate::types::{FlowState, NodeStatus};
6use serde::Serialize;
7
8#[derive(Debug, Clone, Serialize)]
9pub struct GraphNode {
10 pub id: String,
11 /// Topological layer index (1-based) derived from the frozen
12 /// definition; 0 when the definition no longer parses.
13 pub layer: usize,
14 pub status: &'static str,
15 pub output_bytes: usize,
16}
17
18#[derive(Debug, Clone, Serialize)]
19pub struct RunGraph {
20 pub run_id: String,
21 pub flow_name: String,
22 pub layers_total: usize,
23 pub nodes: Vec<GraphNode>,
24 pub completed: usize,
25 pub failed: usize,
26 pub skipped: usize,
27 pub pending_or_running: usize,
28 pub generated_at: String,
29}
30
31fn status_str(s: NodeStatus) -> &'static str {
32 match s {
33 NodeStatus::Pending => "pending",
34 NodeStatus::Running => "running",
35 NodeStatus::Completed => "completed",
36 NodeStatus::Failed => "failed",
37 NodeStatus::Skipped => "skipped",
38 }
39}
40
41/// Layer assignment mirrors `parse::layers`: repeated passes peel off
42/// ready nodes until exhaustion (cycle-safe).
43fn compute_layers(
44 nodes: &[crate::types::NodeDef],
45) -> (std::collections::HashMap<String, usize>, usize) {
46 let mut map = std::collections::HashMap::new();
47 let mut remaining: Vec<&crate::types::NodeDef> = nodes.iter().collect();
48 let mut layer = 0usize;
49 while !remaining.is_empty() {
50 layer += 1;
51 // Ready = every dep already placed in an EARLIER pass.
52 let placed_now: Vec<String> = Vec::new();
53 let _ = placed_now;
54 let mut progressed = false;
55 let newly: Vec<String> = remaining
56 .iter()
57 .filter(|n| {
58 n.deps
59 .iter()
60 .all(|d| map.contains_key(d) || !remaining.iter().any(|r| &r.id == d))
61 })
62 .map(|n| n.id.clone())
63 .collect();
64 if !newly.is_empty() {
65 for id in &newly {
66 map.insert(id.clone(), layer);
67 }
68 progressed = true;
69 remaining.retain(|n| !newly.contains(&n.id));
70 }
71 if !progressed {
72 // Cycle or dangling deps: pin the rest to the next layer so the
73 // snapshot still renders instead of diverging.
74 for n in remaining {
75 map.insert(n.id.clone(), layer);
76 }
77 break;
78 }
79 }
80 (map, layer)
81}
82
83pub fn graph_snapshot(state: &FlowState) -> RunGraph {
84 let (layer_map, layers_total) = match crate::parse_flow(&state.definition_toml) {
85 Ok(flow) => {
86 let (m, l) = compute_layers(&flow.nodes);
87 (m, l)
88 }
89 Err(_) => (std::collections::HashMap::new(), 0),
90 };
91
92 let mut nodes: Vec<GraphNode> = Vec::new();
93 let (mut completed, mut failed, mut skipped, mut pending) = (0usize, 0, 0, 0);
94 // Deterministic order: by (layer, id).
95 let mut entries: Vec<(&String, &crate::types::NodeResult)> = state.nodes.iter().collect();
96 entries.sort_by(|a, b| {
97 let la = layer_map.get(a.0).copied().unwrap_or(0);
98 let lb = layer_map.get(b.0).copied().unwrap_or(0);
99 la.cmp(&lb).then(a.0.cmp(b.0))
100 });
101 for (id, res) in entries {
102 let status = status_str(res.status);
103 match res.status {
104 NodeStatus::Completed => completed += 1,
105 NodeStatus::Failed => failed += 1,
106 NodeStatus::Skipped => skipped += 1,
107 _ => pending += 1,
108 }
109 nodes.push(GraphNode {
110 id: id.clone(),
111 layer: layer_map.get(id).copied().unwrap_or(0),
112 status,
113 output_bytes: res.output.len(),
114 });
115 }
116
117 RunGraph {
118 run_id: state.run_id.clone(),
119 flow_name: state.flow_name.clone(),
120 layers_total,
121 nodes,
122 completed,
123 failed,
124 skipped,
125 pending_or_running: pending,
126 generated_at: chrono::Utc::now().to_rfc3339(),
127 }
128}
129
130#[cfg(test)]
131#[allow(clippy::unwrap_used, clippy::expect_used)]
132mod tests {
133 use super::*;
134 use crate::types::{FlowState, NodeResult};
135
136 const DEF: &str = "[flow]\nname=\"g\"\ndescription=\"\"\n\n[[nodes]]\nid=\"a\"\ntype=\"bash\"\ncommand=\"echo a\"\n\n[[nodes]]\nid=\"b\"\ntype=\"bash\"\ncommand=\"echo b\"\ndeps=[\"a\"]\n";
137
138 fn state_with(nodes: &[(&str, NodeStatus)]) -> FlowState {
139 let mut st = FlowState {
140 run_id: "r".into(),
141 flow_name: "g".into(),
142 definition_toml: DEF.into(),
143 started_at: chrono::Utc::now(),
144 outcome: None,
145 nodes: Default::default(),
146 };
147 for (id, status) in nodes {
148 st.nodes.insert(
149 (*id).into(),
150 NodeResult {
151 status: *status,
152 output: "x".into(),
153 },
154 );
155 }
156 st
157 }
158
159 #[test]
160 fn snapshot_maps_layers_statuses_and_counts() {
161 let st = state_with(&[("a", NodeStatus::Completed), ("b", NodeStatus::Running)]);
162 let g = graph_snapshot(&st);
163 assert_eq!(g.layers_total, 2);
164 assert_eq!(g.completed, 1);
165 assert_eq!(g.pending_or_running, 1);
166 assert_eq!(g.nodes[0].id, "a");
167 assert_eq!(g.nodes[0].layer, 1);
168 assert_eq!(g.nodes[1].layer, 2);
169 assert_eq!(g.nodes[1].status, "running");
170 }
171
172 #[test]
173 fn unparseable_definition_still_snapshots_with_layer_zero() {
174 let mut st = state_with(&[("a", NodeStatus::Failed)]);
175 st.definition_toml = "not [ valid toml".into();
176 let g = graph_snapshot(&st);
177 assert_eq!(g.layers_total, 0);
178 assert_eq!(g.nodes[0].layer, 0);
179 assert_eq!(g.failed, 1);
180 }
181
182 #[test]
183 fn deterministic_ordering_across_calls() {
184 let st = state_with(&[("b", NodeStatus::Completed), ("a", NodeStatus::Skipped)]);
185 let g1 = graph_snapshot(&st);
186 let g2 = graph_snapshot(&st);
187 assert_eq!(
188 serde_json::to_string(&g1.nodes).unwrap(),
189 serde_json::to_string(&g2.nodes).unwrap()
190 );
191 }
192}
193

Indexing the workspace…