Reject structurally invalid nested dataset collections

For nested collection types like list:paired_or_unpaired and list:paired,
verify in the manager that every element value is a DatasetCollection
whose collection_type matches the expected sub-collection type. Otherwise
the API silently accepted raw HDAs as children, producing collections
that crashed downstream tools (e.g. SplitPairedAndUnpairedTool).

The third tool test in split_paired_and_unpaired.xml relied on the
unvalidated path: el2/el3 were raw datasets rather than paired_or_unpaired
sub-collections wrapping a singleton "unpaired" element. Update the test
to use the canonical wrapped form so it passes against the new validator.

lib/galaxy/managers/collections.py

@@ -354,6 +354,7 @@ class DatasetCollectionManager:
         # else if elements is set, it better be an ordered dict!
 
         if elements is not self.ELEMENTS_UNINITIALIZED:
+            self._validate_nested_collection_elements(collection_type_description, elements)
             type_plugin = collection_type_description.rank_type_plugin()
             dataset_collection = builder.build_collection(
                 type_plugin, elements, fields=fields, column_definitions=column_definitions, rows=rows
@@ -577,6 +578,32 @@ class DatasetCollectionManager:
             session.commit()
         return dataset_collection_instance
 
+    def _validate_nested_collection_elements(self, collection_type_description, elements) -> None:
+        """For nested collection types (e.g. ``list:paired_or_unpaired``), verify that
+        every element value is a :class:`DatasetCollection` whose ``collection_type``
+        matches the expected sub-collection type. Otherwise creating a
+        ``list:paired_or_unpaired`` from raw HDAs would silently produce a
+        structurally invalid collection that downstream tools cannot handle.
+        """
+        if not collection_type_description.has_subcollections():
+            return
+        if not isinstance(elements, dict):
+            return
+        expected_sub = collection_type_description.subcollection_type_description().collection_type
+        for identifier, value in elements.items():
+            if isinstance(value, DatasetCollection) and value.collection_type == expected_sub:
+                continue
+            if isinstance(value, DatasetCollection):
+                actual = f"a collection of type '{value.collection_type}'"
+            elif getattr(value, "history_content_type", None) == "dataset":
+                actual = "a dataset"
+            else:
+                actual = type(value).__name__
+            raise RequestParameterInvalidException(
+                f"Element '{identifier}' of collection type '{collection_type_description.collection_type}' "
+                f"must be a sub-collection of type '{expected_sub}', got {actual}."
+            )
+
     def __recursively_create_collections_for_identifiers(
         self, trans, element_identifiers, hide_source_items: bool, copy_elements: bool, history=None
     ):

lib/galaxy/tools/split_paired_and_unpaired.xml

@@ -81,9 +81,15 @@
               <element name="reverse" value="simple_line_alternative.txt" />
             </collection>
           </element>
-          <element name="el2" value="simple_line.txt">
+          <element name="el2">
+            <collection type="paired_or_unpaired">
+              <element name="unpaired" value="simple_line.txt" />
+            </collection>
           </element>
-          <element name="el3" value="simple_line_alternative.txt">
+          <element name="el3">
+            <collection type="paired_or_unpaired">
+              <element name="unpaired" value="simple_line_alternative.txt" />
+            </collection>
           </element>
         </collection>
       </param>

lib/galaxy_test/api/test_dataset_collections.py

@@ -7,6 +7,7 @@ from urllib.parse import quote
 from galaxy.tool_util_models.sample_sheet import SampleSheetColumnDefinitions
 from galaxy.util import galaxy_root_path
 from galaxy_test.base.api_asserts import (
+    assert_error_message_contains,
     assert_has_key,
     assert_object_id_error,
     assert_status_code_is,
@@ -152,6 +153,30 @@ class TestDatasetCollectionsApi(ApiTestCase):
         returned_collections = dataset_collection["elements"]
         assert len(returned_collections) == 1, dataset_collection
 
+    def test_create_list_paired_or_unpaired_rejects_raw_hda_child(self, history_id):
+        element_identifiers = self.dataset_collection_populator.list_identifiers(history_id, contents=[("el1", "hi")])
+        payload = dict(
+            instance_type="history",
+            history_id=history_id,
+            element_identifiers=element_identifiers,
+            collection_type="list:paired_or_unpaired",
+        )
+        create_response = self._post("dataset_collections", payload, json=True)
+        assert_status_code_is(create_response, 400)
+        assert_error_message_contains(create_response, "sub-collection of type 'paired_or_unpaired'")
+
+    def test_create_list_paired_rejects_raw_hda_child(self, history_id):
+        element_identifiers = self.dataset_collection_populator.list_identifiers(history_id, contents=[("el1", "hi")])
+        payload = dict(
+            instance_type="history",
+            history_id=history_id,
+            element_identifiers=element_identifiers,
+            collection_type="list:paired",
+        )
+        create_response = self._post("dataset_collections", payload, json=True)
+        assert_status_code_is(create_response, 400)
+        assert_error_message_contains(create_response, "sub-collection of type 'paired'")
+
     def test_create_record(self, history_id):
         contents = [
             ("condition", "1\t2\t3"),