mecha10-core 0.6.2

Mecha10 core runtime and foundational types
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
//! Dependency Resolution for Node Startup/Shutdown Ordering
//!
//! Provides topological sorting and dependency graph analysis to ensure nodes
//! start and stop in the correct order based on their dependencies.
//!
//! # Use Cases
//!
//! - Start nodes in dependency order (dependencies before dependents)
//! - Stop nodes in reverse order (dependents before dependencies)
//! - Detect circular dependencies
//! - Validate dependency references
//! - Group nodes by dependency level for parallel startup
//!
//! # Example
//!
//! ```rust
//! use mecha10::dependency::{DependencyGraph, StartupPlan};
//! use mecha10::types::NodeConfig;
//!
//! # async fn example() -> mecha10::Result<()> {
//! let nodes = vec![
//!     NodeConfig { id: "camera".into(), depends_on: vec![], /* ... */ },
//!     NodeConfig { id: "vision".into(), depends_on: vec!["camera".into()], /* ... */ },
//!     NodeConfig { id: "planner".into(), depends_on: vec!["vision".into()], /* ... */ },
//! ];
//!
//! // Build dependency graph
//! let graph = DependencyGraph::from_configs(&nodes)?;
//!
//! // Get startup order
//! let startup_order = graph.topological_sort()?;
//! // Result: ["camera", "vision", "planner"]
//!
//! // Get shutdown order (reverse)
//! let shutdown_order = graph.reverse_topological_sort()?;
//! // Result: ["planner", "vision", "camera"]
//!
//! // Get parallel startup plan (nodes grouped by level)
//! let plan = graph.startup_plan()?;
//! // Level 0: ["camera"] - no dependencies
//! // Level 1: ["vision"] - depends on camera
//! // Level 2: ["planner"] - depends on vision
//! # Ok(())
//! # }
//! ```

use crate::error::{Mecha10Error, Result};
use crate::types::NodeConfig;
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet, VecDeque};
use tracing::{debug, info};

// ============================================================================
// Dependency Graph
// ============================================================================

/// Dependency graph for node relationships
///
/// Maintains a directed graph of node dependencies where an edge from A to B
/// means "A depends on B" (B must start before A).
#[derive(Debug, Clone)]
pub struct DependencyGraph {
    /// Node dependencies: node_id -> vec of dependencies
    dependencies: HashMap<String, Vec<String>>,

    /// Reverse dependencies: node_id -> vec of dependents
    dependents: HashMap<String, Vec<String>>,

    /// All node IDs
    nodes: HashSet<String>,
}

impl DependencyGraph {
    /// Create a new empty dependency graph
    pub fn new() -> Self {
        Self {
            dependencies: HashMap::new(),
            dependents: HashMap::new(),
            nodes: HashSet::new(),
        }
    }

    /// Build dependency graph from node configurations
    ///
    /// # Arguments
    ///
    /// * `configs` - Slice of node configurations
    ///
    /// # Returns
    ///
    /// Dependency graph or error if dependencies are invalid
    ///
    /// # Errors
    ///
    /// - Circular dependencies detected
    /// - Undefined dependency references
    pub fn from_configs(configs: &[NodeConfig]) -> Result<Self> {
        let mut graph = Self::new();

        // Add all nodes first
        for config in configs {
            graph.add_node(&config.id);
        }

        // Add dependencies
        for config in configs {
            for dep in &config.depends_on {
                graph.add_dependency(&config.id, dep)?;
            }
        }

        // Validate graph
        graph.validate()?;

        Ok(graph)
    }

    /// Add a node to the graph
    pub fn add_node(&mut self, node_id: &str) {
        self.nodes.insert(node_id.to_string());
        self.dependencies.entry(node_id.to_string()).or_default();
        self.dependents.entry(node_id.to_string()).or_default();
    }

