tatara_core/domain/
convergence_graph.rs

1//! Typed convergence DAGs.
2//!
3//! A convergence graph is a DAG of typed convergence points with typed edges.
4//! Points are content-addressed via PointId. The graph supports topological
5//! ordering, substrate filtering, and validation.
6
7use serde::{Deserialize, Serialize};
8use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque};
9
10use super::compliance_binding::ComplianceClosure;
11use super::convergence_state::{ConvergencePoint, SubstrateType};
12use super::multi_distance::ConvergenceBandwidth;
13use super::point_id::PointId;
14
15/// The type of relationship between two convergence points.
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
17#[serde(rename_all = "snake_case")]
18pub enum EdgeType {
19    /// Data flows from one point to another.
20    Data,
21    /// Control dependency (must complete before next can start).
22    Control,
23    /// Attestation chain (output attestation feeds input attestation).
24    Attestation,
25}
26
27/// A typed edge between two convergence points.
28#[derive(Debug, Clone, Serialize, Deserialize)]
29pub struct TypedEdge {
30    /// Source point.
31    pub from: PointId,
32    /// Target point.
33    pub to: PointId,
34    /// Type of relationship.
35    pub edge_type: EdgeType,
36}
37
38/// The complete convergence graph across all substrates.
39#[derive(Debug, Clone, Serialize, Deserialize, Default)]
40pub struct ConvergenceGraph {
41    /// All points, keyed by hex-encoded PointId.
42    /// (BTreeMap keys must be strings for JSON serialization.)
43    #[serde(with = "point_id_map")]
44    pub points: BTreeMap<PointId, ConvergencePoint>,
45    /// Typed edges between points.
46    pub edges: Vec<TypedEdge>,
47}
48
49mod point_id_map {
50    use super::*;
51    use serde::de::{self, MapAccess, Visitor};
52    use serde::ser::SerializeMap;
53
54    pub fn serialize<S>(
55        map: &BTreeMap<PointId, ConvergencePoint>,
56        serializer: S,
57    ) -> Result<S::Ok, S::Error>
58    where
59        S: serde::Serializer,
60    {
61        let mut m = serializer.serialize_map(Some(map.len()))?;
62        for (k, v) in map {
63            m.serialize_entry(&k.to_hex(), v)?;
64        }
65        m.end()
66    }
67
68    pub fn deserialize<'de, D>(
69        deserializer: D,
70    ) -> Result<BTreeMap<PointId, ConvergencePoint>, D::Error>
71    where
72        D: serde::Deserializer<'de>,
73    {
74        struct PointIdMapVisitor;
75        impl<'de> Visitor<'de> for PointIdMapVisitor {
76            type Value = BTreeMap<PointId, ConvergencePoint>;
77            fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
78                f.write_str("a map with hex PointId keys")
79            }
80            fn visit_map<M: MapAccess<'de>>(self, mut access: M) -> Result<Self::Value, M::Error> {
81                let mut map = BTreeMap::new();
82                while let Some((key, value)) = access.next_entry::<String, ConvergencePoint>()? {
83                    let id = PointId::from_hex(&key).map_err(de::Error::custom)?;
84                    map.insert(id, value);
85                }
86                Ok(map)
87            }
88        }
89        deserializer.deserialize_map(PointIdMapVisitor)
90    }
91}
92
93impl ConvergenceGraph {
94    /// Create an empty graph.
95    pub fn new() -> Self {
96        Self::default()
97    }
98
99    /// Add a convergence point, returning its PointId.
100    pub fn add_point(&mut self, id: PointId, point: ConvergencePoint) {
101        self.points.insert(id, point);
102    }
103
104    /// Add a typed edge between two points.
105    pub fn add_edge(&mut self, edge: TypedEdge) {
106        self.edges.push(edge);
107    }
108
109    /// Compute topological order using Kahn's algorithm.
110    /// Returns Err if the graph contains a cycle.
111    pub fn topological_order(&self) -> Result<Vec<PointId>, GraphError> {
112        let mut in_degree: HashMap<PointId, usize> = HashMap::new();
113        let mut adjacency: HashMap<PointId, Vec<PointId>> = HashMap::new();
114
115        for id in self.points.keys() {
116            in_degree.entry(*id).or_insert(0);
117            adjacency.entry(*id).or_default();
118        }
119
120        for edge in &self.edges {
121            *in_degree.entry(edge.to).or_insert(0) += 1;
122            adjacency.entry(edge.from).or_default().push(edge.to);
123        }
124
125        let mut queue: VecDeque<PointId> = in_degree
126            .iter()
127            .filter(|(_, &deg)| deg == 0)
128            .map(|(id, _)| *id)
129            .collect();
130
131        let mut order = Vec::new();
132
133        while let Some(id) = queue.pop_front() {
134            order.push(id);
135            if let Some(neighbors) = adjacency.get(&id) {
136                for neighbor in neighbors {
137                    if let Some(deg) = in_degree.get_mut(neighbor) {
138                        *deg -= 1;
139                        if *deg == 0 {
140                            queue.push_back(*neighbor);
141                        }
142                    }
143                }
144            }
145        }
146
147        if order.len() != self.points.len() {
148            Err(GraphError::CycleDetected)
149        } else {
150            Ok(order)
151        }
152    }
153
154    /// Validate the graph: all edge endpoints exist, no cycles.
155    pub fn validate(&self) -> Result<(), GraphError> {
156        for edge in &self.edges {
157            if !self.points.contains_key(&edge.from) {
158                return Err(GraphError::MissingPoint(edge.from));
159            }
160            if !self.points.contains_key(&edge.to) {
161                return Err(GraphError::MissingPoint(edge.to));
162            }
163        }
164        self.topological_order()?;
165        Ok(())
166    }
167
168    /// Filter points by substrate type.
169    pub fn points_by_substrate(
170        &self,
171        substrate: &SubstrateType,
172    ) -> Vec<(&PointId, &ConvergencePoint)> {
173        self.points
174            .iter()
175            .filter(|(_, p)| &p.substrate == substrate)
176            .collect()
177    }
178
179    /// Get the number of points.
180    pub fn point_count(&self) -> usize {
181        self.points.len()
182    }
183
184    /// Get the number of edges.
185    pub fn edge_count(&self) -> usize {
186        self.edges.len()
187    }
188
189    /// Compute forward closure: all points this point transitively depends on.
190    /// Walks backward along edges (from → to) to find all upstream dependencies.
191    /// Like `nix-store --query --requisites`.
192    pub fn forward_closure(&self, point_id: &PointId) -> BTreeSet<PointId> {
193        let mut adjacency: HashMap<PointId, Vec<PointId>> = HashMap::new();
194        for edge in &self.edges {
195            adjacency.entry(edge.to).or_default().push(edge.from);
196        }
197
198        let mut visited = BTreeSet::new();
199        let mut queue = VecDeque::new();
200        queue.push_back(*point_id);
201
202        while let Some(id) = queue.pop_front() {
203            if !visited.insert(id) {
204                continue;
205            }
206            if let Some(deps) = adjacency.get(&id) {
207                for dep in deps {
208                    queue.push_back(*dep);
209                }
210            }
211        }
212
213        visited.remove(point_id);
214        visited
215    }
216
217    /// Compute reverse closure: all points that transitively depend on this point.
218    /// Walks forward along edges (from → to) to find all downstream dependents.
219    /// Like `nix-store --query --referrers` (transitive).
220    pub fn reverse_closure(&self, point_id: &PointId) -> BTreeSet<PointId> {
221        let mut adjacency: HashMap<PointId, Vec<PointId>> = HashMap::new();
222        for edge in &self.edges {
223            adjacency.entry(edge.from).or_default().push(edge.to);
224        }
225
226        let mut visited = BTreeSet::new();
227        let mut queue = VecDeque::new();
228        queue.push_back(*point_id);
229
230        while let Some(id) = queue.pop_front() {
231            if !visited.insert(id) {
232                continue;
233            }
234            if let Some(deps) = adjacency.get(&id) {
235                for dep in deps {
236                    queue.push_back(*dep);
237                }
238            }
239        }
240
241        visited.remove(point_id);
242        visited
243    }
244}
245
246/// A substrate-scoped subgraph with cross-substrate boundary edges.
247#[derive(Debug, Clone, Serialize, Deserialize)]
248pub struct SubstrateDAG {
249    /// The substrate this DAG belongs to.
250    pub substrate: SubstrateType,
251    /// Points in this substrate.
252    pub points: BTreeMap<PointId, ConvergencePoint>,
253    /// Edges within this substrate.
254    pub internal_edges: Vec<TypedEdge>,
255    /// Edges crossing to other substrates.
256    pub cross_edges: Vec<TypedEdge>,
257    /// Maximum convergence velocity for this substrate.
258    pub bandwidth: ConvergenceBandwidth,
259}
260
261/// Pre-execution analysis of a convergence graph.
262#[derive(Debug, Clone, Serialize, Deserialize)]
263pub struct ConvergencePlan {
264    /// The graph being planned.
265    pub graph: ConvergenceGraph,
266    /// Topological execution order.
267    pub execution_order: Vec<PointId>,
268    /// Compliance bindings resolved to specific points.
269    pub compliance: ComplianceClosure,
270    /// Points that can skip re-execution (attestation unchanged).
271    pub cache_hits: Vec<PointId>,
272    /// Longest sequential dependency chain.
273    pub critical_path: Vec<PointId>,
274}
275
276/// Errors in convergence graph operations.
277#[derive(Debug, Clone, thiserror::Error)]
278pub enum GraphError {
279    #[error("cycle detected in convergence graph")]
280    CycleDetected,
281    #[error("edge references missing point: {0}")]
282    MissingPoint(PointId),
283}
284
285#[cfg(test)]
286mod tests {
287    use super::*;
288    use crate::domain::convergence_state::*;
289
290    fn make_point(name: &str, substrate: SubstrateType) -> (PointId, ConvergencePoint) {
291        let id = PointId::compute(name.as_bytes(), &[], b"desired");
292        let point = ConvergencePoint {
293            name: name.into(),
294            description: format!("{name} point"),
295            monotone: true,
296            mechanism: ConvergenceMechanism::Local,
297            state: ConvergenceState::new(name),
298            boundary: ConvergenceBoundary::default(),
299            point_type: ConvergencePointType::Transform,
300            horizon: ConvergenceHorizon::Bounded,
301            substrate,
302            computation_mode: ComputationMode::Mechanical,
303        };
304        (id, point)
305    }
306
307    #[test]
308    fn test_empty_graph() {
309        let g = ConvergenceGraph::new();
310        assert_eq!(g.point_count(), 0);
311        assert!(g.validate().is_ok());
312        assert!(g.topological_order().unwrap().is_empty());
313    }
314
315    #[test]
316    fn test_single_point() {
317        let mut g = ConvergenceGraph::new();
318        let (id, p) = make_point("a", SubstrateType::Compute);
319        g.add_point(id, p);
320        assert_eq!(g.point_count(), 1);
321        assert!(g.validate().is_ok());
322        assert_eq!(g.topological_order().unwrap(), vec![id]);
323    }
324
325    #[test]
326    fn test_linear_chain() {
327        let mut g = ConvergenceGraph::new();
328        let (a, pa) = make_point("a", SubstrateType::Compute);
329        let (b, pb) = make_point("b", SubstrateType::Compute);
330        let (c, pc) = make_point("c", SubstrateType::Compute);
331        g.add_point(a, pa);
332        g.add_point(b, pb);
333        g.add_point(c, pc);
334        g.add_edge(TypedEdge {
335            from: a,
336            to: b,
337            edge_type: EdgeType::Attestation,
338        });
339        g.add_edge(TypedEdge {
340            from: b,
341            to: c,
342            edge_type: EdgeType::Attestation,
343        });
344
345        let order = g.topological_order().unwrap();
346        assert_eq!(order.len(), 3);
347        let pos_a = order.iter().position(|x| *x == a).unwrap();
348        let pos_b = order.iter().position(|x| *x == b).unwrap();
349        let pos_c = order.iter().position(|x| *x == c).unwrap();
350        assert!(pos_a < pos_b);
351        assert!(pos_b < pos_c);
352    }
353
354    #[test]
355    fn test_cycle_detection() {
356        let mut g = ConvergenceGraph::new();
357        let (a, pa) = make_point("a", SubstrateType::Compute);
358        let (b, pb) = make_point("b", SubstrateType::Compute);
359        g.add_point(a, pa);
360        g.add_point(b, pb);
361        g.add_edge(TypedEdge {
362            from: a,
363            to: b,
364            edge_type: EdgeType::Data,
365        });
366        g.add_edge(TypedEdge {
367            from: b,
368            to: a,
369            edge_type: EdgeType::Data,
370        });
371
372        assert!(matches!(
373            g.topological_order(),
374            Err(GraphError::CycleDetected)
375        ));
376    }
377
378    #[test]
379    fn test_missing_point_validation() {
380        let mut g = ConvergenceGraph::new();
381        let (a, pa) = make_point("a", SubstrateType::Compute);
382        let missing = PointId::compute(b"missing", &[], b"state");
383        g.add_point(a, pa);
384        g.add_edge(TypedEdge {
385            from: a,
386            to: missing,
387            edge_type: EdgeType::Data,
388        });
389
390        assert!(matches!(g.validate(), Err(GraphError::MissingPoint(_))));
391    }
392
393    #[test]
394    fn test_points_by_substrate() {
395        let mut g = ConvergenceGraph::new();
396        let (a, pa) = make_point("a", SubstrateType::Compute);
397        let (b, pb) = make_point("b", SubstrateType::Security);
398        let (c, pc) = make_point("c", SubstrateType::Compute);
399        g.add_point(a, pa);
400        g.add_point(b, pb);
401        g.add_point(c, pc);
402
403        assert_eq!(g.points_by_substrate(&SubstrateType::Compute).len(), 2);
404        assert_eq!(g.points_by_substrate(&SubstrateType::Security).len(), 1);
405        assert_eq!(g.points_by_substrate(&SubstrateType::Financial).len(), 0);
406    }
407
408    #[test]
409    fn test_forward_closure() {
410        let mut g = ConvergenceGraph::new();
411        let (a, pa) = make_point("a", SubstrateType::Compute);
412        let (b, pb) = make_point("b", SubstrateType::Compute);
413        let (c, pc) = make_point("c", SubstrateType::Compute);
414        g.add_point(a, pa);
415        g.add_point(b, pb);
416        g.add_point(c, pc);
417        g.add_edge(TypedEdge {
418            from: a,
419            to: b,
420            edge_type: EdgeType::Data,
421        });
422        g.add_edge(TypedEdge {
423            from: b,
424            to: c,
425            edge_type: EdgeType::Data,
426        });
427
428        let closure = g.forward_closure(&c);
429        assert!(closure.contains(&a));
430        assert!(closure.contains(&b));
431        assert!(!closure.contains(&c));
432    }
433
434    #[test]
435    fn test_reverse_closure() {
436        let mut g = ConvergenceGraph::new();
437        let (a, pa) = make_point("a", SubstrateType::Compute);
438        let (b, pb) = make_point("b", SubstrateType::Compute);
439        let (c, pc) = make_point("c", SubstrateType::Compute);
440        g.add_point(a, pa);
441        g.add_point(b, pb);
442        g.add_point(c, pc);
443        g.add_edge(TypedEdge {
444            from: a,
445            to: b,
446            edge_type: EdgeType::Data,
447        });
448        g.add_edge(TypedEdge {
449            from: b,
450            to: c,
451            edge_type: EdgeType::Data,
452        });
453
454        let closure = g.reverse_closure(&a);
455        assert!(closure.contains(&b));
456        assert!(closure.contains(&c));
457        assert!(!closure.contains(&a));
458    }
459
460    #[test]
461    fn test_diamond_dag() {
462        let mut g = ConvergenceGraph::new();
463        let (a, pa) = make_point("a", SubstrateType::Compute);
464        let (b, pb) = make_point("b", SubstrateType::Compute);
465        let (c, pc) = make_point("c", SubstrateType::Compute);
466        let (d, pd) = make_point("d", SubstrateType::Compute);
467        g.add_point(a, pa);
468        g.add_point(b, pb);
469        g.add_point(c, pc);
470        g.add_point(d, pd);
471        g.add_edge(TypedEdge {
472            from: a,
473            to: b,
474            edge_type: EdgeType::Data,
475        });
476        g.add_edge(TypedEdge {
477            from: a,
478            to: c,
479            edge_type: EdgeType::Data,
480        });
481        g.add_edge(TypedEdge {
482            from: b,
483            to: d,
484            edge_type: EdgeType::Data,
485        });
486        g.add_edge(TypedEdge {
487            from: c,
488            to: d,
489            edge_type: EdgeType::Data,
490        });
491
492        let order = g.topological_order().unwrap();
493        assert_eq!(order.len(), 4);
494        let pos_a = order.iter().position(|x| *x == a).unwrap();
495        let pos_d = order.iter().position(|x| *x == d).unwrap();
496        assert!(pos_a < pos_d);
497    }
498
499    #[test]
500    fn test_graph_serde() {
501        let mut g = ConvergenceGraph::new();
502        let (a, pa) = make_point("a", SubstrateType::Compute);
503        let (b, pb) = make_point("b", SubstrateType::Network);
504        g.add_point(a, pa);
505        g.add_point(b, pb);
506        g.add_edge(TypedEdge {
507            from: a,
508            to: b,
509            edge_type: EdgeType::Control,
510        });
511
512        let json = serde_json::to_string(&g).unwrap();
513        let parsed: ConvergenceGraph = serde_json::from_str(&json).unwrap();
514        assert_eq!(parsed.point_count(), 2);
515        assert_eq!(parsed.edge_count(), 1);
516    }
517}
tatara_core/domain/convergence_graph.rs

tatara_core/domain/
convergence_graph.rs
