diff --git a/lib/galaxy/dataset_collections/structure.py b/lib/galaxy/dataset_collections/structure.py
index 19058b15b42..0caf7f2fac5 100644
--- a/lib/galaxy/dataset_collections/structure.py
+++ b/lib/galaxy/dataset_collections/structure.py
@@ -177,19 +177,14 @@ def tool_output_to_structure(get_sliced_input_collection_type, tool_output, coll
     if not tool_output.collection:
         tree = leaf
     else:
-        collection_type_descriptions = collections_manager.collection_type_descriptions
         # Okay this is ToolCollectionOutputStructure not a Structure - different
         # concepts of structure.
-        if tool_output.dynamic_structure:
-            # Two cases collection_type_source and collection_type right?
-            tree = UnitializedTree(collection_type_descriptions.for_type_description("list"))  # list is obviously wrong...
+        structured_like = tool_output.structure.structured_like
+        if structured_like:
+            collection_type = get_sliced_input_collection_type(structured_like)
         else:
-            structured_like = tool_output.structure.structured_like
-            if structured_like:
-                collection_type = get_sliced_input_collection_type(structured_like)
-            else:
-                collection_type = tool_output.structure.collection_type
-            tree = UnitializedTree(collection_type)
+            collection_type = tool_output.structure.collection_type
+        tree = UnitializedTree(collection_type)
 
     return tree
 
diff --git a/lib/galaxy/tools/parameters/output_collect.py b/lib/galaxy/tools/parameters/output_collect.py
index 3e6ecc645df..b5a15de3a62 100644
--- a/lib/galaxy/tools/parameters/output_collect.py
+++ b/lib/galaxy/tools/parameters/output_collect.py
@@ -164,6 +164,9 @@ def collect_dynamic_collections(
         else:
             collection = has_collection
 
+        # We are adding dynamic collections, which may be precreated, but their actually state is still new!
+        collection.populated_state = collection.populated_states.NEW
+
         try:
             collection_builder = collections_service.collection_builder_for(
                 collection
diff --git a/test/api/test_tools.py b/test/api/test_tools.py
index 4ce2ffc893f..8da38b4388a 100644
--- a/test/api/test_tools.py
+++ b/test/api/test_tools.py
@@ -787,6 +787,22 @@ class ToolsTestCase(api.ApiTestCase):
         assert output1_content.startswith("chr1")
         assert output2_content.startswith("chr1")
 
+    @skip_without_tool("collection_creates_dynamic_list_of_pairs")
+    def test_map_over_with_discovered_output_collection_elements(self):
+        with self.dataset_populator.test_history() as history_id:
+            hdca_id = self.dataset_collection_populator.create_list_in_history(history_id).json()["id"]
+            inputs = {
+                "input": {"batch": True, "values": [{"src": "hdca", "id": hdca_id}]}
+            }
+            create = self._run('collection_creates_dynamic_list_of_pairs', history_id, inputs).json()
+            implicit_collections = create['implicit_collections']
+            self.assertEquals(len(implicit_collections), 1)
+            self.assertEquals(implicit_collections[0]['collection_type'], 'list:list:paired')
+            self.assertEquals(implicit_collections[0]['elements'][0]['object']['element_count'], None)
+            self.dataset_populator.wait_for_job(create["jobs"][0]["id"], assert_ok=True)
+            hdca = self._get("histories/%s/contents/dataset_collections/%s" % (history_id, implicit_collections[0]['id'])).json()
+            self.assertEquals(hdca['elements'][0]['object']['elements'][0]['object']['elements'][0]['element_identifier'], 'forward')
+
     def _bed_list(self, history_id):
         bed1_contents = open(self.get_filename("1.bed"), "r").read()
         bed2_contents = open(self.get_filename("2.bed"), "r").read()
diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py
index 85725a22695..3cadfea5e4d 100644
--- a/test/api/test_workflows.py
+++ b/test/api/test_workflows.py
@@ -1814,11 +1814,9 @@ test_data:
         self.assertEqual("chr5\t131424298\t131424460\tCCDS4149.1_cds_0_0_chr5_131424299_f\t0\t+\n", content)
 
     def wait_for_invocation_and_jobs(self, history_id, workflow_id, invocation_id, assert_ok=True):
-        # Revert after https://github.com/galaxyproject/galaxy/issues/5146 is fixed.
-        # state = self.workflow_populator.wait_for_invocation(workflow_id, invocation_id)
-        # if assert_ok:
-        #    assert state == "scheduled", state
-        self.workflow_populator.wait_for_invocation(workflow_id, invocation_id)
+        state = self.workflow_populator.wait_for_invocation(workflow_id, invocation_id)
+        if assert_ok:
+            assert state == "scheduled", state
         time.sleep(.5)
         self.dataset_populator.wait_for_history_jobs(history_id, assert_ok=assert_ok)
         time.sleep(.5)
diff --git a/test/functional/tools/collection_creates_dynamic_list_of_pairs.xml b/test/functional/tools/collection_creates_dynamic_list_of_pairs.xml
index bd56401b0a3..b6450bb9f20 100644
--- a/test/functional/tools/collection_creates_dynamic_list_of_pairs.xml
+++ b/test/functional/tools/collection_creates_dynamic_list_of_pairs.xml
@@ -14,6 +14,7 @@
   ]]></command>
   <inputs>
     <param name="foo" type="text" label="Dummy Parameter" />
+    <param name="file" type="data" label="Dummy File Parameter" />
   </inputs>
   <outputs>
     <collection name="list_output" type="list:paired" label="Duplicate List">