    /// Add a dependency relationship
    ///
    /// # Arguments
    ///
    /// * `node_id` - The dependent node
    /// * `dependency` - The node that `node_id` depends on
    ///
    /// # Errors
    ///
    /// Returns error if dependency node doesn't exist
    pub fn add_dependency(&mut self, node_id: &str, dependency: &str) -> Result<()> {
        // Ensure both nodes exist
        if !self.nodes.contains(node_id) {
            return Err(Mecha10Error::Configuration(format!(
                "Node '{}' not found in graph",
                node_id
            )));
        }

        if !self.nodes.contains(dependency) {
            return Err(Mecha10Error::Configuration(format!(
                "Dependency '{}' not found (required by '{}')",
                dependency, node_id
            )));
        }

        // Add to dependencies map
        self.dependencies
            .entry(node_id.to_string())
            .or_default()
            .push(dependency.to_string());

        // Add to reverse dependencies map
        self.dependents
            .entry(dependency.to_string())
            .or_default()
            .push(node_id.to_string());

        Ok(())
    }

    /// Validate the dependency graph
    ///
    /// Checks for:
    /// - Circular dependencies
    /// - Undefined dependency references
    pub fn validate(&self) -> Result<()> {
        // Check for circular dependencies
        self.check_circular_dependencies()?;

        // Check all dependencies exist
        for (node_id, deps) in &self.dependencies {
            for dep in deps {
                if !self.nodes.contains(dep) {
                    return Err(Mecha10Error::Configuration(format!(
                        "Node '{}' depends on undefined node '{}'",
                        node_id, dep
                    )));
                }
            }
        }

        Ok(())
    }

    /// Check for circular dependencies using DFS
    fn check_circular_dependencies(&self) -> Result<()> {
        let mut visited = HashSet::new();
        let mut rec_stack = HashSet::new();

        for node in &self.nodes {
            if !visited.contains(node) && self.has_cycle_dfs(node, &mut visited, &mut rec_stack)? {
                return Err(Mecha10Error::Configuration(format!(
                    "Circular dependency detected involving node '{}'",
                    node
                )));
            }
        }

        Ok(())
    }

    /// DFS helper for cycle detection
    fn has_cycle_dfs(
        &self,
        node: &str,
        visited: &mut HashSet<String>,
        rec_stack: &mut HashSet<String>,
    ) -> Result<bool> {
        visited.insert(node.to_string());
        rec_stack.insert(node.to_string());

        if let Some(deps) = self.dependencies.get(node) {
            for dep in deps {
                if !visited.contains(dep) {
                    if self.has_cycle_dfs(dep, visited, rec_stack)? {
                        return Ok(true);
                    }
                } else if rec_stack.contains(dep) {
                    return Ok(true); // Cycle detected
                }
            }
        }

        rec_stack.remove(node);
        Ok(false)
    }

    /// Perform topological sort using Kahn's algorithm
    ///
    /// Returns nodes in dependency order (dependencies before dependents).
    /// This is the order for starting nodes.
    ///
    /// # Returns
    ///
    /// Vector of node IDs in topological order
    ///
    /// # Errors
    ///
    /// Returns error if graph has cycles
    pub fn topological_sort(&self) -> Result<Vec<String>> {
        let mut in_degree: HashMap<String, usize> = HashMap::new();
        let mut result = Vec::new();
        let mut queue = VecDeque::new();

        // Calculate in-degree for each node
        for node in &self.nodes {
            let degree = self.dependencies.get(node).map(|d| d.len()).unwrap_or(0);
            in_degree.insert(node.clone(), degree);

            // Nodes with no dependencies can start immediately
            if degree == 0 {
                queue.push_back(node.clone());
            }
        }

        // Process nodes with no dependencies
        while let Some(node) = queue.pop_front() {
            result.push(node.clone());

            // Reduce in-degree for all dependents
            if let Some(dependents) = self.dependents.get(&node) {
                for dependent in dependents {
                    if let Some(degree) = in_degree.get_mut(dependent) {
                        *degree -= 1;
                        if *degree == 0 {
                            queue.push_back(dependent.clone());
                        }
                    }
                }
            }
        }

        // Check if all nodes were processed
        if result.len() != self.nodes.len() {
            return Err(Mecha10Error::Configuration(
                "Circular dependency detected during topological sort".to_string(),
            ));
        }

        debug!("Topological sort result: {:?}", result);
        Ok(result)
    }

