diff --git a/client/src/api/datasets.ts b/client/src/api/datasets.ts
index bd811255691..32826c40f49 100644
--- a/client/src/api/datasets.ts
+++ b/client/src/api/datasets.ts
@@ -88,6 +88,7 @@ export async function copyDataset(
             type,
             copy_elements: true,
             // TODO: Investigate. These should be optional, but the API requires explicit null values?
+            fields: null,
             hide_source_items: null,
             instance_type: null,
         },
diff --git a/client/src/api/schema/schema.ts b/client/src/api/schema/schema.ts
index f9d3d4ad2e2..f3aa397fa3d 100644
--- a/client/src/api/schema/schema.ts
+++ b/client/src/api/schema/schema.ts
@@ -7067,6 +7067,12 @@ export interface components {
              * @description List of elements that should be in the new collection.
              */
             element_identifiers?: components["schemas"]["CollectionElementIdentifier"][] | null;
+            /**
+             * Fields
+             * @description List of fields to create for this collection. Set to 'auto' to guess fields from identifiers.
+             * @default []
+             */
+            fields: string | components["schemas"]["FieldDict"][] | null;
             /**
              * Folder Id
              * @description The ID of the library folder that will contain the collection. Required if `instance_type=library`.
@@ -7265,6 +7271,12 @@ export interface components {
              * @description List of elements that should be in the new collection.
              */
             element_identifiers?: components["schemas"]["CollectionElementIdentifier"][] | null;
+            /**
+             * Fields
+             * @description List of fields to create for this collection. Set to 'auto' to guess fields from identifiers.
+             * @default []
+             */
+            fields: string | components["schemas"]["FieldDict"][] | null;
             /**
              * Folder Id
              * @description The ID of the library folder that will contain the collection. Required if `instance_type=library`.
@@ -9249,6 +9261,17 @@ export interface components {
             /** Hash Value */
             hash_value: string;
         };
