jorm 0.1.1

A lightweight DAG execution engine with natural language processing
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

use crate::core::task::Task;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Dag {
    pub name: String,
    pub tasks: HashMap<String, Task>,
    pub dependencies: HashMap<String, Vec<String>>,
}

impl Dag {
    pub fn new(name: String) -> Self {
        Self {
            name,
            tasks: HashMap::new(),
            dependencies: HashMap::new(),
        }
    }

    pub fn add_task(&mut self, task: Task) {
        self.tasks.insert(task.name.clone(), task);
    }

    pub fn add_dependency(&mut self, task_name: String, depends_on: Vec<String>) {
        self.dependencies.insert(task_name, depends_on);
    }

    pub fn validate(&self) -> Result<(), crate::core::error::JormError> {
        // Check for circular dependencies
        self.check_circular_dependencies()?;

        // Validate task references in dependencies
        self.validate_task_references()?;

        // Validate all tasks have required parameters
        self.validate_task_parameters()?;

        // Validate DAG has at least one task
        if self.tasks.is_empty() {
            return Err(crate::core::error::JormError::ParseError(
                "DAG must contain at least one task".to_string(),
            ));
        }

        Ok(())
    }

    fn check_circular_dependencies(&self) -> Result<(), crate::core::error::JormError> {
        // Simple cycle detection using DFS with path tracking
        let mut visited = HashMap::new();
        let mut rec_stack = HashMap::new();
        let mut path = Vec::new();

        for task_name in self.tasks.keys() {
            if !visited.get(task_name).unwrap_or(&false) {
                if let Some(cycle_path) =
                    self.find_cycle(task_name, &mut visited, &mut rec_stack, &mut path)?
                {
                    return Err(crate::core::error::JormError::ParseError(format!(
                        "Circular dependency detected: {}",
                        cycle_path.join(" -> ")
                    )));
                }
            }
        }
        Ok(())
    }

    fn find_cycle(
        &self,
        task_name: &str,
        visited: &mut HashMap<String, bool>,
        rec_stack: &mut HashMap<String, bool>,
        path: &mut Vec<String>,
    ) -> Result<Option<Vec<String>>, crate::core::error::JormError> {
        visited.insert(task_name.to_string(), true);
        rec_stack.insert(task_name.to_string(), true);
        path.push(task_name.to_string());

        if let Some(deps) = self.dependencies.get(task_name) {
            for dep in deps {
                if !visited.get(dep).unwrap_or(&false) {
                    if let Some(cycle) = self.find_cycle(dep, visited, rec_stack, path)? {
                        return Ok(Some(cycle));
                    }
                } else if *rec_stack.get(dep).unwrap_or(&false) {
                    // Found cycle - construct the cycle path
                    let cycle_start = path.iter().position(|x| x == dep).unwrap();
                    let mut cycle_path = path[cycle_start..].to_vec();
                    cycle_path.push(dep.to_string());
                    return Ok(Some(cycle_path));
                }
            }
        }

        rec_stack.insert(task_name.to_string(), false);
        path.pop();
        Ok(None)
    }

    fn validate_task_references(&self) -> Result<(), crate::core::error::JormError> {
        for (task_name, deps) in &self.dependencies {
            if !self.tasks.contains_key(task_name) {
                return Err(crate::core::error::JormError::ParseError(format!(
                    "Task '{}' referenced in dependencies but not defined",
                    task_name
                )));
            }

            for dep in deps {
                if !self.tasks.contains_key(dep) {
                    return Err(crate::core::error::JormError::ParseError(format!(
                        "Dependency '{}' for task '{}' not found",
                        dep, task_name
                    )));
                }
            }
        }
        Ok(())
    }

    fn validate_task_parameters(&self) -> Result<(), crate::core::error::JormError> {
        for (task_name, task) in &self.tasks {
            // Validate each task's parameters
            task.validate().map_err(|e| {
                crate::core::error::JormError::ParseError(format!(
                    "Task '{}' validation failed: {}",
                    task_name, e
                ))
            })?;
        }
        Ok(())
    }
}