diff --git a/lib/galaxy/config.py b/lib/galaxy/config.py
index a7187b5ebae..263e02f752f 100644
--- a/lib/galaxy/config.py
+++ b/lib/galaxy/config.py
@@ -382,6 +382,8 @@ class Configuration(object):
         # workflows built using these modules may not function in the
         # future.
         self.enable_beta_workflow_modules = string_as_bool(kwargs.get('enable_beta_workflow_modules', 'False'))
+        # Enable use of gxformat2 workflows.
+        self.enable_beta_workflow_format = string_as_bool(kwargs.get('enable_beta_workflow_format', 'False'))
         # These are not even beta - just experiments - don't use them unless
         # you want yours tools to be broken in the future.
         self.enable_beta_tool_formats = string_as_bool(kwargs.get('enable_beta_tool_formats', 'False'))
diff --git a/lib/galaxy/managers/workflows.py b/lib/galaxy/managers/workflows.py
index b2618e73684..319b01dc872 100644
--- a/lib/galaxy/managers/workflows.py
+++ b/lib/galaxy/managers/workflows.py
@@ -5,6 +5,12 @@ import logging
 import uuid
 from collections import namedtuple
 
+from gxformat2 import (
+    from_galaxy_native,
+    ImporterGalaxyInterface,
+    ImportOptions,
+    python_to_workflow,
+)
 from six import string_types
 from sqlalchemy import and_
 from sqlalchemy.orm import joinedload, subqueryload
@@ -235,6 +241,27 @@ class WorkflowContentsManager(UsesAnnotations):
         self.app = app
         self._resource_mapper_function = get_resource_mapper_function(app)
 
+    def normalize_workflow_format(self, as_dict):
+        """Process incoming workflow descriptions for consumption by other methods.
+
+        Currently this mostly means converting format 2 workflows into standard Galaxy
+        workflow JSON for consumption for the rest of this module. In the future we will
+        want to be a lot more percise about this - preserve the original description along
+        side the data model and apply updates in a way that largely preserves YAML structure
+        so workflows can be extracted.
+        """
+        workflow_class = as_dict.get("class", None)
+        if workflow_class == "GalaxyWorkflow" or "$graph" in as_dict or "yaml_content" in as_dict:
+            if not self.app.config.enable_beta_workflow_format:
+                raise exceptions.ConfigDoesNotAllowException("Format2 workflows not enabled.")
+
+            # Format 2 Galaxy workflow.
+            galaxy_interface = Format2ConverterGalaxyInterface()
+            import_options = ImportOptions()
+            import_options.deduplicate_subworkflows = True
+            as_dict = python_to_workflow(as_dict, galaxy_interface, workflow_directory=None, import_options=import_options)
+        return as_dict
+
     def build_workflow_from_dict(
         self,
         trans,
@@ -346,12 +373,21 @@ class WorkflowContentsManager(UsesAnnotations):
         # but do need to use to make connections
         steps_by_external_id = {}
 
+        # Preload dependent workflows with locally defined content_ids.
+        subworkflows = data.get("subworkflows")
+        subworkflow_id_map = None
+        if subworkflows:
+            subworkflow_id_map = {}
+            for key, subworkflow_dict in subworkflows.items():
+                subworkflow = self.__build_embedded_subworkflow(trans, subworkflow_dict, **kwds)
+                subworkflow_id_map[key] = subworkflow
+
         # Keep track of tools required by the workflow that are not available in
         # the local Galaxy instance.  Each tuple in the list of missing_tool_tups
         # will be ( tool_id, tool_name, tool_version ).
         missing_tool_tups = []
         for step_dict in self.__walk_step_dicts(data):
-            self.__load_subworkflows(trans, step_dict)
+            self.__load_subworkflows(trans, step_dict, subworkflow_id_map, **kwds)
 
         for step_dict in self.__walk_step_dicts(data):
             module, step = self.__module_from_dict(trans, steps, steps_by_external_id, step_dict, **kwds)
@@ -380,6 +416,12 @@ class WorkflowContentsManager(UsesAnnotations):
         option describes the workflow in a context more tied to the current Galaxy instance and includes
         fields like 'url' and 'url' and actual unencoded step ids instead of 'order_index'.
         """
+
+        def to_format_2(wf_dict, **kwds):
+            if not trans.app.config.enable_beta_workflow_format:
+                raise exceptions.ConfigDoesNotAllowException("Format2 workflows not enabled.")
+            return from_galaxy_native(wf_dict, None, **kwds)
+
         if version == '':
             version = None
         if version is not None:
@@ -393,6 +435,12 @@ class WorkflowContentsManager(UsesAnnotations):
             wf_dict = self._workflow_to_dict_instance(stored, workflow=workflow, legacy=False)
         elif style == "run":
             wf_dict = self._workflow_to_dict_run(trans, stored, workflow=workflow)
+        elif style == "format2":
+            wf_dict = self._workflow_to_dict_export(trans, stored, workflow=workflow)
+            wf_dict = to_format_2(wf_dict)
+        elif style == "format2_wrapped_yaml":
+            wf_dict = self._workflow_to_dict_export(trans, stored, workflow=workflow)
+            wf_dict = to_format_2(wf_dict, json_wrapper=True)
         else:
             wf_dict = self._workflow_to_dict_export(trans, stored, workflow=workflow)
         if version:
@@ -896,11 +944,11 @@ class WorkflowContentsManager(UsesAnnotations):
 
             yield step_dict
 
-    def __load_subworkflows(self, trans, step_dict):
+    def __load_subworkflows(self, trans, step_dict, subworkflow_id_map, **kwds):
         step_type = step_dict.get("type", None)
         if step_type == "subworkflow":
             subworkflow = self.__load_subworkflow_from_step_dict(
-                trans, step_dict
+                trans, step_dict, subworkflow_id_map, **kwds
             )
             step_dict["subworkflow"] = subworkflow
 
@@ -930,7 +978,8 @@ class WorkflowContentsManager(UsesAnnotations):
 
         # Create the model class for the step
         steps.append(step)
-        steps_by_external_id[step_dict['id']] = step
+        external_id = step_dict["id"]
+        steps_by_external_id[external_id] = step
         if 'workflow_outputs' in step_dict:
             workflow_outputs = step_dict['workflow_outputs']
             found_output_names = set([])
@@ -954,7 +1003,7 @@ class WorkflowContentsManager(UsesAnnotations):
                 trans.sa_session.add(m)
         return module, step
 
-    def __load_subworkflow_from_step_dict(self, trans, step_dict):
+    def __load_subworkflow_from_step_dict(self, trans, step_dict, subworkflow_id_map, **kwds):
         embedded_subworkflow = step_dict.get("subworkflow", None)
         subworkflow_id = step_dict.get("content_id", None)
         if embedded_subworkflow and subworkflow_id:
@@ -964,11 +1013,10 @@ class WorkflowContentsManager(UsesAnnotations):
             raise Exception("Subworkflow step must define either subworkflow or content_id.")
 
         if embedded_subworkflow:
-            subworkflow = self.build_workflow_from_dict(
-                trans,
-                embedded_subworkflow,
-                create_stored_workflow=False,
-            ).workflow
+            subworkflow = self.__build_embedded_subworkflow(trans, embedded_subworkflow, **kwds)
+        elif subworkflow_id_map is not None:
+            # Interpret content_id as a workflow local thing.
+            subworkflow = subworkflow_id_map[subworkflow_id[1:]]
         else:
             workflow_manager = WorkflowsManager(self.app)
             subworkflow = workflow_manager.get_owned_workflow(
@@ -977,6 +1025,12 @@ class WorkflowContentsManager(UsesAnnotations):
 
         return subworkflow
 
+    def __build_embedded_subworkflow(self, trans, data, **kwds):
+        subworkflow = self.build_workflow_from_dict(
+            trans, data, create_stored_workflow=False, fill_defaults=kwds.get("fill_defaults", False)
+        ).workflow
+        return subworkflow
+
     def __connect_workflow_steps(self, steps, steps_by_external_id):
         """ Second pass to deal with connections between steps.
 
@@ -999,7 +1053,10 @@ class WorkflowContentsManager(UsesAnnotations):
                     conn.input_step = step
                     conn.input_name = input_name
                     conn.output_name = conn_dict['output_name']
-                    conn.output_step = steps_by_external_id[conn_dict['id']]
+                    external_id = conn_dict['id']
+                    if external_id not in steps_by_external_id:
+                        raise KeyError("Failed to find external id %s in %s" % (external_id, steps_by_external_id.keys()))
+                    conn.output_step = steps_by_external_id[external_id]
 
                     input_subworkflow_step_index = conn_dict.get('input_subworkflow_step_id', None)
                     if input_subworkflow_step_index is not None:
@@ -1023,3 +1080,9 @@ class MissingToolsException(exceptions.MessageException):
     def __init__(self, workflow, errors):
         self.workflow = workflow
         self.errors = errors
+
+
+class Format2ConverterGalaxyInterface(ImporterGalaxyInterface):
+
+    def import_workflow(self, workflow, **kwds):
+        raise NotImplementedError("Direct format 2 import of nested workflows is not yet implemented, use bioblend client.")
diff --git a/lib/galaxy/webapps/galaxy/api/workflows.py b/lib/galaxy/webapps/galaxy/api/workflows.py
index ad6f4082052..17ccb03773d 100644
--- a/lib/galaxy/webapps/galaxy/api/workflows.py
+++ b/lib/galaxy/webapps/galaxy/api/workflows.py
@@ -492,6 +492,7 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesAnn
         stored_workflow = self.__get_stored_workflow(trans, id)
         workflow_dict = payload.get('workflow') or payload
         if workflow_dict:
+            workflow_dict = self.__normalize_workflow(workflow_dict)
             new_workflow_name = workflow_dict.get('name') or workflow_dict.get('name')
             if new_workflow_name and new_workflow_name != stored_workflow.name:
                 sanitized_name = sanitize_html(new_workflow_name)
@@ -572,6 +573,7 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesAnn
             raise exceptions.MessageException("The data content does not appear to be a valid workflow.")
         if not data:
             raise exceptions.MessageException("The data content is missing.")
+        data = self.__normalize_workflow(data)
         workflow, missing_tool_tups = self._workflow_from_dict(trans, data, source=source)
         workflow = workflow.latest_workflow
         if workflow.has_errors:
@@ -584,6 +586,7 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesAnn
 
     def __api_import_new_workflow(self, trans, payload, **kwd):
         data = payload['workflow']
+        data = self.__normalize_workflow(data)
         import_tools = util.string_as_bool(payload.get("import_tools", False))
         if import_tools and not trans.user_is_admin:
             raise exceptions.AdminRequiredException()
@@ -651,6 +654,9 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesAnn
             'fill_defaults': fill_defaults,
         }
 
