diff --git a/client/src/api/schema/schema.ts b/client/src/api/schema/schema.ts
index b888d567942d..cdc9d31b8dd3 100644
--- a/client/src/api/schema/schema.ts
+++ b/client/src/api/schema/schema.ts
@@ -4555,6 +4555,23 @@ export interface components {
              */
             user_email?: string | null;
         };
+        /** EncodedJobParameterHistoryItem */
+        EncodedJobParameterHistoryItem: {
+            /** Hid */
+            hid?: number | null;
+            /**
+             * Id
+             * @example 0123456789ABCDEF
+             */
+            id: string;
+            /** Name */
+            name: string;
+            /**
+             * Source
+             * @description The source of this dataset, either `hda`, `ldda`, `hdca`, `dce` or `dc` depending of its origin.
+             */
+            src: components["schemas"]["DataItemSourceType"];
+        };
         /** ExportHistoryArchivePayload */
         ExportHistoryArchivePayload: {
             /**
@@ -7735,7 +7752,13 @@ export interface components {
              * Value
              * @description The values of the job parameter
              */
-            value?: Record<string, never> | null;
+            value?:
+                | components["schemas"]["EncodedJobParameterHistoryItem"][]
+                | number
+                | number
+                | boolean
+                | string
+                | null;
         };
         /**
          * JobSourceType
diff --git a/client/src/components/Tool/ToolForm.vue b/client/src/components/Tool/ToolForm.vue
index 1a88b40654e4..9c524b2c0511 100644
--- a/client/src/components/Tool/ToolForm.vue
+++ b/client/src/components/Tool/ToolForm.vue
@@ -75,6 +75,12 @@
                         title="Attempt to re-use jobs with identical parameters?"
                         help="This may skip executing jobs that you have already run."
                         type="boolean" />
+                    <FormSelect
+                        v-if="formConfig.model_class === 'DataManagerTool'"
+                        id="data_manager_mode"
+                        v-model="dataManagerMode"
+                        :options="bundleOptions"
+                        title="Create dataset bundle instead of adding data table to loc file ?"></FormSelect>
                 </div>
             </template>
             <template v-slot:header-buttons>
@@ -120,6 +126,8 @@ import { getToolFormData, submitJob, updateToolFormData } from "./services";
 import ToolCard from "./ToolCard";
 import { allowCachedJobs } from "./utilities";
 
+import FormSelect from "@/components/Form/Elements/FormSelect.vue";
+
 export default {
     components: {
         ButtonSpinner,
@@ -127,6 +135,7 @@ export default {
         FormDisplay,
         ToolCard,
         FormElement,
+        FormSelect,
         ToolEntryPoints,
         ToolRecommendation,
         Heading,
@@ -175,6 +184,7 @@ export default {
             useCachedJobs: false,
             useEmail: false,
             useJobRemapping: false,
+            dataManagerMode: "populate",
             entryPoints: [],
             jobDef: {},
             jobResponse: {},
@@ -182,6 +192,10 @@ export default {
             validationScrollTo: null,
             currentVersion: this.version,
             preferredObjectStoreId: null,
+            bundleOptions: [
+                { label: "populate", value: "populate" },
+                { label: "bundle", value: "bundle" },
+            ],
         };
     },
     computed: {
@@ -326,6 +340,9 @@ export default {
             if (this.preferredObjectStoreId) {
                 jobDef.preferred_object_store_id = this.preferredObjectStoreId;
             }
+            if (this.dataManagerMode === "bundle") {
+                jobDef.data_manager_mode = this.dataManagerMode;
+            }
             console.debug("toolForm::onExecute()", jobDef);
             const prevRoute = this.$route.fullPath;
             submitJob(jobDef).then(
diff --git a/lib/galaxy/config/sample/datatypes_conf.xml.sample b/lib/galaxy/config/sample/datatypes_conf.xml.sample
index dfae056f4cb9..7baf1e748fcf 100644
--- a/lib/galaxy/config/sample/datatypes_conf.xml.sample
+++ b/lib/galaxy/config/sample/datatypes_conf.xml.sample
@@ -121,7 +121,7 @@
     <datatype extension="d3_hierarchy" type="galaxy.datatypes.text:Json" mimetype="application/json" subclass="true" display_in_upload="true"/>
     <datatype extension="imgt.json" type="galaxy.datatypes.text:ImgtJson" mimetype="application/json" display_in_upload="True"/>
     <datatype extension="geojson" type="galaxy.datatypes.text:GeoJson" mimetype="application/json" display_in_upload="True"/>
-    <datatype extension="data_manager_json" type="galaxy.datatypes.text:Json" mimetype="application/json" subclass="true" display_in_upload="false"/>
+    <datatype extension="data_manager_json" type="galaxy.datatypes.text:DataManagerJson" mimetype="application/json" subclass="true" display_in_upload="false"/>
     <datatype extension="dbn" type="galaxy.datatypes.sequence:DotBracket" display_in_upload="true" description="Dot-Bracket format is a text-based format for storing both an RNA sequence and its corresponding 2D structure." description_url="https://wiki.galaxyproject.org/Learn/Datatypes#Dbn"/>
     <datatype extension="fai" type="galaxy.datatypes.tabular:Tabular" display_in_upload="true" subclass="true" description="A Fasta Index File is a text file consisting of lines each with five TAB-delimited columns : Name, Length, offset, linebases, Linewidth" description_url="http://www.htslib.org/doc/faidx.html"/>
     <datatype extension="fasta" auto_compressed_types="gz" type="galaxy.datatypes.sequence:Fasta" display_in_upload="true" description="A sequence in FASTA format consists of a single-line description, followed by lines of sequence data. The first character of the description line is a greater-than ('&gt;') symbol in the first column. All lines should be shorter than 80 characters." description_url="https://wiki.galaxyproject.org/Learn/Datatypes#Fasta">
diff --git a/lib/galaxy/datatypes/text.py b/lib/galaxy/datatypes/text.py
index f62ff3ace52d..54a8f28aa4b9 100644
--- a/lib/galaxy/datatypes/text.py
+++ b/lib/galaxy/datatypes/text.py
@@ -133,6 +133,19 @@ def display_peek(self, dataset: DatasetProtocol) -> str:
             return f"JSON file ({nice_size(dataset.get_size())})"
 
 
+class DataManagerJson(Json):
+    file_ext = "data_manager_json"
+    MetadataElement(
+        name="data_tables", default=None, desc="Data tables represented by this dataset", readonly=True, visible=True
+    )
+
+    def set_meta(self, dataset: DatasetProtocol, overwrite: bool = True, **kwd):
+        super().set_meta(dataset=dataset, overwrite=overwrite, **kwd)
+        with open(dataset.get_file_name()) as fh:
+            data_tables = json.load(fh)["data_tables"]
+        dataset.metadata.data_tables = data_tables
+
+
 class ExpressionJson(Json):
     """Represents the non-data input or output to a tool or workflow."""
 
diff --git a/lib/galaxy/managers/jobs.py b/lib/galaxy/managers/jobs.py
index dc46534bb253..102d8b04f591 100644
--- a/lib/galaxy/managers/jobs.py
+++ b/lib/galaxy/managers/jobs.py
@@ -901,23 +901,24 @@ def inputs_recursive(input_params, param_values, depth=1, upgrade_messages=None)
 
         for input in input_params.values():
             if input.name in param_values:
+                input_value = param_values[input.name]
                 if input.type == "repeat":
-                    for i in range(len(param_values[input.name])):
-                        rval.extend(inputs_recursive(input.inputs, param_values[input.name][i], depth=depth + 1))
+                    for i in range(len(input_value)):
+                        rval.extend(inputs_recursive(input.inputs, input_value[i], depth=depth + 1))
                 elif input.type == "section":
                     # Get the value of the current Section parameter
                     rval.append(dict(text=input.name, depth=depth))
                     rval.extend(
                         inputs_recursive(
                             input.inputs,
-                            param_values[input.name],
+                            input_value,
                             depth=depth + 1,
                             upgrade_messages=upgrade_messages.get(input.name),
                         )
                     )
                 elif input.type == "conditional":
                     try:
-                        current_case = param_values[input.name]["__current_case__"]
+                        current_case = input_value["__current_case__"]
                         is_valid = True
                     except Exception:
                         current_case = None
@@ -929,7 +930,7 @@ def inputs_recursive(input_params, param_values, depth=1, upgrade_messages=None)
                         rval.extend(
                             inputs_recursive(
                                 input.cases[current_case].inputs,
-                                param_values[input.name],
+                                input_value,
                                 depth=depth + 1,
                                 upgrade_messages=upgrade_messages.get(input.name),
                             )
@@ -948,12 +949,16 @@ def inputs_recursive(input_params, param_values, depth=1, upgrade_messages=None)
                         dict(
                             text=input.group_title(param_values),
                             depth=depth,
-                            value=f"{len(param_values[input.name])} uploaded datasets",
+                            value=f"{len(input_value)} uploaded datasets",
                         )
                     )
-                elif input.type == "data" or input.type == "data_collection":
+                elif (
+                    input.type == "data"
+                    or input.type == "data_collection"
+                    or isinstance(input_value, model.HistoryDatasetAssociation)
+                ):
                     value = []
-                    for element in listify(param_values[input.name]):
+                    for element in listify(input_value):
                         element_id = element.id
                         if isinstance(element, model.HistoryDatasetAssociation):
                             hda = element
@@ -977,7 +982,7 @@ def inputs_recursive(input_params, param_values, depth=1, upgrade_messages=None)
                         dict(
                             text=label,
                             depth=depth,
-                            value=input.value_to_display_text(param_values[input.name]),
+                            value=input.value_to_display_text(input_value),
                             notes=upgrade_messages.get(input.name, ""),
                         )
                     )
diff --git a/lib/galaxy/managers/workflows.py b/lib/galaxy/managers/workflows.py
index acb43475ef6d..109ade27acef 100644
--- a/lib/galaxy/managers/workflows.py
+++ b/lib/galaxy/managers/workflows.py
@@ -92,8 +92,8 @@
     DataCollectionToolParameter,
     DataToolParameter,
     RuntimeValue,
-    workflow_building_modes,
 )
+from galaxy.tools.parameters.workflow_building_modes import workflow_building_modes
 from galaxy.util.hash_util import md5_hash_str
 from galaxy.util.json import (
     safe_dumps,
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 17a7647b86ba..25a60f1e6040 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -792,6 +792,27 @@ def __init__(self, email=None, password=None, username=None):
         self.active = False
         self.username = username
 
+    def get_user_data_tables(self, data_table: str):
+        session = object_session(self)
+        assert session
+        metadata_select = (
+            select(HistoryDatasetAssociation)
+            .join(Dataset)
+            .join(History)
+            .where(
+                HistoryDatasetAssociation.deleted == false(),
+                HistoryDatasetAssociation.extension == "data_manager_json",
+                History.user_id == self.id,
+                Dataset.state == "ok",
+                # excludes data manager runs that actually populated tables.
+                # maybe track this formally by creating a different datatype for bundles ?
+                Dataset.total_size != Dataset.file_size,
+                HistoryDatasetAssociation._metadata.contains(data_table),
+            )
+            .order_by(HistoryDatasetAssociation.id)
+        )
+        return session.execute(metadata_select).scalars().all()
+
     @property
     def extra_preferences(self):
         data = defaultdict(lambda: None)
diff --git a/lib/galaxy/model/custom_types.py b/lib/galaxy/model/custom_types.py
index 3db767af281a..f36b4b9a6130 100644
--- a/lib/galaxy/model/custom_types.py
+++ b/lib/galaxy/model/custom_types.py
@@ -343,11 +343,12 @@ class MetadataType(JSONType):
     def process_bind_param(self, value, dialect):
         if value is not None:
             if MAX_METADATA_VALUE_SIZE is not None:
-                for k, v in list(value.items()):
-                    sz = total_size(v)
-                    if sz > MAX_METADATA_VALUE_SIZE:
-                        del value[k]
-                        log.warning(f"Refusing to bind metadata key {k} due to size ({sz})")
+                if hasattr(value, "items"):
+                    for k, v in list(value.items()):
+                        sz = total_size(v)
+                        if sz > MAX_METADATA_VALUE_SIZE:
+                            del value[k]
+                            log.warning(f"Refusing to bind metadata key {k} due to size ({sz})")
             value = json_encoder.encode(value).encode()
         return value
 
diff --git a/lib/galaxy/schema/jobs.py b/lib/galaxy/schema/jobs.py
index 5f8e2e79d8d0..fe6316262983 100644
--- a/lib/galaxy/schema/jobs.py
+++ b/lib/galaxy/schema/jobs.py
@@ -4,6 +4,7 @@
     Dict,
     List,
     Optional,
+    Union,
 )
 
 from pydantic import (
@@ -21,6 +22,7 @@
 from galaxy.schema.schema import (
     DataItemSourceType,
     EncodedDataItemSourceId,
+    EncodedJobParameterHistoryItem,
     JobMetricCollection,
     JobState,
     JobSummary,
@@ -221,7 +223,9 @@ class JobParameter(Model):
         title="Depth",
         description="The depth of the job parameter.",
     )
-    value: Optional[Any] = Field(default=None, title="Value", description="The values of the job parameter")
+    value: Optional[Union[List[EncodedJobParameterHistoryItem], float, int, bool, str]] = Field(
+        default=None, title="Value", description="The values of the job parameter", union_mode="left_to_right"
+    )
     notes: Optional[str] = Field(default=None, title="Notes", description="Notes associated with the job parameter.")
 
 
diff --git a/lib/galaxy/schema/schema.py b/lib/galaxy/schema/schema.py
index f94061a887c1..97069a0dd2cc 100644
--- a/lib/galaxy/schema/schema.py
+++ b/lib/galaxy/schema/schema.py
@@ -1965,6 +1965,11 @@ class EncodedDataItemSourceId(Model):
     )
 
 
+class EncodedJobParameterHistoryItem(EncodedDataItemSourceId):
+    hid: Optional[int] = None
+    name: str
+
+
 class DatasetJobInfo(DatasetSourceId):
     uuid: UuidField
 
diff --git a/lib/galaxy/tool_util/data/__init__.py b/lib/galaxy/tool_util/data/__init__.py
index 655597b39f39..921dc100fd38 100644
--- a/lib/galaxy/tool_util/data/__init__.py
+++ b/lib/galaxy/tool_util/data/__init__.py
@@ -489,7 +489,7 @@ def handle_found_index_file(self, filename):
 
     # This method is used in tools, so need to keep its API stable
     def get_fields(self) -> List[List[str]]:
-        return self.data
+        return self.data.copy()
 
     def get_field(self, value):
         rval = None
diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py
index 250ddb87b256..311987a399ec 100644
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -127,7 +127,6 @@
     SelectTagParameter,
     SelectToolParameter,
     ToolParameter,
-    workflow_building_modes,
 )
 from galaxy.tools.parameters.dataset_matcher import (
     set_dataset_matcher_factory,
@@ -143,6 +142,7 @@
 )
 from galaxy.tools.parameters.input_translation import ToolInputTranslator
 from galaxy.tools.parameters.meta import expand_meta_parameters
+from galaxy.tools.parameters.workflow_building_modes import workflow_building_modes
 from galaxy.tools.parameters.wrapped_json import json_wrap
 from galaxy.tools.test import parse_tests
 from galaxy.util import (
diff --git a/lib/galaxy/tools/actions/__init__.py b/lib/galaxy/tools/actions/__init__.py
index 80604f7976fa..cfce9b540692 100644
--- a/lib/galaxy/tools/actions/__init__.py
+++ b/lib/galaxy/tools/actions/__init__.py
@@ -32,6 +32,7 @@
     DataCollectionToolParameter,
     DataToolParameter,
     RuntimeValue,
+    SelectToolParameter,
 )
 from galaxy.tools.parameters.wrapped import (
     LegacyUnprefixedDict,
@@ -283,6 +284,8 @@ def process_dataset(data, formats=None):
                         value.child_collection = new_collection
                     else:
                         value.collection = new_collection
+            elif isinstance(input, SelectToolParameter) and isinstance(value, HistoryDatasetAssociation):
+                input_datasets[prefixed_name] = value
 
         tool.visit_inputs(param_values, visitor)
         return input_datasets, all_permissions
diff --git a/lib/galaxy/tools/evaluation.py b/lib/galaxy/tools/evaluation.py
index 1c202a3031f9..451790a69e5e 100644
--- a/lib/galaxy/tools/evaluation.py
+++ b/lib/galaxy/tools/evaluation.py
@@ -407,7 +407,7 @@ def __populate_input_dataset_wrappers(self, param_dict, input_datasets):
                 if wrapper:
                     param_dict[name] = wrapper
                     continue
-            if not isinstance(param_dict_value, (DatasetFilenameWrapper, DatasetListWrapper)):
+            if not isinstance(param_dict_value, ToolParameterValueWrapper):
                 wrapper_kwds = dict(
                     datatypes_registry=self.app.datatypes_registry,
                     tool=self.tool,
diff --git a/lib/galaxy/tools/parameters/basic.py b/lib/galaxy/tools/parameters/basic.py
index 3cf9e7d377d6..3e2f54053008 100644
--- a/lib/galaxy/tools/parameters/basic.py
+++ b/lib/galaxy/tools/parameters/basic.py
@@ -8,7 +8,9 @@
 import os
 import os.path
 import re
+import typing
 import urllib.parse
+from collections.abc import MutableMapping
 from typing import (
     Any,
     Dict,
@@ -39,6 +41,7 @@
 from galaxy.model.dataset_collections import builder
 from galaxy.schema.fetch_data import FilesPayload
 from galaxy.tool_util.parser import get_input_source as ensure_input_source
+from galaxy.tools.parameters.workflow_building_modes import workflow_building_modes
 from galaxy.util import (
     sanitize_param,
     string_as_bool,
@@ -66,12 +69,6 @@
 log = logging.getLogger(__name__)
 
 
-class workflow_building_modes:
-    DISABLED = False
-    ENABLED = True
-    USE_HISTORY = 1
-
-
 WORKFLOW_PARAMETER_REGULAR_EXPRESSION = re.compile(r"\$\{.+?\}")
 
 
@@ -91,7 +88,7 @@ def contains_workflow_parameter(value, search=False):
 
 def is_runtime_value(value):
     return isinstance(value, RuntimeValue) or (
-        isinstance(value, dict) and value.get("__class__") in ["RuntimeValue", "ConnectedValue"]
+        isinstance(value, MutableMapping) and value.get("__class__") in ["RuntimeValue", "ConnectedValue"]
     )
 
 
@@ -242,7 +239,7 @@ def value_from_basic(self, value, app, ignore_errors=False):
             if isinstance(self, HiddenToolParameter):
                 raise ParameterValueError(message_suffix="Runtime Parameter not valid", parameter_name=self.name)
             return runtime_to_object(value)
-        elif isinstance(value, dict) and value.get("__class__") == "UnvalidatedValue":
+        elif isinstance(value, MutableMapping) and value.get("__class__") == "UnvalidatedValue":
             return value["value"]
         # Delegate to the 'to_python' method
         if ignore_errors:
@@ -668,7 +665,7 @@ def from_json(self, value, trans=None, other_values=None):
         if isinstance(value, FilesPayload):
             # multi-part upload handled and persisted in service layer
             return value.model_dump()
-        elif isinstance(value, dict):
+        elif isinstance(value, MutableMapping):
             if "session_id" in value:
                 # handle api upload
                 session_id = value["session_id"]
@@ -700,7 +697,7 @@ def to_json(self, value, app, use_security):
             return None
         elif isinstance(value, str):
             return value
-        elif isinstance(value, dict):
+        elif isinstance(value, MutableMapping):
             # or should we jsonify?
             try:
                 return value["local_filename"]
@@ -774,7 +771,7 @@ def to_python(self, value, app, validate=False):
             if val in [None, ""]:
                 lst = []
                 break
-            if isinstance(val, dict):
+            if isinstance(val, MutableMapping):
                 lst.append(val["name"])
             else:
                 lst.append(val)
@@ -979,7 +976,10 @@ def get_legal_values(self, trans, other_values, value):
         """
         determine the set of values of legal options
         """
-        return {v for _, v, _ in self.get_options(trans, other_values)}
+        return {
+            history_item_dict_to_python(v, trans.app, self.name) or v
+            for _, v, _ in self.get_options(trans, other_values)
+        }
 
     def get_legal_names(self, trans, other_values):
         """
@@ -1094,8 +1094,13 @@ def to_param_dict_string(self, value, other_values=None):
         return value
 
     def to_json(self, value, app, use_security):
+        if isinstance(value, HistoryDatasetAssociation):
+            return history_item_to_json(value, app, use_security=use_security)
         return value
 
+    def to_python(self, value, app):
+        return history_item_dict_to_python(value, app, self.name) or super().to_python(value, app)
+
     def get_initial_value(self, trans, other_values):
         try:
             options = list(self.get_options(trans, other_values))
@@ -1918,45 +1923,22 @@ def get_initial_value(self, trans, other_values):
                         return hdca
 
     def to_json(self, value, app, use_security):
-        def single_to_json(value):
-            src = None
-            if isinstance(value, dict) and "src" in value and "id" in value:
-                return value
-            elif isinstance(value, DatasetCollectionElement):
-                src = "dce"
-            elif isinstance(value, HistoryDatasetCollectionAssociation):
-                src = "hdca"
-            elif isinstance(value, LibraryDatasetDatasetAssociation):
-                src = "ldda"
-            elif isinstance(value, HistoryDatasetAssociation) or hasattr(value, "id"):
-                # hasattr 'id' fires a query on persistent objects after a flush so better
-                # to do the isinstance check. Not sure we need the hasattr check anymore - it'd be
-                # nice to drop it.
-                src = "hda"
-            if src is not None:
-                object_id = cached_id(value)
-                return {"id": app.security.encode_id(object_id) if use_security else object_id, "src": src}
 
         if value not in [None, "", "None"]:
             if isinstance(value, list) and len(value) > 0:
-                values = [single_to_json(v) for v in value]
+                values = [history_item_to_json(v, app, use_security) for v in value]
             else:
-                values = [single_to_json(value)]
+                values = [history_item_to_json(value, app, use_security)]
             return {"values": values}
         return None
 
     def to_python(self, value, app):
-        def single_to_python(value):
-            if isinstance(value, dict) and "src" in value:
-                if value["src"] not in ("hda", "dce", "ldda", "hdca"):
-                    raise ParameterValueError(f"Invalid value {value}", self.name)
-                return src_id_to_item(sa_session=app.model.context, security=app.security, value=value)
 
-        if isinstance(value, dict) and "values" in value:
+        if isinstance(value, MutableMapping) and "values" in value:
             if hasattr(self, "multiple") and self.multiple is True:
-                return [single_to_python(v) for v in value["values"]]
+                return [history_item_dict_to_python(v, app, self.name) for v in value["values"]]
             elif len(value["values"]) > 0:
-                return single_to_python(value["values"][0])
+                return history_item_dict_to_python(value["values"][0], app, self.name)
 
         # Handle legacy string values potentially stored in databases
         none_values = [None, "", "None"]
@@ -2029,7 +2011,9 @@ def do_validate(v):
                 raise ValueError("At most %d datasets are required for %s" % (self.max, self.name))
 
 
-def src_id_to_item(sa_session: "Session", value: Dict[str, Any], security: "IdEncodingHelper") -> Union[
+def src_id_to_item(
+    sa_session: "Session", value: typing.MutableMapping[str, Any], security: "IdEncodingHelper"
+) -> Union[
     DatasetCollectionElement,
     HistoryDatasetAssociation,
     HistoryDatasetCollectionAssociation,
@@ -2121,7 +2105,7 @@ def from_json(self, value, trans, other_values=None):
             if self.default_object:
                 return raw_to_galaxy(trans, self.default_object)
             return None
-        if isinstance(value, dict) and "values" in value:
+        if isinstance(value, MutableMapping) and "values" in value:
             value = self.to_python(value, trans.app)
         if isinstance(value, str) and value.find(",") > 0:
             value = [int(value_part) for value_part in value.split(",")]
@@ -2136,7 +2120,7 @@ def from_json(self, value, trans, other_values=None):
         if isinstance(value, list):
             found_srcs = set()
             for single_value in value:
-                if isinstance(single_value, dict) and "src" in single_value and "id" in single_value:
+                if isinstance(single_value, MutableMapping) and "src" in single_value and "id" in single_value:
                     found_srcs.add(single_value["src"])
                     rval.append(
                         src_id_to_item(sa_session=trans.sa_session, value=single_value, security=trans.security)
@@ -2165,7 +2149,7 @@ def from_json(self, value, trans, other_values=None):
                     )
         elif isinstance(value, (HistoryDatasetAssociation, LibraryDatasetDatasetAssociation)):
             rval.append(value)
-        elif isinstance(value, dict) and "src" in value and "id" in value:
+        elif isinstance(value, MutableMapping) and "src" in value and "id" in value:
             rval.append(src_id_to_item(sa_session=trans.sa_session, value=value, security=trans.security))
         elif str(value).startswith("__collection_reduce__|"):
             encoded_ids = [v[len("__collection_reduce__|") :] for v in str(value).split(",")]
@@ -2461,7 +2445,7 @@ def from_json(self, value, trans, other_values=None):
             if self.default_object:
                 return raw_to_galaxy(trans, self.default_object)
             return None
-        if isinstance(value, dict) and "values" in value:
+        if isinstance(value, MutableMapping) and "values" in value:
             value = self.to_python(value, trans.app)
         if isinstance(value, str) and value.find(",") > 0:
             value = [int(value_part) for value_part in value.split(",")]
@@ -2472,13 +2456,13 @@ def from_json(self, value, trans, other_values=None):
             # a DatasetCollectionElement instead of a
             # HistoryDatasetCollectionAssociation.
             rval = value
-        elif isinstance(value, dict) and "src" in value and "id" in value:
+        elif isinstance(value, MutableMapping) and "src" in value and "id" in value:
             if value["src"] == "hdca":
                 rval = session.get(HistoryDatasetCollectionAssociation, trans.security.decode_id(value["id"]))
         elif isinstance(value, list):
             if len(value) > 0:
                 value = value[0]
-                if isinstance(value, dict) and "src" in value and "id" in value:
+                if isinstance(value, MutableMapping) and "src" in value and "id" in value:
                     if value["src"] == "hdca":
                         rval = session.get(HistoryDatasetCollectionAssociation, trans.security.decode_id(value["id"]))
                     elif value["src"] == "dce":
@@ -2656,7 +2640,7 @@ def to_dict(self, trans, other_values=None):
 
     def validate(self, value, trans=None):
         super().validate(value, trans=trans)
-        if not isinstance(value, dict):
+        if not isinstance(value, MutableMapping):
             raise ValueError("No rules specified for rules parameter.")
 
         if "rules" not in value:
@@ -2783,7 +2767,7 @@ def write_elements_to_collection(has_elements, collection_builder):
 
 def runtime_to_json(runtime_value):
     if isinstance(runtime_value, ConnectedValue) or (
-        isinstance(runtime_value, dict) and runtime_value["__class__"] == "ConnectedValue"
+        isinstance(runtime_value, MutableMapping) and runtime_value["__class__"] == "ConnectedValue"
     ):
         return {"__class__": "ConnectedValue"}
     else:
@@ -2792,7 +2776,7 @@ def runtime_to_json(runtime_value):
 
 def runtime_to_object(runtime_value):
     if isinstance(runtime_value, ConnectedValue) or (
-        isinstance(runtime_value, dict) and runtime_value["__class__"] == "ConnectedValue"
+        isinstance(runtime_value, MutableMapping) and runtime_value["__class__"] == "ConnectedValue"
     ):
         return ConnectedValue()
     else:
@@ -2809,3 +2793,30 @@ class ConnectedValue(RuntimeValue):
     """
     Wrapper to note a value that is not yet set, but will be inferred from a connection.
     """
+
+
+def history_item_dict_to_python(value, app, name):
+    if isinstance(value, MutableMapping) and "src" in value:
+        if value["src"] not in ("hda", "dce", "ldda", "hdca"):
+            raise ParameterValueError(f"Invalid value {value}", name)
+        return src_id_to_item(sa_session=app.model.context, security=app.security, value=value)
+
+
+def history_item_to_json(value, app, use_security):
+    src = None
+    if isinstance(value, MutableMapping) and "src" in value and "id" in value:
+        return value
+    elif isinstance(value, DatasetCollectionElement):
+        src = "dce"
+    elif isinstance(value, HistoryDatasetCollectionAssociation):
+        src = "hdca"
+    elif isinstance(value, LibraryDatasetDatasetAssociation):
+        src = "ldda"
+    elif isinstance(value, HistoryDatasetAssociation) or hasattr(value, "id"):
+        # hasattr 'id' fires a query on persistent objects after a flush so better
+        # to do the isinstance check. Not sure we need the hasattr check anymore - it'd be
+        # nice to drop it.
+        src = "hda"
+    if src is not None:
+        object_id = cached_id(value)
+        return {"id": app.security.encode_id(object_id) if use_security else object_id, "src": src}
diff --git a/lib/galaxy/tools/parameters/dynamic_options.py b/lib/galaxy/tools/parameters/dynamic_options.py
index f02c1dbe0bef..c93665945d26 100644
--- a/lib/galaxy/tools/parameters/dynamic_options.py
+++ b/lib/galaxy/tools/parameters/dynamic_options.py
@@ -28,6 +28,7 @@
     User,
 )
 from galaxy.tools.expressions import do_eval
+from galaxy.tools.parameters.workflow_building_modes import workflow_building_modes
 from galaxy.util import (
     Element,
     string_as_bool,
@@ -748,6 +749,8 @@ def get_fields(self, trans, other_values):
                     options += self.parse_file_fields(StringIO(contents))
         elif self.tool_data_table:
             options = self.tool_data_table.get_fields()
+            if trans and trans.user and trans.workflow_building_mode != workflow_building_modes.ENABLED:
+                options += self.get_user_options(trans.user)
         elif self.file_fields:
             options = list(self.file_fields)
         else:
@@ -756,6 +759,39 @@ def get_fields(self, trans, other_values):
             options = filter.filter_options(options, trans, other_values)
         return options
 
+    def get_user_options(self, user: User):
+        # stored metadata are key: value pairs, turn into flat lists of correct order
+        fields = []
+        if self.tool_data_table_name:
+            hdas = user.get_user_data_tables(self.tool_data_table_name)
+            by_dbkey = {}
+            for hda in hdas:
+                by_dbkey.update(self.hda_to_table_entries(hda, self.tool_data_table_name))
+            for data_table_entry in by_dbkey.values():
+                field_entry = []
+                for column_key in self.tool_data_table.columns.keys():
+                    field_entry.append(data_table_entry[column_key])
+                fields.append(field_entry)
+        return fields
+
+    @staticmethod
+    def hda_to_table_entries(hda, table_name):
+        table_entries = {}
+        for value in hda._metadata["data_tables"][table_name]:
+            if dbkey := value.get("dbkey"):
+                table_entries[dbkey] = value
+            if path := value.get("path"):
+                # maybe a hack, should probably pass around dataset or src id combinations ?
+                value["path"] = os.path.join(hda.extra_files_path, path)
+                value["value"] = {"src": "hda", "id": hda.id}
+        return table_entries
+
+    def get_option_from_dataset(self, dataset):
+        # TODO: we may have to pass the name/id in case there are multiple entries produced by a single dm run
+        entries = self.hda_to_table_entries(dataset, self.tool_data_table_name)
+        assert len(entries) == 1, "Cannot pass tool data bundle with more than 1 data entry per table"
+        return next(iter(entries.values()))
+
     def get_fields_by_value(self, value, trans, other_values):
         """
         Return a list of fields with column 'value' matching provided value.
diff --git a/lib/galaxy/tools/parameters/workflow_building_modes.py b/lib/galaxy/tools/parameters/workflow_building_modes.py
new file mode 100644
index 000000000000..ff3be9c7d9fd
--- /dev/null
+++ b/lib/galaxy/tools/parameters/workflow_building_modes.py
@@ -0,0 +1,4 @@
+class workflow_building_modes:
+    DISABLED = False
+    ENABLED = True
+    USE_HISTORY = 1
diff --git a/lib/galaxy/tools/recommendations.py b/lib/galaxy/tools/recommendations.py
index 6443ea2f987b..f4041988afd2 100644
--- a/lib/galaxy/tools/recommendations.py
+++ b/lib/galaxy/tools/recommendations.py
@@ -10,7 +10,7 @@
 import yaml
 
 from galaxy.tools.parameters import populate_state
-from galaxy.tools.parameters.basic import workflow_building_modes
+from galaxy.tools.parameters.workflow_building_modes import workflow_building_modes
 from galaxy.util import DEFAULT_SOCKET_TIMEOUT
 from galaxy.workflow.modules import module_factory
 
diff --git a/lib/galaxy/tools/wrappers.py b/lib/galaxy/tools/wrappers.py
index c3c7ec40bce0..547bf6cbc2a2 100644
--- a/lib/galaxy/tools/wrappers.py
+++ b/lib/galaxy/tools/wrappers.py
@@ -214,14 +214,17 @@ def __init__(
             self._input = input
             self._value = value
             self._other_values = other_values
-            self._fields: Dict[str, str] = {}
+            self._fields: Dict[str, List[str]] = {}
             self._compute_environment = compute_environment
 
         def __getattr__(self, name: str) -> Any:
             if name not in self._fields:
-                self._fields[name] = self._input.options.get_field_by_name_for_value(
-                    name, self._value, None, self._other_values
-                )
+                if isinstance(self._value, DatasetInstance):
+                    self._fields[name] = [self._input.options.get_option_from_dataset(self._value)[name]]
+                else:
+                    self._fields[name] = self._input.options.get_field_by_name_for_value(
+                        name, self._value, None, self._other_values
+                    )
             values = map(str, self._fields[name])
             if name in PATH_ATTRIBUTES and self._compute_environment:
                 # If we infer this is a path, rewrite it if needed.
diff --git a/lib/galaxy/webapps/galaxy/api/workflows.py b/lib/galaxy/webapps/galaxy/api/workflows.py
index 66bbeecb076c..98b1f34cb86f 100644
--- a/lib/galaxy/webapps/galaxy/api/workflows.py
+++ b/lib/galaxy/webapps/galaxy/api/workflows.py
@@ -77,7 +77,7 @@
 from galaxy.tool_shed.galaxy_install.install_manager import InstallRepositoryManager
 from galaxy.tools import recommendations
 from galaxy.tools.parameters import populate_state
-from galaxy.tools.parameters.basic import workflow_building_modes
+from galaxy.tools.parameters.workflow_building_modes import workflow_building_modes
 from galaxy.util.sanitize_html import sanitize_html
 from galaxy.version import VERSION
 from galaxy.web import (
diff --git a/lib/galaxy/webapps/galaxy/controllers/workflow.py b/lib/galaxy/webapps/galaxy/controllers/workflow.py
index 2d2eec42e5d2..5ace9ae80e93 100644
--- a/lib/galaxy/webapps/galaxy/controllers/workflow.py
+++ b/lib/galaxy/webapps/galaxy/controllers/workflow.py
@@ -17,7 +17,7 @@
 )
 from galaxy.model.base import transaction
 from galaxy.model.item_attrs import UsesItemRatings
-from galaxy.tools.parameters.basic import workflow_building_modes
+from galaxy.tools.parameters.workflow_building_modes import workflow_building_modes
 from galaxy.util import FILENAME_VALID_CHARS
 from galaxy.util.sanitize_html import sanitize_html
 from galaxy.web import url_for
diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py
index b823ce3d5921..da53d9de6159 100644
--- a/lib/galaxy/workflow/modules.py
+++ b/lib/galaxy/workflow/modules.py
@@ -80,13 +80,13 @@
     runtime_to_json,
     SelectToolParameter,
     TextToolParameter,
-    workflow_building_modes,
 )
 from galaxy.tools.parameters.grouping import (
     Conditional,
     ConditionalWhen,
 )
 from galaxy.tools.parameters.history_query import HistoryQuery
+from galaxy.tools.parameters.workflow_building_modes import workflow_building_modes
 from galaxy.tools.parameters.wrapped import make_dict_copy
 from galaxy.util import (
     listify,
diff --git a/test/functional/tools/sample_datatypes_conf.xml b/test/functional/tools/sample_datatypes_conf.xml
index d115b669ec28..802c286be68f 100644
--- a/test/functional/tools/sample_datatypes_conf.xml
+++ b/test/functional/tools/sample_datatypes_conf.xml
@@ -48,7 +48,7 @@
     <datatype extension="bgzip" type="galaxy.datatypes.binary:Binary" subclass="true" />
     <datatype extension="vcf_bgzip" type="galaxy.datatypes.tabular:VcfGz" display_in_upload="true"/>
     <datatype extension="html" type="galaxy.datatypes.text:Html" mimetype="text/html"/>
-    <datatype extension="data_manager_json" type="galaxy.datatypes.text:Json" mimetype="application/json" subclass="true" display_in_upload="false"/>
+    <datatype extension="data_manager_json" type="galaxy.datatypes.text:DataManagerJson" mimetype="application/json" subclass="true" display_in_upload="false"/>
     <datatype extension="data" type="galaxy.datatypes.data:Data" mimetype="application/octet-stream" max_optional_metadata_filesize="1048576" />
     <datatype extension="binary" type="galaxy.datatypes.binary:Binary" mimetype="application/octet-stream" max_optional_metadata_filesize="1048576" />
     <datatype extension="shp" type="galaxy.datatypes.gis:Shapefile" mimetype="application/octet-stream" display_in_upload="true"/>
diff --git a/test/integration/test_data_manager_workflow_bundle.py b/test/integration/test_data_manager_workflow_bundle.py
new file mode 100644
index 000000000000..dd7262906646
--- /dev/null
+++ b/test/integration/test_data_manager_workflow_bundle.py
@@ -0,0 +1,105 @@
+"""Integration tests for workflow syncing."""
+
+from galaxy_test.base.populators import (
+    DatasetPopulator,
+    WorkflowPopulator,
+)
+from galaxy_test.base.uses_shed_api import UsesShedApi
+from .test_containerized_jobs import ContainerizedIntegrationTestCase
+
+
+class TestDataManagerWorkflowInvocation(ContainerizedIntegrationTestCase, UsesShedApi):
+    dataset_populator: DatasetPopulator
+    framework_tool_and_types = True
+    require_admin_user = False
+    admin_user_email = "data-manager@galaxyproject.org"
+    container_type = "docker"
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        super().handle_galaxy_config_kwds(config)
+        config["check_migrate_databases"] = False
+        config["admin_users"] = cls.admin_user_email
+
+    def setUp(self):
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+        self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
+
+    def test_run_data_manager_workflow(self):
+        self.install_repository("devteam", "data_manager_fetch_genome_dbkeys_all_fasta", "4d3eff1bc421")
+        self.install_repository("devteam", "data_manager_bwa_mem_index_builder", "9e993022c762")
+        with self._different_user(email=self.admin_user_email), self.dataset_populator.test_history() as history_id:
+            self.workflow_populator.run_workflow(
+                """
+class: GalaxyWorkflow
+label: Indexing workflow
+inputs:
+  dbkey:
+    optional: false
+    type: string
+  display name:
+    optional: false
+    type: string
+  name:
+    optional: false
+    type: string
+  id:
+    optional: false
+    type: string
+  url:
+    optional: false
+    type: string
+steps:
+  fasta_index:
+    tool_id: toolshed.g2.bx.psu.edu/repos/devteam/data_manager_fetch_genome_dbkeys_all_fasta/data_manager_fetch_genome_all_fasta_dbkey/0.0.4
+    tool_version: 0.0.4
+    tool_state:
+      dbkey_source:
+        dbkey_source_selector: new
+      reference_source:
+        reference_source_selector: url
+      sorting:
+        sort_selector: as_is
+      __data_manager_mode: bundle
+    in:
+      dbkey_source|dbkey:
+        source: dbkey
+      dbkey_source|dbkey_name:
+        source: display name
+      reference_source|user_url:
+        source: url
+      sequence_id:
+        source: id
+      sequence_name:
+        source: name
+
+  bwa_mem_index:
+    tool_id: toolshed.g2.bx.psu.edu/repos/devteam/data_manager_bwa_mem_index_builder/bwa_mem_index_builder_data_manager/0.0.5
+    tool_version: 0.0.5
+    tool_state:
+      __data_manager_mode: bundle
+    in:
+      all_fasta_source: fasta_index/out_file
+
+test_data:
+    dbkey:
+        value: wf_bundle
+        type: raw
+    display name:
+        value: display name value
+        type: raw
+    id:
+        value: wf_bundle_id
+        type: raw
+    name:
+        value: workflow bundle
+        type: raw
+    url:
+        value: https://raw.githubusercontent.com/galaxyproject/galaxy/dev/test-data/1.fasta
+        type: raw
+""",
+                history_id=history_id,
+                wait=True,
+                assert_ok=True,
+            )
diff --git a/test/integration/test_workflow_refactoring.py b/test/integration/test_workflow_refactoring.py
index 4d9066d988ad..ba26b2a856ba 100644
--- a/test/integration/test_workflow_refactoring.py
+++ b/test/integration/test_workflow_refactoring.py
@@ -21,7 +21,7 @@
     WorkflowStepConnection,
 )
 from galaxy.model.base import transaction
-from galaxy.tools.parameters.basic import workflow_building_modes
+from galaxy.tools.parameters.workflow_building_modes import workflow_building_modes
 from galaxy.workflow.refactor.schema import RefactorActionExecutionMessageTypeEnum
 from galaxy_test.base.populators import WorkflowPopulator
 from galaxy_test.base.uses_shed_api import UsesShedApi
diff --git a/test/unit/app/tools/test_select_parameters.py b/test/unit/app/tools/test_select_parameters.py
index b13abb191d29..1f3bff215c96 100644
--- a/test/unit/app/tools/test_select_parameters.py
+++ b/test/unit/app/tools/test_select_parameters.py
@@ -68,6 +68,7 @@ def setUp(self):
             get_current_user_roles=lambda: [],
             workflow_building_mode=False,
             webapp=Mock(name="galaxy"),
+            user=None,
         )
         self.type = "select"
         self.set_data_ref = False