+        /** FieldDict */
+        FieldDict: {
+            /** Format */
+            format?: string | null;
+            /** Name */
+            name: string;
+            /** Type */
+            type:
+                | ("File" | "null" | "boolean" | "int" | "float" | "string")
+                | ("File" | "null" | "boolean" | "int" | "float" | "string")[];
+        };
         /** FileDataElement */
         FileDataElement: {
             /** Md5 */
diff --git a/client/src/components/History/model/queries.ts b/client/src/components/History/model/queries.ts
index 0c8a0415c3f..21ee64e4ddb 100644
--- a/client/src/components/History/model/queries.ts
+++ b/client/src/components/History/model/queries.ts
@@ -87,6 +87,7 @@ export async function createDatasetCollection(history: HistorySummary, inputs =
         copy_elements: true,
         name: "list",
         element_identifiers: [],
+        fields: "auto",
         hide_source_items: true,
         history_id: history.id,
     };
diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py
index 18232ea3df7..5525e4de9a7 100644
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -1961,8 +1961,9 @@ class MinimalJobWrapper(HasResourceParameters):
                 dataset.mark_unhidden()
         elif not purged:
             # If the tool was expected to set the extension, attempt to retrieve it
-            if dataset.ext == "auto":
-                dataset.extension = context.get("ext", "data")
+            context_ext = context.get("ext", "data")
+            if dataset.ext == "auto" or (dataset.ext == "data" and context_ext != "data"):
+                dataset.extension = context_ext
                 dataset.init_meta(copy_from=dataset)
             # if a dataset was copied, it won't appear in our dictionary:
             # either use the metadata from originating output dataset, or call set_meta on the copies
diff --git a/lib/galaxy/managers/collections.py b/lib/galaxy/managers/collections.py
index 04825abc66a..d711f558a1d 100644
--- a/lib/galaxy/managers/collections.py
+++ b/lib/galaxy/managers/collections.py
@@ -178,6 +178,7 @@ class DatasetCollectionManager:
         flush=True,
         completed_job=None,
         output_name=None,
+        fields=None,
     ):
         """
         PRECONDITION: security checks on ability to add to parent
@@ -202,6 +203,7 @@ class DatasetCollectionManager:
                 hide_source_items=hide_source_items,
                 copy_elements=copy_elements,
                 history=history,
+                fields=fields,
             )
 
         implicit_inputs = []
@@ -288,17 +290,20 @@ class DatasetCollectionManager:
         hide_source_items=None,
         copy_elements=False,
         history=None,
+        fields=None,
     ):
         # Make sure at least one of these is None.
         assert element_identifiers is None or elements is None
-
         if element_identifiers is None and elements is None:
             raise RequestParameterInvalidException(ERROR_INVALID_ELEMENTS_SPECIFICATION)
         if not collection_type:
             raise RequestParameterInvalidException(ERROR_NO_COLLECTION_TYPE)
 
-        collection_type_description = self.collection_type_descriptions.for_collection_type(collection_type)
+        collection_type_description = self.collection_type_descriptions.for_collection_type(
+            collection_type, fields=fields
+        )
         has_subcollections = collection_type_description.has_subcollections()
+
         # If we have elements, this is an internal request, don't need to load
         # objects from identifiers.
         if elements is None:
@@ -322,8 +327,9 @@ class DatasetCollectionManager:
 
         if elements is not self.ELEMENTS_UNINITIALIZED:
             type_plugin = collection_type_description.rank_type_plugin()
-            dataset_collection = builder.build_collection(type_plugin, elements)
+            dataset_collection = builder.build_collection(type_plugin, elements, fields=fields)
         else:
+            # TODO: Pass fields here - need test case first.
             dataset_collection = model.DatasetCollection(populated=False)
         dataset_collection.collection_type = collection_type
         return dataset_collection
diff --git a/lib/galaxy/managers/collections_util.py b/lib/galaxy/managers/collections_util.py
index 32feab23889..7f129992c75 100644
--- a/lib/galaxy/managers/collections_util.py
+++ b/lib/galaxy/managers/collections_util.py
@@ -39,6 +39,7 @@ def api_payload_to_create_params(payload):
         name=payload.get("name", None),
         hide_source_items=string_as_bool(payload.get("hide_source_items", False)),
         copy_elements=string_as_bool(payload.get("copy_elements", False)),
+        fields=payload.get("fields", None),
     )
     return params
 
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index d3f6177faf9..20c9e77e5dd 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -6614,6 +6614,7 @@ class DatasetCollection(Base, Dictifiable, UsesAnnotations, Serializable):
     element_count: Mapped[Optional[int]]
     create_time: Mapped[datetime] = mapped_column(default=now, nullable=True)
     update_time: Mapped[datetime] = mapped_column(default=now, onupdate=now, nullable=True)
+    fields: Mapped[Optional[bytes]] = mapped_column(JSONType, nullable=True)
 
     elements: Mapped[List["DatasetCollectionElement"]] = relationship(
         primaryjoin=(lambda: DatasetCollection.id == DatasetCollectionElement.dataset_collection_id),
@@ -6626,12 +6627,21 @@ class DatasetCollection(Base, Dictifiable, UsesAnnotations, Serializable):
 
     populated_states = DatasetCollectionPopulatedState
 
-    def __init__(self, id=None, collection_type=None, populated=True, element_count=None):
+    def __init__(
+        self,
+        id=None,
+        collection_type=None,
+        populated=True,
+        element_count=None,
+        fields=None,
+    ):
         self.id = id
         self.collection_type = collection_type
         if not populated:
             self.populated_state = DatasetCollection.populated_states.NEW
         self.element_count = element_count
+        # TODO: persist fields...
+        self.fields = fields
 
     def _build_nested_collection_attributes_stmt(
         self,
@@ -6806,6 +6816,10 @@ class DatasetCollection(Base, Dictifiable, UsesAnnotations, Serializable):
 
         return self._populated_optimized
 
+    @property
+    def allow_implicit_mapping(self):
+        return self.collection_type != "record"
+
     @property
     def populated(self):
         top_level_populated = self.populated_state == DatasetCollection.populated_states.OK
diff --git a/lib/galaxy/model/dataset_collections/builder.py b/lib/galaxy/model/dataset_collections/builder.py
index 2ae001f33a2..8c7f2b1e500 100644
--- a/lib/galaxy/model/dataset_collections/builder.py
+++ b/lib/galaxy/model/dataset_collections/builder.py
@@ -1,28 +1,37 @@
+from typing import (
+    Any,
+    Dict,
+    List,
+)
+
 from galaxy import model
 from galaxy.model.orm.util import add_object_to_object_session
+from galaxy.tool_util_models.tool_source import FieldDict
 from galaxy.util.oset import OrderedSet
 from .type_description import COLLECTION_TYPE_DESCRIPTION_FACTORY
 
 
-def build_collection(type, dataset_instances, collection=None, associated_identifiers=None):
+def build_collection(type, dataset_instances, collection=None, associated_identifiers=None, fields=None):
     """
     Build DatasetCollection with populated DatasetcollectionElement objects
     corresponding to the supplied dataset instances or throw exception if
     this is not a valid collection of the specified type.
     """
-    dataset_collection = collection or model.DatasetCollection()
+    dataset_collection = collection or model.DatasetCollection(fields=fields)
     associated_identifiers = associated_identifiers or set()
-    set_collection_elements(dataset_collection, type, dataset_instances, associated_identifiers)
+    set_collection_elements(dataset_collection, type, dataset_instances, associated_identifiers, fields=fields)
     return dataset_collection
 
 
-def set_collection_elements(dataset_collection, type, dataset_instances, associated_identifiers):
+def set_collection_elements(dataset_collection, type, dataset_instances, associated_identifiers, fields=None):
     new_element_keys = OrderedSet(dataset_instances.keys()) - associated_identifiers
     new_dataset_instances = {k: dataset_instances[k] for k in new_element_keys}
     dataset_collection.element_count = dataset_collection.element_count or 0
     element_index = dataset_collection.element_count
     elements = []
-    for element in type.generate_elements(new_dataset_instances):
+    if type.collection_type == "record" and fields == "auto":
+        fields = guess_fields(dataset_instances)
+    for element in type.generate_elements(new_dataset_instances, fields=fields):
         element.element_index = element_index
         add_object_to_object_session(element, dataset_collection)
         element.collection = dataset_collection
@@ -35,6 +44,16 @@ def set_collection_elements(dataset_collection, type, dataset_instances, associa
     return dataset_collection
 
 
+def guess_fields(dataset_instances: Dict[str, Any]) -> List[FieldDict]:
+    fields: List[FieldDict] = []
+    for identifier, element in dataset_instances.items():
+        # TODO: Make generic enough to handle nested record types.
+        assert element.history_content_type == "dataset"
+        fields.append({"type": "File", "name": identifier})
+
+    return fields
+
+
 class CollectionBuilder:
     """Purely functional builder pattern for building a dataset collection."""
 
diff --git a/lib/galaxy/model/dataset_collections/registry.py b/lib/galaxy/model/dataset_collections/registry.py
index 9c849dfdad6..bd148edafd2 100644
--- a/lib/galaxy/model/dataset_collections/registry.py
+++ b/lib/galaxy/model/dataset_collections/registry.py
@@ -2,9 +2,14 @@ from galaxy import model
 from .types import (
     list,
     paired,
+    record,
 )
 
-PLUGIN_CLASSES = [list.ListDatasetCollectionType, paired.PairedDatasetCollectionType]
+PLUGIN_CLASSES = [
+    list.ListDatasetCollectionType,
+    paired.PairedDatasetCollectionType,
+    record.RecordDatasetCollectionType,
+]
 
 
 class DatasetCollectionTypesRegistry:
@@ -14,13 +19,13 @@ class DatasetCollectionTypesRegistry:
     def get(self, plugin_type):
         return self.__plugins[plugin_type]
 
-    def prototype(self, plugin_type):
+    def prototype(self, plugin_type, fields=None):
         plugin_type_object = self.get(plugin_type)
         if not hasattr(plugin_type_object, "prototype_elements"):
             raise Exception(f"Cannot pre-determine structure for collection of type {plugin_type}")
 
         dataset_collection = model.DatasetCollection()
-        for e in plugin_type_object.prototype_elements():
+        for e in plugin_type_object.prototype_elements(fields=fields):
             e.collection = dataset_collection
         return dataset_collection
 
diff --git a/lib/galaxy/model/dataset_collections/type_description.py b/lib/galaxy/model/dataset_collections/type_description.py
index cade102453c..172ac1976b2 100644
--- a/lib/galaxy/model/dataset_collections/type_description.py
+++ b/lib/galaxy/model/dataset_collections/type_description.py
@@ -9,9 +9,9 @@ class CollectionTypeDescriptionFactory:
         # I think.
         self.type_registry = type_registry
 
-    def for_collection_type(self, collection_type):
+    def for_collection_type(self, collection_type, fields=None):
         assert collection_type is not None
-        return CollectionTypeDescription(collection_type, self)
+        return CollectionTypeDescription(collection_type, self, fields=fields)
 
 
 class CollectionTypeDescription:
@@ -47,12 +47,15 @@ class CollectionTypeDescription:
 
     collection_type: str
 
-    def __init__(self, collection_type: Union[str, "CollectionTypeDescription"], collection_type_description_factory):
+    def __init__(
+        self, collection_type: Union[str, "CollectionTypeDescription"], collection_type_description_factory, fields=None
+    ):
         if isinstance(collection_type, CollectionTypeDescription):
             self.collection_type = collection_type.collection_type
         else:
             self.collection_type = collection_type
         self.collection_type_description_factory = collection_type_description_factory
+        self.fields = fields
         self.__has_subcollections = self.collection_type.find(":") > 0
 
     def child_collection_type(self):
diff --git a/lib/galaxy/model/dataset_collections/types/__init__.py b/lib/galaxy/model/dataset_collections/types/__init__.py
index bfcf7bae79a..c294f6957be 100644
--- a/lib/galaxy/model/dataset_collections/types/__init__.py
+++ b/lib/galaxy/model/dataset_collections/types/__init__.py
@@ -11,7 +11,7 @@ log = logging.getLogger(__name__)
 
 class DatasetCollectionType(metaclass=ABCMeta):
     @abstractmethod
-    def generate_elements(self, dataset_instances):
+    def generate_elements(self, dataset_instances: dict, **kwds):
         """Generate DatasetCollectionElements with corresponding
         to the supplied dataset instances or throw exception if
         this is not a valid collection of the specified type.
