fixed workflow step input

JensKrumsieck · JensKrumsieck · commit 37f4dc21050b · 2025-05-19T16:27:56.000+02:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -8,7 +8,7 @@
 - fixed a bug with Dockerfile path resolution
 - handle NetworkAccess Requirment in runner
 - inherit parents requirements correclty
-- ramping up runner conformance from 160/378 to 186/378
+- ramping up runner conformance from 160/378 to 188/378
 
 # v0.5.2
 ## 🐛 Bugfixes
diff --git a/crates/cwl-execution/src/runner.rs b/crates/cwl-execution/src/runner.rs
@@ -18,7 +18,7 @@ use crate::{
 };
 use cwl::{
     clt::{Argument, Command, CommandLineTool},
-    inputs::{CommandLineBinding, WorkflowStepInput},
+    inputs::CommandLineBinding,
     requirements::{DockerRequirement, InlineJavascriptRequirement, StringOrInclude},
     types::{CWLType, DefaultValue, Directory, Entry, File, PathItem},
     wf::{StringOrDocument, Workflow},
@@ -75,44 +75,29 @@ pub fn run_workflow(
 
             //map inputs to correct fields
             let mut step_inputs = HashMap::new();
-            for (key, input) in &step.in_ {
-                match input {
-                    WorkflowStepInput::String(in_string) => {
-                        let parts: Vec<&str> = in_string.split('/').collect();
-                        if parts.len() == 2 {
-                            if let Some(out_value) = outputs.get(in_string) {
-                                step_inputs.insert(key.to_string(), out_value.to_default_value());
-                            }
-                        } else if let Some(input) = workflow.inputs.iter().find(|i| i.id == *in_string) {
-                            let value = evaluate_input(input, &input_values.inputs)?;
-                            step_inputs.insert(key.to_string(), value.to_owned());
-                        }
+            for parameter in &step.in_ {
+                let source = parameter.source.as_deref().unwrap_or_default();
+                let source_parts: Vec<&str> = source.split('/').collect();
+
+                //try output
+                if source_parts.len() == 2 {
+                    if let Some(out_value) = outputs.get(source) {
+                        step_inputs.insert(parameter.id.to_string(), out_value.to_default_value());
+                        continue;
                     }
-                    WorkflowStepInput::Parameter(parameter) => {
-                        let source = parameter.source.as_deref().unwrap_or_default();
-                        let source_parts: Vec<&str> = source.split('/').collect();
-
-                        if source_parts.len() == 2 {
-                            //try output
-                            if let Some(out_value) = outputs.get(source) {
-                                step_inputs.insert(key.to_string(), out_value.to_default_value());
-                                continue;
-                            }
-                        }
-
-                        //try default
-                        if let Some(default) = &parameter.default {
-                            step_inputs.entry(key.to_string()).or_insert(default.to_owned());
-                        }
+                }
+                //try default
+                if let Some(default) = &parameter.default {
+                    step_inputs.entry(parameter.id.to_string()).or_insert(default.to_owned());
+                }
 
-                        if let Some(input) = workflow.inputs.iter().find(|i| i.id == *source) {
-                            let value = evaluate_input(input, &input_values.inputs)?;
-                            match value {
-                                DefaultValue::Any(val) if val.is_null() => continue,
-                                _ => {
-                                    step_inputs.insert(key.to_string(), value.to_owned());
-                                }
-                            }
+                //try input
+                if let Some(input) = workflow.inputs.iter().find(|i| i.id == *source) {
+                    let value = evaluate_input(input, &input_values.inputs)?;
+                    match value {
+                        DefaultValue::Any(val) if val.is_null() => continue,
+                        _ => {
+                            step_inputs.insert(parameter.id.to_string(), value.to_owned());
                         }
                     }
                 }
@@ -244,7 +229,7 @@ pub fn run_tool(
 
     //build runtime object
     let mut runtime = RuntimeEnvironment::initialize(tool, &input_values, dir.path(), tool_path, tmp_dir.path())?;
-    
+
     //replace inputs and runtime placeholders in tool with the actual values
     set_placeholder_values(tool, &runtime, &mut input_values);
     runtime.environment = collect_environment(&input_values);
diff --git a/crates/cwl/src/inputs.rs b/crates/cwl/src/inputs.rs
@@ -121,28 +121,34 @@ where
     Ok(parameters)
 }
 
-#[derive(Serialize, Deserialize, Debug, PartialEq, Clone)]
-#[serde(untagged)]
-pub enum WorkflowStepInput {
-    String(String),
-    Parameter(Box<WorkflowStepInputParameter>),
-}
-
-impl Default for WorkflowStepInput {
-    fn default() -> Self {
-        WorkflowStepInput::String(String::default())
-    }
-}
-
 #[derive(Serialize, Deserialize, Debug, Default, PartialEq, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct WorkflowStepInputParameter {
+    #[serde(default)]
+    pub id: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub source: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub default: Option<DefaultValue>,
 }
 
+impl WorkflowStepInputParameter {
+    pub fn with_id(mut self, id: impl ToString) -> Self {
+        self.id = id.to_string();
+        self
+    }
+
+    pub fn with_source(mut self, source: impl ToString) -> Self {
+        self.source = Some(source.to_string());
+        self
+    }
+
+    pub fn with_default(mut self, f: DefaultValue) -> Self {
+        self.default = Some(f);
+        self
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -154,10 +160,4 @@ mod tests {
         input.set_id("test".to_string());
         assert_eq!(input.id(), "test");
     }
-
-    #[test]
-    pub fn test_workflow_step_input_default() {
-        let input = WorkflowStepInput::default();
-        assert_eq!(input, WorkflowStepInput::String(String::default()));
-    }
 }
diff --git a/crates/cwl/src/wf.rs b/crates/cwl/src/wf.rs
@@ -1,11 +1,12 @@
 use super::{
     deserialize::{deserialize_list, Identifiable},
-    inputs::WorkflowStepInput,
+    inputs::WorkflowStepInputParameter,
     outputs::WorkflowOutputParameter,
     requirements::{deserialize_hints, deserialize_requirements, Requirement},
     CWLDocument, DocumentBase,
 };
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Deserializer, Serialize};
+use serde_yaml::Value;
 use std::{
     collections::{HashMap, VecDeque},
     ops::{Deref, DerefMut},
@@ -79,15 +80,9 @@ impl Workflow {
 
     /// Checks whether the `Workflow` has a `WorkflowStep` with an input of id `id`
     pub fn has_step_input(&self, id: &str) -> bool {
-        self.steps.iter().any(|step| {
-            step.in_.clone().into_values().any(|val| {
-                let src = match val {
-                    WorkflowStepInput::String(str) => str,
-                    WorkflowStepInput::Parameter(par) => par.source.unwrap_or_default(),
-                };
-                src == id
-            })
-        })
+        self.steps
+            .iter()
+            .any(|step| step.in_.iter().any(|val| val.source == Some(id.to_string())))
     }
 
     /// Checks whether the `Workflow` has a `WorkflowStep` with an ouput of id `id`
@@ -117,16 +112,11 @@ impl Workflow {
         for step in &self.steps {
             in_degree.entry(step.id.clone()).or_insert(0);
 
-            for input in step.in_.values() {
-                let parts: Vec<&str> = match input {
-                    WorkflowStepInput::String(string) => string.split('/').collect(),
-                    WorkflowStepInput::Parameter(parameter) => {
-                        if let Some(source) = &parameter.source {
-                            source.split('/').collect()
-                        } else {
-                            vec![]
-                        }
-                    }
+            for input in &step.in_ {
+                let parts: Vec<&str> = if let Some(source) = &input.source {
+                    source.split('/').collect()
+                } else {
+                    vec![]
                 };
 
                 if parts.len() == 2 {
@@ -167,8 +157,10 @@ impl Workflow {
 pub struct WorkflowStep {
     #[serde(default)]
     pub id: String,
-    pub run: StringOrDocument,
-    pub in_: HashMap<String, WorkflowStepInput>,
+    pub run: StringOrDocument,    
+    #[serde(deserialize_with = "deserialize_workflow_inputs")]
+    pub in_: Vec<WorkflowStepInputParameter>,
+    //pub in_: HashMap<String, WorkflowStepInput>,
     pub out: Vec<String>,
 
     #[serde(skip_serializing_if = "Vec::is_empty")]
@@ -191,6 +183,41 @@ impl Identifiable for WorkflowStep {
     }
 }
 
+pub fn deserialize_workflow_inputs<'de, D>(deserializer: D) -> Result<Vec<WorkflowStepInputParameter>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value: Value = Deserialize::deserialize(deserializer)?;
+
+    let parameters = match value {
+        Value::Sequence(seq) => seq
+            .into_iter()
+            .map(|item| {
+                let param: WorkflowStepInputParameter = serde_yaml::from_value(item).map_err(serde::de::Error::custom)?;
+                Ok(param)
+            })
+            .collect::<Result<Vec<_>, _>>()?,
+        Value::Mapping(map) => map
+            .into_iter()
+            .map(|(key, value)| {
+                let id = key.as_str().ok_or_else(|| serde::de::Error::custom("Expected string key"))?;
+                let param = if let Value::String(source_str) = value {
+                    WorkflowStepInputParameter::default().with_id(id).with_source(source_str)
+                } else {
+                    let mut param: WorkflowStepInputParameter = serde_yaml::from_value(value).map_err(serde::de::Error::custom)?;
+                    param.id = id.to_string();
+                    param
+                };
+
+                Ok(param)
+            })
+            .collect::<Result<Vec<_>, _>>()?,
+        _ => return Err(serde::de::Error::custom("Expected sequence or mapping for inputs")),
+    };
+
+    Ok(parameters)
+}
+
 #[derive(Serialize, Deserialize, Debug, PartialEq, Clone)]
 #[serde(untagged)]
 pub enum StringOrDocument {
diff --git a/src/commands/workflow.rs b/src/commands/workflow.rs
@@ -7,7 +7,6 @@ use clap::{Args, Subcommand};
 use colored::Colorize;
 use cwl::{
     format::format_cwl,
-    inputs::WorkflowStepInput,
     load_tool, load_workflow,
     wf::{StringOrDocument, Workflow},
     CWLDocument,
@@ -220,7 +219,7 @@ pub fn get_workflow_status(args: &CreateWorkflowArgs) -> Result<(), Box<dyn Erro
             .inputs
             .iter()
             .map(|input| {
-                if step.in_.contains_key(&input.id) {
+                if step.in_.iter().any(|i| i.id == input.id) {
                     format!("✅    {}", input.id)
                 } else if input.default.is_some() {
                     format!("🔘    {}", input.id)
@@ -235,15 +234,11 @@ pub fn get_workflow_status(args: &CreateWorkflowArgs) -> Result<(), Box<dyn Erro
             .outputs
             .iter()
             .map(|output| {
-                if workflow.steps.iter().any(|s| {
-                    s.in_.clone().into_values().any(|v| {
-                        let src = match v {
-                            WorkflowStepInput::String(str) => str,
-                            WorkflowStepInput::Parameter(par) => par.source.unwrap_or_default(),
-                        };
-                        src == format!("{}/{}", step.id, output.id)
-                    })
-                }) || workflow.outputs.iter().any(|o| o.output_source == format!("{}/{}", step.id, output.id))
+                if workflow
+                    .steps
+                    .iter()
+                    .any(|s| s.in_.clone().iter().any(|v| v.source == Some(format!("{}/{}", step.id, output.id))))
+                    || workflow.outputs.iter().any(|o| o.output_source == format!("{}/{}", step.id, output.id))
                 {
                     format!("✅    {}", output.id)
                 } else {
diff --git a/src/cwl.rs b/src/cwl.rs
@@ -1,7 +1,7 @@
 use crate::io::{get_workflows_folder, resolve_path};
 use cwl::{
     clt::CommandLineTool,
-    inputs::{CommandInputParameter, WorkflowStepInput},
+    inputs::{CommandInputParameter, WorkflowStepInputParameter},
     load_tool,
     outputs::WorkflowOutputParameter,
     requirements::{Requirement, WorkDirItem},
@@ -98,7 +98,11 @@ impl Connectable for Workflow {
             .find(|step| step.id == to_parts[0])
             .unwrap()
             .in_
-            .insert(to_parts[1].to_string(), WorkflowStepInput::String(from_input.to_owned()));
+            .push(WorkflowStepInputParameter {
+                id: to_parts[1].to_string(),
+                source: Some(from_input.to_owned()),
+                ..Default::default()
+            });
 
         info!("➕ Added or updated connection from inputs.{from_input} to {to} in workflow");
 
@@ -160,7 +164,11 @@ impl Connectable for Workflow {
         }
 
         let step = self.steps.iter_mut().find(|s| s.id == to_parts[0]).unwrap(); //safe here!
-        step.in_.insert(to_parts[1].to_string(), WorkflowStepInput::String(from.to_string()));
+        step.in_.push(WorkflowStepInputParameter {
+            id: to_parts[1].to_string(),
+            source: Some(from.to_string()),
+            ..Default::default()
+        });
 
         Ok(())
     }
@@ -182,7 +190,8 @@ impl Connectable for Workflow {
         // If the step is found, try to remove the connection by removing input from `tool_y` that uses output of `tool_x`
         //Input is empty, change that?
         if let Some(step) = step {
-            if step.in_.remove(to_parts[1]).is_some() {
+            if step.in_.iter().any(|v| v.id == to_parts[1]) {
+                step.in_.retain(|v| v.id != to_parts[1]);
                 info!("🔗 Successfully disconnected {from} from {to}");
             } else {
                 warn!("No connection found between {from} and {to}. Nothing to disconnect.");
@@ -203,7 +212,8 @@ impl Connectable for Workflow {
             self.inputs.remove(index);
         }
         if let Some(step) = self.steps.iter_mut().find(|s| s.id == to_parts[0]) {
-            if step.in_.remove(to_parts[1]).is_some() {
+            if step.in_.iter().any(|v| v.id == to_parts[1]) {
+                step.in_.retain(|v| v.id != to_parts[1]);
                 info!("➖ Successfully disconnected input {from_input} from {to}");
             } else {
                 warn!("No input connection found for {from_input} to disconnect.");
