Pipeline Runner Implementation

.clippy.toml

@@ -1,3 +1,3 @@
-excessive-nesting-threshold = 4
+excessive-nesting-threshold = 6
 too-many-arguments-threshold = 10
 allowed-idents-below-min-chars = ["..", "k", "v", "f", "re", "id", "Ok", "'_"]

.devcontainer/devcontainer.json

@@ -21,6 +21,7 @@
 	},
 	"runArgs": [
 		"--name=${localWorkspaceFolderBasename}_devcontainer",
+		"--gpus=all",
 		"--privileged",
 		"--cgroupns=host"
 	],

.devcontainer/gpu/devcontainer.json

@@ -21,7 +21,6 @@
 	},
 	"runArgs": [
 		"--name=${localWorkspaceFolderBasename}_devcontainer",
-		"--gpus=all",
 		"--privileged",
 		"--cgroupns=host"
 	],

.vscode/settings.json

@@ -21,4 +21,4 @@
     "python.terminal.activateEnvironment": false,
     "notebook.formatOnSave.enabled": true,
     "notebook.output.scrolling": true
-}
+}

Cargo.toml

@@ -49,10 +49,13 @@ getset = { version = "0.1.5", git = "https://github.com/guzman-raphael/getset.gi
 glob = "0.3.1"
 # strings to snake_case
 heck = "0.5.0"
+hostname = "0.4.1"
 # convert bytes to hex strings
 hex = "0.4.3"
 # hashmaps that preserve insertion order
 indexmap = { version = "2.9.0", features = ["serde"] }
+itertools = "0.14.0"
+layout-rs = "0.1.3"
 # random name generator
 names = "0.14.0"
 # graph algorithms
@@ -73,6 +76,7 @@ sha2 = "0.10.8"
 snafu = { version = "0.8.5", features = ["futures"] }
 # a runtime for async applications
 tokio = { version = "1.41.0", features = ["full"] }
+tokio-stream = "0.1.17"
 # utilities for async calls
 tokio-util = "0.7.13"
 # automated CFFI + bindings in other languages

cspell.json

@@ -78,7 +78,14 @@
         "strsim",
         "getrandom",
         "wasi",
-        "patchelf"
+        "petgraph",
+        "rfind",
+        "itertools",
+        "oneshot",
+        "patchelf",
+        "colinianking",
+        "bitcode",
+        "pathset"
     ],
     "useGitignore": false,
     "ignorePaths": [

output.txt

@@ -0,0 +1,3 @@
+1
+2
+2

src/core/error.rs

@@ -10,6 +10,7 @@ use std::{
     io,
     path::{self},
 };
+use tokio::sync::oneshot;
 use tokio::task;
 
 impl From<BollardError> for OrcaError {
@@ -22,6 +23,16 @@ impl From<BollardError> for OrcaError {
         }
     }
 }