diff --git a/lib/galaxy/model/dataset_collections/types/list.py b/lib/galaxy/model/dataset_collections/types/list.py
index 18ce4db7653..d4421d009c3 100644
--- a/lib/galaxy/model/dataset_collections/types/list.py
+++ b/lib/galaxy/model/dataset_collections/types/list.py
@@ -7,8 +7,8 @@ class ListDatasetCollectionType(BaseDatasetCollectionType):
 
     collection_type = "list"
 
-    def generate_elements(self, elements):
-        for identifier, element in elements.items():
+    def generate_elements(self, dataset_instances, **kwds):
+        for identifier, element in dataset_instances.items():
             association = DatasetCollectionElement(
                 element=element,
                 element_identifier=identifier,
diff --git a/lib/galaxy/model/dataset_collections/types/paired.py b/lib/galaxy/model/dataset_collections/types/paired.py
index 4ae95a1442a..e774ab67aac 100644
--- a/lib/galaxy/model/dataset_collections/types/paired.py
+++ b/lib/galaxy/model/dataset_collections/types/paired.py
@@ -15,21 +15,21 @@ class PairedDatasetCollectionType(BaseDatasetCollectionType):
 
     collection_type = "paired"
 
-    def generate_elements(self, elements):
-        if forward_dataset := elements.get(FORWARD_IDENTIFIER):
+    def generate_elements(self, dataset_instances, **kwds):
+        if forward_dataset := dataset_instances.get(FORWARD_IDENTIFIER):
             left_association = DatasetCollectionElement(
                 element=forward_dataset,
                 element_identifier=FORWARD_IDENTIFIER,
             )
             yield left_association
-        if reverse_dataset := elements.get(REVERSE_IDENTIFIER):
+        if reverse_dataset := dataset_instances.get(REVERSE_IDENTIFIER):
             right_association = DatasetCollectionElement(
                 element=reverse_dataset,
                 element_identifier=REVERSE_IDENTIFIER,
             )
             yield right_association
 
-    def prototype_elements(self):
+    def prototype_elements(self, **kwds):
         left_association = DatasetCollectionElement(
             element=HistoryDatasetAssociation(),
             element_identifier=FORWARD_IDENTIFIER,
diff --git a/lib/galaxy/model/dataset_collections/types/record.py b/lib/galaxy/model/dataset_collections/types/record.py
new file mode 100644
index 00000000000..193509f439e
--- /dev/null
+++ b/lib/galaxy/model/dataset_collections/types/record.py
@@ -0,0 +1,45 @@
+from galaxy.exceptions import RequestParameterMissingException
+from galaxy.model import (
+    DatasetCollectionElement,
+    HistoryDatasetAssociation,
+)
+from ..types import BaseDatasetCollectionType
+
+
+class RecordDatasetCollectionType(BaseDatasetCollectionType):
+    """Arbitrary CWL-style record type."""
+
+    collection_type = "record"
+
+    def generate_elements(self, dataset_instances, **kwds):
+        fields = kwds.get("fields", None)
+        if fields is None:
+            raise RequestParameterMissingException("Missing or null parameter 'fields' required for record types.")
+        if len(dataset_instances) != len(fields):
+            self._validation_failed("Supplied element do not match fields.")
+        index = 0
+        for identifier, element in dataset_instances.items():
+            field = fields[index]
+            if field["name"] != identifier:
+                self._validation_failed("Supplied element do not match fields.")
+
+            # TODO: validate type and such.
+            association = DatasetCollectionElement(
+                element=element,
+                element_identifier=identifier,
+            )
+            yield association
+            index += 1
+
+    def prototype_elements(self, fields=None, **kwds):
+        if fields is None:
+            raise RequestParameterMissingException("Missing or null parameter 'fields' required for record types.")
+        for field in fields:
+            name = field.get("name", None)
+            assert name
+            assert field.get("type", "File")  # NS: this assert doesn't make sense as it is
+            field_dataset = DatasetCollectionElement(
+                element=HistoryDatasetAssociation(),
+                element_identifier=name,
+            )
+            yield field_dataset
diff --git a/lib/galaxy/schema/schema.py b/lib/galaxy/schema/schema.py
index fcf28ebe061..6ab318d11b3 100644
--- a/lib/galaxy/schema/schema.py
+++ b/lib/galaxy/schema/schema.py
@@ -50,6 +50,7 @@ from galaxy.schema.types import (
     OffsetNaiveDatetime,
     RelativeUrl,
 )
+from galaxy.tool_util_models.tool_source import FieldDict
 from galaxy.util.hash_util import HashFunctionNameEnum
 from galaxy.util.sanitize_html import sanitize_html
 
@@ -1715,6 +1716,11 @@ class CreateNewCollectionPayload(Model):
         default=None,
         description="The ID of the library folder that will contain the collection. Required if `instance_type=library`.",
     )
+    fields_: Optional[Union[str, List[FieldDict]]] = Field(
+        default=[],
+        description="List of fields to create for this collection. Set to 'auto' to guess fields from identifiers.",
+        alias="fields",
+    )
 
 
 class ModelStoreFormat(str, Enum):
diff --git a/lib/galaxy/tool_util/parser/interface.py b/lib/galaxy/tool_util/parser/interface.py
index 35087f97f5f..b0594f4acc0 100644
--- a/lib/galaxy/tool_util/parser/interface.py
+++ b/lib/galaxy/tool_util/parser/interface.py
@@ -31,6 +31,7 @@ from galaxy.tool_util_models.tool_source import (
     BaseJsonTestCollectionDefCollectionElementDict,
     Citation,
     DrillDownOptionsDict,
+    FieldDict,
     HelpContent,
     JsonTestCollectionDefCollectionElementDict,
     JsonTestCollectionDefDatasetElementDict,
@@ -597,6 +598,7 @@ class XmlTestCollectionDefDict(TypedDict):
     model_class: Literal["TestCollectionDef"]
     attributes: TestCollectionAttributeDict
     collection_type: CollectionType
+    fields: Optional[List[FieldDict]]
     elements: List[TestCollectionDefElementDict]
     name: str
 
@@ -634,12 +636,17 @@ def _copy_if_exists(attributes, as_dict, name: str, as_name: Optional[str] = Non
 class TestCollectionDef:
     __test__ = False  # Prevent pytest from discovering this class (issue #12071)
     elements: List[TestCollectionDefElementInternal]
+    collection_type: Optional[str]
+    fields: Optional[List[FieldDict]]
 
-    def __init__(self, attrib, name, collection_type, elements):
+    def __init__(
+        self, attrib, name, collection_type: Optional[str], elements, fields: Optional[List[FieldDict]] = None
+    ):
         self.attrib = attrib
         self.collection_type = collection_type
         self.elements = elements
         self.name = name
+        self.fields = fields
 
     def _test_format_to_dict(self) -> "BaseJsonTestCollectionDefCollectionElementDict":
 
@@ -696,6 +703,7 @@ class TestCollectionDef:
             "collection_type": self.collection_type,
             "elements": list(map(element_to_dict, self.elements or [])),
             "name": self.name,
+            "fields": self.fields,
         }
 
     @staticmethod
@@ -719,6 +727,7 @@ class TestCollectionDef:
                 name=xml_as_dict.get("name", "Unnamed Collection"),
                 elements=list(map(element_from_dict, xml_as_dict["elements"] or [])),
                 collection_type=xml_as_dict["collection_type"],
+                fields=xml_as_dict.get("fields", None),
             )
         else:
             json_as_dict = cast(JsonTestCollectionDefDict, as_dict)
diff --git a/lib/galaxy/tool_util/parser/output_objects.py b/lib/galaxy/tool_util/parser/output_objects.py
index 090b38ac63b..655c0051854 100644
--- a/lib/galaxy/tool_util/parser/output_objects.py
+++ b/lib/galaxy/tool_util/parser/output_objects.py
@@ -425,12 +425,14 @@ class ToolOutputCollectionStructure:
         collection_type_from_rules: Optional[str] = None,
         structured_like: Optional[str] = None,
         dataset_collector_descriptions: Optional[List[DatasetCollectionDescription]] = None,
+        fields=None,
     ) -> None:
         self.collection_type = collection_type
         self.collection_type_source = collection_type_source
         self.collection_type_from_rules = collection_type_from_rules
         self.structured_like = structured_like
         self.dataset_collector_descriptions = dataset_collector_descriptions or []
