Merge pull request #21452 from mvdbeek/fix_lost_delayed_input_parameters

[25.1] Record input parameter invocation inputs

lib/galaxy/workflow/modules.py

@@ -1029,13 +1029,6 @@ class InputModule(WorkflowModule):
                     step_outputs["input_ds_copy"] = new_hdca
                 else:
                     raise Exception("Unknown history content encountered")
-        # If coming from UI - we haven't registered invocation inputs yet,
-        # so do that now so dependent steps can be recalculated. In the future
-        # everything should come in from the API and this can be eliminated.
-        if not invocation.has_input_for_step(step.id):
-            content = next(iter(step_outputs.values()))
-            if content and content is not NO_REPLACEMENT:
-                invocation.add_input(content, step.id)
         progress.set_outputs_for_input(invocation_step, step_outputs)
         return None
 

lib/galaxy/workflow/run.py

@@ -623,6 +623,12 @@ class WorkflowProgress:
 
         if step.label and step.type == "parameter_input" and "output" in outputs:
             self.runtime_replacements[step.label] = str(outputs["output"])
+        invocation = invocation_step.workflow_invocation
+        if not invocation.has_input_for_step(step.id):
+            content = outputs.get("output", NO_REPLACEMENT)
+            if content is not NO_REPLACEMENT:
+                log.info("ADDING INPUT FOR STEP %s: %s", step.id, content, exc_info=True)
+                invocation.add_input(content, step.id)
         self.set_step_outputs(invocation_step, outputs, already_persisted=already_persisted)
 
     def effective_replacement_dict(self):

lib/galaxy_test/api/test_workflows.py

@@ -5648,6 +5648,98 @@ test_data:
             messages = subworkflow_invocation.get("messages", [])
             assert len(messages) == 0, f"Expected no error messages, got: {messages}"
 
+    def test_run_subworkflow_with_boolean_parameter_in_when_condition(self):
+        """Test boolean false parameter passed to subworkflow with when condition.
+
+        This test verifies that boolean parameters (especially false) are properly
+        passed from parent to subworkflow when the subworkflow has:
+        1. Delayed scheduling (via $link)
+        2. A when condition that uses the boolean parameter
+
+        Previously, false values were converted to None in the when expression evaluation,
+        causing "when_not_boolean" errors.
+        """
+        with self.dataset_populator.test_history() as history_id:
+            workflow = """
+class: GalaxyWorkflow
+inputs:
+  should_run:
+    type: boolean
+  some_file:
+    type: data
+steps:
+  nested_workflow:
+    in:
+      subworkflow_should_run: should_run
+      subworkflow_file: some_file
+    run:
+      class: GalaxyWorkflow
+      inputs:
+        subworkflow_should_run: boolean
+        subworkflow_file: data
+      steps:
+        expression:
+          tool_id: expression_forty_two
+          state: {}
+        conditional_step:
+          tool_id: cheetah_casting
+          in:
+            subworkflow_should_run: subworkflow_should_run
+          state:
+            floattest: 3.14
+            inttest:
+              $link: expression/out1
+          when: $(inputs.subworkflow_should_run)
+test_data:
+  some_file:
+    value: 1.bed
+    type: File
+  should_run:
+    value: false
+    type: raw
+"""
+            summary = self._run_workflow(workflow, history_id=history_id, wait=True, assert_ok=True)
+
+            # Verify parent workflow executed successfully
+            parent_invocation = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True)
+            assert parent_invocation["state"] == "scheduled"
+
+            # Find the subworkflow step and get its invocation
+            subworkflow_step = None
+            for step in parent_invocation["steps"]:
+                if step.get("subworkflow_invocation_id"):
+                    subworkflow_step = step
+                    break
+
+            assert subworkflow_step is not None, "No subworkflow step found"
+            subworkflow_invocation_id = subworkflow_step["subworkflow_invocation_id"]
+            subworkflow_invocation = self.workflow_populator.get_invocation(
+                subworkflow_invocation_id, step_details=True
+            )
+
+            # The subworkflow should have succeeded
+            assert (
+                subworkflow_invocation["state"] == "scheduled"
+            ), f"Expected subworkflow to succeed, got state: {subworkflow_invocation['state']}"
+
+            # Should not have error messages (previously failed with "when_not_boolean")
+            messages = subworkflow_invocation.get("messages", [])
+            assert len(messages) == 0, f"Expected no error messages, got: {messages}"
+
+            # Find the conditional step in the subworkflow and verify it was skipped
+            # (when condition was false, so step should not execute)
+            conditional_step = None
+            for step in subworkflow_invocation["steps"]:
+                if step.get("workflow_step_label") == "conditional_step":
+                    conditional_step = step
+                    break
+
+            assert conditional_step is not None, "Conditional step not found in subworkflow"
+            # The step should have been skipped because should_run=false
+            assert len(conditional_step["jobs"]) == 0 or all(
+                j["state"] == "skipped" for j in conditional_step["jobs"]
+            ), "Expected conditional step to be skipped when should_run=false"
+
     def test_run_with_non_optional_data_unspecified_fails_invocation(self):
         with self.dataset_populator.test_history() as history_id:
             error = self._run_jobs(

test/unit/workflows/test_workflow_progress.py

@@ -94,8 +94,7 @@ class TestWorkflowProgress(TestCase):
                 workflow_invocation_step.state = "scheduled"
                 workflow_invocation_step.workflow_step = self._step(i)
                 assert step_id == self._step(i).id
-                # workflow_invocation_step.workflow_invocation = self.invocation
-                self.invocation.steps.append(workflow_invocation_step)
+                workflow_invocation_step.workflow_invocation = self.invocation
 
             workflow_invocation_step_state = model.WorkflowRequestStepState()
             workflow_invocation_step_state.workflow_step_id = step_id
@@ -111,6 +110,7 @@ class TestWorkflowProgress(TestCase):
         else:
             workflow_invocation_step = model.WorkflowInvocationStep()
             workflow_invocation_step.workflow_step = self._step(index)
+            workflow_invocation_step.workflow_invocation = self.invocation
             return workflow_invocation_step
 
     def test_connect_data_input(self):
@@ -211,6 +211,7 @@ class TestWorkflowProgress(TestCase):
         subworkflow_invocation_step.workflow_step_id = subworkflow_input_step.id
         subworkflow_invocation_step.state = "new"
         subworkflow_invocation_step.workflow_step = subworkflow_input_step
+        subworkflow_invocation_step.workflow_invocation = subworkflow_invocation
 
         subworkflow_progress.set_outputs_for_input(subworkflow_invocation_step)