+impl From<oneshot::error::RecvError> for OrcaError {
+    fn from(error: oneshot::error::RecvError) -> Self {
+        Self {
+            kind: Kind::ChannelReceiveError {
+                source: error,
+                backtrace: Some(Backtrace::capture()),
+            },
+        }
+    }
+}
 impl From<PestError> for OrcaError {
     fn from(error: PestError) -> Self {
         Self {
@@ -113,9 +124,11 @@ impl fmt::Debug for OrcaError {
         match &self.kind {
             Kind::AgentCommunicationFailure { backtrace, .. }
             | Kind::EmptyResponseWhenLoadingContainerAltImage { backtrace, .. }
+            | Kind::FailedToParseDot { backtrace, .. }
             | Kind::GeneratedNamesOverflow { backtrace, .. }
             | Kind::IncompletePacket { backtrace, .. }
             | Kind::InvalidFilepath { backtrace, .. }
+            | Kind::InvalidIndex { backtrace, .. }
             | Kind::InvalidPodResultTerminatedDatetime { backtrace, .. }
             | Kind::KeyMissing { backtrace, .. }
             | Kind::NoAnnotationFound { backtrace, .. }
@@ -124,11 +137,17 @@ impl fmt::Debug for OrcaError {
             | Kind::NoMatchingPodRun { backtrace, .. }
             | Kind::NoRemainingServices { backtrace, .. }
             | Kind::NoTagFoundInContainerAltImage { backtrace, .. }
+            | Kind::PodJobSubmissionFailed { backtrace, .. }
+            | Kind::PodJobProcessingError { backtrace, .. }
+            | Kind::StatusConversionFailure { backtrace, .. }
+            | Kind::UnsupportedPathType { backtrace, .. }
             | Kind::BollardError { backtrace, .. }
+            | Kind::ChannelReceiveError { backtrace, .. }
             | Kind::DOTError { backtrace, .. }
             | Kind::GlobPatternError { backtrace, .. }
             | Kind::IoError { backtrace, .. }
             | Kind::PathPrefixError { backtrace, .. }
+            | Kind::SendError { backtrace, .. }
             | Kind::SerdeJsonError { backtrace, .. }
             | Kind::SerdeYamlError { backtrace, .. }
             | Kind::TokioTaskJoinError { backtrace, .. } => {

src/core/graph.rs

@@ -1,5 +1,5 @@
 use crate::{
-    core::{pipeline::PipelineNode, util::get},
+    core::{model::pipeline::PipelineNode, util::get},
     uniffi::{error::Result, model::pipeline::Kernel},
 };
 use dot_parser::ast::Graph as DOTGraph;

src/core/mod.rs

@@ -2,10 +2,9 @@
 pub mod crypto;
 pub(crate) mod error;
 pub(crate) mod graph;
-/// Components of the data model.
+/// Model definitions and utilities.
 pub mod model;
 pub(crate) mod orchestrator;
-pub(crate) mod pipeline;
 pub(crate) mod store;
 pub(crate) mod util;
 pub(crate) mod validation;

src/core/model.rs → src/core/model/mod.rs

@@ -1,19 +1,13 @@
-use crate::{
-    core::util::get_type_name,
-    uniffi::{
-        error::Result,
-        model::pod::{Pod, PodJob},
-    },
-};
+use crate::{core::util::get_type_name, uniffi::error::Result};
 use heck::ToSnakeCase as _;
 use indexmap::IndexMap;
-use serde::{Deserialize as _, Deserializer, Serialize, Serializer};
+use serde::{Serialize, Serializer};
 use serde_yaml::{self, Value};
 use std::{
     collections::{BTreeMap, HashMap},
     result,
-    sync::Arc,
 };
+
 /// Converts a model instance into a consistent yaml.
 ///
 /// # Errors
@@ -63,54 +57,5 @@ where
     sorted.serialize(serializer)
 }
 
-#[expect(
-    clippy::expect_used,
-    reason = "Function signature required by serde API."
-)]
-pub(crate) fn deserialize_pod<'de, D>(deserializer: D) -> result::Result<Arc<Pod>, D::Error>
-where
-    D: Deserializer<'de>,
-{
-    let value = Value::deserialize(deserializer)?;
-    (value).as_str().map_or_else(
-        || {
-            Ok(serde_yaml::from_value(value.clone())
-                .expect("Failed to convert from serde value to specific type."))
-        },
-        |hash| {
-            Ok({
-                Pod {
-                    hash: hash.to_owned(),
-                    ..Pod::default()
-                }
-                .into()
-            })
-        },
-    )
-}
-
-#[expect(
-    clippy::expect_used,
-    reason = "Function signature required by serde API."
-)]
-pub(crate) fn deserialize_pod_job<'de, D>(deserializer: D) -> result::Result<Arc<PodJob>, D::Error>
-where
-    D: Deserializer<'de>,
-{
-    let value = Value::deserialize(deserializer)?;
-    (value).as_str().map_or_else(
-        || {
-            Ok(serde_yaml::from_value(value.clone())
-                .expect("Failed to convert from serde value to specific type."))
-        },
-        |hash| {
-            Ok({
-                PodJob {
-                    hash: hash.to_owned(),
-                    ..PodJob::default()
-                }
-                .into()
-            })
-        },
-    )
-}
+pub(crate) mod pipeline;
+pub(crate) mod pod;

src/core/model/pipeline.rs

@@ -0,0 +1,43 @@
+use std::collections::HashSet;
+
+use crate::uniffi::model::pipeline::{Kernel, Pipeline};
+use petgraph::Direction::Incoming;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
+pub struct PipelineNode {
+    pub name: String,
+    pub kernel: Kernel,
+}
+
+impl Pipeline {
+    /// Function to get the parents of a node
+    pub(crate) fn get_node_parents(
+        &self,
+        node: &PipelineNode,
+    ) -> impl Iterator<Item = &PipelineNode> {
+        // Find the NodeIndex for the given node_key
+        let node_index = self
+            .graph
+            .node_indices()
+            .find(|&idx| self.graph[idx] == *node);
+        node_index.into_iter().flat_map(move |idx| {
+            self.graph
+                .neighbors_directed(idx, Incoming)
+                .map(move |parent_idx| &self.graph[parent_idx])
+        })
+    }
+
+    /// Return a vec of `node_names` that takes in inputs based on the `input_spec`ec
+    pub(crate) fn get_input_nodes(&self) -> HashSet<&String> {
+        let mut input_nodes = HashSet::new();
+
+        self.input_spec.iter().for_each(|(_, node_uris)| {
+            for node_uri in node_uris {
+                input_nodes.insert(&node_uri.node_name);
+            }
+        });
+
+        input_nodes
+    }
+}

src/core/model/pod.rs

@@ -0,0 +1,56 @@
+use crate::uniffi::model::pod::{Pod, PodJob};
+use serde::{Deserialize as _, Deserializer};
+use serde_yaml::{self, Value};
+use std::{result, sync::Arc};
+
+#[expect(
+    clippy::expect_used,
+    reason = "Function signature required by serde API."
+)]
+pub fn deserialize_pod<'de, D>(deserializer: D) -> result::Result<Arc<Pod>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value = Value::deserialize(deserializer)?;
+    (value).as_str().map_or_else(
+        || {
+            Ok(serde_yaml::from_value(value.clone())
+                .expect("Failed to convert from serde value to specific type."))
+        },
+        |hash| {
+            Ok({
+                Pod {
+                    hash: hash.to_owned(),
+                    ..Pod::default()
+                }
+                .into()
+            })
+        },
+    )
+}
+
+#[expect(
+    clippy::expect_used,
+    reason = "Function signature required by serde API."
+)]
+pub fn deserialize_pod_job<'de, D>(deserializer: D) -> result::Result<Arc<PodJob>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value = Value::deserialize(deserializer)?;
+    (value).as_str().map_or_else(
+        || {
+            Ok(serde_yaml::from_value(value.clone())
+                .expect("Failed to convert from serde value to specific type."))
+        },
+        |hash| {
+            Ok({
+                PodJob {
+                    hash: hash.to_owned(),
+                    ..PodJob::default()
+                }
+                .into()
+            })
+        },
+    )
+}

src/core/orchestrator/agent.rs

@@ -28,7 +28,7 @@ static RE_PODJOB_ACTION: LazyLock<Regex> = LazyLock::new(|| {
                 group\/(?<group>[a-z_\-]+)\/
                     (?<action>request|reservation|success|failure)\/
                         pod_job\/(?<pod_job_hash>[0-9a-f]+)\/
-                            host\/(?<host>[a-z_]+)\/
+                            host\/(?<host>[0-9a-z_]+)\/
                                 timestamp\/(?<timestamp>.*?)
             $
             ",
@@ -101,8 +101,8 @@ impl AgentClient {
 }
 
 #[expect(
-    clippy::excessive_nesting,
     clippy::let_underscore_must_use,
+    clippy::excessive_nesting,
     reason = "`result::Result<(), SendError<_>>` is the only uncaptured result since it would mean we can't transmit results over mpsc."
 )]
 pub async fn start_service<
@@ -165,6 +165,7 @@ where
                         subgroup: metadata["pod_job_hash"].to_string(),
                     };
                     let _event_payload = event_classifier(&input);
+                    println!("Sending it to request task.");
                     tasks.spawn({
                         let inner_request_task = request_task.clone();
                         let inner_inner_agent = Arc::clone(&inner_agent);