+        self.fields = fields
         if collection_type and collection_type_source:
             raise ValueError("Cannot set both type and type_source on collection output.")
         if (
@@ -447,6 +449,10 @@ class ToolOutputCollectionStructure:
             raise ValueError(
                 "Cannot specify dynamic structure (discover_datasets) and collection type attributes structured_like or collection_type_from_rules."
             )
+        if collection_type == "record" and fields is None:
+            raise ValueError("If record outputs are defined, fields must be defined as well.")
+        if fields is not None and collection_type != "record":
+            raise ValueError("If fields are specified for outputs, the collection type must be record.")
         self.dynamic = bool(dataset_collector_descriptions)
 
     def collection_prototype(self, inputs, type_registry):
@@ -456,7 +462,7 @@ class ToolOutputCollectionStructure:
         else:
             collection_type = self.collection_type
             assert collection_type
-            collection_prototype = type_registry.prototype(collection_type)
+            collection_prototype = type_registry.prototype(collection_type, fields=self.fields)
             collection_prototype.collection_type = collection_type
         return collection_prototype
 
diff --git a/lib/galaxy/tool_util/parser/xml.py b/lib/galaxy/tool_util/parser/xml.py
index 1b942b3070a..1d753fb4db3 100644
--- a/lib/galaxy/tool_util/parser/xml.py
+++ b/lib/galaxy/tool_util/parser/xml.py
@@ -1045,13 +1045,18 @@ def _test_collection_def_dict(elem: Element) -> XmlTestCollectionDefDict:
         else:
             element_definition = __parse_param_elem(element)
         elements.append({"element_identifier": element_identifier, "element_definition": element_definition})
-
+    fields_json = "null"
+    fields_el = elem.find("fields")
+    if fields_el is not None:
+        fields_json = fields_el.text or "null"
+    fields = json.loads(fields_json)
     return XmlTestCollectionDefDict(
         model_class="TestCollectionDef",
         attributes=attrib,
         collection_type=collection_type,
         elements=elements,
         name=name,
+        fields=fields,
     )
 
 
diff --git a/lib/galaxy/tool_util/verify/interactor.py b/lib/galaxy/tool_util/verify/interactor.py
index 9d9f87a6f4a..650e5c80682 100644
--- a/lib/galaxy/tool_util/verify/interactor.py
+++ b/lib/galaxy/tool_util/verify/interactor.py
@@ -693,6 +693,8 @@ class GalaxyInteractorApi:
             collection_type=collection_def.collection_type,
             history_id=history_id,
         )