    /// Get reverse topological sort (for shutdown order)
    ///
    /// Returns nodes in reverse dependency order (dependents before dependencies).
    /// This is the order for stopping nodes.
    pub fn reverse_topological_sort(&self) -> Result<Vec<String>> {
        let mut order = self.topological_sort()?;
        order.reverse();
        debug!("Reverse topological sort result: {:?}", order);
        Ok(order)
    }

    /// Get startup plan with nodes grouped by dependency level
    ///
    /// Nodes at the same level have no dependencies on each other and can
    /// be started in parallel.
    ///
    /// # Returns
    ///
    /// StartupPlan with nodes grouped by level
    pub fn startup_plan(&self) -> Result<StartupPlan> {
        let mut levels: Vec<Vec<String>> = Vec::new();
        let mut processed = HashSet::new();
        let mut current_level = Vec::new();

        // Find nodes with no dependencies (level 0)
        for node in &self.nodes {
            if self.dependencies.get(node).map(|d| d.is_empty()).unwrap_or(true) {
                current_level.push(node.clone());
                processed.insert(node.clone());
            }
        }

        if !current_level.is_empty() {
            levels.push(current_level.clone());
        }

        // Process remaining levels
        while processed.len() < self.nodes.len() {
            let mut next_level = Vec::new();

            for node in &self.nodes {
                if processed.contains(node) {
                    continue;
                }

                // Check if all dependencies are processed
                if let Some(deps) = self.dependencies.get(node) {
                    if deps.iter().all(|dep| processed.contains(dep)) {
                        next_level.push(node.clone());
                        processed.insert(node.clone());
                    }
                }
            }

            if next_level.is_empty() && processed.len() < self.nodes.len() {
                // No progress - must be a cycle
                return Err(Mecha10Error::Configuration(
                    "Circular dependency detected in startup plan".to_string(),
                ));
            }

            if !next_level.is_empty() {
                levels.push(next_level);
            }
        }

        info!("Startup plan created with {} levels", levels.len());
        for (i, level) in levels.iter().enumerate() {
            debug!("  Level {}: {:?}", i, level);
        }

        Ok(StartupPlan { levels })
    }

    /// Get shutdown plan (reverse of startup plan)
    pub fn shutdown_plan(&self) -> Result<ShutdownPlan> {
        let startup = self.startup_plan()?;
        let mut levels = startup.levels;
        levels.reverse();

        info!("Shutdown plan created with {} levels", levels.len());
        for (i, level) in levels.iter().enumerate() {
            debug!("  Level {}: {:?}", i, level);
        }

        Ok(ShutdownPlan { levels })
    }

    /// Get all dependencies of a node (transitive closure)
    ///
    /// Returns all nodes that must be started before this node.
    pub fn get_all_dependencies(&self, node_id: &str) -> HashSet<String> {
        let mut result = HashSet::new();
        let mut to_visit = vec![node_id.to_string()];

        while let Some(current) = to_visit.pop() {
            if let Some(deps) = self.dependencies.get(&current) {
                for dep in deps {
                    if result.insert(dep.clone()) {
                        to_visit.push(dep.clone());
                    }
                }
            }
        }

        result
    }

    /// Get all dependents of a node (transitive closure)
    ///
    /// Returns all nodes that must be stopped before this node can be stopped.
    pub fn get_all_dependents(&self, node_id: &str) -> HashSet<String> {
        let mut result = HashSet::new();
        let mut to_visit = vec![node_id.to_string()];

        while let Some(current) = to_visit.pop() {
            if let Some(deps) = self.dependents.get(&current) {
                for dep in deps {
                    if result.insert(dep.clone()) {
                        to_visit.push(dep.clone());
                    }
                }
            }
        }

        result
    }

    /// Check if a node can be started
    ///
    /// Returns true if all dependencies are in the provided running set.
    pub fn can_start(&self, node_id: &str, running: &HashSet<String>) -> bool {
        if let Some(deps) = self.dependencies.get(node_id) {
            deps.iter().all(|dep| running.contains(dep))
        } else {
            true // No dependencies
        }
    }