+    def __normalize_workflow(self, as_dict):
+        return self.workflow_contents_manager.normalize_workflow_format(as_dict)
+
     @expose_api
     def import_shared_workflow_deprecated(self, trans, payload, **kwd):
         """
diff --git a/lib/galaxy/webapps/galaxy/config_schema.yml b/lib/galaxy/webapps/galaxy/config_schema.yml
index 80573180e12..9bbdadfa459 100644
--- a/lib/galaxy/webapps/galaxy/config_schema.yml
+++ b/lib/galaxy/webapps/galaxy/config_schema.yml
@@ -2107,6 +2107,13 @@ mapping:
           Enable beta workflow modules that should not yet be considered part of Galaxy's
           stable API.
 
+      enable_beta_workflow_format:
+        type: bool
+        default: false
+        required: false
+        desc: |
+          Enable import and export of workflows as Galaxy Format 2 workflows.
+
       force_beta_workflow_scheduled_min_steps:
         type: int
         default: 250
diff --git a/run.sh b/run.sh
index 62fc61c188e..26013382ec6 100755
--- a/run.sh
+++ b/run.sh
@@ -38,6 +38,7 @@ if [ ! -z "$GALAXY_RUN_WITH_TEST_TOOLS" ];
 then
     export GALAXY_CONFIG_OVERRIDE_TOOL_CONFIG_FILE="test/functional/tools/samples_tool_conf.xml"
     export GALAXY_CONFIG_ENABLE_BETA_WORKFLOW_MODULES="true"
+    export GALAXY_CONFIG_ENABLE_BETA_WORKFLOW_FORMAT="true"
     export GALAXY_CONFIG_OVERRIDE_ENABLE_BETA_TOOL_FORMATS="true"
     export GALAXY_CONFIG_OVERRIDE_WEBHOOKS_DIR="test/functional/webhooks"
 fi
diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py
index a89e21dc2f4..dd397c5fdc5 100644
--- a/test/api/test_workflows.py
+++ b/test/api/test_workflows.py
@@ -38,18 +38,20 @@ outputs:
   outer_output:
     outputSource: second_cat/out_file1
 steps:
-  - tool_id: cat1
-    label: first_cat
+  first_cat:
+    tool_id: cat1
     in:
       input1: outer_input
-  - run:
+  nested_workflow:
+    run:
       class: GalaxyWorkflow
       inputs:
         - id: inner_input
       outputs:
         - source: 1#out_file1
       steps:
-        - tool_id: random_lines1
+        random:
+          tool_id: random_lines1
           state:
             num_lines: 1
             input:
@@ -57,11 +59,10 @@ steps:
             seed_source:
               seed_source_selector: set_seed
               seed: asdf
-    label: nested_workflow
     in:
       inner_input: first_cat/out_file1
-  - tool_id: cat1
-    label: second_cat
+  second_cat:
+    tool_id: cat1
     state:
       input1:
         $link: nested_workflow#1:out_file1
@@ -500,26 +501,23 @@ steps:
         workflow_id = self._upload_yaml_workflow("""
 class: GalaxyWorkflow
 inputs:
-  - id: outer_input
+  outer_input: data
 steps:
-  - run:
+  inner_workflow:
+    run:
       class: GalaxyWorkflow
       inputs:
-        - id: inner_input
+        inner_input:
           type: collection
           collection_type: "list:paired"
       outputs:
-        - id: workflow_output
-          source: collection_type_source#list_output
+        workflow_output:
+          outputSource: collection_type_source/list_output
       steps:
-        - label: text_input1
-          type: input_collection
-          collection_type: "list:paired"
-        - tool_id: collection_type_source
-          label: collection_type_source
+        collection_type_source:
+          tool_id: collection_type_source
           in:
             input_collect: inner_input
-    label: inner_workflow
     in:
       inner_input: outer_input
 """)
@@ -586,7 +584,8 @@ steps:
             workflow_version_01 = self._upload_yaml_workflow("""
 class: GalaxyWorkflow
 steps:
-  - tool_id: multiple_versions
+  multiple:
+    tool_id: multiple_versions
     tool_version: "0.1"
     state:
       inttest: 0
