Don't create workflow outputs to recover input parameter outputs

Just eval the code that determines the outputs of the step again.
Avoids modifying the workflow as it executes, which is bad.

Author: mvdbeek

lib/galaxy/workflow/modules.py

@@ -1620,21 +1620,13 @@ def get_all_outputs(self, data_only=False):
         ]
 
     def execute(
-        self, trans, progress: "WorkflowProgress", invocation_step, use_cached_job: bool = False
+        self,
+        trans,
+        progress: "WorkflowProgress",
+        invocation_step: "WorkflowInvocationStep",
+        use_cached_job: bool = False,
     ) -> Optional[bool]:
-        step = invocation_step.workflow_step
-        if step.id in progress.inputs_by_step_id:
-            input_value = progress.inputs_by_step_id[step.id]
-        else:
-            input_value = step.state.inputs["input"]
-        if input_value is NO_REPLACEMENT:
-            default_value = step.get_input_default_value(NO_REPLACEMENT)
-            # TODO: look at parameter type and infer if value should be a dictionary
-            # instead. Guessing only field parameter types in CWL branch would have
-            # default as dictionary like this.
-            if not isinstance(default_value, dict):
-                default_value = {"value": default_value}
-            input_value = default_value.get("value", NO_REPLACEMENT)
+        input_value = self.get_input_value(progress, invocation_step)
         input_param = self.get_runtime_inputs(self)["input"]
         # TODO: raise DelayedWorkflowEvaluation if replacement not ready ? Need test
         try:
@@ -1654,13 +1646,37 @@ def execute(
         except ValueError as e:
             raise FailWorkflowEvaluation(
                 why=InvocationFailureWorkflowParameterInvalid(
-                    reason=FailureReason.workflow_parameter_invalid, workflow_step_id=step.id, details=str(e)
+                    reason=FailureReason.workflow_parameter_invalid,
+                    workflow_step_id=invocation_step.workflow_step_id,
+                    details=str(e),
                 )
             )
         step_outputs = dict(output=input_value)
         progress.set_outputs_for_input(invocation_step, step_outputs)
         return None
 
+    def get_input_value(self, progress: "WorkflowProgress", invocation_step: "WorkflowInvocationStep"):
+        step = invocation_step.workflow_step
+        if step.id in progress.inputs_by_step_id:
+            input_value = progress.inputs_by_step_id[step.id]
+        else:
+            assert step.state
+            input_value = step.state.inputs["input"]
+        if input_value is NO_REPLACEMENT:
+            default_value = step.get_input_default_value(NO_REPLACEMENT)
+            # TODO: look at parameter type and infer if value should be a dictionary
+            # instead. Guessing only field parameter types in CWL branch would have
+            # default as dictionary like this.
+            if not isinstance(default_value, dict):
+                default_value = {"value": default_value}
+            input_value = default_value.get("value", NO_REPLACEMENT)
+        return input_value
+
+    def recover_mapping(self, invocation_step: "WorkflowInvocationStep", progress: "WorkflowProgress"):
+        input_value = self.get_input_value(progress, invocation_step)
+        step_outputs = dict(output=input_value)
+        progress.set_outputs_for_input(invocation_step, step_outputs, already_persisted=True)
+
     def step_state_to_tool_state(self, state):
         state = safe_loads(state)
         default_set, default_value = False, None

lib/galaxy/workflow/run.py

@@ -611,17 +611,9 @@ def set_step_outputs(
             outputs[invocation_step.output_value.workflow_output.output_name] = invocation_step.output_value.value
         self.outputs[step.id] = outputs
         if not already_persisted:
-            workflow_outputs_by_name = {wo.output_name: wo for wo in step.workflow_outputs}
             for output_name, output_object in outputs.items():
                 if hasattr(output_object, "history_content_type"):
                     invocation_step.add_output(output_name, output_object)
-                else:
-                    # Add this non-data, non workflow-output output to the workflow outputs.
-                    # This is required for recovering the output in the next scheduling iteration,
-                    # and should be replaced with a WorkflowInvocationStepOutputValue ASAP.
-                    if not workflow_outputs_by_name.get(output_name) and output_object is not NO_REPLACEMENT:
-                        workflow_output = model.WorkflowOutput(step, output_name=output_name)
-                        step.workflow_outputs.append(workflow_output)
             for workflow_output in step.workflow_outputs:
                 assert workflow_output.output_name
                 output_name = workflow_output.output_name