Don't replace default values too early

We should probably only do this once we're ready to execute the step,
not when injecting modules and building the tool state.

lib/galaxy/model/__init__.py

@@ -8206,17 +8206,18 @@ def setup_inputs_by_name(self):
         # Ensure input_connections has already been set.
 
         # Make connection information available on each step by input name.
-        inputs_by_name = {}
+        inputs_by_name: Dict[str, Any] = {}
         for step_input in self.inputs:
             input_name = step_input.name
             assert input_name not in inputs_by_name
             inputs_by_name[input_name] = step_input
         self._inputs_by_name = inputs_by_name
+        return inputs_by_name
 
     @property
     def inputs_by_name(self):
         if self._inputs_by_name is None:
-            self.setup_inputs_by_name()
+            return self.setup_inputs_by_name()
         return self._inputs_by_name
 
     def get_input(self, input_name):

lib/galaxy/tools/parameters/__init__.py

@@ -195,7 +195,7 @@ def callback_helper(input, input_values, name_prefix, label_prefix, parent_prefi
         elif replace_optional_connections:
             # Only used in workflow context
             has_default = hasattr(input, "value")
-            if new_value is value is NO_REPLACEMENT:
+            if new_value is value is NO_REPLACEMENT or is_runtime_value(value):
                 # NO_REPLACEMENT means value was connected but left unspecified
                 if has_default:
                     # Use default if we have one
@@ -206,9 +206,6 @@ def callback_helper(input, input_values, name_prefix, label_prefix, parent_prefi
                     # We might want to raise an exception here, instead of depending on a tool parameter value error.
                     input_values[input.name] = None
 
-            elif is_runtime_value(value) and has_default:
-                input_values[input.name] = input.value
-
     def get_current_case(input, input_values):
         test_parameter = input.test_param
         test_parameter_name = test_parameter.name

lib/galaxy/workflow/modules.py

@@ -410,14 +410,11 @@ def get_runtime_inputs(self, step, connections: Optional[Iterable[WorkflowStepCo
         """
         return {}
 
-    def compute_runtime_state(self, trans, step=None, step_updates=None):
+    def compute_runtime_state(self, trans, step=None, step_updates=None, replace_default_values=False):
         """Determine the runtime state (potentially different from self.state
         which describes configuration state). This (again unlike self.state) is
         currently always a `DefaultToolState` object.
 
-        If `step` is not `None`, it will be used to search for default values
-        defined in workflow input steps.
-
         If `step_updates` is `None`, this is likely for rendering the run form
         for instance and no runtime properties are available and state must be
         solely determined by the default runtime state described by the step.
@@ -426,6 +423,8 @@ def compute_runtime_state(self, trans, step=None, step_updates=None):
         supplied by the workflow runner.
         """
         state = self.get_runtime_state()
+        if replace_default_values and step:
+            state.inputs = step.state.inputs
         step_errors = {}
 
         if step is not None:
@@ -435,8 +434,11 @@ def update_value(input, context, prefixed_name, **kwargs):
                 if step_input is None:
                     return NO_REPLACEMENT
 
-                if step_input.default_value_set:
-                    return step_input.default_value
+                if replace_default_values and step_input.default_value_set:
+                    input_value = step_input.default_value
+                    if isinstance(input, BaseDataToolParameter):
+                        input_value = raw_to_galaxy(trans.app, trans.history, input_value)
+                    return input_value
 
                 return NO_REPLACEMENT
 
@@ -2227,13 +2229,14 @@ def get_runtime_state(self):
     def get_runtime_inputs(self, step, connections: Optional[Iterable[WorkflowStepConnection]] = None):
         return self.get_inputs()
 
-    def compute_runtime_state(self, trans, step=None, step_updates=None):
+    def compute_runtime_state(self, trans, step=None, step_updates=None, replace_default_values=False):
         # Warning: This method destructively modifies existing step state.
         if self.tool:
             step_errors = {}
             state = self.state
-            self.runtime_post_job_actions = {}
-            state, step_errors = super().compute_runtime_state(trans, step, step_updates)
+            state, step_errors = super().compute_runtime_state(
+                trans, step, step_updates, replace_default_values=replace_default_values
+            )
             if step_updates:
                 self.runtime_post_job_actions = step_updates.get(RUNTIME_POST_JOB_ACTIONS_KEY, {})
                 step_metadata_runtime_state = self.__step_meta_runtime_state()
@@ -2273,6 +2276,10 @@ def execute(
             # TODO: why do we even create an invocation, seems like something we could check on submit?
             message = f"Specified tool [{tool.id}] in step {step.order_index + 1} is not workflow-compatible."
             raise exceptions.MessageException(message)
+        self.state, _ = self.compute_runtime_state(
+            trans, step, step_updates=progress.param_map.get(step.id), replace_default_values=True
+        )
+        step.state = self.state
         tool_state = step.state
         assert tool_state is not None
         tool_inputs = tool.inputs.copy()

lib/galaxy/workflow/run.py

@@ -39,6 +39,7 @@
 )
 from galaxy.tools.parameters.basic import raw_to_galaxy
 from galaxy.tools.parameters.workflow_utils import (
+    is_runtime_value,
     NO_REPLACEMENT,
     NoReplacement,
 )
@@ -462,15 +463,18 @@ def replacement_for_input(self, trans, step: "WorkflowStep", input_dict: Dict[st
                     replacement = temp
             else:
                 replacement = self.replacement_for_connection(connection[0], is_data=is_data)
-        elif step.state and (state_input := get_path(step.state.inputs, nested_key_to_path(prefixed_name), None)):
+        elif (
+            step.state
+            and (state_input := get_path(step.state.inputs, nested_key_to_path(prefixed_name), None))
+            and not is_runtime_value(state_input)
+        ):
             # workflow submitted with step parameters populates state directly
             # via populate_module_and_state
             replacement = state_input
-        else:
-            for step_input in step.inputs:
-                if step_input.name == prefixed_name and step_input.default_value_set:
-                    if is_data:
-                        replacement = raw_to_galaxy(trans.app, trans.history, step_input.default_value)
+        elif (step_input := step.inputs_by_name.get(prefixed_name)) and step_input.default_value_set:
+            replacement = step_input.default_value
+            if is_data:
+                replacement = raw_to_galaxy(trans.app, trans.history, step_input.default_value)
         return replacement
 
     def replacement_for_connection(self, connection: "WorkflowStepConnection", is_data: bool = True):