@@ -598,7 +597,8 @@ steps:
             workflow_version_02 = self._upload_yaml_workflow("""
 class: GalaxyWorkflow
 steps:
-  - tool_id: multiple_versions
+  multiple:
+    tool_id: multiple_versions
     tool_version: "0.2"
     state:
       inttest: 1
@@ -634,11 +634,13 @@ steps:
         workflow_id = self._upload_yaml_workflow("""
 class: GalaxyWorkflow
 steps:
-  - tool_id: job_properties
+  job_props:
+    tool_id: job_properties
     state:
       thebool: true
       failbool: true
-  - tool_id: identifier_multiple_in_conditional
+  identifier:
+    tool_id: identifier_multiple_in_conditional
     state:
       outer_cond:
         cond_param_outer: true
@@ -670,14 +672,15 @@ steps:
         workflow_id = self._upload_yaml_workflow("""
 class: GalaxyWorkflow
 steps:
-  - tool_id: job_properties
+  job_props:
+    tool_id: job_properties
     state:
       thebool: true
       failbool: true
-  - tool_id: identifier_collection
-    state:
-      input1:
-        $link: 0#list_output
+  identifier:
+    tool_id: identifier_collection
+    in:
+      input1: job_props/list_output
 """)
         with self.dataset_populator.test_history() as history_id:
             invocation_id = self.__invoke_workflow(history_id, workflow_id)
@@ -704,16 +707,16 @@ steps:
             job_summary = self._run_jobs("""
 class: GalaxyWorkflow
 inputs:
-  - id: input_datasets
-    type: collection
+  input_datasets: collection
 steps:
-  - label: fail_identifier_1
+  fail_identifier_1:
     tool_id: fail_identifier
     state:
       failbool: true
     in:
       input1: input_datasets
-  - tool_id: identifier_collection
+  identifier:
+    tool_id: identifier_collection
     in:
       input1: fail_identifier_1/out_file1
 test_data:
@@ -784,27 +787,20 @@ test_data:
         with self.dataset_populator.test_history() as history_id:
             workflow_id = self._upload_yaml_workflow("""
 class: GalaxyWorkflow
+inputs:
+  test_input_1: data
+  test_input_2: data
+  test_input_3: data
 steps:
-  - label: test_input_1
-    type: input
-  - label: test_input_2
-    type: input
-  - label: test_input_3
-    type: input
-  - label: split_up
+  split_up:
     tool_id: collection_split_on_column
-    state:
-      input1:
-        $link: test_input_2
-  - label: min_repeat
+    in:
+      input1: test_input_2
+  min_repeat:
     tool_id: min_repeat
-    state:
-      queries:
-        - input:
-            $link: test_input_1
-      queries2:
-        - input2:
-            $link: split_up#split_output
+    in:
+      queries_0|input: test_input_1
+      queries2_0|input2: split_up/split_output
 """)
             hda1 = self.dataset_populator.new_dataset(history_id, content="samp1\t10.0\nsamp2\t20.0\n")
             hda2 = self.dataset_populator.new_dataset(history_id, content="samp1\t20.0\nsamp2\t40.0\n")
@@ -829,33 +825,27 @@ steps:
         with self.dataset_populator.test_history() as history_id:
             workflow_id = self._upload_yaml_workflow("""
 class: GalaxyWorkflow
+inputs:
+  text_input1: data
+  text_input2: data
 steps:
-  - label: text_input1
-    type: input
-  - label: text_input2
-    type: input
-  - label: cat_inputs
+  cat_inputs:
     tool_id: cat1
-    state:
-      input1:
-        $link: text_input1
-      queries:
-        - input2:
-            $link: text_input2
-  - label: split_up_1
+    in:
+      input1: text_input1
+      queries_0|input2: text_input2
+  split_up_1:
     tool_id: collection_split_on_column
-    state:
-      input1:
-        $link: cat_inputs#out_file1
-  - label: split_up_2
+    in:
+      input1: cat_inputs/out_file1
+  split_up_2:
     tool_id: collection_split_on_column
-    state:
-      input1:
-        $link: split_up_1#split_output
-  - tool_id: cat
-    state:
-      input1:
-        $link: split_up_2#split_output
+    in:
+      input1: split_up_1/split_output
+  cat_output:
+    tool_id: cat
+    in:
+      input1: split_up_2/split_output
 """)
             hda1 = self.dataset_populator.new_dataset(history_id, content="samp1\t10.0\nsamp2\t20.0\n")
             hda2 = self.dataset_populator.new_dataset(history_id, content="samp1\t30.0\nsamp2\t40.0\n")
@@ -875,25 +865,20 @@ steps:
         with self.dataset_populator.test_history() as history_id:
             self._run_jobs("""
 class: GalaxyWorkflow
+inputs:
+  input_fastqs: collection
+  reference: data
 steps:
-  - label: input_fastqs
-    type: input_collection
-  - label: reference
-    type: input
-  - label: map_over_mapper
+  map_over_mapper:
     tool_id: mapper
-    state:
-      input1:
-        $link: input_fastqs
-      reference:
-        $link: reference
-  - label: pileup
+    in:
+      input1: input_fastqs
+      reference: reference
+  pileup:
     tool_id: pileup
-    state:
-      input1:
-        $link: map_over_mapper#out_file1
-      reference:
-        $link: reference
+    in:
+      input1: map_over_mapper/out_file1
+      reference: reference
 test_data:
   input_fastqs:
     type: list
@@ -995,14 +980,12 @@ test_data:
 
     def test_run_subworkflow_auto_labels(self):
             history_id = self.dataset_populator.new_history()
-            workflow_run_description = """%s
-
-test_data:
-  outer_input:
-    value: 1.bed
-    type: File
-""" % NESTED_WORKFLOW_AUTO_LABELS
-            job_summary = self._run_jobs(workflow_run_description, history_id=history_id)
+            test_data = """
+outer_input:
+  value: 1.bed
+  type: File
+"""
+            job_summary = self._run_jobs(NESTED_WORKFLOW_AUTO_LABELS, test_data=test_data, history_id=history_id)
             assert len(job_summary.jobs) == 4, "4 jobs expected, got %d jobs" % len(job_summary.jobs)
 
             content = self.dataset_populator.get_history_dataset_content(history_id)
@@ -1020,23 +1003,19 @@ inputs:
   test_input_1: data
   test_input_2: data
 steps:
-  - label: first_cat
+  first_cat:
     tool_id: cat1
-    state:
-      input1:
-        $link: test_input_1
-  - label: zip_it
+    in:
+      input1: test_input_1
+  zip_it:
     tool_id: "__ZIP_COLLECTION__"
-    state:
-      input_forward:
-        $link: first_cat#out_file1
-      input_reverse:
-        $link: test_input_2
-  - label: concat_pair
+    in:
+      input_forward: first_cat/out_file1
+      input_reverse: test_input_2
+  concat_pair:
     tool_id: collection_paired_test
-    state:
-      f1:
-        $link: zip_it#output
+    in:
+      f1: zip_it/output
 """)
             hda1 = self.dataset_populator.new_dataset(history_id, content="samp1\t10.0\nsamp2\t20.0\n")
             hda2 = self.dataset_populator.new_dataset(history_id, content="samp1\t20.0\nsamp2\t40.0\n")
@@ -1056,18 +1035,18 @@ steps:
             self._run_jobs("""
 class: GalaxyWorkflow
 steps:
-  - tool_id: 'collection_creates_dynamic_nested'
-    label: 'nested'
+  nested:
+    tool_id: collection_creates_dynamic_nested
     state:
       sleep_time: 0
       foo: 'dummy'
-  - tool_id: '__FLATTEN__'
+  flatten:
+    tool_id: '__FLATTEN__'
     state:
       input:
         $link: nested#list_output
       join_identifier: '-'
-test_data: {}
-""", history_id=history_id)
+""", test_data={}, history_id=history_id)
             details = self.dataset_populator.get_history_collection_details(history_id, hid=14)
             assert details['collection_type'] == "list"
             elements = details["elements"]
@@ -1078,7 +1057,7 @@ test_data: {}
     @skip_without_tool("__APPLY_RULES__")
     def test_workflow_run_apply_rules(self):
         with self.dataset_populator.test_history() as history_id:
-            self._run_jobs(WORKFLOW_WITH_RULES_1, history_id=history_id, wait=True, assert_ok=True)
+            self._run_jobs(WORKFLOW_WITH_RULES_1, history_id=history_id, wait=True, assert_ok=True, round_trip_format_conversion=True)
             output_content = self.dataset_populator.get_history_collection_details(history_id, hid=6)
             rules_test_data.check_example_2(output_content, self.dataset_populator)
 
@@ -1090,30 +1069,31 @@ inputs:
   input_c: collection
 
 steps:
-  - label: mixed_collection
+  mixed_collection:
     tool_id: exit_code_from_file
     state:
        input:
          $link: input_c
 
-  - label: filtered_collection
+  filtered_collection:
     tool_id: "__FILTER_FAILED_DATASETS__"
     state:
       input:
         $link: mixed_collection#out_file1
 
-  - tool_id: cat1
+  cat:
+    tool_id: cat1
     state:
       input1:
         $link: filtered_collection
-test_data:
-  input_c:
-    type: list
-    elements:
-      - identifier: i1
-        content: "0"
-      - identifier: i2
-        content: "1"
+""", test_data="""
+input_c:
+  type: list
+  elements:
+    - identifier: i1
+      content: "0"
+    - identifier: i2
+      content: "1"
 """, history_id=history_id, wait=True, assert_ok=False)
             jobs = summary.jobs
 
@@ -1148,8 +1128,8 @@ outputs:
   wf_output_1:
     outputSource: first_cat/out_file1
 steps:
-  - tool_id: cat1
-    label: first_cat
+  first_cat:
+    tool_id: cat1
     in:
       input1: input1
 """, test_data={"input1": "hello world"}, history_id=history_id)
@@ -1177,8 +1157,8 @@ outputs:
   wf_output_1:
     outputSource: first_cat/out_file1
 steps:
-  - tool_id: cat
-    label: first_cat
+  first_cat:
+    tool_id: cat
     in:
       input1: input1
 """, test_data="""
@@ -1189,7 +1169,7 @@ input1:
     - identifier: el1
       value: 1.fastq
       type: File
-""", history_id=history_id)
+""", history_id=history_id, round_trip_format_conversion=True)
             workflow_id = summary.workflow_id
             invocation_id = summary.invocation_id
             invocation_response = self._get("workflows/%s/invocations/%s" % (workflow_id, invocation_id))
@@ -1217,8 +1197,8 @@ outputs:
   wf_output_1:
     outputSource: first_cat/out_file1
 steps:
-  - tool_id: cat
-    label: first_cat
+  first_cat:
+    tool_id: cat
     in:
       input1: input1
 """, test_data="""
@@ -1259,7 +1239,7 @@ outputs:
   wf_output_1:
     outputSource: split_up/paired_output
 steps:
-  - label: split_up
+  split_up:
     tool_id: collection_creates_pair
     in:
       input1: text_input
@@ -1341,11 +1321,12 @@ outputs:
   outer_output:
     outputSource: second_cat/out_file1
 steps:
-  - tool_id: cat1
-    label: first_cat
+  first_cat:
+    tool_id: cat1
     in:
       input1: outer_input
-  - run:
+  nested_workflow:
+    run:
       class: GalaxyWorkflow
       inputs:
         inner_input: data
@@ -1353,8 +1334,8 @@ steps:
         workflow_output:
           outputSource: random_lines/out_file1
       steps:
-        - tool_id: random_lines1
-          label: random_lines
+        random_lines:
+          tool_id: random_lines1
           state:
             num_lines: 2
             input:
@@ -1362,15 +1343,14 @@ steps:
             seed_source:
               seed_source_selector: set_seed
               seed: asdf
-    label: nested_workflow
     in:
       inner_input: first_cat/out_file1
-  - tool_id: split
-    label: split
+  split:
+    tool_id: split
     in:
       input1: nested_workflow/workflow_output
-  - tool_id: cat_list
-    label: second_cat
+  second_cat:
+    tool_id: cat_list
     in:
       input1: split/output
 
@@ -1378,7 +1358,7 @@ test_data:
   outer_input:
     value: 1.bed
     type: File
-""", history_id=history_id, wait=True)
+""", history_id=history_id, wait=True, round_trip_format_conversion=True)
             self.assertEqual("chr6\t108722976\t108723115\tCCDS5067.1_cds_0_0_chr6_108722977_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n", self.dataset_populator.get_history_dataset_content(history_id))
             # self.assertEqual("chr16\t142908\t143003\tCCDS10397.1_cds_0_0_chr16_142909_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n", self.dataset_populator.get_history_dataset_content(history_id))
 
@@ -1399,11 +1379,12 @@ outputs:
   outer_output:
     outputSource: second_cat/out_file1
 steps:
-  - tool_id: cat1
-    label: first_cat
+  first_cat:
+    tool_id: cat1
     in:
       input1: outer_input
-  - run:
+  nested_workflow:
+    run:
       class: GalaxyWorkflow
       inputs:
         inner_input: data
@@ -1411,8 +1392,8 @@ steps:
         workflow_output:
           outputSource: inner_cat/out_file1
       steps:
-        - tool_id: random_lines1
-          label: random_lines
+        random_lines:
+          tool_id: random_lines1
           state:
             num_lines: 2
             input:
@@ -1420,27 +1401,25 @@ steps:
             seed_source:
               seed_source_selector: set_seed
               seed: asdf
-        - tool_id: split
-          label: split
+        split:
+          tool_id: split
           in:
             input1: random_lines/out_file1
-        - tool_id: cat1
-          label: inner_cat
+        inner_cat:
+          tool_id: cat1
           in:
             input1: split/output
-
-    label: nested_workflow
     in:
       inner_input: first_cat/out_file1
-  - tool_id: cat_list
-    label: second_cat
+  second_cat:
+    tool_id: cat_list
     in:
       input1: nested_workflow/workflow_output
 """, test_data="""
 outer_input:
   value: 1.bed
   type: File
-""", history_id=history_id, wait=True)
+""", history_id=history_id, wait=True, round_trip_format_conversion=True)
             self.assertEqual("chr6\t108722976\t108723115\tCCDS5067.1_cds_0_0_chr6_108722977_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n", self.dataset_populator.get_history_dataset_content(history_id))
 
     @skip_without_tool("cat_list")
@@ -1456,11 +1435,12 @@ outputs:
   outer_output:
     outputSource: second_cat/out_file1
 steps:
-  - tool_id: cat1
-    label: first_cat
+  first_cat:
+    tool_id: cat1
     in:
       input1: outer_input
-  - run:
+  nested_workflow:
+    run:
       class: GalaxyWorkflow
       inputs:
         inner_input: data
@@ -1468,8 +1448,8 @@ steps:
         workflow_output:
           outputSource: split/output
       steps:
-        - tool_id: random_lines1
-          label: random_lines
+        random_lines:
+          tool_id: random_lines1
           state:
             num_lines: 2
             input:
@@ -1477,22 +1457,21 @@ steps:
             seed_source:
               seed_source_selector: set_seed
               seed: asdf
-        - tool_id: split
-          label: split
+        split:
+          tool_id: split
           in:
             input1: random_lines/out_file1
-    label: nested_workflow
     in:
       inner_input: first_cat/out_file1
-  - tool_id: cat_list
-    label: second_cat
+  second_cat:
+    tool_id: cat_list
     in:
       input1: nested_workflow/workflow_output
 """, test_data="""
 outer_input:
   value: 1.bed
   type: File
-""", history_id=history_id, wait=True)
+""", history_id=history_id, wait=True, round_trip_format_conversion=True)
             self.assertEqual("chr6\t108722976\t108723115\tCCDS5067.1_cds_0_0_chr6_108722977_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n", self.dataset_populator.get_history_dataset_content(history_id))
 
     @skip_without_tool("empty_list")
@@ -1508,12 +1487,12 @@ outputs:
   count_list:
     outputSource: count_list/out_file1
 steps:
-  - tool_id: empty_list
-    label: empty_list
+  empty_list:
+    tool_id: empty_list
     in:
       input1: input1
-  - tool_id: random_lines1
-    label: random_lines
+  random_lines:
+    tool_id: random_lines1
     state:
       num_lines: 2
       input:
@@ -1521,8 +1500,8 @@ steps:
       seed_source:
         seed_source_selector: set_seed
         seed: asdf
-  - tool_id: count_list
-    label: count_list
+  count_list:
+    tool_id: count_list
     in:
       input1: random_lines/out_file1
 """, test_data="""
@@ -1540,7 +1519,8 @@ class: GalaxyWorkflow
 inputs:
   text_input1: collection
 steps:
-  - tool_id: collection_type_source_map_over
+  map_over:
+    tool_id: collection_type_source_map_over
     in:
       input_collect: text_input1
 """, test_data="""
@@ -1564,12 +1544,12 @@ outputs:
   count_multi_file:
     outputSource: count_multi_file/out_file1
 steps:
-  - tool_id: empty_list
-    label: empty_list
+  empty_list:
+    tool_id: empty_list
     in:
       input1: input1
-  - tool_id: random_lines1
-    label: random_lines
+  random_lines:
+    tool_id: random_lines1
     state:
       num_lines: 2
       input:
@@ -1577,15 +1557,15 @@ steps:
       seed_source:
         seed_source_selector: set_seed
         seed: asdf
-  - tool_id: count_multi_file
-    label: count_multi_file
+  count_multi_file:
+    tool_id: count_multi_file
     in:
       input1: random_lines/out_file1
 """, test_data="""
 input1:
   value: 1.bed
   type: File
-""", history_id=history_id, wait=True)
+""", history_id=history_id, wait=True, round_trip_format_conversion=True)
             self.assertEqual("0\n", self.dataset_populator.get_history_dataset_content(history_id))
 
     @skip_without_tool("cat")
@@ -1738,12 +1718,12 @@ inputs:
     type: collection
     collection_type: list
 steps:
-  - tool_id: cat
-    label: first_cat
+  first_cat:
+    tool_id: cat
     in:
       input1: input1
