diff --git a/client/src/components/Workflow/Editor/Draggable.vue b/client/src/components/Workflow/Editor/Draggable.vue
index bc883d56aa4..ed955f83ab1 100644
--- a/client/src/components/Workflow/Editor/Draggable.vue
+++ b/client/src/components/Workflow/Editor/Draggable.vue
@@ -28,6 +28,7 @@ const props = defineProps({
 
 const emit = defineEmits(["mousedown", "mouseup", "move", "dragstart", "start", "stop"]);
 
+let dragImg: HTMLImageElement | undefined;
 const draggable = ref();
 const size = reactive(useAnimationFrameSize(draggable));
 const transform: Ref<ZoomTransform> | undefined = inject("transform");
@@ -40,8 +41,16 @@ const onStart = (position: Position, event: DragEvent) => {
     emit("start");
     emit("mousedown", event);
     if (event.type == "dragstart") {
+        dragImg = document.createElement("img");
+        dragImg.src = "data:image/gif;base64,R0lGODlhAQABAIAAAAAAAP///yH5BAEAAAAALAAAAAABAAEAAAIBRAA7";
+        document.body.appendChild(dragImg);
         // I guess better than copy ?
         event.dataTransfer!.effectAllowed = "link";
+        try {
+            event.dataTransfer!.setDragImage(dragImg, 0, 0);
+        } catch (e) {
+            console.error(e);
+        }
         if (props.dragData) {
             event.dataTransfer!.setData("text/plain", JSON.stringify(props.dragData));
         }
@@ -66,6 +75,9 @@ const onMove = (position: Position, event: DragEvent) => {
 };
 
 const onEnd = (position: Position, event: DragEvent) => {
+    if (dragImg) {
+        document.body.removeChild(dragImg);
+    }
     emit("stop");
     emit("mouseup");
 };
diff --git a/client/src/components/Workflow/Editor/Node.vue b/client/src/components/Workflow/Editor/Node.vue
index 6fc29e70f17..2801d92a933 100644
--- a/client/src/components/Workflow/Editor/Node.vue
+++ b/client/src/components/Workflow/Editor/Node.vue
@@ -49,11 +49,13 @@
                     triggers="hover"
                     placement="bottom"
                     :show.sync="popoverShow">
-                    <Recommendations
-                        v-if="popoverShow"
-                        :step-id="id"
-                        :datatypes-mapper="datatypesMapper"
-                        @onCreate="onCreate" />
+                    <div>
+                        <Recommendations
+                            v-if="popoverShow"
+                            :step-id="id"
+                            :datatypes-mapper="datatypesMapper"
+                            @onCreate="onCreate" />
+                    </div>
                 </b-popover>
             </b-button-group>
             <i :class="iconClass" />
@@ -232,13 +234,7 @@ const invalidOutputs = computed(() => {
     });
 });
 const outputs = computed(() => {
-    let stepOutputs = props.step.outputs;
-    if (props.step.when) {
-        stepOutputs = stepOutputs.map((output) => {
-            return { ...output, optional: true };
-        });
-    }
-    return [...stepOutputs, ...invalidOutputs.value];
+    return [...props.step.outputs, ...invalidOutputs.value];
 });
 
 function onDragConnector(dragPosition: TerminalPosition, terminal: OutputTerminals) {
diff --git a/client/src/components/Workflow/Editor/NodeOutput.vue b/client/src/components/Workflow/Editor/NodeOutput.vue
index fe624685679..f6dd4d0589a 100644
--- a/client/src/components/Workflow/Editor/NodeOutput.vue
+++ b/client/src/components/Workflow/Editor/NodeOutput.vue
@@ -15,6 +15,7 @@ import {
 } from "@/stores/workflowStepStore";
 import { assertDefined, ensureDefined } from "@/utils/assertions";
 import type { UseScrollReturn } from "@vueuse/core";
+import { NULL_COLLECTION_TYPE_DESCRIPTION, type CollectionTypeDescriptor } from "./modules/collectionTypeDescription";
 
 const props = defineProps<{
     output: OutputTerminalSource;
@@ -204,6 +205,50 @@ const terminalClass = computed(() => {
     return cls;
 });
 
+function collectionTypeToDescription(collectionTypeDescription: CollectionTypeDescriptor) {
+    let collectionDescription = collectionTypeDescription.collectionType;
+    if (
+        collectionTypeDescription &&
+        collectionTypeDescription.isCollection &&
+        collectionTypeDescription.collectionType
+    ) {
+        // we'll give a prettier label to the must common nested lists
+        switch (collectionTypeDescription.collectionType) {
+            case "list:paired": {
+                collectionDescription = "list of pairs dataset collection";
+                break;
+            }
+            case "list:list": {
+                collectionDescription = "list of lists dataset collection";
+                break;
+            }
+            default: {
+                if (collectionTypeDescription.rank > 1) {
+                    collectionDescription = `dataset collection with ${collectionTypeDescription.rank} levels of nesting`;
+                }
+                break;
+            }
+        }
+    }
+    return collectionDescription;
+}
+
+const outputDetails = computed(() => {
+    let collectionType = "collectionType" in terminal.value && terminal.value.collectionType;
+    const outputType =
+        collectionType && collectionType.isCollection && collectionType.collectionType
+            ? `output is ${collectionTypeToDescription(collectionType)}`
+            : `output is dataset`;
+    if (isMultiple.value) {
+        if (!collectionType) {
+            collectionType = NULL_COLLECTION_TYPE_DESCRIPTION;
+        }
+        const effectiveOutputType = terminal.value.mapOver.append(collectionType);
+        return `${outputType} and mapped-over to produce a ${collectionTypeToDescription(effectiveOutputType)} `;
+    }
+    return outputType;
+});
+
 onBeforeUnmount(() => {
     stateStore.deleteOutputTerminalPosition(props.stepId, props.output.name);
 });
@@ -245,6 +290,7 @@ onBeforeUnmount(() => {
             @move="onMove">
             <div
                 ref="icon"
+                v-b-tooltip.hover="outputDetails"
                 class="icon prevent-zoom"
                 tabindex="0"
                 :aria-label="`Connect output ${output.name} to input. Press space to see a list of available inputs`"
diff --git a/client/src/components/Workflow/Editor/composables/useTerminal.ts b/client/src/components/Workflow/Editor/composables/useTerminal.ts
index 7fada01f8ec..2ef2a00388f 100644
--- a/client/src/components/Workflow/Editor/composables/useTerminal.ts
+++ b/client/src/components/Workflow/Editor/composables/useTerminal.ts
@@ -11,10 +11,11 @@ export function useTerminal(
 ) {
     const terminal: Ref<ReturnType<typeof terminalFactory> | null> = ref(null);
     const stepStore = useWorkflowStepStore();
+    const step = computed(() => stepStore.getStep(stepId.value));
     const isMappedOver = computed(() => stepStore.stepMapOver[stepId.value]?.isCollection ?? false);
 
     watch(
-        [stepId, terminalSource, datatypesMapper],
+        [step, terminalSource, datatypesMapper],
         () => {
             // rebuild terminal if any of the tracked dependencies change
             const newTerminal = terminalFactory(stepId.value, terminalSource.value, datatypesMapper.value);
diff --git a/client/src/components/Workflow/Editor/modules/terminals.test.ts b/client/src/components/Workflow/Editor/modules/terminals.test.ts
index fd14fce4933..f789d8094da 100644
--- a/client/src/components/Workflow/Editor/modules/terminals.test.ts
+++ b/client/src/components/Workflow/Editor/modules/terminals.test.ts
@@ -380,6 +380,18 @@ describe("canAccept", () => {
         dataInTwo.disconnect(dataOut);
         expect(dataIn.mapOver).toEqual(NULL_COLLECTION_TYPE_DESCRIPTION);
     });
+    it("maintains step map over state when disconnecting output", () => {
+        const listListOut = terminals["list:list input"]["output"] as OutputCollectionTerminal;
+        const filterFailedInput = terminals["filter_failed"]["input"] as InputCollectionTerminal;
+        const filterFailedOutput = terminals["filter_failed"]["output"] as OutputCollectionTerminal;
+        const dataIn = terminals["simple data"]["input"] as InputTerminal;
+        filterFailedInput.connect(listListOut);
+        dataIn.connect(filterFailedOutput);
+        expect(filterFailedInput.isMappedOver()).toBe(true);
+        expect(stepStore.stepMapOver[filterFailedOutput.stepId].isCollection).toBe(true);
+        dataIn.disconnect(filterFailedOutput);
+        expect(stepStore.stepMapOver[filterFailedOutput.stepId].isCollection).toBe(true);
+    });
     it("rejects connecting incompatible connection types", () => {
         const pairedOut = terminals["paired input"]!["output"] as OutputCollectionTerminal;
         const collectionIn = terminals["list collection input"]!["input1"] as InputCollectionTerminal;
diff --git a/client/src/components/Workflow/Editor/modules/terminals.ts b/client/src/components/Workflow/Editor/modules/terminals.ts
index 1ce23b5606f..4949969b76e 100644
--- a/client/src/components/Workflow/Editor/modules/terminals.ts
+++ b/client/src/components/Workflow/Editor/modules/terminals.ts
@@ -239,7 +239,12 @@ class BaseInputTerminal extends Terminal {
                 const terminalSource = step.outputs[0];
                 if (terminalSource) {
                     const terminal = terminalFactory(step.id, terminalSource, this.datatypesMapper);
+                    // drop mapping restrictions
                     terminal.resetMappingIfNeeded();
+                    // re-establish map over through inputs
+                    step.inputs.forEach((input) => {
+                        terminalFactory(step.id, input, this.datatypesMapper).getStepMapOver();
+                    });
                 }
             } else {
                 console.error(`Invalid step. Could not fine step with id ${stepId} in store.`);
@@ -359,9 +364,6 @@ class BaseInputTerminal extends Terminal {
                 });
             }
             const postJobActionKey = `ChangeDatatypeAction${connection.output.name}`;
-            if (outputStep.when) {
-                terminalSource = { ...terminalSource, optional: true };
-            }
             if (
                 "extensions" in terminalSource &&
                 outputStep.post_job_actions &&
@@ -626,7 +628,7 @@ class BaseOutputTerminal extends Terminal {
     constructor(attr: BaseOutputTerminalArgs) {
         super(attr);
         this.datatypes = attr.datatypes;
-        this.optional = attr.optional;
+        this.optional = attr.optional || Boolean(this.stepStore.getStep(this.stepId)?.when);
         this.terminalType = "output";
     }
     getConnectedTerminals(): InputTerminalsAndInvalid[] {
@@ -743,7 +745,15 @@ export class OutputCollectionTerminal extends BaseOutputTerminal {
                                 otherCollectionType.canMapOver(collectionType)
                         );
                         if (connectedCollectionType) {
-                            return connectedCollectionType;
+                            if (connectedCollectionType.collectionType === "any") {
+                                // if the input collection type is "any" this output's collection type
+                                // is exactly the same as the connected output
+                                return otherCollectionType;
+                            } else {
+                                // else we pick the matching input collection type
+                                // so that the map over logic applies correctly
+                                return connectedCollectionType;
+                            }
                         }
                     }
                 }
diff --git a/client/src/schema/schema.ts b/client/src/schema/schema.ts
index 41e24cab40c..4d987e8084b 100644
--- a/client/src/schema/schema.ts
+++ b/client/src/schema/schema.ts
@@ -3767,6 +3767,8 @@ export interface components {
              * @description The current state of this dataset.
              */
             state: components["schemas"]["galaxy__model__Dataset__states"];
+            /** Tags */
+            tags: string[];
         };
         /**
          * HDASummary
diff --git a/client/src/utils/navigation/navigation.yml b/client/src/utils/navigation/navigation.yml
index 962e809c055..fa355ca5a49 100644
--- a/client/src/utils/navigation/navigation.yml
+++ b/client/src/utils/navigation/navigation.yml
@@ -11,6 +11,7 @@ _:  # global stuff
   selectors:
     editable_text: '.editable-text'
     tooltip_balloon: '.tooltip'
+    tooltip_inner: .tooltip-inner
     left_panel_drag: '#left > .unified-panel-footer > .drag'
     left_panel_collapse: '#left > .unified-panel-footer > .panel-collapse'
     right_panel_drag: '#right > .unified-panel-footer > .drag'
diff --git a/config/plugins/webhooks/demo/search/styles.css b/config/plugins/webhooks/demo/search/styles.css
index 58c408b4302..9f56d5da63d 100644
--- a/config/plugins/webhooks/demo/search/styles.css
+++ b/config/plugins/webhooks/demo/search/styles.css
@@ -4,7 +4,7 @@
 
 .search-screen {
     position: fixed;
-    z-index: 202;
+    z-index: 5100;
     width:100%;
     height:100%;
     display:none;
@@ -15,7 +15,7 @@
     top:0;
     left:0;
     background: rgba(224, 224, 224, 0.75);
-    z-index: 201;
+    z-index: 5000;
     width:100%;
     height:100%;
     display:none;
@@ -64,7 +64,7 @@
     position: fixed;
     top: 34px;
     width: 100%;
-    z-index: 203;
+    z-index: 5200;
 }
 
 .txtbx-search-data {
diff --git a/config/plugins/webhooks/gtn/styles.css b/config/plugins/webhooks/gtn/styles.css
index 9bb1d6bbf93..a4d3fe8f478 100644
--- a/config/plugins/webhooks/gtn/styles.css
+++ b/config/plugins/webhooks/gtn/styles.css
@@ -1,6 +1,6 @@
 #gtn-screen {
 	position: fixed;
-	z-index: 202;
+	z-index: 5100;
 	width:100%;
 	height:100%;
 }
@@ -10,7 +10,7 @@
 	top:0;
 	left:0;
 	background: rgba(224, 224, 224, 0.75);
-	z-index: 201;
+	z-index: 5000;
 	width:100%;
 	height:100%;
 	opacity: 2;
@@ -20,7 +20,7 @@
 	position: fixed;
 	height: 100%;
 	width: 100%;
-	z-index: 203;
+	z-index: 5200;
 	display: flex;
 	justify-content: center;
 	flex-direction: column;
diff --git a/config/plugins/webhooks/news/styles.css b/config/plugins/webhooks/news/styles.css
index 7ac3231209d..55b8ce1d996 100644
--- a/config/plugins/webhooks/news/styles.css
+++ b/config/plugins/webhooks/news/styles.css
@@ -1,6 +1,6 @@
 #news-screen {
 	position: fixed;
-	z-index: 202;
+	z-index: 5100;
 	width:100%;
 	height:100%;
 }
@@ -10,7 +10,7 @@
 	top:0;
 	left:0;
 	background: rgba(224, 224, 224, 0.75);
-	z-index: 201;
+	z-index: 5000;
 	width:100%;
 	height:100%;
 	opacity: 2;
@@ -20,7 +20,7 @@
 	position: fixed;
 	height: 100%;
 	width: 100%;
-	z-index: 203;
+	z-index: 5200;
 	display: flex;
 	justify-content: center;
 	flex-direction: column;
diff --git a/lib/galaxy/dependencies/pinned-requirements.txt b/lib/galaxy/dependencies/pinned-requirements.txt
index b8e598287df..d4b5e379665 100644
--- a/lib/galaxy/dependencies/pinned-requirements.txt
+++ b/lib/galaxy/dependencies/pinned-requirements.txt
@@ -73,7 +73,7 @@ fsspec==2023.1.0 ; python_version >= "3.7" and python_version < "3.12"
 future==0.18.3 ; python_version >= "3.7" and python_version < "3.12"
 galaxy-sequence-utils==1.1.5 ; python_version >= "3.7" and python_version < "3.12"
 galaxy2cwl==0.1.4 ; python_version >= "3.7" and python_version < "3.12"
-gravity==1.0.1 ; python_version >= "3.7" and python_version < "3.12"
+gravity==1.0.2 ; python_version >= "3.7" and python_version < "3.12"
 greenlet==2.0.2 ; python_version >= "3.7" and platform_machine == "aarch64" and python_version < "3.12" or python_version >= "3.7" and platform_machine == "ppc64le" and python_version < "3.12" or python_version >= "3.7" and platform_machine == "x86_64" and python_version < "3.12" or python_version >= "3.7" and platform_machine == "amd64" and python_version < "3.12" or python_version >= "3.7" and platform_machine == "AMD64" and python_version < "3.12" or python_version >= "3.7" and platform_machine == "win32" and python_version < "3.12" or python_version >= "3.7" and platform_machine == "WIN32" and python_version < "3.12"
 gunicorn==20.1.0 ; python_version >= "3.7" and python_version < "3.12"
 gxformat2==0.17.0 ; python_version >= "3.7" and python_version < "3.12"
diff --git a/lib/galaxy/jobs/runners/aws.py b/lib/galaxy/jobs/runners/aws.py
index 9a7278ea668..bed35ec19c3 100644
--- a/lib/galaxy/jobs/runners/aws.py
+++ b/lib/galaxy/jobs/runners/aws.py
@@ -597,6 +597,7 @@ class AWSBatchJobRunner(AsynchronousJobRunner):
             "exit_code_path": exit_code_path,
             "working_directory": job_wrapper.working_directory,
             "shell": job_wrapper.shell,
+            "galaxy_virtual_env": None,
         }
         job_file_contents = self.get_job_file(job_wrapper, **job_script_props)
         self.write_executable_script(job_file, job_file_contents, job_io=job_wrapper.job_io)
diff --git a/lib/galaxy/managers/cloud.py b/lib/galaxy/managers/cloud.py
index 0c2edcf2d1d..671ec216989 100644
--- a/lib/galaxy/managers/cloud.py
+++ b/lib/galaxy/managers/cloud.py
@@ -293,7 +293,7 @@ class CloudManager(sharable.SharableModelManager):
             incoming = params.__dict__
             history = trans.sa_session.query(trans.app.model.History).get(history_id)
             if not history:
-                raise ObjectNotFound(f"History with ID `{trans.app.security.encode_id(history_id)}` not found.")
+                raise ObjectNotFound("History with the ID provided was not found.")
             output = trans.app.toolbox.get_tool("upload1").handle_input(trans, incoming, history=history)
 
             job_errors = output.get("job_errors", [])
@@ -360,7 +360,7 @@ class CloudManager(sharable.SharableModelManager):
 
         history = trans.sa_session.query(trans.app.model.History).get(history_id)
         if not history:
-            raise ObjectNotFound(f"History with ID `{trans.app.security.encode_id(history_id)}` not found.")
+            raise ObjectNotFound("History with the provided ID not found.")
 
         sent = []
         failed = []
diff --git a/lib/galaxy/managers/collections.py b/lib/galaxy/managers/collections.py
index 1cd673b50c0..a338532d2a9 100644
--- a/lib/galaxy/managers/collections.py
+++ b/lib/galaxy/managers/collections.py
@@ -786,7 +786,7 @@ class DatasetCollectionManager:
             instance_id
         )
         if not collection_instance:
-            raise RequestParameterInvalidException(f"History dataset collection association {id} not found")
+            raise RequestParameterInvalidException("History dataset collection association not found")
         history = getattr(trans, "history", collection_instance.history)
         if check_ownership:
             self.history_manager.error_unless_owner(collection_instance.history, trans.user, current_history=history)
@@ -808,7 +808,7 @@ class DatasetCollectionManager:
             instance_id
         )
         if not collection_instance:
-            raise RequestParameterInvalidException(f"Library dataset collection association {id} not found")
+            raise RequestParameterInvalidException("Library dataset collection association not found")
         if check_accessible:
             if not trans.app.security_agent.can_access_library_item(
                 trans.get_current_user_roles(), collection_instance, trans.user
diff --git a/lib/galaxy/managers/collections_util.py b/lib/galaxy/managers/collections_util.py
index dcf5c68c4f9..60b47c3cdff 100644
--- a/lib/galaxy/managers/collections_util.py
+++ b/lib/galaxy/managers/collections_util.py
@@ -178,6 +178,7 @@ def dictify_element_reference(element, rank_fuzzy_counts=None, recursive=True, s
             object_details["state"] = element_object.state
             object_details["hda_ldda"] = "hda"
             object_details["history_id"] = element_object.history_id
+            object_details["tags"] = element_object.make_tag_string_list()
 
         dictified["object"] = object_details
     else:
diff --git a/lib/galaxy/managers/group_roles.py b/lib/galaxy/managers/group_roles.py
index 630b068f886..12e6d4c3698 100644
--- a/lib/galaxy/managers/group_roles.py
+++ b/lib/galaxy/managers/group_roles.py
@@ -7,7 +7,6 @@ from typing import (
 from galaxy import model
 from galaxy.exceptions import ObjectNotFound
 from galaxy.managers.context import ProvidesAppContext
-from galaxy.schema.fields import DecodedDatabaseIdField
 from galaxy.structured_app import MinimalManagerApp
 
 log = logging.getLogger(__name__)
@@ -63,13 +62,13 @@ class GroupRolesManager:
     def _get_group(self, trans: ProvidesAppContext, group_id: int) -> model.Group:
         group = trans.sa_session.query(model.Group).get(group_id)
         if not group:
-            raise ObjectNotFound(f"Group with id {DecodedDatabaseIdField.encode(group_id)} was not found.")
+            raise ObjectNotFound("Group with the id provided was not found.")
         return group
 
     def _get_role(self, trans: ProvidesAppContext, role_id: int) -> model.Role:
         role = trans.sa_session.query(model.Role).get(role_id)
         if not role:
-            raise ObjectNotFound(f"Role with id {DecodedDatabaseIdField.encode(role_id)} was not found.")
+            raise ObjectNotFound("Role with the id provided was not found.")
         return role
 
     def _get_group_role(
diff --git a/lib/galaxy/managers/group_users.py b/lib/galaxy/managers/group_users.py
index 25451679df3..9ceff766ba9 100644
--- a/lib/galaxy/managers/group_users.py
+++ b/lib/galaxy/managers/group_users.py
@@ -7,7 +7,6 @@ from typing import (
 from galaxy import model
 from galaxy.exceptions import ObjectNotFound
 from galaxy.managers.context import ProvidesAppContext
-from galaxy.schema.fields import DecodedDatabaseIdField
 from galaxy.structured_app import MinimalManagerApp
 
 log = logging.getLogger(__name__)
@@ -63,13 +62,13 @@ class GroupUsersManager:
     def _get_group(self, trans: ProvidesAppContext, group_id: int) -> model.Group:
         group = trans.sa_session.query(model.Group).get(group_id)
         if group is None:
-            raise ObjectNotFound(f"Group with id {DecodedDatabaseIdField.encode(group_id)} was not found.")
+            raise ObjectNotFound("Group with the id provided was not found.")
         return group
 
     def _get_user(self, trans: ProvidesAppContext, user_id: int) -> model.User:
         user = trans.sa_session.query(model.User).get(user_id)
         if user is None:
-            raise ObjectNotFound(f"User with id {DecodedDatabaseIdField.encode(user_id)} was not found.")
+            raise ObjectNotFound("User with the id provided was not found.")
         return user
 
     def _get_group_user(
diff --git a/lib/galaxy/managers/groups.py b/lib/galaxy/managers/groups.py
index a617b70dcd1..8566aff0a2f 100644
--- a/lib/galaxy/managers/groups.py
+++ b/lib/galaxy/managers/groups.py
@@ -104,7 +104,7 @@ class GroupsManager:
     def _get_group(self, sa_session: galaxy_scoped_session, group_id: int) -> model.Group:
         group = sa_session.query(model.Group).get(group_id)
         if group is None:
-            raise ObjectNotFound(f"Group with id {DecodedDatabaseIdField.encode(group_id)} was not found.")
+            raise ObjectNotFound("Group with the provided id was not found.")
         return group
 
     def _get_users_by_encoded_ids(
diff --git a/lib/galaxy/model/store/ro_crate_utils.py b/lib/galaxy/model/store/ro_crate_utils.py
index 0efbdddefb6..d990af972e6 100644
--- a/lib/galaxy/model/store/ro_crate_utils.py
+++ b/lib/galaxy/model/store/ro_crate_utils.py
@@ -3,7 +3,6 @@ import os
 from typing import (
     Any,
     Dict,
-    Optional,
 )
 
 from rocrate.model.computationalworkflow import (
@@ -11,28 +10,47 @@ from rocrate.model.computationalworkflow import (
     WorkflowDescription,
 )
 from rocrate.model.contextentity import ContextEntity
+from rocrate.model.file import File
 from rocrate.model.softwareapplication import SoftwareApplication
 from rocrate.rocrate import ROCrate
 
 from galaxy.model import (
-    JobParameter,
-    JobToInputDatasetAssociation,
-    JobToOutputDatasetAssociation,
+    HistoryDatasetAssociation,
+    HistoryDatasetCollectionAssociation,
     Workflow,
     WorkflowInvocation,
     WorkflowInvocationStep,
-    WorkflowRequestInputStepParameter,
-    WorkflowStep,
 )
 
 logger = logging.getLogger(__name__)
 
 
+PROFILES_VERSION = "0.1"
+WROC_PROFILE_VERSION = "1.0"
+
+GALAXY_EXPORT_VERSION = "2.0"
+
+ATTRS_FILENAME_HISTORY = "history_attrs.txt"
+ATTRS_FILENAME_DATASETS = "datasets_attrs.txt"
+ATTRS_FILENAME_JOBS = "jobs_attrs.txt"
+ATTRS_FILENAME_IMPLICIT_COLLECTION_JOBS = "implicit_collection_jobs_attrs.txt"
+ATTRS_FILENAME_COLLECTIONS = "collections_attrs.txt"
+ATTRS_FILENAME_EXPORT = "export_attrs.txt"
+ATTRS_FILENAME_LIBRARIES = "libraries_attrs.txt"
+ATTRS_FILENAME_LIBRARY_FOLDERS = "library_folders_attrs.txt"
+ATTRS_FILENAME_INVOCATIONS = "invocation_attrs.txt"
+
+
 class WorkflowRunCrateProfileBuilder:
     def __init__(self, model_store: Any):
         self.model_store = model_store
         self.invocation: WorkflowInvocation = model_store.included_invocations[0]
         self.workflow: Workflow = self.invocation.workflow
+        self.collection_type_mapping = {
+            "list": "https://training.galaxyproject.org/training-material/faqs/galaxy/collections_build_list.html",
+            "paired": "https://training.galaxyproject.org/training-material/faqs/galaxy/collections_build_list_paired.html",
+            None: "https://training.galaxyproject.org/training-material/faqs/galaxy/collections_build_list.html",
+        }
         self.param_type_mapping = {
             "integer": "Integer",
             "text": "Text",
@@ -54,7 +72,7 @@ class WorkflowRunCrateProfileBuilder:
             "rules": "Text",
             "directory_uri": "URI",
             "drill_down": "Text",
-            None: "None",
+            None: "Text",
         }
 
         self.ignored_parameter_type = [
@@ -67,88 +85,148 @@ class WorkflowRunCrateProfileBuilder:
             "dbkey",
             "__input_ext",
         ]
+        self.workflow_entities: Dict[int, Any] = {}
+        self.collection_entities: Dict[int, Any] = {}
+        self.file_entities: Dict[int, Any] = {}
+        self.param_entities: Dict[int, Any] = {}
         self.pv_entities: Dict[str, Any] = {}
 
     def build_crate(self):
         crate = ROCrate()
-        file_entities = self._add_files(crate)
-        self._add_collections(crate, file_entities)
         self._add_workflows(crate)
         self._add_engine_run(crate)
-        self._add_actions(crate, file_entities)
+        self._add_create_action(crate)
+        self._add_collections(crate)
+        self._add_files(crate)
+        self._add_profiles(crate)
+        self._add_parameters(crate)
+        self._add_attrs_files(crate)
         return crate
 
-    def _add_files(self, crate: ROCrate) -> Dict[int, Any]:
-        file_entities: Dict[int, Any] = {}
-        for dataset, _ in self.model_store.included_datasets.values():
-            if dataset.dataset.id in self.model_store.dataset_id_to_path:
-                filename, _ = self.model_store.dataset_id_to_path[dataset.dataset.id]
-                if not filename:
-                    filename = f"datasets/dataset_{dataset.dataset.uuid}"
-                name = dataset.name
-                encoding_format = dataset.datatype.get_mime()
-                properties = {
-                    "name": name,
-                    "encodingFormat": encoding_format,
-                    "exampleOfWork": {"@id": dataset.dataset.uuid.urn},
-                }
-                file_entity = crate.add_file(
-                    filename,
-                    dest_path=filename,
-                    properties=properties,
-                )
-                file_entities[dataset.dataset.id] = file_entity
-        return file_entities
+    def _add_file(self, dataset: HistoryDatasetAssociation, properties: Dict[Any, Any], crate: ROCrate) -> File:
+        if dataset.dataset.id in self.model_store.dataset_id_to_path:
+            filename, _ = self.model_store.dataset_id_to_path[dataset.dataset.id]
+            if not filename:
+                filename = f"datasets/dataset_{dataset.dataset.uuid}"
+            name = dataset.name
+            encoding_format = dataset.datatype.get_mime()
+            properties["name"] = name
+            properties["encodingFormat"] = encoding_format
+            file_entity = crate.add_file(
+                filename,
+                dest_path=filename,
+                properties=properties,
+            )
+            self.file_entities[dataset.dataset.id] = file_entity
 
-    def _add_collections(self, crate: ROCrate, file_entities: Dict[int, Any]) -> Dict[int, Any]:
-        collection_entities: Dict[int, Any] = {}
-        for collection in self.model_store.included_collections:
-            name = collection.name
-            dataset_ids = []
-            for dataset in collection.dataset_instances:
-                if dataset.dataset:
-                    dataset_id = file_entities.get(dataset.dataset.id)
-                    if dataset_id:
+            return file_entity
+
+    def _add_files(self, crate: ROCrate):
+        for wfda in self.invocation.input_datasets:
+            if not self.file_entities.get(wfda.dataset.dataset.id):
+                properties = {
+                    "exampleOfWork": {"@id": f"#{wfda.dataset.dataset.uuid}"},
+                }
+                file_entity = self._add_file(wfda.dataset, properties, crate)
+                dataset_formal_param = self._add_dataset_formal_parameter(wfda.dataset, crate)
+                crate.mainEntity.append_to("input", dataset_formal_param)
+                self.create_action.append_to("object", file_entity)
+
+        for wfda in self.invocation.output_datasets:
+            if not self.file_entities.get(wfda.dataset.dataset.id):
+                properties = {
+                    "exampleOfWork": {"@id": f"#{wfda.dataset.dataset.uuid}"},
+                }
+                file_entity = self._add_file(wfda.dataset, properties, crate)
+                dataset_formal_param = self._add_dataset_formal_parameter(wfda.dataset, crate)
+                crate.mainEntity.append_to("output", dataset_formal_param)
+                self.create_action.append_to("result", file_entity)
+
+    def _add_collection(self, hdca: HistoryDatasetCollectionAssociation, crate: ROCrate) -> ContextEntity:
+        name = hdca.name
+        dataset_ids = []
+        for hda in hdca.dataset_instances:
+            if hda.dataset:
+                properties: Dict[Any, Any] = {}
+                self._add_file(hda, properties, crate)
+                dataset_id = self.file_entities.get(hda.dataset.id)
+                if dataset_id:
+                    if {"@id": dataset_id.id} not in dataset_ids:
                         dataset_ids.append({"@id": dataset_id.id})
 
-            properties = {
-                "name": name,
-                "@type": "Collection",
-                "additionalType": collection.collection.collection_type,
-                "hasPart": dataset_ids,
-            }
-            collection_entity = crate.add(
-                ContextEntity(
-                    crate,
-                    collection.type_id,
-                    properties=properties,
-                )
+        collection_properties = {
+            "name": name,
+            "@type": "Collection",
+            "additionalType": self.collection_type_mapping[hdca.collection.collection_type],
+            "hasPart": dataset_ids,
+            "exampleOfWork": {"@id": f"#{hdca.type_id}-param"},
+        }
+        collection_entity = crate.add(
+            ContextEntity(
+                crate,
+                hdca.type_id,
+                properties=collection_properties,
             )
-            collection_entities[collection.collection.id] = collection_entity
+        )
+        self.collection_entities[hdca.collection.id] = collection_entity
 
-        crate.root_dataset["mentions"] = [{"@id": coll.id} for coll in collection_entities.values() if coll]
-        return collection_entities
+        crate.root_dataset.append_to("mentions", collection_entity)
+
+        return collection_entity
+
+    def _add_collections(self, crate: ROCrate):
+        for wfdca in self.invocation.input_dataset_collections:
+            collection_entity = self._add_collection(wfdca.dataset_collection, crate)
+            collection_formal_param = self._add_collection_formal_parameter(wfdca.dataset_collection, crate)
+            crate.mainEntity.append_to("input", collection_formal_param)
+            self.create_action.append_to("object", collection_entity)
+
+        for wfdca in self.invocation.output_dataset_collections:
+            collection_entity = self._add_collection(wfdca.dataset_collection, crate)
+            collection_formal_param = self._add_collection_formal_parameter(wfdca.dataset_collection, crate)
+            crate.mainEntity.append_to("output", collection_formal_param)
+            self.create_action.append_to("result", collection_entity)
 
     def _add_workflows(self, crate: ROCrate):
         workflows_directory = self.model_store.workflows_directory
 
         if os.path.exists(workflows_directory):
             for filename in os.listdir(workflows_directory):
+                is_main_wf = filename.endswith(".gxwf.yml")
                 is_computational_wf = not filename.endswith(".cwl")
                 workflow_cls = ComputationalWorkflow if is_computational_wf else WorkflowDescription
                 lang = "galaxy" if not filename.endswith(".cwl") else "cwl"
                 dest_path = os.path.join("workflows", filename)
 
-                crate.add_workflow(
+                wf = crate.add_workflow(
                     source=os.path.join(workflows_directory, filename),
                     dest_path=dest_path,
-                    main=is_computational_wf,
+                    main=is_main_wf,
                     cls=workflow_cls,
                     lang=lang,
                 )
+                self.workflow_entities[wf.id] = wf
+                if lang == "cwl":
+                    cwl_wf = wf
 
             crate.license = self.workflow.license or ""
             crate.mainEntity["name"] = self.workflow.name
+            crate.mainEntity["subjectOf"] = cwl_wf
+
+    def _add_create_action(self, crate: ROCrate):
+        self.create_action = crate.add(
+            ContextEntity(
+                crate,
+                properties={
+                    "@type": "CreateAction",
+                    "name": self.workflow.name,
+                    "startTime": self.invocation.workflow.create_time.isoformat(),
+                    "endTime": self.invocation.workflow.update_time.isoformat(),
+                    "instrument": {"@id": crate.mainEntity["@id"]},
+                },
+            )
+        )
+        crate.root_dataset.append_to("mentions", self.create_action)
 
     def _add_engine_run(self, crate: ROCrate):
         roc_engine = crate.add(SoftwareApplication(crate, properties={"name": "Galaxy workflow engine"}))
@@ -159,91 +237,119 @@ class WorkflowRunCrateProfileBuilder:
                     "@type": "OrganizeAction",
                     "name": f"Run of {roc_engine['name']}",
                     "startTime": self.invocation.create_time.isoformat(),
+                    "endTime": self.invocation.update_time.isoformat(),
                 },
             )
         )
         roc_engine_run["instrument"] = roc_engine
         self.roc_engine_run = roc_engine_run
 
-    def _add_actions(self, crate: ROCrate, file_entities: Dict[int, Any]):
-        input_formal_params: Dict[int, Any] = {}
-        output_formal_params = []
-        workflow_inputs = []
-        workflow_outputs = []
+    def _add_attrs_files(self, crate: ROCrate):
+        targets = [
+            ATTRS_FILENAME_HISTORY,
+            ATTRS_FILENAME_DATASETS,
+            ATTRS_FILENAME_JOBS,
+            ATTRS_FILENAME_IMPLICIT_COLLECTION_JOBS,
+            ATTRS_FILENAME_COLLECTIONS,
+            ATTRS_FILENAME_EXPORT,
+            ATTRS_FILENAME_LIBRARIES,
+            ATTRS_FILENAME_LIBRARY_FOLDERS,
+            ATTRS_FILENAME_INVOCATIONS,
+        ]
+        for attrs in targets:
+            attrs_path = os.path.join(self.model_store.export_directory, attrs)
+            description = " ".join(attrs.split("_")[:-1])
+            if os.path.exists(attrs_path):
+                properties = {
+                    "@type": "File",
+                    "version": GALAXY_EXPORT_VERSION,
+                    "description": f"{description} properties",
+                    "encodingFormat": "application/json",
+                }
+                crate.add_file(
+                    attrs,
+                    dest_path=attrs,
+                    properties=properties,
+                )
 
-        for param in self.invocation.input_step_parameters:
-            property_value = self._add_wf_property_value(crate, param)
-            workflow_inputs.append(property_value)
-            formal_param = self._add_formal_parameter(crate, param.workflow_step)
-            input_formal_params[formal_param.id] = formal_param
+        prov_target = f"{ATTRS_FILENAME_DATASETS}.provenance"
+        provenance_attrs_path = os.path.join(self.model_store.export_directory, prov_target)
+        description = " ".join(prov_target.split("_")[:-1])
+        if os.path.exists(provenance_attrs_path):
+            crate.add(
+                ContextEntity(
+                    crate,
+                    prov_target,
+                    properties={
+                        "@type": "CreativeWork",
+                        "version": GALAXY_EXPORT_VERSION,
+                        "description": f"{description} provenance properties",
+                        "encodingFormat": "application/json",
+                    },
+                )
+            )
 
-        for output_step in self.invocation.steps:
-            for job in output_step.jobs:
-                for job_input in job.input_datasets:
-                    formal_param = self._add_formal_parameter_input(crate, job_input)
-                    input_formal_params[formal_param.id] = formal_param
-                    dataset_id = job_input.dataset.dataset.id
-                    input_file_entity = file_entities.get(dataset_id)
-                    workflow_inputs.append(input_file_entity)
-                for job_output in job.output_datasets:
-                    formal_param = self._add_formal_parameter_output(crate, job_output)
-                    output_formal_params.append(formal_param)
-                    dataset_id = job_output.dataset.dataset.id
-                    output_file_entity = file_entities.get(dataset_id)
-                    workflow_outputs.append(output_file_entity)
-                for param in job.parameters:
-                    if param.name not in self.ignored_parameter_type and not param.name.startswith("__"):
-                        property_value = self._add_job_property_value(crate, param)
-                        workflow_inputs.append(property_value)
-                        formal_param = self._add_formal_parameter(crate, output_step.workflow_step, param.name)
-                        input_formal_params[formal_param.id] = formal_param
-            if output_step.workflow_step.type == "parameter_input":
-                property_value = self._add_param_property_value(crate, output_step)
-                workflow_inputs.append(property_value)
-                formal_param = self._add_formal_parameter(crate, output_step.workflow_step)
-                input_formal_params[formal_param.id] = formal_param
-            if output_step.workflow_step.type == "tool":
-                for tool_input in output_step.workflow_step.tool_inputs.keys():
-                    if tool_input not in self.ignored_parameter_type and not tool_input.startswith("__"):
-                        property_value = self._add_tool_property_value(crate, output_step, tool_input)
-                        workflow_inputs.append(property_value)
-                        formal_param = self._add_formal_parameter(crate, output_step.workflow_step, tool_input)
-                        input_formal_params[formal_param.id] = formal_param
+    def _add_profiles(self, crate: ROCrate):
+        profiles = []
+        for p in "process", "workflow":
+            id_ = f"https://w3id.org/ro/wfrun/{p}/{PROFILES_VERSION}"
+            profiles.append(
+                crate.add(
+                    ContextEntity(
+                        crate,
+                        id_,
+                        properties={
+                            "@type": "CreativeWork",
+                            "name": f"{p.title()} Run Crate",
+                            "version": PROFILES_VERSION,
+                        },
+                    )
+                )
+            )
+        # FIXME: in the future, this could go out of sync with the wroc
+        # profile added by ro-crate-py to the metadata descriptor
+        wroc_profile_id = f"https://w3id.org/workflowhub/workflow-ro-crate/{WROC_PROFILE_VERSION}"
+        profiles.append(
+            crate.add(
+                ContextEntity(
+                    crate,
+                    wroc_profile_id,
+                    properties={
+                        "@type": "CreativeWork",
+                        "name": "Workflow RO-Crate",
+                        "version": WROC_PROFILE_VERSION,
+                    },
+                )
+            )
+        )
+        crate.root_dataset["conformsTo"] = profiles
 
-        wf_input_param_ids = [{"@id": entity.id} for entity in input_formal_params.values()]
-        crate.mainEntity["input"] = wf_input_param_ids
-        wf_input_ids = [{"@id": input.id} for input in workflow_inputs if input]
-        wf_output_param_ids = [{"@id": entity.id} for entity in output_formal_params]
-        crate.mainEntity["output"] = wf_output_param_ids
-        wf_output_ids = [{"@id": output.id} for output in workflow_outputs if output]
+    def _add_parameters(self, crate: ROCrate):
+        for step in self.invocation.steps:
+            if step.workflow_step.type == "parameter_input":
+                property_value = self._add_step_parameter_pv(step, crate)
+                formal_param = self._add_step_parameter_fp(step, crate)
+                crate.mainEntity.append_to("input", formal_param)
+                self.create_action.append_to("object", property_value)
 
-        input_param_value = crate.add(
+    def _add_step_parameter_pv(self, step: WorkflowInvocationStep, crate: ROCrate):
+        param_id = step.workflow_step.label
+        return crate.add(
             ContextEntity(
                 crate,
+                f"{param_id}-pv",
                 properties={
-                    "@type": "CreateAction",
-                    "name": self.workflow.name,
-                    "instrument": {"@id": crate.mainEntity["@id"]},
-                    "object": wf_input_ids,
-                    "result": wf_output_ids,
+                    "@type": "PropertyValue",
+                    "name": f"{param_id}",
+                    "value": step.output_value.value,
+                    "exampleOfWork": {"@id": f"#{param_id}-param"},
                 },
             )
         )
-        self.main_action = input_param_value
-
-    def _add_formal_parameter(self, crate: ROCrate, step: WorkflowStep, tool_input: Optional[str] = None):
-        param_id = ""
-        param_type = None
-        if not tool_input:
-            if step.output_connections:
-                param_id = step.output_connections[0].input_name
-            if step.annotations:
-                param_type = step.annotations[0].workflow_step.tool_inputs.get("parameter_type")
-            if step.tool_inputs:
-                param_type = param_type or step.tool_inputs.get("parameter_type")
-        else:
-            param_id = tool_input
 
+    def _add_step_parameter_fp(self, step: WorkflowInvocationStep, crate: ROCrate):
+        param_id = step.workflow_step.label
+        param_type = step.workflow_step.tool_inputs["parameter_type"]
         return crate.add(
             ContextEntity(
                 crate,
@@ -251,116 +357,74 @@ class WorkflowRunCrateProfileBuilder:
                 properties={
                     "@type": "FormalParameter",
                     "additionalType": self.param_type_mapping[param_type],
-                    "description": step.annotations[0].annotation if step.annotations else "",
-                    "name": f"{step.label} parameter",
-                    "valueRequired": not step.input_optional,
+                    "description": self._get_association_description(step.workflow_step),
+                    "name": f"{param_id}",
+                    "valueRequired": str(not step.workflow_step.input_optional),
                 },
             )
         )
 
-    def _add_formal_parameter_input(self, crate: ROCrate, input: JobToInputDatasetAssociation):
+    def _add_step_tool_pv(self, step: WorkflowInvocationStep, tool_input: str, crate: ROCrate):
+        param_id = tool_input
         return crate.add(
             ContextEntity(
                 crate,
-                input.dataset.dataset.uuid.urn,
+                f"{param_id}-pv",
                 properties={
-                    "@type": "FormalParameter",
-                    "additionalType": "File",  # TODO: always a dataset/File?
-                    "description": input.dataset.annotations[0].annotation
-                    if input.dataset.annotations
-                    else input.dataset.info or "",
-                    "name": input.name,
+                    "@type": "PropertyValue",
+                    "name": f"{step.workflow_step.label}",
+                    "value": step.workflow_step.tool_inputs[tool_input],
+                    "exampleOfWork": {"@id": f"#{param_id}-param"},
                 },
             )
         )
 
-    def _add_formal_parameter_output(self, crate: ROCrate, output: JobToOutputDatasetAssociation):
+    def _add_step_tool_fp(self, step: WorkflowInvocationStep, tool_input: str, crate: ROCrate):
+        param_id = tool_input
+        param_type = "text"
+        return ContextEntity(
+            crate,
+            f"{param_id}-param",
+            properties={
+                "@type": "FormalParameter",
+                "additionalType": self.param_type_mapping[param_type],
+                "description": self._get_association_description(step.workflow_step),
+                "name": f"{step.workflow_step.label}",
+                "valueRequired": str(not step.workflow_step.input_optional),
+            },
+        )
+
+    def _add_dataset_formal_parameter(self, hda: HistoryDatasetAssociation, crate: ROCrate):
         return crate.add(
             ContextEntity(
                 crate,
-                output.dataset.dataset.uuid.urn,
+                str(hda.dataset.uuid),
                 properties={
                     "@type": "FormalParameter",
-                    "additionalType": "File",  # TODO: always a dataset/File?
-                    "description": output.dataset.annotations[0].annotation
-                    if output.dataset.annotations
-                    else output.dataset.info or "",
-                    "name": output.name,
+                    "additionalType": "File",
+                    "description": self._get_association_description(hda),
+                    "name": hda.name,
                 },
             )
         )
 
-    def _add_job_property_value(self, crate: ROCrate, param: JobParameter):
-        input_name = param.name
-        if self.pv_entities.get(input_name):
-            return
-        job_pv = crate.add(
+    def _add_collection_formal_parameter(self, hdca: HistoryDatasetCollectionAssociation, crate: ROCrate):
+        return crate.add(
             ContextEntity(
                 crate,
-                f"{input_name}-pv",
+                f"{hdca.type_id}-param",
                 properties={
-                    "@type": "PropertyValue",
-                    "name": input_name,
-                    "value": param.value,
-                    "exampleOfWork": {"@id": f"#{input_name}-param"},
+                    "@type": "FormalParameter",
+                    "additionalType": "Collection",
+                    "description": self._get_association_description(hdca),
+                    "name": hdca.name,
                 },
             )
         )
-        self.pv_entities[input_name] = job_pv
-        return job_pv
 
-    def _add_wf_property_value(self, crate: ROCrate, param: WorkflowRequestInputStepParameter):
-        input_name = param.workflow_step.output_connections[0].input_name
-        if self.pv_entities.get(input_name):
-            return
-        wf_pv = crate.add(
-            ContextEntity(
-                crate,
-                f"{input_name}-pv",
-                properties={
-                    "@type": "PropertyValue",
-                    "name": input_name,
-                    "value": param.parameter_value,
-                    "exampleOfWork": {"@id": f"#{input_name}-param"},
-                },
-            )
-        )
-        self.pv_entities[input_name] = wf_pv
-        return wf_pv
-
-    def _add_tool_property_value(self, crate: ROCrate, invocation_step: WorkflowInvocationStep, tool_input: str):
-        if self.pv_entities.get(tool_input):
-            return
-        tool_pv = crate.add(
-            ContextEntity(
-                crate,
-                f"{tool_input}-pv",
-                properties={
-                    "@type": "PropertyValue",
-                    "name": tool_input,
-                    "value": invocation_step.workflow_step.tool_inputs[tool_input],
-                    "exampleOfWork": {"@id": f"#{tool_input}-param"},
-                },
-            )
-        )
-        self.pv_entities[tool_input] = tool_pv
-        return tool_pv
-
-    def _add_param_property_value(self, crate: ROCrate, invocation_step: WorkflowInvocationStep):
-        input_name = invocation_step.workflow_step.output_connections[0].input_name
-        if self.pv_entities.get(input_name):
-            return
-        param_pv = crate.add(
-            ContextEntity(
-                crate,
-                f"{input_name}-pv",
-                properties={
-                    "@type": "PropertyValue",
-                    "name": invocation_step.workflow_step.label,
-                    "value": invocation_step.output_value.value,
-                    "exampleOfWork": {"@id": f"#{input_name}-param"},
-                },
-            )
-        )
-        self.pv_entities[input_name] = param_pv
-        return param_pv
+    def _get_association_description(self, association: Any) -> str:
+        if hasattr(association, "annotations"):
+            return association.annotations[0].annotation if association.annotations else ""
+        elif hasattr(association, "info"):
+            return association.info
+        return ""
diff --git a/lib/galaxy/schema/schema.py b/lib/galaxy/schema/schema.py
index 7466e187f6b..cc76441d07f 100644
--- a/lib/galaxy/schema/schema.py
+++ b/lib/galaxy/schema/schema.py
@@ -647,6 +647,7 @@ class HDAObject(Model):
     state: Dataset.states = DatasetStateField
     hda_ldda: DatasetSourceType = HdaLddaField
     history_id: DecodedDatabaseIdField = HistoryIdField
+    tags: List[str]
 
     class Config:
         extra = Extra.allow  # Can contain more fields like metadata_*
diff --git a/lib/galaxy/tool_util/deps/container_resolvers/explicit.py b/lib/galaxy/tool_util/deps/container_resolvers/explicit.py
index 1809e8a2a5b..ce06d08f320 100644
--- a/lib/galaxy/tool_util/deps/container_resolvers/explicit.py
+++ b/lib/galaxy/tool_util/deps/container_resolvers/explicit.py
@@ -1,4 +1,5 @@
 """This module describes the :class:`ExplicitContainerResolver` ContainerResolver plugin."""
+import copy
 import logging
 import os
 from typing import cast
@@ -79,11 +80,10 @@ class CachedExplicitSingularityContainerResolver(CliContainerResolver):
         hence the container_description hack here.
         """
         for container_description in tool_info.container_descriptions:  # type: ContainerDescription
+            container_description = copy.copy(container_description)
             if container_description.type == "docker":
-                desc_dict = container_description.to_dict()
-                desc_dict["type"] = self.container_type
-                desc_dict["identifier"] = f"docker://{container_description.identifier}"
-                container_description = container_description.from_dict(desc_dict)
+                container_description.type = self.container_type
+                container_description.identifier = f"docker://{container_description.identifier}"
             if not self._container_type_enabled(container_description, enabled_container_types):
                 return None
             if not self.cli_available:
diff --git a/lib/galaxy/tool_util/xsd/galaxy.xsd b/lib/galaxy/tool_util/xsd/galaxy.xsd
index 7c75844cc1d..33224247363 100644
--- a/lib/galaxy/tool_util/xsd/galaxy.xsd
+++ b/lib/galaxy/tool_util/xsd/galaxy.xsd
@@ -2344,6 +2344,11 @@ $attribute_list::5
     </xs:annotation>
     <xs:attributeGroup ref="AssertAttributeN"/>
     <xs:attributeGroup ref="AssertAttributeNegate"/>
+    <xs:attribute name="comment" type="xs:string">
+      <xs:annotation>
+        <xs:documentation xml:lang="en">Comment character(s) used to skip comment lines (which should not be used for counting columns)</xs:documentation>
+      </xs:annotation>
+    </xs:attribute>
   </xs:complexType>
   <xs:complexType name="AssertHasArchiveMember">
     <xs:annotation>
diff --git a/lib/galaxy/tools/wrappers.py b/lib/galaxy/tools/wrappers.py
index 9b9dd56fcb6..04343c4d2af 100644
--- a/lib/galaxy/tools/wrappers.py
+++ b/lib/galaxy/tools/wrappers.py
@@ -47,7 +47,6 @@ if TYPE_CHECKING:
     from galaxy.model.metadata import MetadataCollection
     from galaxy.tools import Tool
     from galaxy.tools.parameters.basic import (
-        BaseDataToolParameter,
         SelectToolParameter,
         ToolParameter,
     )
@@ -357,22 +356,12 @@ class DatasetFilenameWrapper(ToolParameterValueWrapper):
         io_type: str = "input",
         formats: Optional[List[str]] = None,
     ) -> None:
+        dataset_instance: Optional[DatasetInstance] = None
         if not dataset:
-            dataset_instance: Optional[DatasetInstance] = None
-            ext = "data"
-            if tool is not None and name is not None:
-                try:
-                    tool_input = tool.inputs[name]
-                    if TYPE_CHECKING:
-                        assert isinstance(tool_input, BaseDataToolParameter)
-                    # TODO: allow this to work when working with grouping
-                    ext = tool_input.extensions[0]
-                except Exception:
-                    pass
             self.dataset = cast(
                 DatasetInstance,
                 wrap_with_safe_string(
-                    NoneDataset(datatypes_registry=datatypes_registry, ext=ext),
+                    NoneDataset(datatypes_registry=datatypes_registry),
                     no_wrap_classes=ToolParameterValueWrapper,
                 ),
             )
diff --git a/lib/galaxy/webapps/base/api.py b/lib/galaxy/webapps/base/api.py
index d87e2224593..2835a2340fa 100644
--- a/lib/galaxy/webapps/base/api.py
+++ b/lib/galaxy/webapps/base/api.py
@@ -152,6 +152,12 @@ class GalaxyFileResponse(FileResponse):
             await self.background()
 
 
+def add_sentry_middleware(app: FastAPI) -> None:
+    from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
+
+    app.add_middleware(SentryAsgiMiddleware)
+
+
 def get_error_response_for_request(request: Request, exc: MessageException) -> JSONResponse:
     error_dict = api_error_message(None, exception=exc)
     status_code = exc.status_code
diff --git a/lib/galaxy/webapps/galaxy/api/jobs.py b/lib/galaxy/webapps/galaxy/api/jobs.py
index 4dfb1bfc9e9..e558e5e2443 100644
--- a/lib/galaxy/webapps/galaxy/api/jobs.py
+++ b/lib/galaxy/webapps/galaxy/api/jobs.py
@@ -322,7 +322,7 @@ class JobController(BaseGalaxyAPIController, UsesVisualizationMixin):
         """
         job = self.__get_job(trans, id)
         if not job:
-            raise exceptions.ObjectNotFound(f"Could not access job with id '{id}'")
+            raise exceptions.ObjectNotFound("Could not access job with the given id")
         if job.state == job.states.PAUSED:
             job.resume()
         else:
@@ -417,7 +417,7 @@ class JobController(BaseGalaxyAPIController, UsesVisualizationMixin):
 
         job = self.__get_job(trans, id)
         if not job:
-            raise exceptions.ObjectNotFound(f"Could not access job with id '{id}'")
+            raise exceptions.ObjectNotFound("Could not access job with the given id")
         tool = self.app.toolbox.get_tool(job.tool_id, kwd.get("tool_version") or job.tool_version)
         if tool is None:
             raise exceptions.ObjectNotFound("Requested tool not found")
diff --git a/lib/galaxy/webapps/galaxy/api/tool_entry_points.py b/lib/galaxy/webapps/galaxy/api/tool_entry_points.py
index 832124c9397..f6590ef0734 100644
--- a/lib/galaxy/webapps/galaxy/api/tool_entry_points.py
+++ b/lib/galaxy/webapps/galaxy/api/tool_entry_points.py
@@ -96,8 +96,8 @@ class ToolEntryPointsAPIController(BaseGalaxyAPIController):
             entry_point_id = self.decode_id(id)
             entry_point = trans.sa_session.query(InteractiveToolEntryPoint).get(entry_point_id)
         except Exception:
-            raise exceptions.RequestParameterInvalidException("entry point '{id}' invalid")
+            raise exceptions.RequestParameterInvalidException("entry point invalid")
         if self.app.interactivetool_manager.can_access_entry_point(trans, entry_point):
             self.app.interactivetool_manager.stop(trans, entry_point)
         else:
-            raise exceptions.ItemAccessibilityException(f"entry point '{id}' is not accessible")
+            raise exceptions.ItemAccessibilityException("entry point is not accessible")
diff --git a/lib/galaxy/webapps/galaxy/api/users.py b/lib/galaxy/webapps/galaxy/api/users.py
index a431cd7b319..e5e2f928639 100644
--- a/lib/galaxy/webapps/galaxy/api/users.py
+++ b/lib/galaxy/webapps/galaxy/api/users.py
@@ -443,7 +443,7 @@ class UserAPIController(BaseGalaxyAPIController, UsesTagsMixin, BaseUIController
             if trans.user == user_to_update:
                 self.user_manager.delete(user_to_update)
             else:
-                raise exceptions.InsufficientPermissionsException("You may only delete your own account.", id=id)
+                raise exceptions.InsufficientPermissionsException("You may only delete your own account.")
         return self.user_serializer.serialize_to_view(user_to_update, view="detailed")
 
     @web.require_admin
@@ -1169,7 +1169,7 @@ class UserAPIController(BaseGalaxyAPIController, UsesTagsMixin, BaseUIController
     def _get_user(self, trans, id):
         user = self.get_user(trans, id)
         if not user:
-            raise exceptions.RequestParameterInvalidException(f"Invalid user ({id}).")
+            raise exceptions.RequestParameterInvalidException("Invalid user id specified.")
         if user != trans.user and not trans.user_is_admin:
             raise exceptions.InsufficientPermissionsException("Access denied.")
         return user
@@ -1196,9 +1196,9 @@ def get_user_full(trans: ProvidesUserContext, user_id: Union[FlexibleUserIdType,
         # check that the user is requesting themselves (and they aren't del'd) unless admin
         if not trans.user_is_admin:
             if trans.user != user or user.deleted:
-                raise exceptions.RequestParameterInvalidException("Invalid user id specified", id=user_id)
+                raise exceptions.RequestParameterInvalidException("Invalid user id specified")
         return user
     except exceptions.MessageException:
         raise
     except Exception:
-        raise exceptions.RequestParameterInvalidException("Invalid user id specified", id=user_id)
+        raise exceptions.RequestParameterInvalidException("Invalid user id specified")
diff --git a/lib/galaxy/webapps/galaxy/api/workflows.py b/lib/galaxy/webapps/galaxy/api/workflows.py
index fce6cc9530c..33794507d4c 100644
--- a/lib/galaxy/webapps/galaxy/api/workflows.py
+++ b/lib/galaxy/webapps/galaxy/api/workflows.py
@@ -699,7 +699,7 @@ class WorkflowsAPIController(
         try:
             stored_workflow = self.get_stored_workflow(trans, workflow_id, check_ownership=False)
         except Exception:
-            raise exceptions.ObjectNotFound(f"Malformed workflow id ( {workflow_id} ) specified.")
+            raise exceptions.ObjectNotFound("Malformed workflow id specified.")
         if stored_workflow.importable is False:
             raise exceptions.ItemAccessibilityException(
                 "The owner of this workflow has disabled imports via this link."
diff --git a/lib/galaxy/webapps/galaxy/services/base.py b/lib/galaxy/webapps/galaxy/services/base.py
index 75245809063..7cae6d5c060 100644
--- a/lib/galaxy/webapps/galaxy/services/base.py
+++ b/lib/galaxy/webapps/galaxy/services/base.py
@@ -1,3 +1,4 @@
+import mimetypes
 from tempfile import NamedTemporaryFile
 from typing import (
     Any,
@@ -135,14 +136,12 @@ class ServedExportStore(NamedTuple):
 def model_store_storage_target(
     short_term_storage_allocator: ShortTermStorageAllocator, file_name: str, model_store_format: str
 ) -> ShortTermStorageTarget:
-    cleaned_filename = ready_name_for_url(f"{file_name}.{model_store_format}")
-    if model_store_format.endswith("gz"):
-        mime_type = "application/x-gzip"
-    else:
-        mime_type = "application/x-tar"
+    cleaned_filename = ready_name_for_url(file_name)
+    filename_with_extension = f"{cleaned_filename}.{model_store_format}"
+    mime_type = mimetypes.guess_type(filename_with_extension)[0] or "application/octet-stream"
 
     return short_term_storage_allocator.new_target(
-        cleaned_filename,
+        filename_with_extension,
         mime_type,
     )
 
diff --git a/lib/galaxy/webapps/galaxy/services/history_contents.py b/lib/galaxy/webapps/galaxy/services/history_contents.py
index 65720ceec22..e2b16635896 100644
--- a/lib/galaxy/webapps/galaxy/services/history_contents.py
+++ b/lib/galaxy/webapps/galaxy/services/history_contents.py
@@ -1191,9 +1191,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto
         decoded_ldda_id = ldda_id
         ld = self.ldda_manager.get(trans, decoded_ldda_id)
         if type(ld) is not LibraryDataset:
-            raise exceptions.RequestParameterInvalidException(
-                f"Library content id ( {self.encode_id(ldda_id)} ) is not a dataset"
-            )
+            raise exceptions.RequestParameterInvalidException("Library content id is not a dataset")
         hda = ld.library_dataset_dataset_association.to_history_dataset_association(history, add_to_history=True)
         return hda
 
diff --git a/lib/galaxy_test/api/test_workflows.py b/lib/galaxy_test/api/test_workflows.py
index 368c8471b86..4d53810a9c8 100644
--- a/lib/galaxy_test/api/test_workflows.py
+++ b/lib/galaxy_test/api/test_workflows.py
@@ -2751,8 +2751,7 @@ steps:
       advanced:
         conflict:
           __current_case__: 0
-          duplicate_options: suffix_conflict
-          suffix_pattern: _#
+          duplicate_options: keep_first
       inputs:
       - __index__: 0
         input:
@@ -2816,26 +2815,28 @@ input collection 2:
             crate = self.workflow_populator.get_ro_crate(invocation_id, include_files=True)
             workflow = crate.mainEntity
             root = crate.root_dataset
-            assert len(root["mentions"]) == 3
+            assert len(root["mentions"]) == 4
             actions = [_ for _ in crate.contextual_entities if "CreateAction" in _.type]
             assert len(actions) == 1
             wf_action = actions[0]
             wf_objects = wf_action["object"]
-            assert len(workflow["input"]) == 7
-            assert len(workflow["output"]) == 2
+            assert len(workflow["input"]) == 3
+            assert len(workflow["output"]) == 3
             collections = [_ for _ in crate.contextual_entities if "Collection" in _.type]
             assert len(collections) == 3
             collection = collections[0]
-            assert collection["additionalType"] == "list"
+            assert (
+                collection["additionalType"]
+                == "https://training.galaxyproject.org/training-material/faqs/galaxy/collections_build_list.html"
+            )
             assert collection.type == "Collection"
             assert len(collection["hasPart"]) == 2
-            for dataset in collection["hasPart"]:
-                assert dataset in wf_objects
+            assert collection in wf_objects
 
             coll_dataset = collection["hasPart"][0].id
             assert coll_dataset in [_.id for _ in collections[2]["hasPart"]]
             property_values = [_ for _ in crate.contextual_entities if "PropertyValue" in _.type]
-            assert len(property_values) == 2
+            assert len(property_values) == 1
             for pv in property_values:
                 assert pv in wf_objects
                 assert pv["exampleOfWork"] in workflow["input"]
diff --git a/lib/galaxy_test/selenium/test_workflow_editor.py b/lib/galaxy_test/selenium/test_workflow_editor.py
index 27935713507..ab6c79333a7 100644
--- a/lib/galaxy_test/selenium/test_workflow_editor.py
+++ b/lib/galaxy_test/selenium/test_workflow_editor.py
@@ -1,4 +1,5 @@
 import json
+from typing import Optional
 
 import pytest
 import yaml
@@ -1032,6 +1033,55 @@ steps:
         # should not show error
         editor.duplicate_label_error(output="out_file1").wait_for_absent()
 
+    @selenium_test
+    def test_map_over_output_indicator(self):
+        self.open_in_workflow_editor(
+            """
+class: GalaxyWorkflow
+inputs:
+  list:
+    type: collection
+    collection_type: "list"
+  nested_list:
+    type: collection
+    collection_type: "list:list"
+steps:
+  filter:
+    tool_id: __FILTER_FROM_FILE__
+"""
+        )
+        self.assert_node_output_is("filter#output_filtered", "any")
+        self.workflow_editor_connect("list#output", "filter#input")
+        self.assert_node_output_is("filter#output_filtered", "list")
+        self.workflow_editor_connect("nested_list#output", "filter#how|filter_source")
+        self.assert_node_output_is("filter#output_filtered", "list", "list:list:list")
+        self.workflow_editor_destroy_connection("filter#how|filter_source")
+        self.assert_node_output_is("filter#output_filtered", "list")
+
+    def assert_node_output_is(self, label: str, output_type: str, map_over_type: Optional[str] = None):
+        editor = self.components.workflow_editor
+        node_label, output_name = label.split("#")
+        node = editor.node._(label=node_label)
+        node.wait_for_present()
+        output_element = node.output_terminal(name=output_name).wait_for_visible()
+        self.hover_over(output_element)
+        element = self.components._.tooltip_inner.wait_for_present()
+        assert f"output is {output_type}" in element.text, element.text
+        if map_over_type is None:
+            assert "mapped-over" not in element.text
+        else:
+            fragment = " and mapped-over to produce a "
+            if map_over_type == "list:paired":
+                fragment += "list of pairs dataset collection"
+            elif map_over_type == "list:list":
+                fragment += "list of lists dataset collection"
+            elif map_over_type.count(":") > 1:
+                fragment += f"dataset collection with {map_over_type.count(':') + 1} levels of nesting"
+            else:
+                fragment += f"{map_over_type}"
+            assert fragment in element.text
+        self.click_center()
+
     def workflow_editor_maximize_center_pane(self, collapse_left=True, collapse_right=True):
         if collapse_left:
             self.components._.left_panel_collapse.wait_for_and_click()
@@ -1091,8 +1141,8 @@ steps:
         output_element = output_terminal.wait_for_present()
         input_element = input_terminal.wait_for_present()
 
-        source_id = output_element.get_attribute("id")
-        sink_id = input_element.get_attribute("id")
+        source_id = output_element.get_attribute("id").replace("|", r"\|")
+        sink_id = input_element.get_attribute("id").replace("|", r"\|")
 
         return source_id, sink_id
 
diff --git a/lib/tool_shed/webapp/api/groups.py b/lib/tool_shed/webapp/api/groups.py
index 1f1cbff1114..99be57d64d8 100644
--- a/lib/tool_shed/webapp/api/groups.py
+++ b/lib/tool_shed/webapp/api/groups.py
@@ -97,7 +97,7 @@ class GroupsController(BaseAPIController):
         decoded_id = trans.security.decode_id(encoded_id)
         group = self.group_manager.get(trans, decoded_id)
         if group is None:
-            raise ObjectNotFound(f"Unable to locate group record for id {str(encoded_id)}.")
+            raise ObjectNotFound("Unable to locate group record with the given id.")
         return self._populate(trans, group)
 
     def _populate(self, trans, group):
diff --git a/scripts/cleanup_datasets/admin_cleanup_datasets.py b/scripts/cleanup_datasets/admin_cleanup_datasets.py
index 5cc3a99f69e..eb46ac70f8b 100755
--- a/scripts/cleanup_datasets/admin_cleanup_datasets.py
+++ b/scripts/cleanup_datasets/admin_cleanup_datasets.py
@@ -202,13 +202,13 @@ def administrative_delete_datasets(
     # We really only need the id column here, but sqlalchemy barfs when
     # trying to select only 1 column
     hda_ids_query = sa.select(
-        (app.model.HistoryDatasetAssociation.table.c.id, app.model.HistoryDatasetAssociation.table.c.deleted),
+        (app.model.HistoryDatasetAssociation.__table__.c.id, app.model.HistoryDatasetAssociation.__table__.c.deleted),
         whereclause=and_(
-            app.model.Dataset.table.c.deleted == false(),
-            app.model.HistoryDatasetAssociation.table.c.update_time < cutoff_time,
-            app.model.HistoryDatasetAssociation.table.c.deleted == false(),
+            app.model.Dataset.__table__.c.deleted == false(),
+            app.model.HistoryDatasetAssociation.__table__.c.update_time < cutoff_time,
+            app.model.HistoryDatasetAssociation.__table__.c.deleted == false(),
         ),
-        from_obj=[sa.outerjoin(app.model.Dataset.table, app.model.HistoryDatasetAssociation.table)],
+        from_obj=[sa.outerjoin(app.model.Dataset.__table__, app.model.HistoryDatasetAssociation.__table__)],
     )
 
     # Add all datasets associated with Histories to our list
@@ -230,16 +230,21 @@ def administrative_delete_datasets(
     # Process each of the Dataset objects
     for hda_id in hda_ids:
         user_query = sa.select(
-            [app.model.HistoryDatasetAssociation.table, app.model.History.table, app.model.User.table],
-            whereclause=and_(app.model.HistoryDatasetAssociation.table.c.id == hda_id),
+            [app.model.HistoryDatasetAssociation.__table__, app.model.History.__table__, app.model.User.__table__],
+            whereclause=and_(app.model.HistoryDatasetAssociation.__table__.c.id == hda_id),
             from_obj=[
-                sa.join(app.model.User.table, app.model.History.table).join(app.model.HistoryDatasetAssociation.table)
+                sa.join(app.model.User.__table__, app.model.History.__table__).join(
+                    app.model.HistoryDatasetAssociation.__table__
+                )
             ],
             use_labels=True,
         )
         for result in app.sa_session.execute(user_query):
-            user_notifications[result[app.model.User.table.c.email]].append(
-                (result[app.model.HistoryDatasetAssociation.table.c.name], result[app.model.History.table.c.name])
+            user_notifications[result[app.model.User.__table__.c.email]].append(
+                (
+                    result[app.model.HistoryDatasetAssociation.__table__.c.name],
+                    result[app.model.History.__table__.c.name],
+                )
             )
             deleted_instance_count += 1
             if not info_only and not email_only:
@@ -280,7 +285,7 @@ def _get_tool_id_for_hda(app, hda_id):
     job = (
         app.sa_session.query(app.model.Job)
         .join(app.model.JobToOutputDatasetAssociation)
-        .filter(app.model.JobToOutputDatasetAssociation.table.c.dataset_id == hda_id)
+        .filter(app.model.JobToOutputDatasetAssociation.__table__.c.dataset_id == hda_id)
         .first()
     )
     if job is not None:
diff --git a/scripts/cleanup_datasets/cleanup_datasets.py b/scripts/cleanup_datasets/cleanup_datasets.py
index eee06b243b5..e6ded97ce8e 100755
--- a/scripts/cleanup_datasets/cleanup_datasets.py
+++ b/scripts/cleanup_datasets/cleanup_datasets.py
@@ -223,13 +223,13 @@ def delete_userless_histories(app, cutoff_time, info_only=False, force_retry=Fal
     start = time.time()
     if force_retry:
         histories = app.sa_session.query(app.model.History).filter(
-            and_(app.model.History.table.c.user_id == null(), app.model.History.update_time < cutoff_time)
+            and_(app.model.History.__table__.c.user_id == null(), app.model.History.update_time < cutoff_time)
         )
     else:
         histories = app.sa_session.query(app.model.History).filter(
             and_(
-                app.model.History.table.c.user_id == null(),
-                app.model.History.table.c.deleted == false(),
+                app.model.History.__table__.c.user_id == null(),
+                app.model.History.__table__.c.deleted == false(),
                 app.model.History.update_time < cutoff_time,
             )
         )
@@ -257,7 +257,7 @@ def purge_histories(app, cutoff_time, remove_from_disk, info_only=False, force_r
     if force_retry:
         histories = (
             app.sa_session.query(app.model.History)
-            .filter(and_(app.model.History.table.c.deleted == true(), app.model.History.update_time < cutoff_time))
+            .filter(and_(app.model.History.__table__.c.deleted == true(), app.model.History.update_time < cutoff_time))
             .options(joinedload("datasets"))
         )
     else:
@@ -265,8 +265,8 @@ def purge_histories(app, cutoff_time, remove_from_disk, info_only=False, force_r
             app.sa_session.query(app.model.History)
             .filter(
                 and_(
-                    app.model.History.table.c.deleted == true(),
-                    app.model.History.table.c.purged == false(),
+                    app.model.History.__table__.c.deleted == true(),
+                    app.model.History.__table__.c.purged == false(),
                     app.model.History.update_time < cutoff_time,
                 )
             )
@@ -307,14 +307,16 @@ def purge_libraries(app, cutoff_time, remove_from_disk, info_only=False, force_r
     start = time.time()
     if force_retry:
         libraries = app.sa_session.query(app.model.Library).filter(
-            and_(app.model.Library.table.c.deleted == true(), app.model.Library.table.c.update_time < cutoff_time)
+            and_(
+                app.model.Library.__table__.c.deleted == true(), app.model.Library.__table__.c.update_time < cutoff_time
+            )
         )
     else:
         libraries = app.sa_session.query(app.model.Library).filter(
             and_(
-                app.model.Library.table.c.deleted == true(),
-                app.model.Library.table.c.purged == false(),
-                app.model.Library.table.c.update_time < cutoff_time,
+                app.model.Library.__table__.c.deleted == true(),
+                app.model.Library.__table__.c.purged == false(),
+                app.model.Library.__table__.c.update_time < cutoff_time,
             )
         )
     for library in libraries:
@@ -342,16 +344,16 @@ def purge_folders(app, cutoff_time, remove_from_disk, info_only=False, force_ret
     if force_retry:
         folders = app.sa_session.query(app.model.LibraryFolder).filter(
             and_(
-                app.model.LibraryFolder.table.c.deleted == true(),
-                app.model.LibraryFolder.table.c.update_time < cutoff_time,
+                app.model.LibraryFolder.__table__.c.deleted == true(),
+                app.model.LibraryFolder.__table__.c.update_time < cutoff_time,
             )
         )
     else:
         folders = app.sa_session.query(app.model.LibraryFolder).filter(
             and_(
-                app.model.LibraryFolder.table.c.deleted == true(),
-                app.model.LibraryFolder.table.c.purged == false(),
-                app.model.LibraryFolder.table.c.update_time < cutoff_time,
+                app.model.LibraryFolder.__table__.c.deleted == true(),
+                app.model.LibraryFolder.__table__.c.purged == false(),
+                app.model.LibraryFolder.__table__.c.update_time < cutoff_time,
             )
         )
     for folder in folders:
@@ -368,34 +370,34 @@ def delete_datasets(app, cutoff_time, remove_from_disk, info_only=False, force_r
     start = time.time()
     if force_retry:
         history_dataset_ids_query = sa.select(
-            (app.model.Dataset.table.c.id, app.model.Dataset.table.c.state),
-            whereclause=app.model.HistoryDatasetAssociation.table.c.update_time < cutoff_time,
-            from_obj=[sa.outerjoin(app.model.Dataset.table, app.model.HistoryDatasetAssociation.table)],
+            (app.model.Dataset.__table__.c.id, app.model.Dataset.__table__.c.state),
+            whereclause=app.model.HistoryDatasetAssociation.__table__.c.update_time < cutoff_time,
+            from_obj=[sa.outerjoin(app.model.Dataset.__table__, app.model.HistoryDatasetAssociation.__table__)],
         )
         library_dataset_ids_query = sa.select(
-            (app.model.LibraryDataset.table.c.id, app.model.LibraryDataset.table.c.deleted),
-            whereclause=app.model.LibraryDataset.table.c.update_time < cutoff_time,
-            from_obj=[app.model.LibraryDataset.table],
+            (app.model.LibraryDataset.__table__.c.id, app.model.LibraryDataset.__table__.c.deleted),
+            whereclause=app.model.LibraryDataset.__table__.c.update_time < cutoff_time,
+            from_obj=[app.model.LibraryDataset.__table__],
         )
     else:
         # We really only need the id column here, but sqlalchemy barfs when trying to select only 1 column
         history_dataset_ids_query = sa.select(
-            (app.model.Dataset.table.c.id, app.model.Dataset.table.c.state),
+            (app.model.Dataset.__table__.c.id, app.model.Dataset.__table__.c.state),
             whereclause=and_(
-                app.model.Dataset.table.c.deleted == false(),
-                app.model.HistoryDatasetAssociation.table.c.update_time < cutoff_time,
-                app.model.HistoryDatasetAssociation.table.c.deleted == true(),
+                app.model.Dataset.__table__.c.deleted == false(),
+                app.model.HistoryDatasetAssociation.__table__.c.update_time < cutoff_time,
+                app.model.HistoryDatasetAssociation.__table__.c.deleted == true(),
             ),
-            from_obj=[sa.outerjoin(app.model.Dataset.table, app.model.HistoryDatasetAssociation.table)],
+            from_obj=[sa.outerjoin(app.model.Dataset.__table__, app.model.HistoryDatasetAssociation.__table__)],
         )
         library_dataset_ids_query = sa.select(
-            (app.model.LibraryDataset.table.c.id, app.model.LibraryDataset.table.c.deleted),
+            (app.model.LibraryDataset.__table__.c.id, app.model.LibraryDataset.__table__.c.deleted),
             whereclause=and_(
-                app.model.LibraryDataset.table.c.deleted == true(),
-                app.model.LibraryDataset.table.c.purged == false(),
-                app.model.LibraryDataset.table.c.update_time < cutoff_time,
+                app.model.LibraryDataset.__table__.c.deleted == true(),
+                app.model.LibraryDataset.__table__.c.purged == false(),
+                app.model.LibraryDataset.__table__.c.update_time < cutoff_time,
             ),
-            from_obj=[app.model.LibraryDataset.table],
+            from_obj=[app.model.LibraryDataset.__table__],
         )
     deleted_dataset_count = 0
     deleted_instance_count = 0
@@ -471,18 +473,18 @@ def purge_datasets(app, cutoff_time, remove_from_disk, info_only=False, force_re
     if force_retry:
         datasets = app.sa_session.query(app.model.Dataset).filter(
             and_(
-                app.model.Dataset.table.c.deleted == true(),
-                app.model.Dataset.table.c.purgable == true(),
-                app.model.Dataset.table.c.update_time < cutoff_time,
+                app.model.Dataset.__table__.c.deleted == true(),
+                app.model.Dataset.__table__.c.purgable == true(),
+                app.model.Dataset.__table__.c.update_time < cutoff_time,
             )
         )
     else:
         datasets = app.sa_session.query(app.model.Dataset).filter(
             and_(
-                app.model.Dataset.table.c.deleted == true(),
-                app.model.Dataset.table.c.purgable == true(),
-                app.model.Dataset.table.c.purged == false(),
-                app.model.Dataset.table.c.update_time < cutoff_time,
+                app.model.Dataset.__table__.c.deleted == true(),
+                app.model.Dataset.__table__.c.purgable == true(),
+                app.model.Dataset.__table__.c.purged == false(),
+                app.model.Dataset.__table__.c.update_time < cutoff_time,
             )
         )
     for dataset in datasets:
@@ -555,12 +557,12 @@ def _delete_dataset(dataset, app, remove_from_disk, info_only=False, is_deletabl
         # lets create a list of metadata files, then perform actions on them
         for hda in dataset.history_associations:
             for metadata_file in app.sa_session.query(app.model.MetadataFile).filter(
-                app.model.MetadataFile.table.c.hda_id == hda.id
+                app.model.MetadataFile.__table__.c.hda_id == hda.id
             ):
                 metadata_files.append(metadata_file)
         for ldda in dataset.library_associations:
             for metadata_file in app.sa_session.query(app.model.MetadataFile).filter(
-                app.model.MetadataFile.table.c.lda_id == ldda.id
+                app.model.MetadataFile.__table__.c.lda_id == ldda.id
             ):
                 metadata_files.append(metadata_file)
         for metadata_file in metadata_files:
diff --git a/test/functional/tools/data_optional.xml b/test/functional/tools/data_optional.xml
new file mode 100644
index 00000000000..a692999e83c
--- /dev/null
+++ b/test/functional/tools/data_optional.xml
@@ -0,0 +1,31 @@
+<tool id="data_optional" name="data select" version="1.0.0">
+    <command><![CDATA[
+        echo INPUT $names >> '$output' &&
+
+        ## verify that ext is "data" for absent optional input
+        echo INPUT ext $names.ext >> '$output' &&
+
+        ## verify that is_of_type for absent optional input
+        ## does not evaluate to True for any of the allowed datatypes
+        echo ISOFTYPE mothur.names $names.is_of_type("mothur.names") >> '$output' &&
+        echo ISOFTYPE mothur.count_table $names.is_of_type("mothur.count_table") >> '$output'
+    ]]></command>
+    <inputs>
+        <param name="names" type="data" format="mothur.names,mothur.count_table" optional="true"/>
+    </inputs>
+    <outputs>
+        <data name="output" format="txt"/>
+    </outputs>
+    <tests>
+        <test>
+            <output name="output">
+                <assert_contents>
+                    <has_line line="INPUT None" />
+                    <has_line line="INPUT ext data" />
+                    <has_line line="ISOFTYPE mothur.names False" />
+                    <has_line line="ISOFTYPE mothur.count_table False" />
+                </assert_contents>
+            </output>
+        </test>
+    </tests>
+</tool>
diff --git a/test/functional/tools/sample_tool_conf.xml b/test/functional/tools/sample_tool_conf.xml
index bdfd6e58b78..d73e80cdf8f 100644
--- a/test/functional/tools/sample_tool_conf.xml
+++ b/test/functional/tools/sample_tool_conf.xml
@@ -45,6 +45,7 @@
   <tool file="tool_provided_metadata_12.xml" />
   <tool file="empty_datasets.xml" />
   <tool file="is_of_type.xml" />
+  <tool file="data_optional.xml" />
   <tool file="inputs_as_json.xml" />
   <tool file="inputs_as_json_profile.xml" />
   <tool file="inputs_as_json_with_paths.xml" />
diff --git a/test/unit/data/model/test_model_store.py b/test/unit/data/model/test_model_store.py
index 1238464438a..6164b2974e5 100644
--- a/test/unit/data/model/test_model_store.py
+++ b/test/unit/data/model/test_model_store.py
@@ -21,6 +21,7 @@ from sqlalchemy.orm.scoping import scoped_session
 from galaxy import model
 from galaxy.model import store
 from galaxy.model.metadata import MetadataTempFile
+from galaxy.model.orm.now import now
 from galaxy.model.unittest_utils import GalaxyDataTestApp
 from galaxy.model.unittest_utils.store_fixtures import (
     deferred_hda_model_store_dict,
@@ -396,7 +397,7 @@ def validate_main_entity(ro_crate: ROCrate):
     assert workflow["name"] == "Test Workflow"
     assert "SoftwareSourceCode" in workflow.type
     assert "ComputationalWorkflow" in workflow.type
-    assert len(workflow["input"]) == 2
+    assert len(workflow["input"]) == 1
     assert len(workflow["output"]) == 1
 
 
@@ -409,7 +410,7 @@ def validate_create_action(ro_crate: ROCrate):
     assert wf_action["instrument"] is workflow
     wf_objects = wf_action["object"]
     wf_results = wf_action["result"]
-    assert len(wf_objects) == 2
+    assert len(wf_objects) == 1
     assert len(wf_results) == 1
     for entity in wf_results:
         if entity.id.endswith(".txt"):
@@ -424,7 +425,6 @@ def validate_other_entities(ro_crate: ROCrate):
     inputs = workflow["input"]
     outputs = workflow["output"]
     assert inputs[0]["additionalType"] == "File"
-    assert inputs[1]["additionalType"] == "None"
     assert outputs[0]["additionalType"] == "File"
 
     for entity in inputs + outputs:
@@ -456,18 +456,21 @@ def validate_invocation_collection_crate_directory(crate_directory):
     actions = [_ for _ in ro_crate.contextual_entities if "CreateAction" in _.type]
     assert len(actions) == 1
     wf_action = actions[0]
-    wf_objects = wf_action["object"]
-    assert len(workflow["input"]) == 3
+    assert wf_action in root["mentions"]
+    assert len(workflow["input"]) == 2
     assert len(workflow["output"]) == 1
-    assert len(root["mentions"]) == 3
+    assert len(root["mentions"]) == 4
     collections = [_ for _ in ro_crate.contextual_entities if "Collection" in _.type]
     assert len(collections) == 3
     collection = collections[0]
     assert collection.type == "Collection"
-    assert collection["additionalType"] == "list"
+    assert (
+        collection["additionalType"]
+        == "https://training.galaxyproject.org/training-material/faqs/galaxy/collections_build_list.html"
+    )
     assert len(collection["hasPart"]) == 2
     for dataset in collection["hasPart"]:
-        assert dataset in wf_objects
+        assert dataset in root["hasPart"]
 
 
 def test_export_history_to_ro_crate(tmp_path):
@@ -489,6 +492,15 @@ def test_export_invocation_to_ro_crate(tmp_path):
     validate_invocation_crate_directory(crate_directory)
 
 
+def test_export_simple_invocation_to_ro_crate(tmp_path):
+    app = _mock_app()
+    workflow_invocation = _setup_simple_invocation(app)
+    crate_directory = tmp_path / "crate"
+    with store.ROCrateModelExportStore(crate_directory, app=app) as export_store:
+        export_store.export_workflow_invocation(workflow_invocation)
+    validate_invocation_crate_directory(crate_directory)
+
+
 def test_export_collection_invocation_to_ro_crate(tmp_path):
     app = _mock_app()
     workflow_invocation = _setup_collection_invocation(app)
@@ -842,7 +854,6 @@ def _setup_invocation(app):
     workflow_step_1 = model.WorkflowStep()
     workflow_step_1.order_index = 0
     workflow_step_1.type = "data_input"
-    workflow_step_1.tool_inputs = {}
     sa_session.add(workflow_step_1)
     workflow_1 = _workflow_from_steps(u, [workflow_step_1])
     workflow_1.license = "MIT"
@@ -853,16 +864,14 @@ def _setup_invocation(app):
     invocation_step.workflow_step = workflow_step_1
     invocation_step.job = j
     sa_session.add(invocation_step)
-    input_assoc = model.WorkflowRequestToInputDatasetAssociation()
-    input_assoc.workflow_invocation = workflow_invocation
-    input_assoc.workflow_step = workflow_step_1
-    input_assoc.dataset = d1
-    invocation_step.input_datasets = [input_assoc]
     output_assoc = model.WorkflowInvocationStepOutputDatasetAssociation()
     output_assoc.dataset = d2
     invocation_step.output_datasets = [output_assoc]
     workflow_invocation.steps = [invocation_step]
     workflow_invocation.user = u
+    workflow_invocation.add_input(d1, step=workflow_step_1)
+    wf_output = model.WorkflowOutput(workflow_step_1, label="output_label")
+    workflow_invocation.add_output(wf_output, workflow_step_1, d2)
     sa_session.add(workflow_invocation)
     sa_session.flush()
     return workflow_invocation
@@ -930,29 +939,44 @@ def _setup_collection_invocation(app):
     workflow_1.name = "Test Workflow"
     sa_session.add(workflow_1)
     workflow_invocation = _invocation_for_workflow(u, workflow_1)
-    invocation_step = model.WorkflowInvocationStep()
-    invocation_step.workflow_step = workflow_step_1
-    invocation_step.job = j
-    sa_session.add(invocation_step)
-    input_assoc = model.WorkflowRequestToInputDatasetCollectionAssociation()
-    input_assoc.workflow_invocation = workflow_invocation
-    input_assoc.workflow_step = workflow_step_1
-    input_assoc.dataset_collection = hc1
-    input_assoc1 = model.WorkflowRequestToInputDatasetCollectionAssociation()
-    input_assoc1.workflow_invocation = workflow_invocation
-    input_assoc1.workflow_step = workflow_step_1
-    input_assoc1.dataset_collection = hc2
-    invocation_step.input_dataset_collections = [input_assoc, input_assoc1]
-    output_assoc = model.WorkflowInvocationStepOutputDatasetCollectionAssociation()
-    output_assoc.dataset_collection = hc3
-    invocation_step.output_dataset_collections = [output_assoc]
-    workflow_invocation.steps = [invocation_step]
     workflow_invocation.user = u
+    workflow_invocation.add_input(hc1, step=workflow_step_1)
+    workflow_invocation.add_input(hc2, step=workflow_step_1)
+    wf_output = model.WorkflowOutput(workflow_step_1, label="output_label")
+    workflow_invocation.add_output(wf_output, workflow_step_1, hc3)
+
     sa_session.add(workflow_invocation)
     sa_session.flush()
     return workflow_invocation
 
 
+def _setup_simple_invocation(app):
+    sa_session = app.model.context
+
+    u, h, d1, d2, j = _setup_simple_cat_job(app)
+    j.parameters = [model.JobParameter(name="index_path", value='"/old/path/human"')]
+
+    workflow_step_1 = model.WorkflowStep()
+    workflow_step_1.order_index = 0
+    workflow_step_1.type = "data_input"
+    workflow_step_1.tool_inputs = {}
+    sa_session.add(workflow_step_1)
+    workflow = _workflow_from_steps(u, [workflow_step_1])
+    workflow.license = "MIT"
+    workflow.name = "Test Workflow"
+    workflow.create_time = now()
+    workflow.update_time = now()
+    sa_session.add(workflow)
+    invocation = _invocation_for_workflow(u, workflow)
+    invocation.create_time = now()
+    invocation.update_time = now()
+
+    invocation.add_input(d1, step=workflow_step_1)
+    wf_output = model.WorkflowOutput(workflow_step_1, label="output_label")
+    invocation.add_output(wf_output, workflow_step_1, d2)
+    return invocation
+
+
 def _import_export_history(app, h, dest_export=None, export_files=None, import_options=None, include_hidden=False):
     if dest_export is None:
         dest_parent = mkdtemp()
diff --git a/test/unit/webapps/test_service_base.py b/test/unit/webapps/test_service_base.py
new file mode 100644
index 00000000000..59d267025e9
--- /dev/null
+++ b/test/unit/webapps/test_service_base.py
@@ -0,0 +1,37 @@
+from typing import Tuple
+
+import pytest
+
+from galaxy.schema.schema import ModelStoreFormat
+from galaxy.web.short_term_storage import ShortTermStorageAllocator
+from galaxy.webapps.galaxy.services.base import model_store_storage_target
+
+
+class MockShortTermStorageAllocator(ShortTermStorageAllocator):
+    def new_target(self, filename, mime_type):
+        return filename, mime_type
+
+
+@pytest.mark.parametrize(
+    "file_name, model_store_format, expected",
+    [
+        ("My Cool Object", "txt", ("My-Cool-Object.txt", "text/plain")),
+        ("!My Cool Object!", "json", ("My-Cool-Object.json", "application/json")),
+        ("Hello₩◎ґʟⅾ", "xml", ("Hello.xml", "application/xml")),
+        ("test", ModelStoreFormat.ROCRATE_ZIP.value, ("test.rocrate.zip", "application/zip")),
+        ("test", ModelStoreFormat.TAR.value, ("test.tar", "application/x-tar")),
+        ("test", ModelStoreFormat.TGZ.value, ("test.tgz", "application/x-tar")),
+        ("test", ModelStoreFormat.TAR_DOT_GZ.value, ("test.tar.gz", "application/x-tar")),
+        ("test", ModelStoreFormat.BAG_DOT_ZIP.value, ("test.bag.zip", "application/zip")),
+        ("test", ModelStoreFormat.BAG_DOT_TAR.value, ("test.bag.tar", "application/x-tar")),
+        ("test", ModelStoreFormat.BAG_DOT_TGZ.value, ("test.bag.tgz", "application/x-tar")),
+        ("test", ModelStoreFormat.BCO_JSON.value, ("test.bco.json", "application/json")),
+    ],
+)
+def test_model_store_storage_target(file_name: str, model_store_format: str, expected: Tuple[str, str]):
+    mock_sts_allocator = MockShortTermStorageAllocator()
+    actual = model_store_storage_target(
+        short_term_storage_allocator=mock_sts_allocator, file_name=file_name, model_store_format=model_store_format
+    )
+
+    assert actual == expected