+        if collection_def.fields:
+            create_payload["fields"] = collection_def.fields
         return self._post("dataset_collections", data=create_payload, json=True).json()["id"]
 
     def _element_identifiers(self, collection_def):
diff --git a/lib/galaxy/tool_util_models/tool_source.py b/lib/galaxy/tool_util_models/tool_source.py
index d7bb207b58d..c7f76d86326 100644
--- a/lib/galaxy/tool_util_models/tool_source.py
+++ b/lib/galaxy/tool_util_models/tool_source.py
@@ -44,6 +44,18 @@ class DrillDownOptionsDict(TypedDict):
     selected: bool
 
 
+# For fields... just implementing a subset of CWL for Galaxy flavors of these objects
+# so far.
+CwlType = Literal["File", "null", "boolean", "int", "float", "string"]
+FieldType = Union[CwlType, List[CwlType]]
+
+
+class FieldDict(TypedDict):
+    name: str
+    type: FieldType
+    format: NotRequired[Optional[str]]
+
+
 JsonTestDatasetDefDict = TypedDict(
     "JsonTestDatasetDefDict",
     {
@@ -81,7 +93,7 @@ BaseJsonTestCollectionDefCollectionElementDict = TypedDict(
     "BaseJsonTestCollectionDefCollectionElementDict",
     {
         "class": Literal["Collection"],
-        "collection_type": str,
+        "collection_type": Optional[str],
         "elements": NotRequired[Optional[List[JsonTestCollectionDefElementDict]]],
     },
 )
@@ -91,7 +103,7 @@ JsonTestCollectionDefCollectionElementDict = TypedDict(
     {
         "identifier": str,
         "class": Literal["Collection"],
-        "collection_type": str,
+        "collection_type": Optional[str],
         "elements": NotRequired[Optional[List[JsonTestCollectionDefElementDict]]],
     },
 )
@@ -100,7 +112,7 @@ JsonTestCollectionDefDict = TypedDict(
     "JsonTestCollectionDefDict",
     {
         "class": Literal["Collection"],
-        "collection_type": str,
+        "collection_type": Optional[str],
         "elements": NotRequired[Optional[List[JsonTestCollectionDefElementDict]]],
         "name": NotRequired[Optional[str]],
     },
diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py
index a97fdb4d4af..e3706e0c4f9 100644
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -3760,7 +3760,7 @@ class ExtractDatasetCollectionTool(DatabaseOperationTool):
             collection = has_collection.collection
 
         collection_type = collection.collection_type
-        assert collection_type in ["list", "paired"]
+        assert collection_type in ["list", "paired", "record"]
         how = incoming["which"]["which_dataset"]
         if how == "first":
             extracted_element = collection.first_dataset_element
diff --git a/lib/galaxy/tools/actions/__init__.py b/lib/galaxy/tools/actions/__init__.py
index d434df8b287..1ad5b688411 100644
--- a/lib/galaxy/tools/actions/__init__.py
+++ b/lib/galaxy/tools/actions/__init__.py
@@ -686,7 +686,10 @@ class DefaultToolAction(ToolAction):
                         assert not element_identifiers  # known_outputs must have been empty
                         element_kwds = dict(elements=collections_manager.ELEMENTS_UNINITIALIZED)
                     else:
-                        element_kwds = dict(element_identifiers=element_identifiers)
+                        element_kwds = dict(
+                            element_identifiers=element_identifiers,
+                            fields=output.structure.fields,
+                        )
                     output_collections.create_collection(
                         output=output, name=name, completed_job=completed_job, **element_kwds
                     )
diff --git a/lib/galaxy/tools/extract_dataset.xml b/lib/galaxy/tools/extract_dataset.xml
index 37986eb722f..5c7ece160ab 100644
--- a/lib/galaxy/tools/extract_dataset.xml
+++ b/lib/galaxy/tools/extract_dataset.xml
@@ -10,7 +10,7 @@
     <edam_operation>operation_3695</edam_operation>
   </edam_operations>
   <inputs>
-    <param type="data_collection" collection_type="list,paired" name="input" label="Input List" />
+    <param type="data_collection" collection_type="list,paired,record" name="input" label="Input List" />
     <conditional name="which">
       <param name="which_dataset" type="select" label="How should a dataset be selected?">
         <option value="first">The first dataset</option>
@@ -73,6 +73,18 @@ This tool will create new history datasets from your collection but your quota u
         <metadata name="name" value="foo" />
       </output>
     </test>
+    <test>
+      <param name="input">
+        <collection type="record">
+          <fields>[{"name": "parent", "type": "File"}, {"name": "child", "type": "File"}]</fields>
+          <element name="parent" value="simple_line.txt" />
+          <element name="child" value="simple_line_alternative.txt" />
+        </collection>
+      </param>
+      <output name="output" file="simple_line.txt">
+        <metadata name="name" value="parent" />
+      </output>
+    </test>
     <test>
       <conditional name="which">
         <param name="which_dataset" value="by_identifier" />
diff --git a/lib/galaxy/webapps/galaxy/services/dataset_collections.py b/lib/galaxy/webapps/galaxy/services/dataset_collections.py
index 273012052ad..c237d04f84f 100644
--- a/lib/galaxy/webapps/galaxy/services/dataset_collections.py
+++ b/lib/galaxy/webapps/galaxy/services/dataset_collections.py
@@ -123,7 +123,7 @@ class DatasetCollectionsService(ServiceBase, UsesLibraryMixinItems):
         :returns:   element view of new dataset collection
         """
         # TODO: Error handling...
-        create_params = api_payload_to_create_params(payload.dict(exclude_unset=True))
+        create_params = api_payload_to_create_params(payload.dict(exclude_unset=True, by_alias=True))
         if payload.instance_type == "history":
             if payload.history_id is None:
                 raise exceptions.RequestParameterInvalidException("Parameter history_id is required.")
diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py
index f7b0c188c64..4a9d1d2cc5b 100644
--- a/lib/galaxy/workflow/modules.py
+++ b/lib/galaxy/workflow/modules.py
@@ -565,7 +565,7 @@ class WorkflowModule:
         for input_dict in all_inputs:
             name = input_dict["name"]
             data = progress.replacement_for_input(self.trans, step, input_dict)
-            can_map_over = hasattr(data, "collection")  # and data.collection.allow_implicit_mapping
+            can_map_over = hasattr(data, "collection") and data.collection.allow_implicit_mapping
 
             if not can_map_over:
                 continue
diff --git a/lib/galaxy_test/api/test_dataset_collections.py b/lib/galaxy_test/api/test_dataset_collections.py
index d7710c57b2f..372d693d4fd 100644
--- a/lib/galaxy_test/api/test_dataset_collections.py
+++ b/lib/galaxy_test/api/test_dataset_collections.py
@@ -1,3 +1,4 @@
+import json
 import zipfile
 from io import BytesIO
 from typing import List
@@ -100,6 +101,106 @@ class TestDatasetCollectionsApi(ApiTestCase):
             pair_1_element_1 = pair_elements[0]
             assert pair_1_element_1["element_index"] == 0
 
+    def test_create_record(self, history_id):
+        contents = [
+            ("condition", "1\t2\t3"),
+            ("control1", "4\t5\t6"),
+            ("control2", "7\t8\t9"),
+        ]
+        record_identifiers = self.dataset_collection_populator.list_identifiers(history_id, contents)
+        fields = [
+            {"name": "condition", "type": "File"},
+            {"name": "control1", "type": "File"},
+            {"name": "control2", "type": "File"},
+        ]
+        payload = dict(
+            name="a record",
+            instance_type="history",
+            history_id=history_id,
+            element_identifiers=record_identifiers,
+            collection_type="record",
+            fields=fields,
+        )
+        create_response = self._post("dataset_collections", payload, json=True)
+        dataset_collection = self._check_create_response(create_response)
+        assert dataset_collection["collection_type"] == "record"
+        assert dataset_collection["name"] == "a record"
+        returned_collections = dataset_collection["elements"]
+        assert len(returned_collections) == 3, dataset_collection
+        record_pos_0_element = returned_collections[0]
+        self._assert_has_keys(record_pos_0_element, "element_index")
+        record_pos_0_object = record_pos_0_element["object"]
+        self._assert_has_keys(record_pos_0_object, "name", "history_content_type")
+
+    def test_record_requires_fields(self, history_id):
+        contents = [
+            ("condition", "1\t2\t3"),
+            ("control1", "4\t5\t6"),
+            ("control2", "7\t8\t9"),
+        ]
+        record_identifiers = self.dataset_collection_populator.list_identifiers(history_id, contents)
+        payload = dict(
+            name="a record",
+            instance_type="history",
+            history_id=history_id,
+            element_identifiers=json.dumps(record_identifiers),
+            collection_type="record",
+        )
+        create_response = self._post("dataset_collections", payload)
+        self._assert_status_code_is(create_response, 400)
+
+    def test_record_auto_fields(self, history_id):
+        contents = [
+            ("condition", "1\t2\t3"),
+            ("control1", "4\t5\t6"),
+            ("control2", "7\t8\t9"),
+        ]
+        record_identifiers = self.dataset_collection_populator.list_identifiers(history_id, contents)
+        payload = dict(
+            name="a record",
+            instance_type="history",
+            history_id=history_id,
+            element_identifiers=record_identifiers,
+            collection_type="record",
+            fields="auto",
+        )
+        create_response = self._post("dataset_collections", payload, json=True)
+        self._check_create_response(create_response)
+
+    def test_record_field_validation(self, history_id):
+        contents = [
+            ("condition", "1\t2\t3"),
+            ("control1", "4\t5\t6"),
+            ("control2", "7\t8\t9"),
+        ]
+        record_identifiers = self.dataset_collection_populator.list_identifiers(history_id, contents)
+        too_few_fields = [
+            {"name": "condition", "type": "File"},
+            {"name": "control1", "type": "File"},
+        ]
+        too_many_fields = [
+            {"name": "condition", "type": "File"},
+            {"name": "control1", "type": "File"},
+            {"name": "control2", "type": "File"},
+            {"name": "control3", "type": "File"},
+        ]
+        wrong_name_fields = [
+            {"name": "condition", "type": "File"},
+            {"name": "control1", "type": "File"},
+            {"name": "control3", "type": "File"},
+        ]
+        for fields in [too_few_fields, too_many_fields, wrong_name_fields]:
+            payload = dict(
+                name="a record",
+                instance_type="history",
+                history_id=history_id,
+                element_identifiers=json.dumps(record_identifiers),
+                collection_type="record",
+                fields=json.dumps(fields),
+            )
+            create_response = self._post("dataset_collections", payload)
+            self._assert_status_code_is(create_response, 400)
+
     def test_list_download(self):
         with self.dataset_populator.test_history(require_new=False) as history_id:
             fetch_response = self.dataset_collection_populator.create_list_in_history(
diff --git a/test/functional/tools/collection_record_test_two_files.xml b/test/functional/tools/collection_record_test_two_files.xml
new file mode 100644
index 00000000000..ae69549fe00
--- /dev/null
+++ b/test/functional/tools/collection_record_test_two_files.xml
@@ -0,0 +1,32 @@
+<tool id="collection_record_test_two_files" name="collection_record_test_two_files" version="0.1.0">
+  <command>
+    cat $f1.parent $f1['child'] >> $out1;
+    echo 'Collection name: $f1.name'
+  </command>
+  <inputs>
+    <param name="f1" type="data_collection" collection_type="record" label="Input collection" />
+  </inputs>
+  <outputs>
+    <data format="txt" name="out1" />
+  </outputs>
+  <tests>
+    <test>
+      <param name="f1">
+        <collection type="record" name="collection name">
+          <fields>[{"name": "parent", "type": "File"}, {"name": "child", "type": "File"}]</fields>
+          <element name="parent" value="simple_line.txt" />
+          <element name="child" value="simple_line_alternative.txt" />
+        </collection>
+      </param>
+      <output name="out1">
+        <assert_contents>
+          <has_line line="This is a line of text." />
+          <has_line line="This is a different line of text." />
+        </assert_contents>
+      </output>
+      <assert_stdout>
+          <has_line line="Collection name: collection name" />
+      </assert_stdout>
+    </test>
+  </tests>
+</tool>
diff --git a/test/functional/tools/sample_tool_conf.xml b/test/functional/tools/sample_tool_conf.xml
index 20aa9225491..f557120cb62 100644
--- a/test/functional/tools/sample_tool_conf.xml
+++ b/test/functional/tools/sample_tool_conf.xml
@@ -215,6 +215,7 @@
   <tool file="collection_paired_or_list_paired_input.xml" />
   <tool file="collection_cat_group_tag.xml" />
   <tool file="collection_cat_group_tag_multiple.xml" />
+  <tool file="collection_field_test_two_files.xml" />
   <tool file="discover_sort_by.xml" />
   <tool file="discover_default_ext.xml" />
   <tool file="expression_forty_two.xml" />
diff --git a/test/unit/tool_util/test_test_definition_parsing.py b/test/unit/tool_util/test_test_definition_parsing.py
index ed5fbdaaee9..77cda95c737 100644
--- a/test/unit/tool_util/test_test_definition_parsing.py
+++ b/test/unit/tool_util/test_test_definition_parsing.py
@@ -58,6 +58,13 @@ COLLECTION_TYPE_SOURCE_EXPECTATIONS = [
 BIGWIG_TO_WIG_EXPECTATIONS = [
     (["inputs", "chrom"], "chr21"),
 ]
+RECORD_TWO_FILES_EXPECTATIONS = [
+    (["inputs", "f1", "model_class"], "TestCollectionDef"),
+    (["inputs", "f1", "collection_type"], "record"),
+    (["inputs", "f1", "fields", 0, "type"], "File"),
+    (["inputs", "f1", "fields", 0, "name"], "parent"),
+    (["inputs", "f1", "fields", 1, "type"], "File"),
+]
 
 
 class TestTestParsing(TestCase):
@@ -118,6 +125,12 @@ class TestTestParsing(TestCase):
         test_0 = test_dicts[0].to_dict()
         assert test_0["error"] is True
 
+    def test_field_collection_inputs(self):
+        self._init_tool_for_path(functional_test_tool_path("collection_record_test_two_files.xml"))
+        test_dicts = self._parse_tests()
+        test_0 = test_dicts[0].to_dict()
+        self._verify_each(test_0, RECORD_TWO_FILES_EXPECTATIONS)
+
     def test_bigwigtowig_converter(self):
         # defines
         if in_packages():