-  - tool_id: cat
-    label: second_cat
+  second_cat:
+    tool_id: cat
     in:
       input1: first_cat/out_file1
 """)
@@ -1777,30 +1757,30 @@ class: GalaxyWorkflow
 inputs:
   test_input: data
 steps:
-- label: first_cat
-  tool_id: cat1
-  in:
-    input1: test_input
-- label: the_pause
-  type: pause
-  in:
-    input: first_cat/out_file1
-- label: second_cat
-  tool_id: cat1
-  in:
-    input1: the_pause
-- label: third_cat
-  tool_id: random_lines1
-  connect:
-    $step: second_cat
-  state:
-    num_lines: 1
-    input:
-      $link: test_input
-    seed_source:
-      seed_source_selector: set_seed
-      seed: asdf
-""", test_data={"test_input": "hello world"}, history_id=history_id, wait=False)
+  first_cat:
+    tool_id: cat1
+    in:
+      input1: test_input
+  the_pause:
+    type: pause
+    in:
+      input: first_cat/out_file1
+  second_cat:
+    tool_id: cat1
+    in:
+      input1: the_pause
+  third_cat:
+    tool_id: random_lines1
+    in:
+      $step: second_cat
+    state:
+      num_lines: 1
+      input:
+        $link: test_input
+      seed_source:
+        seed_source_selector: set_seed
+        seed: asdf
+""", test_data={"test_input": "hello world"}, history_id=history_id, wait=False, round_trip_format_conversion=True)
             history_id = run_summary.history_id
             workflow_id = run_summary.workflow_id
             invocation_id = run_summary.invocation_id
@@ -1824,18 +1804,18 @@ class: GalaxyWorkflow
 inputs:
   text_input: text
 steps:
-- tool_id: validation_repeat
-  state:
-    r2:
-     - text:
-        $link: text_input
+  validation:
+    tool_id: validation_repeat
+    state:
+      r2:
+      - text:
+          $link: text_input
 """, test_data="""
 text_input:
   value: "abd"
   type: raw
-""", history_id=history_id, wait=True)
-            time.sleep(10)
-            self.workflow_populator.wait_for_invocation(run_summary.workflow_id, run_summary.invocation_id)
+""", history_id=history_id, wait=True, round_trip_format_conversion=True)
+            self.wait_for_invocation_and_jobs(history_id, run_summary.workflow_id, run_summary.invocation_id)
             jobs = self._history_jobs(history_id)
             assert len(jobs) == 1
 
@@ -1846,11 +1826,12 @@ class: GalaxyWorkflow
 inputs:
   text_input: text
 steps:
-- tool_id: validation_repeat
-  state:
-    r2:
-     - text:
-        $link: text_input
+  validation:
+    tool_id: validation_repeat
+    state:
+      r2:
+      - text:
+          $link: text_input
 """, test_data="""
 text_input:
   value: ""
@@ -1865,16 +1846,16 @@ inputs:
   data_input: data
   text_input: text
 steps:
-- label: randomlines
-  tool_id: random_lines1
-  state:
-    num_lines: 1
-    input:
-      $link: data_input
-    seed_source:
-      seed_source_selector: set_seed
-      seed:
-        $link: text_input
+  randomlines:
+    tool_id: random_lines1
+    state:
+      num_lines: 1
+      input:
+        $link: data_input
+      seed_source:
+        seed_source_selector: set_seed
+        seed:
+          $link: text_input
 """, test_data="""
 data_input:
   value: 1.bed
@@ -2048,8 +2029,8 @@ outputs:
   wf_output_1:
     outputSource: output_filter/out_1
 steps:
-  - tool_id: output_filter
-    label: output_filter
+  output_filter:
+    tool_id: output_filter
     state:
       produce_out_1: False
       filter_text_1: '1'
@@ -2070,8 +2051,8 @@ inputs:
     type: collection
     collection_type: list
 steps:
-  - tool_id: cat
-    label: first_cat
+  first_cat:
+    tool_id: cat
     in:
       input1: input1
     outputs:
@@ -2105,8 +2086,8 @@ inputs:
     type: collection
     collection_type: list
 steps:
-  - tool_id: cat
-    label: first_cat
+  first_cat:
+    tool_id: cat
     in:
       input1: input1
     outputs:
@@ -2158,7 +2139,8 @@ input1:
 class: GalaxyWorkflow
 inputs: []
 steps:
-  - tool_id: create_2
+  create_2:
+    tool_id: create_2
     state:
       sleep_time: 0
     outputs:
@@ -2190,8 +2172,8 @@ class: GalaxyWorkflow
 inputs:
   input1: data
 steps:
-  - tool_id: fail_identifier
-    label: first_fail
+  first_fail:
+    tool_id: fail_identifier
     state:
       failbool: true
       input1:
@@ -2199,7 +2181,8 @@ steps:
     outputs:
       out_file1:
         rename: "cat1 out"
-  - tool_id: cat
+  cat:
+    tool_id: cat
     in:
       input1: first_fail/out_file1
     outputs:
@@ -2237,8 +2220,8 @@ class: GalaxyWorkflow
 inputs:
   input1: data
 steps:
-  - tool_id: cat
-    label: first_cat
+  first_cat:
+    tool_id: cat
     in:
       input1: input1
     outputs:
@@ -2263,8 +2246,8 @@ inputs:
   input1: data
   input2: data
 steps:
-  - tool_id: cat
-    label: first_cat
+  first_cat:
+    tool_id: cat
     state:
       input1:
         $link: input1
@@ -2297,7 +2280,8 @@ inputs:
   fasta_input: data
   fastq_input: data
 steps:
-  - tool_id: mapper2
+  mapping:
+    tool_id: mapper2
     state:
       fastq_input:
         fastq_input_selector: single
@@ -2335,7 +2319,8 @@ inputs:
   fasta_input: data
   fastq_inputs: data
 steps:
-  - tool_id: mapper2
+  mapping:
+    tool_id: mapper2
     state:
       fastq_input:
         fastq_input_selector: paired_collection
@@ -2377,7 +2362,8 @@ class: GalaxyWorkflow
 inputs:
   input1: data
 steps:
-  - tool_id: collection_creates_pair
+  create_pair:
+    tool_id: collection_creates_pair
     state:
       input1:
         $link: input1
@@ -2405,8 +2391,8 @@ inputs:
     type: data_collection_input
     collection_type: list
 steps:
-  - tool_id: cat
-    label: first_cat
+  first_cat:
+    tool_id: cat
     in:
       input1: input1
     outputs:
@@ -2438,7 +2424,7 @@ class: GalaxyWorkflow
 inputs:
   input1: data
 steps:
-  - label: first_cat
+  first_cat:
     tool_id: cat
     in:
       input1: input1
@@ -2449,7 +2435,7 @@ steps:
             - "group:condition:treated"
             - "group:type:single-read"
             - "machine:illumina"
-  - label: second_cat
+  second_cat:
     tool_id: cat
     in:
       input1: first_cat/out_file1
@@ -2458,7 +2444,7 @@ input1:
   value: 1.fasta
   type: File
   name: fasta1
-""", history_id=history_id)
+""", history_id=history_id, round_trip_format_conversion=True)
 
             details0 = self.dataset_populator.get_history_dataset_details(history_id, hid=2, wait=True, assert_ok=True)
             tags = details0["tags"]
@@ -2481,7 +2467,8 @@ class: GalaxyWorkflow
 inputs:
   input1: data
 steps:
-  - tool_id: collection_creates_pair
+  create_pair:
+    tool_id: collection_creates_pair
     in:
       input1: input1
     outputs:
@@ -2493,7 +2480,7 @@ input1:
   value: 1.fasta
   type: File
   name: fasta1
-""", history_id=history_id)
+""", history_id=history_id, round_trip_format_conversion=True)
             details1 = self.dataset_populator.get_history_collection_details(history_id, hid=4, wait=True, assert_ok=True)
 
             assert details1["history_content_type"] == "dataset_collection"
@@ -2509,8 +2496,8 @@ inputs:
     type: collection
     collection_type: list
 steps:
-  - tool_id: cat
-    label: first_cat
+  first_cat:
+    tool_id: cat
     in:
       input1: input1
     outputs:
@@ -2525,7 +2512,7 @@ input1:
     - identifier: el1
       value: 1.fastq
       type: File
-""", history_id=history_id)
+""", history_id=history_id, round_trip_format_conversion=True)
             details1 = self.dataset_populator.get_history_collection_details(history_id, hid=3, wait=True, assert_ok=True)
 
             assert details1["history_content_type"] == "dataset_collection"
@@ -2540,15 +2527,16 @@ class: GalaxyWorkflow
 inputs:
   input1: data
 steps:
-  - tool_id: cat
-    label: first_cat
+  first_cat:
+    tool_id: cat
     in:
       input1: input1
     outputs:
       out_file1:
         add_tags:
           - "name:foo"
-  - tool_id: collection_creates_pair
+  create_pair:
+    tool_id: collection_creates_pair
     in:
       input1: first_cat#out_file1
     outputs:
@@ -2560,7 +2548,7 @@ input1:
   value: 1.fasta
   type: File
   name: fasta1
-""", history_id=history_id)
+""", history_id=history_id, round_trip_format_conversion=True)
             details_dataset_with_tag = self.dataset_populator.get_history_dataset_details(history_id, hid=2, wait=True, assert_ok=True)
 
             assert details_dataset_with_tag["history_content_type"] == "dataset", details_dataset_with_tag
@@ -2610,19 +2598,19 @@ class: GalaxyWorkflow
 inputs:
   test_input: data
 steps:
-  - label: first_cat
+  first_cat:
     tool_id: cat1
     in:
       input1: test_input
-  - label: the_pause
+  the_pause:
     type: pause
     in:
       input: first_cat/out_file1
-  - label: second_cat
+  second_cat:
     tool_id: cat1
     in:
       input1: the_pause
-""")
+""", round_trip_format_conversion=True)
         downloaded_workflow = self._download_workflow(workflow_id)
         uuid_dict = dict((int(index), step["uuid"]) for index, step in downloaded_workflow["steps"].items())
         with self.dataset_populator.test_history() as history_id:
@@ -2666,16 +2654,16 @@ outputs:
   wf_output_1:
     outputSource: third_cat/out_file1
 steps:
-  - tool_id: cat1
-    label: first_cat
+  first_cat:
+    tool_id: cat1
     in:
       input1: input1
-  - tool_id: cat1
-    label: second_cat
+  second_cat:
+    tool_id: cat1
     in:
       input1: first_cat/out_file1
-  - tool_id: cat1
-    label: third_cat
+  third_cat:
+    tool_id: cat1
     in:
       input1: second_cat/out_file1
     outputs:
@@ -2772,10 +2760,11 @@ steps:
             workflow_id = self._upload_yaml_workflow("""
 class: GalaxyWorkflow
 steps:
- - tool_id: validation_repeat
-   state:
-     r2:
-      - text: "abd"
+  validation:
+    tool_id: validation_repeat
+    state:
+      r2:
+        - text: "abd"
 """)
             workflow_request = dict(
                 history="hist_id=%s" % history_id,
@@ -2799,10 +2788,11 @@ steps:
             self._run_jobs("""
 class: GalaxyWorkflow
 steps:
- - tool_id: validation_repeat
-   state:
-     r2:
-     - text: ""
+  validation:
+    tool_id: validation_repeat
+    state:
+      r2:
+      - text: ""
 """, history_id=history_id, wait=False, expected_response=400)
 
     def _run_validation_workflow_with_substitions(self, substitions):
@@ -2965,7 +2955,7 @@ class: GalaxyWorkflow
 inputs:
   input_c: collection
 steps:
-  - label: cat1
+  cat1:
     tool_id: cat1
     in:
        input1: input_c
diff --git a/test/api/test_workflows_from_yaml.py b/test/api/test_workflows_from_yaml.py
index 1b2d1e791b9..9a10f32bea2 100644
--- a/test/api/test_workflows_from_yaml.py
+++ b/test/api/test_workflows_from_yaml.py
@@ -22,13 +22,17 @@ class WorkflowsFromYamlApiTestCase(BaseWorkflowsApiTestCase):
     def setUp(self):
         super(WorkflowsFromYamlApiTestCase, self).setUp()
 
-    def _upload_and_download(self, yaml_workflow):
-        workflow_id = self._upload_yaml_workflow(yaml_workflow)
-        workflow = self._get("workflows/%s/download" % workflow_id).json()
-        return workflow
+    def _upload_and_download(self, yaml_workflow, **kwds):
+        style = None
+        if "style" in kwds:
+            style = kwds.pop("style")
+        workflow_id = self._upload_yaml_workflow(yaml_workflow, **kwds)
+        return self.workflow_populator.download_workflow(workflow_id, style=style)
 
     def test_simple_upload(self):
-        workflow = self._upload_and_download(WORKFLOW_SIMPLE_CAT_AND_RANDOM_LINES)
+        workflow = self._upload_and_download(WORKFLOW_SIMPLE_CAT_AND_RANDOM_LINES, client_convert=False)
+
+        assert workflow["annotation"].startswith("Simple workflow that ")
 
         tool_count = {'random_lines1': 0, 'cat1': 0}
         input_found = False
@@ -47,6 +51,10 @@ class WorkflowsFromYamlApiTestCase(BaseWorkflowsApiTestCase):
         assert tool_count['random_lines1'] == 1
         assert tool_count['cat1'] == 2
 
+        workflow_as_format2 = self._upload_and_download(WORKFLOW_SIMPLE_CAT_AND_RANDOM_LINES, client_convert=False, style="format2")
+        assert workflow_as_format2["doc"].startswith("Simple workflow that")
+
+
 # FIXME:  This test fails on some machines due to (we're guessing) yaml.safe_loading
 # order being not guaranteed and inconsistent across platforms.  The workflow
 # yaml.safe_loader probably needs to enforce order using something like the
@@ -86,15 +94,16 @@ input1: "hello world"
 
     def test_inputs_to_steps(self):
         history_id = self.dataset_populator.new_history()
-        self._run_jobs(WORKFLOW_SIMPLE_CAT_TWICE, test_data={"input1": "hello world"}, history_id=history_id)
+        self._run_jobs(WORKFLOW_SIMPLE_CAT_TWICE, test_data={"input1": "hello world"}, history_id=history_id, round_trip_format_conversion=True)
         contents1 = self.dataset_populator.get_history_dataset_content(history_id)
         self.assertEqual(contents1.strip(), "hello world\nhello world")
 
     def test_outputs(self):
-        workflow_id = self._upload_yaml_workflow(WORKFLOW_WITH_OUTPUTS)
+        workflow_id = self._upload_yaml_workflow(WORKFLOW_WITH_OUTPUTS, round_trip_format_conversion=True)
         workflow = self._get("workflows/%s/download" % workflow_id).json()
         self.assertEqual(workflow["steps"]["1"]["workflow_outputs"][0]["output_name"], "out_file1")
         self.assertEqual(workflow["steps"]["1"]["workflow_outputs"][0]["label"], "wf_output_1")
+        workflow = self.workflow_populator.download_workflow(workflow_id, style="format2")
 
     def test_runtime_inputs(self):
         workflow = self._upload_and_download(WORKFLOW_RUNTIME_PARAMETER_SIMPLE)
@@ -116,11 +125,12 @@ class: GalaxyWorkflow
 inputs:
   outer_input: data
 steps:
-  - tool_id: cat1
-    label: first_cat
+  first_cat:
+    tool_id: cat1
     in:
       input1: outer_input
-  - run:
+  nested_workflow:
+    run:
       class: GalaxyWorkflow
       inputs:
         inner_input: data
@@ -133,16 +143,10 @@ steps:
             seed_source:
               seed_source_selector: set_seed
               seed: asdf
-    label: nested_workflow
     in:
       inner_input: first_cat/out_file1
-
-test_data:
-  outer_input:
-    value: 1.bed
-    type: File
-""")
-        workflow = self._get("workflows/%s/download" % workflow_id).json()
+""", client_convert=False)
+        workflow = self.workflow_populator.download_workflow(workflow_id)
         by_label = self._steps_by_label(workflow)
         if "nested_workflow" not in by_label:
             template = "Workflow [%s] does not contain label 'nested_workflow'."
@@ -173,50 +177,88 @@ test_data:
         # content = self.dataset_populator.get_history_dataset_content( history_id )
         # self.assertEqual("chr5\t131424298\t131424460\tCCDS4149.1_cds_0_0_chr5_131424299_f\t0\t+\n", content)
 
+    def test_subworkflow_duplicate(self):
+        duplicate_subworkflow_invocate_wf = """
+format-version: "v2.0"
+$graph:
+- id: nested
+  class: GalaxyWorkflow
+  inputs:
+    inner_input: data
+  outputs:
+    inner_output:
+      outputSource: inner_cat/out_file1
+  steps:
+    inner_cat:
+      tool_id: cat
+      in:
+        input1: inner_input
+        queries_0|input2: inner_input
+
+- id: main
+  class: GalaxyWorkflow
+  inputs:
+    outer_input: data
+  steps:
+    outer_cat:
+      tool_id: cat
+      in:
+        input1: outer_input
+    nested_workflow_1:
+      run: '#nested'
+      in:
+        inner_input: outer_cat/out_file1
+    nested_workflow_2:
+      run: '#nested'
+      in:
+        inner_input: nested_workflow_1/inner_output
+"""
+        history_id = self.dataset_populator.new_history()
+        self._run_jobs(duplicate_subworkflow_invocate_wf, test_data={"outer_input": "hello world"}, history_id=history_id, client_convert=False)
+        content = self.dataset_populator.get_history_dataset_content(history_id)
+        assert content == "hello world\nhello world\nhello world\nhello world\n"
+
     def test_pause(self):
         workflow_id = self._upload_yaml_workflow("""
 class: GalaxyWorkflow
 steps:
-  - label: test_input
+  test_input:
     type: input
-  - label: first_cat
+  first_cat:
     tool_id: cat1
     state:
       input1:
         $link: test_input
-  - label: the_pause
+  the_pause:
     type: pause
     in:
       input: first_cat/out_file1
-  - label: second_cat
+  second_cat:
     tool_id: cat1
     in:
       input1: the_pause
 """)
-        print(self._get("workflows/%s/download" % workflow_id).json())
+        self.workflow_populator.dump_workflow(workflow_id)
 
     def test_implicit_connections(self):
         workflow_id = self._upload_yaml_workflow("""
 class: GalaxyWorkflow
+inputs:
+  test_input: data
 steps:
-  - label: test_input
-    type: input
-  - label: first_cat
+  first_cat:
     tool_id: cat1
-    state:
-      input1:
-        $link: test_input
-  - label: the_pause
+    in:
+      input1: test_input
+  the_pause:
     type: pause
-    connect:
-      input:
-      - first_cat#out_file1
-  - label: second_cat
+    in:
+      input: first_cat/out_file1
+  second_cat:
     tool_id: cat1
-    state:
-      input1:
-        $link: the_pause
-  - label: third_cat
+    in:
+      input1: the_pause
+  third_cat:
     tool_id: cat1
     connect:
       $step: second_cat
@@ -224,22 +266,21 @@ steps:
       input1:
         $link: test_input
 """)
-        workflow = self._get("workflows/%s/download" % workflow_id).json()
-        print(workflow)
+        self.workflow_populator.dump_workflow(workflow_id)
 
     @uses_test_history()
     def test_conditional_ints(self, history_id):
         self._run_jobs("""
 class: GalaxyWorkflow
 steps:
-  - label: test_input
+  test_input:
     tool_id: disambiguate_cond
     state:
       p3:
         use: true
       files:
         attach_files: false
-""", test_data={}, history_id=history_id)
+""", test_data={}, history_id=history_id, round_trip_format_conversion=True)
         content = self.dataset_populator.get_history_dataset_content(history_id)
         assert "no file specified" in content
         assert "7 7 4" in content
@@ -247,7 +288,7 @@ steps:
         self._run_jobs("""
 class: GalaxyWorkflow
 steps:
-  - label: test_input
+  test_input:
     tool_id: disambiguate_cond
     state:
       p3:
@@ -255,7 +296,7 @@ steps:
         p3v: 5
       files:
         attach_files: false
-""", test_data={}, history_id=history_id)
+""", test_data={}, history_id=history_id, round_trip_format_conversion=True)
         content = self.dataset_populator.get_history_dataset_content(history_id)
         assert "no file specified" in content
         assert "7 7 5" in content
diff --git a/test/base/driver_util.py b/test/base/driver_util.py
index 55c9be9b300..cae48fdc5c9 100644
--- a/test/base/driver_util.py
+++ b/test/base/driver_util.py
@@ -201,6 +201,7 @@ def setup_galaxy_config(
         cleanup_job='onsuccess',
         data_manager_config_file=data_manager_config_file,
         enable_beta_tool_formats=True,
+        enable_beta_workflow_format=True,
         expose_dataset_path=True,
         file_path=file_path,
         ftp_upload_purge=False,
diff --git a/test/base/populators.py b/test/base/populators.py
index a465fbd5025..99627f93e3c 100644
--- a/test/base/populators.py
+++ b/test/base/populators.py
@@ -547,8 +547,18 @@ class BaseWorkflowPopulator(object):
         return upload_response
 
     def upload_yaml_workflow(self, has_yaml, **kwds):
+        round_trip_conversion = kwds.get("round_trip_format_conversion", False)
+        client_convert = kwds.pop("client_convert", not round_trip_conversion)
+        kwds["convert"] = client_convert
         workflow = convert_and_import_workflow(has_yaml, galaxy_interface=self, **kwds)
-        return workflow["id"]
+        workflow_id = workflow["id"]
+        if round_trip_conversion:
+            workflow_yaml_wrapped = self.download_workflow(workflow_id, style="format2_wrapped_yaml")
+            assert "yaml_content" in workflow_yaml_wrapped, workflow_yaml_wrapped
+            round_trip_converted_content = workflow_yaml_wrapped["yaml_content"]
+            workflow_id = self.upload_yaml_workflow(round_trip_converted_content, client_convert=False, round_trip_conversion=False)
+
+        return workflow_id
 
     def wait_for_invocation(self, workflow_id, invocation_id, timeout=DEFAULT_TIMEOUT):
         url = "workflows/%s/usage/%s" % (workflow_id, invocation_id)
@@ -578,7 +588,15 @@ class BaseWorkflowPopulator(object):
         else:
             return invocation_response
 
-    def run_workflow(self, has_workflow, test_data=None, history_id=None, wait=True, source_type=None, jobs_descriptions=None, expected_response=200, assert_ok=True):
+    def download_workflow(self, workflow_id, style=None):
+        params = {}
+        if style is not None:
+            params["style"] = style
+        response = self._get("workflows/%s/download" % workflow_id, data=params)
+        api_asserts.assert_status_code_is(response, 200)
+        return response.json()
+
+    def run_workflow(self, has_workflow, test_data=None, history_id=None, wait=True, source_type=None, jobs_descriptions=None, expected_response=200, assert_ok=True, client_convert=None, round_trip_format_conversion=False, raw_yaml=False):
         """High-level wrapper around workflow API, etc. to invoke format 2 workflows."""
         workflow_populator = self
 
@@ -588,7 +606,10 @@ class BaseWorkflowPopulator(object):
             content = open(filename, "r").read()
             return content
 
-        workflow_id = workflow_populator.upload_yaml_workflow(has_workflow, source_type=source_type)
+        if client_convert is None:
+            client_convert = not round_trip_format_conversion
+
+        workflow_id = workflow_populator.upload_yaml_workflow(has_workflow, source_type=source_type, client_convert=client_convert, round_trip_format_conversion=round_trip_format_conversion, raw_yaml=raw_yaml)
 
         if test_data is None:
             if jobs_descriptions is None:
@@ -636,6 +657,13 @@ class BaseWorkflowPopulator(object):
                 workflow_request=workflow_request
             )
 
+    def dump_workflow(self, workflow_id, style=None):
+        raw_workflow = self.download_workflow(workflow_id, style=style)
+        if style == "format2_wrapped_yaml":
+            print(raw_workflow["yaml_content"])
+        else:
+            print(json.dumps(raw_workflow, sort_keys=True, indent=2))
+
 
 RunJobsSummary = namedtuple('RunJobsSummary', ['history_id', 'workflow_id', 'invocation_id', 'inputs', 'jobs', 'invocation', 'workflow_request'])
 
@@ -662,7 +690,7 @@ class WorkflowPopulator(BaseWorkflowPopulator, ImporterGalaxyInterface):
         }
         data.update(**kwds)
         upload_response = self._post("workflows", data=data)
-        assert upload_response.status_code == 200, upload_response
+        assert upload_response.status_code == 200, upload_response.content
         return upload_response.json()
 
 
diff --git a/test/base/workflow_fixtures.py b/test/base/workflow_fixtures.py
index 34bdaedb16d..519dd750131 100644
--- a/test/base/workflow_fixtures.py
+++ b/test/base/workflow_fixtures.py
@@ -2,10 +2,15 @@
 
 WORKFLOW_SIMPLE_CAT_AND_RANDOM_LINES = """
 class: GalaxyWorkflow
