Fix broken subworkflow reodering on workflow import

We started reordering workflow steps on import in
https://github.com/galaxyproject/galaxy/pull/13641.

If subworkflow steps are reordered before the input connections are
set the wrong subworkflow steps are referenced when we build the
workflow connection. This broke the IWC tests for gromacs.

We now do not reorder subworkflow steps until the parent workflow is
being reordered, which is after all connections are set up.

lib/galaxy/managers/workflows.py

@@ -572,6 +572,7 @@ class WorkflowContentsManager(UsesAnnotations):
         source=None,
         add_to_menu=False,
         hidden=False,
+        is_subworkflow=False,
     ):
         data = raw_workflow_description.as_dict
         # Put parameters in workflow mode
@@ -591,6 +592,7 @@ class WorkflowContentsManager(UsesAnnotations):
             raw_workflow_description,
             workflow_create_options,
             name=name,
+            is_subworkflow=is_subworkflow,
         )
         if "uuid" in data:
             workflow.uuid = data["uuid"]
@@ -680,7 +682,7 @@ class WorkflowContentsManager(UsesAnnotations):
         return workflow, errors
 
     def _workflow_from_raw_description(
-        self, trans, raw_workflow_description, workflow_state_resolution_options, name, **kwds
+        self, trans, raw_workflow_description, workflow_state_resolution_options, name, is_subworkflow=False, **kwds
     ):
         # don't commit the workflow or attach its part to the sa session - just build a
         # a transient model to operate on or render.
@@ -755,8 +757,12 @@ class WorkflowContentsManager(UsesAnnotations):
         # Second pass to deal with connections between steps
         self.__connect_workflow_steps(steps, steps_by_external_id, dry_run)
 
-        # Order the steps if possible
-        attach_ordered_steps(workflow, steps)
+        workflow.has_cycles = True
+        workflow.steps = steps
+        # we can't reorder subworkflows, as step connections would become invalid
+        if not is_subworkflow:
+            # Order the steps if possible
+            attach_ordered_steps(workflow)
 
         return workflow, missing_tool_tups
 
@@ -848,7 +854,7 @@ class WorkflowContentsManager(UsesAnnotations):
         """
         if len(workflow.steps) == 0:
             raise exceptions.MessageException("Workflow cannot be run because it does not have any steps.")
-        if attach_ordered_steps(workflow, workflow.steps):
+        if attach_ordered_steps(workflow):
             raise exceptions.MessageException("Workflow cannot be run because it contains cycles.")
         trans.workflow_building_mode = workflow_building_modes.USE_HISTORY
         module_injector = WorkflowModuleInjector(trans)
@@ -940,7 +946,7 @@ class WorkflowContentsManager(UsesAnnotations):
         """
         if len(workflow.steps) == 0:
             raise exceptions.MessageException("Workflow cannot be run because it does not have any steps.")
-        if attach_ordered_steps(workflow, workflow.steps):
+        if attach_ordered_steps(workflow):
             raise exceptions.MessageException("Workflow cannot be run because it contains cycles.")
 
         # Ensure that the user has a history
@@ -1637,6 +1643,7 @@ class WorkflowContentsManager(UsesAnnotations):
         steps.append(step)
         external_id = step_dict["id"]
         steps_by_external_id[external_id] = step
+        step.order_index = external_id
         if "workflow_outputs" in step_dict:
             workflow_outputs = step_dict["workflow_outputs"]
             found_output_names = set()
@@ -1703,7 +1710,11 @@ class WorkflowContentsManager(UsesAnnotations):
     def __build_embedded_subworkflow(self, trans, data, workflow_state_resolution_options):
         raw_workflow_description = self.ensure_raw_description(data)
         subworkflow = self.build_workflow_from_raw_description(
-            trans, raw_workflow_description, workflow_state_resolution_options, hidden=True
+            trans,
+            raw_workflow_description,
+            workflow_state_resolution_options,
+            hidden=True,
+            is_subworkflow=True,
         ).workflow
         return subworkflow
 

lib/galaxy/workflow/extract.py

@@ -52,8 +52,9 @@ def extract_workflow(
     # Workflow to populate
     workflow = model.Workflow()
     workflow.name = workflow_name
+    workflow.steps = steps
     # Order the steps if possible
-    attach_ordered_steps(workflow, steps)
+    attach_ordered_steps(workflow)
     # And let's try to set up some reasonable locations on the canvas
     # (these are pretty arbitrary values)
     levorder = order_workflow_steps_with_levels(steps)

lib/galaxy/workflow/steps.py

@@ -10,11 +10,11 @@ from galaxy.util.topsort import (
 )
 
 
-def attach_ordered_steps(workflow, steps):
+def attach_ordered_steps(workflow):
     """Attempt to topologically order steps and attach to workflow. If this
     fails - the workflow contains cycles so it mark it as such.
     """
-    ordered_steps = order_workflow_steps(steps)
+    ordered_steps = order_workflow_steps(workflow.steps)
     workflow.has_cycles = True
     if ordered_steps:
         workflow.has_cycles = False
@@ -30,6 +30,8 @@ def order_workflow_steps(steps):
     """
     position_data_available = bool(steps)
     for step in steps:
+        if step.subworkflow:
+            attach_ordered_steps(step.subworkflow)
         if not step.position or "left" not in step.position or "top" not in step.position:
             position_data_available = False
     if position_data_available:

lib/galaxy_test/api/test_workflows.py

@@ -5021,6 +5021,65 @@ steps:
         run_workflow_response = self.workflow_populator.invoke_workflow_raw(uploaded_workflow_id, workflow_request)
         return run_workflow_response, history_id
 
+    def test_subworkflow_import_order_maintained(self):
+        summary = self._run_workflow(
+            """
+class: GalaxyWorkflow
+inputs:
+  outer_input_1:
+    type: int
+    default: 1
+    position:
+      left: 0
+      top: 0
+  outer_input_2:
+    type: int
+    default: 2
+    position:
+      left: 100
+      top: 0
+steps:
+  nested_workflow:
+    in:
+      inner_input_1: outer_input_1
+      inner_input_2: outer_input_2
+    run:
+      class: GalaxyWorkflow
+      inputs:
+        inner_input_1:
+          type: int
+          position:
+            left: 100
+            top: 0
+        inner_input_2:
+          type: int
+          position:
+            left: 0
+            top: 0
+      steps: []
+      outputs:
+        - label: nested_out_1
+          outputSource: inner_input_1/output
+        - label: nested_out_2
+          outputSource: inner_input_2/output
+outputs:
+  - label: out_1
+    outputSource: nested_workflow/nested_out_1
+  - label: out_2
+    outputSource: nested_workflow/nested_out_2
+""",
+            assert_ok=False,
+            wait=False,
+        )
+        self.workflow_populator.wait_for_invocation(summary.workflow_id, summary.invocation_id)
+        self.workflow_populator.wait_for_history_workflows(
+            summary.history_id, assert_ok=False, expected_invocation_count=2
+        )
+        invocation = self.workflow_populator.get_invocation(summary.invocation_id)
+        output_values = invocation["output_values"]
+        assert output_values["out_1"] == 1
+        assert output_values["out_2"] == 2
+
     @skip_without_tool("random_lines1")
     def test_run_replace_params_by_steps(self):
         workflow_request, history_id, workflow_id, steps = self._setup_random_x2_workflow_steps(