diff --git a/lib/galaxy/managers/collections.py b/lib/galaxy/managers/collections.py
index bd3f6f2e59a..2a81db4ee3a 100644
--- a/lib/galaxy/managers/collections.py
+++ b/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
     ):
diff --git a/lib/galaxy/tools/split_paired_and_unpaired.xml b/lib/galaxy/tools/split_paired_and_unpaired.xml
index d1d05c88158..e0815e721cc 100644
--- a/lib/galaxy/tools/split_paired_and_unpaired.xml
+++ b/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>
diff --git a/lib/galaxy_test/api/test_dataset_collections.py b/lib/galaxy_test/api/test_dataset_collections.py
index 7a505e7346f..0b21a16cc15 100644
--- a/lib/galaxy_test/api/test_dataset_collections.py
+++ b/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"),