+doc: |
+  Simple workflow that no-op cats a file and then selects 10 random lines.
 inputs:
-  - id: the_input
+  the_input:
+    type: data
+    doc: input doc
 steps:
   - tool_id: cat1
+    doc: cat doc
     in:
       input1: the_input
   - tool_id: cat1
@@ -28,8 +33,8 @@ class: GalaxyWorkflow
 inputs:
   input1: data
 steps:
-  - tool_id: cat
-    label: first_cat
+  first_cat:
+    tool_id: cat
     in:
       input1: input1
       queries_0|input2: input1
@@ -41,7 +46,8 @@ class: GalaxyWorkflow
 inputs:
   input1: data
 steps:
-  - tool_id: multiple_versions
+  mul_versions:
+    tool_id: multiple_versions
     tool_version: "0.0.1"
     state:
       inttest: 8
@@ -53,7 +59,8 @@ class: GalaxyWorkflow
 inputs:
   input1: data
 steps:
-  - tool_id: multiple_versions
+  mul_versions:
+    tool_id: multiple_versions
     tool_version: "0.0.1"
     state:
       inttest: "moocow"
@@ -65,11 +72,12 @@ class: GalaxyWorkflow
 inputs:
   text_input: data
 steps:
-  - label: split_up
+  split_up:
     tool_id: collection_creates_pair
     in:
       input1: text_input
-  - tool_id: collection_paired_test
+  paired:
+    tool_id: collection_paired_test
     in:
       f1: split_up/paired_output
 test_data:
@@ -83,25 +91,21 @@ test_data:
 
 WORKFLOW_WITH_DYNAMIC_OUTPUT_COLLECTION = """
 class: GalaxyWorkflow
+inputs:
+  text_input1: data
+  text_input2: data
 steps:
-  - label: text_input1
-    type: input
-  - label: text_input2
-    type: input
-  - label: cat_inputs
+  cat_inputs:
     tool_id: cat1
-    state:
-      input1:
-        $link: text_input1
-      queries:
-        - input2:
-            $link: text_input2
-  - label: split_up
+    in:
+      input1: text_input1
+      queries_0|input2: text_input2
+  split_up:
     tool_id: collection_split_on_column
-    state:
-      input1:
-        $link: cat_inputs#out_file1
-  - tool_id: cat_list
+    in:
+      input1: cat_inputs/out_file1
+  cat_list:
+    tool_id: cat_list
     in:
       input1: split_up/split_output
 test_data:
@@ -121,8 +125,8 @@ inputs:
     type: collection
     collection_type: list
 steps:
-  - tool_id: cat
-    label: cat
+  cat:
+    tool_id: cat
     in:
       input1: input1
 """
@@ -152,7 +156,7 @@ class: GalaxyWorkflow
 inputs:
   input_c: collection
 steps:
-  - label: apply
+  apply:
     tool_id: __APPLY_RULES__
     state:
       input:
@@ -166,8 +170,8 @@ steps:
         mapping:
           - type: list_identifiers
             columns: [0, 1]
-  - tool_id: random_lines1
-    label: random_lines
+  random_lines:
+    tool_id: random_lines1
     state:
       num_lines: 1
       input:
@@ -191,7 +195,7 @@ class: GalaxyWorkflow
 inputs:
   input_c: collection
 steps:
-  - label: apply
+  apply:
     tool_id: __APPLY_RULES__
     state:
       input:
@@ -205,8 +209,8 @@ steps:
         mapping:
           - type: list_identifiers
             columns: [0, 1]
-  - tool_id: collection_creates_list
-    label: copy_list
+  copy_list:
+    tool_id: collection_creates_list
     in:
       input1: apply/output
 test_data:
@@ -228,11 +232,12 @@ outputs:
   outer_output:
     outputSource: second_cat/out_file1
 steps:
-  - tool_id: cat1
-    label: first_cat
+  first_cat:
+    tool_id: cat1
     in:
       input1: outer_input
-  - run:
+  nested_workflow:
+    run:
       class: GalaxyWorkflow
       inputs:
         inner_input: data
@@ -240,8 +245,8 @@ steps:
         workflow_output:
           outputSource: random_lines/out_file1
       steps:
-        - tool_id: random_lines1
-          label: random_lines
+        random_lines:
+          tool_id: random_lines1
           state:
             num_lines: 1
             input:
@@ -249,17 +254,13 @@ steps:
             seed_source:
               seed_source_selector: set_seed
               seed: asdf
-    label: nested_workflow
     in:
       inner_input: first_cat/out_file1
-  - tool_id: cat1
-    label: second_cat
-    state:
-      input1:
-        $link: nested_workflow#workflow_output
-      queries:
-        - input2:
-            $link: nested_workflow#workflow_output
+  second_cat:
+    tool_id: cat1
+    in:
+      input1: nested_workflow/workflow_output
+      queries_0|input2: nested_workflow/workflow_output
 """
 
 
@@ -271,7 +272,8 @@ outputs:
   outer_output:
     outputSource: nested_workflow/workflow_output
 steps:
-  - run:
+  nested_workflow:
+    run:
       class: GalaxyWorkflow
       inputs:
         inner_input: data
@@ -289,7 +291,6 @@ steps:
             seed_source:
               seed_source_selector: set_seed
               seed: asdf
-    label: nested_workflow
     in:
       inner_input: outer_input
 """
@@ -300,15 +301,16 @@ class: GalaxyWorkflow
 inputs:
   input1: data
 steps:
-  - tool_id: cat1
-    label: first_cat
+  first_cat:
+    tool_id: cat1
     outputs:
        out_file1:
          hide: true
          rename: "the new value"
     in:
       input1: input1
-  - tool_id: cat1
+  second_cat:
+    tool_id: cat1
     in:
       input1: first_cat/out_file1
 """
@@ -319,7 +321,8 @@ class: GalaxyWorkflow
 inputs:
   input1: data
 steps:
-  - tool_id: random_lines1
+  random:
+    tool_id: random_lines1
     runtime_inputs:
       - num_lines
     state:
@@ -336,11 +339,12 @@ class: GalaxyWorkflow
 inputs:
   input1: data
 steps:
-  - label: the_pause
+  the_pause:
     type: pause
     in:
       input: input1
-  - tool_id: random_lines1
+  random:
+    tool_id: random_lines1
     runtime_inputs:
       - num_lines
     state:
@@ -356,8 +360,8 @@ class: GalaxyWorkflow
 inputs:
   input1: data
 steps:
-  - tool_id: cat
-    label: first_cat
+  first_cat:
+    tool_id: cat
     state:
       input1:
         $link: input1
@@ -376,11 +380,10 @@ class: GalaxyWorkflow
 inputs:
   input1: data
 steps:
-  - tool_id: cat
-    label: first_cat
-    state:
-      input1:
-        $link: input1
+  first_cat:
+    tool_id: cat
+    in:
+      input1: input1
     outputs:
       out_file1:
         rename: "${replaceme} suffix"
@@ -394,7 +397,8 @@ outputs:
   outer_output:
     outputSource: nested_workflow/workflow_output
 steps:
-  - run:
+  nested_workflow:
+    run:
       class: GalaxyWorkflow
       inputs:
         inner_input: data
@@ -409,7 +413,6 @@ steps:
           outputs:
             out_file1:
               rename: "${replaceme} suffix"
-    label: nested_workflow
     in:
       inner_input: outer_input
 """
@@ -422,12 +425,9 @@ outputs:
   wf_output_1:
     outputSource: first_cat/out_file1
 steps:
-  - tool_id: cat1
-    label: first_cat
-    state:
-      input1:
-        $link: input1
-      queries:
-        - input2:
-            $link: input1
+  first_cat:
+    tool_id: cat1
+    in:
+      input1: input1
+      queries_0|input2: input1
 """