    /// Check if a node can be stopped
    ///
    /// Returns true if no dependents are in the provided running set.
    pub fn can_stop(&self, node_id: &str, running: &HashSet<String>) -> bool {
        if let Some(deps) = self.dependents.get(node_id) {
            !deps.iter().any(|dep| running.contains(dep))
        } else {
            true // No dependents
        }
    }

    /// Get nodes that depend on this node
    pub fn get_direct_dependents(&self, node_id: &str) -> Vec<String> {
        self.dependents.get(node_id).cloned().unwrap_or_default()
    }

    /// Get nodes this node depends on
    pub fn get_direct_dependencies(&self, node_id: &str) -> Vec<String> {
        self.dependencies.get(node_id).cloned().unwrap_or_default()
    }

    /// Get the number of nodes in the graph
    pub fn node_count(&self) -> usize {
        self.nodes.len()
    }

    /// Get all node IDs
    pub fn all_nodes(&self) -> Vec<String> {
        self.nodes.iter().cloned().collect()
    }
}

impl Default for DependencyGraph {
    fn default() -> Self {
        Self::new()
    }
}

// ============================================================================
// Startup Plan
// ============================================================================

/// Startup plan with nodes grouped by dependency level
///
/// Nodes at the same level can be started in parallel.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StartupPlan {
    /// Nodes grouped by dependency level
    /// Level 0: No dependencies
    /// Level N: All dependencies are in levels 0..N-1
    pub levels: Vec<Vec<String>>,
}

impl StartupPlan {
    /// Get total number of levels
    pub fn level_count(&self) -> usize {
        self.levels.len()
    }

    /// Get nodes at a specific level
    pub fn get_level(&self, level: usize) -> Option<&Vec<String>> {
        self.levels.get(level)
    }

    /// Get total number of nodes
    pub fn node_count(&self) -> usize {
        self.levels.iter().map(|l| l.len()).sum()
    }

    /// Get maximum parallelism (largest level size)
    pub fn max_parallelism(&self) -> usize {
        self.levels.iter().map(|l| l.len()).max().unwrap_or(0)
    }

    /// Flatten to sequential order
    pub fn flatten(&self) -> Vec<String> {
        self.levels.iter().flatten().cloned().collect()
    }
}

// ============================================================================
// Shutdown Plan
// ============================================================================

/// Shutdown plan with nodes grouped by dependency level (reversed)
///
/// Nodes at the same level can be stopped in parallel.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ShutdownPlan {
    /// Nodes grouped by dependency level (reversed)
    pub levels: Vec<Vec<String>>,
}

impl ShutdownPlan {
    /// Get total number of levels
    pub fn level_count(&self) -> usize {
        self.levels.len()
    }

    /// Get nodes at a specific level
    pub fn get_level(&self, level: usize) -> Option<&Vec<String>> {
        self.levels.get(level)
    }

    /// Get total number of nodes
    pub fn node_count(&self) -> usize {
        self.levels.iter().map(|l| l.len()).sum()
    }

    /// Get maximum parallelism (largest level size)
    pub fn max_parallelism(&self) -> usize {
        self.levels.iter().map(|l| l.len()).max().unwrap_or(0)
    }

    /// Flatten to sequential order
    pub fn flatten(&self) -> Vec<String> {
        self.levels.iter().flatten().cloned().collect()
    }
}

// ============================================================================
// Helper Functions
// ============================================================================

/// Create a simple dependency graph from node ID mappings
///
/// # Example
///
/// ```
/// use mecha10::dependency::create_graph;
///
/// let graph = create_graph(&[
///     ("camera", vec![]),
///     ("vision", vec!["camera"]),
///     ("planner", vec!["vision"]),
/// ])?;
/// ```
pub fn create_graph(nodes: &[(&str, Vec<&str>)]) -> Result<DependencyGraph> {
    let mut graph = DependencyGraph::new();

    // Add all nodes
    for (node_id, _) in nodes {
        graph.add_node(node_id);
    }

    // Add dependencies
    for (node_id, deps) in nodes {
        for dep in deps {
            graph.add_dependency(node_id, dep)?;
        }
    }

    graph.validate()?;
    Ok(graph)
}