diff --git a/lib/galaxy/datatypes/text.py b/lib/galaxy/datatypes/text.py
index 872ce2b0d4ab..8a87cf1166f2 100644
--- a/lib/galaxy/datatypes/text.py
+++ b/lib/galaxy/datatypes/text.py
@@ -137,6 +137,7 @@ class DataManagerJson(Json):
     MetadataElement(
         name="data_tables", default=None, desc="Data tables represented by this dataset", readonly=True, visible=True
     )
+    MetadataElement(name="is_bundle", default=False, desc="Dataset represents bundle", readonly=True, visible=True)
 
     def set_meta(self, dataset: DatasetProtocol, overwrite: bool = True, **kwd):
         super().set_meta(dataset=dataset, overwrite=overwrite, **kwd)
diff --git a/lib/galaxy/managers/workflows.py b/lib/galaxy/managers/workflows.py
index 60680798057b..85b2cd981c2c 100644
--- a/lib/galaxy/managers/workflows.py
+++ b/lib/galaxy/managers/workflows.py
@@ -31,8 +31,6 @@
 )
 from sqlalchemy import (
     and_,
-    Cast,
-    ColumnElement,
     desc,
     false,
     func,
@@ -40,7 +38,6 @@
     select,
     true,
 )
-from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import (
     aliased,
     joinedload,
@@ -75,6 +72,7 @@
     StoredWorkflow,
     StoredWorkflowTagAssociation,
     StoredWorkflowUserShareAssociation,
+    to_json,
     User,
     Workflow,
     WorkflowInvocation,
@@ -2070,26 +2068,14 @@ def get_workflow_by_trs_id_and_version(
     ) -> Optional[model.StoredWorkflow]:
         sa_session = self.app.model.session
 
-        def to_json(column, keys: List[str]):
-            assert sa_session.bind
-            if sa_session.bind.dialect.name == "postgresql":
-                cast: Union[ColumnElement[Any], Cast[Any]] = func.cast(func.convert_from(column, "UTF8"), JSONB)
-                for key in keys:
-                    cast = cast.__getitem__(key)
-                return cast.astext
-            else:
-                for key in keys:
-                    column = func.json_extract(column, f"$.{key}")
-                return column
-
         stmnt = (
             select(model.StoredWorkflow)
             .join(model.Workflow, model.Workflow.id == model.StoredWorkflow.latest_workflow_id)
             .filter(
                 and_(
                     model.StoredWorkflow.deleted == false(),
-                    to_json(model.Workflow.source_metadata, ["trs_tool_id"]) == trs_id,
-                    to_json(model.Workflow.source_metadata, ["trs_version_id"]) == trs_version,
+                    to_json(sa_session, model.Workflow.source_metadata, ["trs_tool_id"]) == trs_id,
+                    to_json(sa_session, model.Workflow.source_metadata, ["trs_version_id"]) == trs_version,
                 )
             )
         )
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 119454c93963..26d301024fca 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -68,8 +68,10 @@
     bindparam,
     Boolean,
     case,
+    Cast,
     Column,
     column,
+    ColumnElement,
     DateTime,
     delete,
     desc,
@@ -100,6 +102,7 @@
     update,
     VARCHAR,
 )
+from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.exc import (
     CompileError,
     OperationalError,
@@ -315,6 +318,19 @@ def get_uuid(uuid: Optional[Union[UUID, str]] = None) -> UUID:
     return UUID(str(uuid))
 
 
+def to_json(sa_session, column, keys: List[str]):
+    assert sa_session.bind
+    if sa_session.bind.dialect.name == "postgresql":
+        cast: Union[ColumnElement[Any], Cast[Any]] = func.cast(func.convert_from(column, "UTF8"), JSONB)
+        for key in keys:
+            cast = cast.__getitem__(key)
+        return cast.astext
+    else:
+        for key in keys:
+            column = func.json_extract(column, f"$.{key}")
+        return column
+
+
 class Base(DeclarativeBase, _HasTable):
     __abstract__ = True
     metadata = MetaData(naming_convention=NAMING_CONVENTION)
@@ -898,8 +914,8 @@ def get_user_data_tables(self, data_table: str):
                 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),
+                to_json(session, HistoryDatasetAssociation._metadata, ["is_bundle"]) == "true",
             )
             .order_by(HistoryDatasetAssociation.id)
         )
diff --git a/lib/galaxy/tool_util/xsd/galaxy.xsd b/lib/galaxy/tool_util/xsd/galaxy.xsd
index e2b90090bda7..88eb97f82b4a 100644
--- a/lib/galaxy/tool_util/xsd/galaxy.xsd
+++ b/lib/galaxy/tool_util/xsd/galaxy.xsd
@@ -8093,7 +8093,7 @@ favour of a ``has_size`` assertion.</xs:documentation>
   
   <xs:simpleType name="Format">
     <xs:restriction base="xs:string">
-      <xs:pattern value="[0-9a-z.-_]+"/>
+      <xs:pattern value="[0-9a-z._-]+"/>
     </xs:restriction>
   </xs:simpleType>
   <xs:simpleType name="FormatList">
diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py
index 90ccde24679e..e7125c055380 100644
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -3255,6 +3255,7 @@ def exec_after_process(self, app, inp_data, out_data, param_dict, job, final_job
                     create=True,
                     preserve_symlinks=True,
                 )
+                hda.metadata.is_bundle = True
 
         else:
             raise Exception("Unknown data manager mode encountered type...")
diff --git a/lib/galaxy/tools/parameters/basic.py b/lib/galaxy/tools/parameters/basic.py
index 3befa87ae1e3..f977ab585594 100644
--- a/lib/galaxy/tools/parameters/basic.py
+++ b/lib/galaxy/tools/parameters/basic.py
@@ -13,10 +13,11 @@
 from collections.abc import MutableMapping
 from typing import (
     Any,
+    cast,
     Dict,
     List,
     Optional,
-    Tuple,
+    Sequence,
     TYPE_CHECKING,
     Union,
 )
@@ -43,12 +44,14 @@
 from galaxy.schema.fetch_data import FilesPayload
 from galaxy.tool_util.parameters.factory import get_color_value
 from galaxy.tool_util.parser import get_input_source as ensure_input_source
+from galaxy.tool_util.parser.interface import DrillDownOptionsDict
 from galaxy.tool_util.parser.util import (
     boolean_is_checked,
     boolean_true_and_false_values,
     ParameterParseException,
     text_input_is_optional,
 )
+from galaxy.tools.parameters.options import ParameterOption
 from galaxy.tools.parameters.workflow_utils import (
     NO_REPLACEMENT,
     workflow_building_modes,
@@ -131,6 +134,10 @@ def parse_dynamic_options(param, input_source):
     return dynamic_options.DynamicOptions(options_elem, param)
 
 
+def serialize_options(security: "IdEncodingHelper", options: Sequence[ParameterOption]):
+    return [option.serialize(security) for option in options]
+
+
 # Describe a parameter value error where there is no actual supplied
 # parameter - e.g. just a specification issue.
 NO_PARAMETER_VALUE = object()
@@ -175,7 +182,7 @@ class ToolParameter(UsesDictVisibleKeys):
 
     >>> from galaxy.util.bunch import Bunch
     >>> from galaxy.util import XML
-    >>> trans = Bunch(app=None)
+    >>> trans = Bunch(app=None, security=lambda x: x)
     >>> p = ToolParameter(None, XML('<param argument="--parameter-name" type="text" value="default" />'))
     >>> assert p.name == 'parameter_name'
     >>> assert sorted(p.to_dict(trans).items()) == [('argument', '--parameter-name'), ('help', ''), ('help_format', 'html'), ('hidden', False), ('is_dynamic', False), ('label', ''), ('model_class', 'ToolParameter'), ('name', 'parameter_name'), ('optional', False), ('refresh_on_change', False), ('type', 'text'), ('value', None)]
@@ -924,7 +931,7 @@ class SelectToolParameter(ToolParameter):
 
     >>> from galaxy.util.bunch import Bunch
     >>> from galaxy.util import XML
-    >>> trans = Bunch(app=None, history=Bunch(), workflow_building_mode=False)
+    >>> trans = Bunch(app=None, history=Bunch(), workflow_building_mode=False, security=lambda x: x)
     >>> p = SelectToolParameter(None, XML(
     ... '''
     ... <param name="_name" type="select">
@@ -982,13 +989,15 @@ def _get_dynamic_options_call_other_values(self, trans, other_values):
             call_other_values.update(other_values.dict)
         return call_other_values
 
-    def get_options(self, trans, other_values):
+    def get_options(self, trans, other_values) -> Sequence[Union[ParameterOption, DrillDownOptionsDict]]:
         if self.options:
             return self.options.get_options(trans, other_values)
         elif self.dynamic_options:
             call_other_values = self._get_dynamic_options_call_other_values(trans, other_values)
             try:
-                return eval(self.dynamic_options, self.tool.code_namespace, call_other_values)
+                return [
+                    ParameterOption(*o) for o in eval(self.dynamic_options, self.tool.code_namespace, call_other_values)
+                ]
             except Exception as e:
                 log.debug(
                     "Error determining dynamic options for parameter '%s' in tool '%s':",
@@ -998,22 +1007,21 @@ def get_options(self, trans, other_values):
                 )
                 return []
         else:
-            return self.static_options
+            return [ParameterOption(*o) for o in self.static_options]
 
     def get_legal_values(self, trans, other_values, value):
         """
         determine the set of values of legal options
         """
-        return {
-            history_item_dict_to_python(v, trans.app, self.name) or v
-            for _, v, _ in self.get_options(trans, other_values)
-        }
+        options = cast(List[ParameterOption], self.get_options(trans, other_values))
+        return {option.dataset or option.value for option in options}
 
     def get_legal_names(self, trans, other_values):
         """
-        determine a mapping from names to values for all legal options
+        determine the set of values of legal options
         """
-        return {n: v for n, v, _ in self.get_options(trans, other_values)}
+        options = cast(List[ParameterOption], self.get_options(trans, other_values))
+        return {option.name: option.value for option in options}
 
     def from_json(self, value, trans, other_values=None):
         return self._select_from_json(value, trans, other_values=other_values, require_legal_value=True)
@@ -1135,17 +1143,17 @@ def to_python(self, value, app):
 
     def get_initial_value(self, trans, other_values):
         try:
-            options = list(self.get_options(trans, other_values))
+            options = cast(List[ParameterOption], self.get_options(trans, other_values))
         except ImplicitConversionRequired:
             return None
         if not options:
             return None
-        value = [optval for _, optval, selected in options if selected]
+        value = [option.value for option in options if option.selected]
         if len(value) == 0:
             if not self.optional and not self.multiple and options:
                 # Nothing selected, but not optional and not a multiple select, with some values,
                 # so we have to default to something (the HTML form will anyway)
-                value2 = options[0][1]
+                value2: Optional[Union[str, List[str]]] = options[0].value
             else:
                 value2 = None
         elif len(value) == 1 or not self.multiple:
@@ -1185,8 +1193,8 @@ def to_dict(self, trans, other_values=None):
         d = super().to_dict(trans, other_values)
 
         # Get options, value.
-        options = self.get_options(trans, other_values)
-        d["options"] = options
+        options = cast(List[ParameterOption], self.get_options(trans, other_values))
+        d["options"] = serialize_options(trans.security, options)
         d["display"] = self.display
         d["multiple"] = self.multiple
         d["textable"] = is_runtime_context(trans, other_values)
@@ -1211,7 +1219,7 @@ class GenomeBuildParameter(SelectToolParameter):
     >>> # Create a mock transaction with 'hg17' as the current build
     >>> from galaxy.util.bunch import Bunch
     >>> from galaxy.util import XML
-    >>> trans = Bunch(app=None, history=Bunch(genome_build='hg17'), db_builds=read_dbnames(None))
+    >>> trans = Bunch(app=None, history=Bunch(genome_build='hg17'), db_builds=read_dbnames(None), security=lambda x:x)
     >>> p = GenomeBuildParameter(None, XML('<param name="_name" type="genomebuild" value="hg17" />'))
     >>> print(p.name)
     _name
@@ -1231,12 +1239,14 @@ def __init__(self, *args, **kwds):
             self.static_options = [(value, key, False) for key, value in self._get_dbkey_names()]
         self.is_dynamic = True
 
-    def get_options(self, trans, other_values):
+    def get_options(self, trans, other_values) -> Sequence[ParameterOption]:
         last_used_build = object()
         if trans.history:
             last_used_build = trans.history.genome_build
-        for dbkey, build_name in self._get_dbkey_names(trans=trans):
-            yield build_name, dbkey, (dbkey == last_used_build)
+        return [
+            ParameterOption(build_name, dbkey, (dbkey == last_used_build))
+            for dbkey, build_name in self._get_dbkey_names(trans=trans)
+        ]
 
     def get_legal_values(self, trans, other_values, value):
         return {dbkey for dbkey, _ in self._get_dbkey_names(trans=trans)}
@@ -1246,16 +1256,16 @@ def to_dict(self, trans, other_values=None):
         d = ToolParameter.to_dict(self, trans)
 
         # Get options, value - options is a generator here, so compile to list
-        options = list(self.get_options(trans, {}))
-        value = options[0][1]
+        options = self.get_options(trans, {})
+        value = options[0].value
         for option in options:
-            if option[2]:
+            if option.selected:
                 # Found selected option.
-                value = option[1]
+                value = option.value
 
         d.update(
             {
-                "options": options,
+                "options": serialize_options(trans, options),
                 "value": value,
                 "display": self.display,
                 "multiple": self.multiple,
@@ -1343,13 +1353,13 @@ def get_tag_list(self, other_values):
                         tags.add(tag.user_value)
         return list(tags)
 
-    def get_options(self, trans, other_values):
+    def get_options(self, trans, other_values) -> Sequence[ParameterOption]:
         """
         Show tags
         """
         options = []
         for tag in self.get_tag_list(other_values):
-            options.append((f"Tags: {tag}", tag, False))
+            options.append(ParameterOption(f"Tags: {tag}", tag, False))
         return options
 
     def get_initial_value(self, trans, other_values):
@@ -1516,11 +1526,11 @@ def get_column_list(self, trans, other_values):
                 column_list = [c for c in column_list if c in this_column_list]
         return column_list
 
-    def get_options(self, trans, other_values):
+    def get_options(self, trans, other_values) -> Sequence[ParameterOption]:
         """
         Show column labels rather than c1..cn if use_header_names=True
         """
-        options: List[Tuple[str, Union[str, Tuple[str, str]], bool]] = []
+        options: Sequence[ParameterOption] = []
         column_list = self.get_column_list(trans, other_values)
         if not column_list:
             return options
@@ -1534,7 +1544,10 @@ def get_options(self, trans, other_values):
                 and dataset.metadata.element_is_set("column_names")
             ):
                 try:
-                    options = [(f"c{c}: {dataset.metadata.column_names[int(c) - 1]}", c, False) for c in column_list]
+                    options = [
+                        ParameterOption(f"c{c}: {dataset.metadata.column_names[int(c) - 1]}", c, False)
+                        for c in column_list
+                    ]
                 except IndexError:
                     # ignore and rely on fallback
                     pass
@@ -1543,13 +1556,13 @@ def get_options(self, trans, other_values):
                     with open(dataset.get_file_name()) as f:
                         head = f.readline()
                     cnames = head.rstrip("\n\r ").split("\t")
-                    options = [(f"c{c}: {cnames[int(c) - 1]}", c, False) for c in column_list]
+                    options = [ParameterOption(f"c{c}: {cnames[int(c) - 1]}", c, False) for c in column_list]
                 except Exception:
                     # ignore and rely on fallback
                     pass
         if not options:
             # fallback if no options list could be built so far
-            options = [(f"Column: {col}", col, False) for col in column_list]
+            options = [ParameterOption(f"Column: {col}", col, False) for col in column_list]
         return options
 
     def get_initial_value(self, trans, other_values):
@@ -1612,7 +1625,7 @@ class DrillDownSelectToolParameter(SelectToolParameter):
     >>> from galaxy.util.bunch import Bunch
     >>> app = Bunch(config=Bunch(tool_data_path=None))
     >>> tool = Bunch(app=app)
-    >>> trans = Bunch(app=app, history=Bunch(genome_build='hg17'), db_builds=read_dbnames(None))
+    >>> trans = Bunch(app=app, history=Bunch(genome_build='hg17'), db_builds=read_dbnames(None), security=lambda x: x)
     >>> p = DrillDownSelectToolParameter(tool, XML(
     ... '''
     ... <param name="_name" type="drill_down" display="checkbox" hierarchy="recurse" multiple="true">
@@ -1679,18 +1692,17 @@ def _get_options_from_code(self, trans=None, other_values=None):
         except Exception:
             return []
 
-    def get_options(self, trans=None, other_values=None):
+    def get_options(self, trans=None, other_values=None) -> List[DrillDownOptionsDict]:
         other_values = other_values or {}
         if self.is_dynamic:
             if self.dynamic_options:
-                options = self._get_options_from_code(trans=trans, other_values=other_values)
-            else:
-                options = []
-            return options
+                return self._get_options_from_code(trans=trans, other_values=other_values)
+            return []
+
         return self.options
 
     def get_legal_values(self, trans, other_values, value):
-        def recurse_options(legal_values, options):
+        def recurse_options(legal_values, options: List[DrillDownOptionsDict]):
             for option in options:
                 legal_values.append(option["value"])
                 recurse_options(legal_values, option["options"])
@@ -1736,16 +1748,16 @@ def to_param_dict_string(self, value, other_values=None):
         other_values = other_values or {}
 
         def get_options_list(value):
-            def get_base_option(value, options):
+            def get_base_option(value, options: List[DrillDownOptionsDict]):
                 for option in options:
                     if value == option["value"]:
                         return option
-                    rval = get_base_option(value, option["options"])
+                    rval = get_base_option(value, option["options"] or [])
                     if rval:
                         return rval
                 return None  # not found
 
-            def recurse_option(option_list, option):
+            def recurse_option(option_list, option: DrillDownOptionsDict):
                 if not option["options"]:
                     option_list.append(option["value"])
                 else:
@@ -1753,8 +1765,10 @@ def recurse_option(option_list, option):
                         recurse_option(option_list, opt)
 
             rval: List[str] = []
-            base_option = get_base_option(value, self.get_options(other_values=other_values))
-            recurse_option(rval, base_option)
+            options = self.get_options(other_values=other_values)
+            base_option = get_base_option(value, options)
+            if base_option:
+                recurse_option(rval, base_option)
             return rval or [value]
 
         if value is None:
@@ -1780,11 +1794,11 @@ def recurse_option(option_list, option):
         return rval
 
     def get_initial_value(self, trans, other_values):
-        def recurse_options(initial_values, options):
+        def recurse_options(initial_values, options: List[DrillDownOptionsDict]):
             for option in options:
                 if option["selected"]:
                     initial_values.append(option["value"])
-                recurse_options(initial_values, option["options"])
+                recurse_options(initial_values, option["options"] or [])
 
         # More working around dynamic options for workflow
         options = self.get_options(trans=trans, other_values=other_values)
@@ -1797,11 +1811,11 @@ def recurse_options(initial_values, options):
         return initial_values
 
     def to_text(self, value):
-        def get_option_display(value, options):
+        def get_option_display(value, options: List[DrillDownOptionsDict]):
             for option in options:
                 if value == option["value"]:
                     return option["name"]
-                rval = get_option_display(value, option["options"])
+                rval = get_option_display(value, option["options"] or [])
                 if rval:
                     return rval
             return None  # not found
@@ -1823,7 +1837,7 @@ def get_option_display(value, options):
         else:
             rval = []
             for val in value:
-                rval.append(get_option_display(val, self.options) or val)
+                rval.append(get_option_display(val, self.options or val))
         if rval:
             return "\n".join(map(str, rval))
         return "Nothing selected."
diff --git a/lib/galaxy/tools/parameters/dynamic_options.py b/lib/galaxy/tools/parameters/dynamic_options.py
index bb5f070bd60b..49010bab392b 100644
--- a/lib/galaxy/tools/parameters/dynamic_options.py
+++ b/lib/galaxy/tools/parameters/dynamic_options.py
@@ -15,7 +15,10 @@
     cast,
     Dict,
     get_args,
+    List,
     Optional,
+    Sequence,
+    Set,
 )
 
 from typing_extensions import Literal
@@ -29,6 +32,7 @@
     User,
 )
 from galaxy.tools.expressions import do_eval
+from galaxy.tools.parameters.options import ParameterOption
 from galaxy.tools.parameters.workflow_utils import (
     is_runtime_value,
     workflow_building_modes,
@@ -92,7 +96,7 @@ def __init__(self, d_option, elem):
         self.column = d_option.column_spec_to_index(column)
         self.keep = string_as_bool(elem.get("keep", "True"))
 
-    def filter_options(self, options, trans, other_values):
+    def filter_options(self, options: Sequence[ParameterOption], trans, other_values):
         rval = []
         filter_value = self.value
         try:
@@ -128,7 +132,7 @@ def __init__(self, d_option, elem):
         self.column = d_option.column_spec_to_index(column)
         self.keep = string_as_bool(elem.get("keep", "True"))
 
-    def filter_options(self, options, trans, other_values):
+    def filter_options(self, options: Sequence[ParameterOption], trans, other_values):
         rval = []
         filter_value = self.value
         try:
@@ -184,7 +188,9 @@ def __init__(self, d_option, elem):
     def get_dependency_name(self):
         return self.ref_name
 
-    def filter_options(self, options, trans, other_values):
+    def filter_options(self, options: Sequence[ParameterOption], trans, other_values):
+        options = list(options)
+
         def _add_meta(meta_value, m):
             if isinstance(m, list):
                 meta_value |= set(m)
@@ -224,7 +230,7 @@ def compare_meta_value(file_value, dataset_value):
         # - for data sets: the meta data value
         # in both cases only meta data that is set (i.e. differs from the no_value)
         # is considered
-        meta_value = set()
+        meta_value: Set[Any] = set()
         for r in ref:
             if not r.metadata.element_is_set(self.key):
                 continue
@@ -236,7 +242,7 @@ def compare_meta_value(file_value, dataset_value):
             return copy.deepcopy(options)
 
         if self.column is not None:
-            rval = []
+            rval: List[ParameterOption] = []
             for fields in options:
                 if compare_meta_value(fields[self.column], meta_value):
                     rval.append(fields)
@@ -246,7 +252,7 @@ def compare_meta_value(file_value, dataset_value):
                 self.dynamic_option.columns = {"name": 0, "value": 1, "selected": 2}
                 self.dynamic_option.largest_index = 2
             for value in meta_value:
-                options.append((value, value, False))
+                options.append(ParameterOption(value, value, False))
             return options
 
 
@@ -286,7 +292,7 @@ def __init__(self, d_option, elem):
     def get_dependency_name(self):
         return self.ref_name
 
-    def filter_options(self, options, trans, other_values):
+    def filter_options(self, options: Sequence[ParameterOption], trans, other_values):
         ref = other_values.get(self.ref_name, None)
         if ref is None or is_runtime_value(ref):
             ref = []
@@ -336,7 +342,7 @@ def __init__(self, d_option, elem):
     def get_dependency_name(self):
         return self.dynamic_option.dataset_ref_name
 
-    def filter_options(self, options, trans, other_values):
+    def filter_options(self, options: Sequence[ParameterOption], trans, other_values):
         rval = []
         seen = set()
         for fields in options:
@@ -365,12 +371,17 @@ def __init__(self, d_option, elem):
         assert columns is not None, "Required 'column' attribute missing from filter"
         self.columns = [d_option.column_spec_to_index(column) for column in columns.split(",")]
 
-    def filter_options(self, options, trans, other_values):
+    def filter_options(self, options: Sequence[ParameterOption], trans, other_values):
         rval = []
         for fields in options:
             for column in self.columns:
-                for field in fields[column].split(self.separator):
-                    rval.append(fields[0:column] + [field] + fields[column + 1 :])
+                field = fields[column]
+                if isinstance(field, str):
+                    for split_field in field.split(self.separator):
+                        new_options = list(fields[0:column]) + [split_field] + list(fields[column + 1 :])
+                        # tested in filter_multiple_splitter.xml
+                        option = tuple(new_options)
+                        rval.append(option)
         return rval
 
 
@@ -826,6 +837,19 @@ def get_fields(self, trans, other_values):
             options = filter.filter_options(options, trans, other_values)
         return options
 
+    @staticmethod
+    def to_parameter_options(options):
+        rval: List[ParameterOption] = []
+        for option in options:
+            if isinstance(option, ParameterOption):
+                rval.append(option)
+            else:
+                if len(option) == 1:
+                    rval.append(ParameterOption(option[0], option[0]))
+                else:
+                    rval.append(ParameterOption(*option[:3]))
+        return rval
+
     def get_user_options(self, user: User):
         # stored metadata are key: value pairs, turn into flat lists of correct order
         fields = []
@@ -843,9 +867,19 @@ def get_user_options(self, user: User):
                 by_dbkey.update(table_entries)
             for data_table_entry in by_dbkey.values():
                 field_entry = []
+                if hda := data_table_entry.get("__hda__"):
+                    field_entry.append(hda)
+                missing_columns = False
                 for column_key in self.tool_data_table.columns.keys():
+                    if column_key not in data_table_entry:
+                        # currrent data table definition (as in self.tool_data_table)
+                        # may not match against the data manager bundle.
+                        # Breaking here fixes https://github.com/galaxyproject/galaxy/issues/18749.
+                        missing_columns = True
+                        break
                     field_entry.append(data_table_entry[column_key])
-                fields.append(field_entry)
+                if not missing_columns:
+                    fields.append(field_entry)
         return fields
 
     @staticmethod
@@ -857,7 +891,7 @@ def hda_to_table_entries(hda, table_name):
             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}
+                value["__hda__"] = hda
         return table_entries
 
     def get_option_from_dataset(self, dataset):
@@ -894,15 +928,15 @@ def get_field_by_name_for_value(self, field_name, value, trans, other_values):
                 rval.append(fields[field_index])
         return rval
 
-    def get_options(self, trans, other_values):
+    def get_options(self, trans, other_values) -> Sequence[ParameterOption]:
 
-        rval = []
+        rval: List[ParameterOption] = []
 
-        def to_triple(values):
+        def to_option(values):
             if len(values) == 2:
-                return [str(values[0]), str(values[1]), False]
+                return ParameterOption(str(values[0]), str(values[1]), False)
             else:
-                return [str(values[0]), str(values[1]), bool(values[2])]
+                return ParameterOption(str(values[0]), str(values[1]), bool(values[2]))
 
         if from_url_options := self.from_url_options:
             context = User.user_template_environment(trans.user)
@@ -940,7 +974,7 @@ def to_triple(values):
                     data = []
 
             # We only support the very specific ["name", "value", "selected"] format for now.
-            rval = [to_triple(d) for d in data]
+            rval = [to_option(d) for d in data]
         if (
             self.file_fields is not None
             or self.tool_data_table is not None
@@ -949,11 +983,14 @@ def to_triple(values):
         ):
             options = self.get_fields(trans, other_values)
             for fields in options:
-                rval.append((fields[self.columns["name"]], fields[self.columns["value"]], False))
+                name = fields[self.columns["name"]]
+                value = fields[self.columns["value"]]
+                hda = fields[-1] if isinstance(fields[-1], HistoryDatasetAssociation) else None
+                rval.append(ParameterOption(name, value, False, dataset=hda))
         else:
             for filter in self.filters:
                 rval = filter.filter_options(rval, trans, other_values)
-        return rval
+        return self.to_parameter_options(rval)
 
     def column_spec_to_index(self, column_spec):
         """
diff --git a/lib/galaxy/tools/parameters/options.py b/lib/galaxy/tools/parameters/options.py
new file mode 100644
index 000000000000..4fd23c3fcaed
--- /dev/null
+++ b/lib/galaxy/tools/parameters/options.py
@@ -0,0 +1,29 @@
+from typing import (
+    NamedTuple,
+    Optional,
+)
+
+from galaxy.model import (
+    DatasetInstance,
+    HistoryDatasetAssociation,
+)
+from galaxy.security.idencoding import IdEncodingHelper
+
+
+class ParameterOption(NamedTuple):
+    name: str
+    value: str
+    selected: bool = False
+    dataset: Optional["DatasetInstance"] = None
+
+    def serialize(self, security: "IdEncodingHelper"):
+        if self.dataset:
+            return (
+                self.name,
+                {
+                    "src": "hda" if isinstance(self.dataset, HistoryDatasetAssociation) else "ldda",
+                    "id": security.encode_id(self.dataset.id),
+                },
+                self.selected,
+            )
+        return (self.name, self.value, self.selected)
diff --git a/test/unit/app/tools/test_dynamic_options.py b/test/unit/app/tools/test_dynamic_options.py
index 05ec365a379e..3de57c332e54 100644
--- a/test/unit/app/tools/test_dynamic_options.py
+++ b/test/unit/app/tools/test_dynamic_options.py
@@ -1,5 +1,6 @@
 from galaxy.app_unittest_utils.galaxy_mock import MockApp
 from galaxy.tools.parameters.dynamic_options import DynamicOptions
+from galaxy.tools.parameters.options import ParameterOption
 from galaxy.util import XML
 from galaxy.util.bunch import Bunch
 from galaxy.work.context import WorkRequestContext
@@ -60,4 +61,4 @@ def test_dynamic_option_cache():
             "start_index": 0,
         },
     )
-    assert from_url_option.get_options(trans, {}) == [["chr2L", "23513712", False]]
+    assert from_url_option.get_options(trans, {}) == [ParameterOption("chr2L", "23513712", False)]
diff --git a/test/unit/app/tools/test_evaluation.py b/test/unit/app/tools/test_evaluation.py
index 27e4b4a147b5..f3451cf92d61 100644
--- a/test/unit/app/tools/test_evaluation.py
+++ b/test/unit/app/tools/test_evaluation.py
@@ -27,6 +27,7 @@
     ConditionalWhen,
     Repeat,
 )
+from galaxy.tools.parameters.options import ParameterOption
 from galaxy.util import XML
 from galaxy.util.bunch import Bunch
 from galaxy.util.unittest import TestCase
@@ -171,7 +172,7 @@ def get_field_by_name_for_value(name, value, trans, other_values):
             return ["/old/path/human"]
 
         def get_options(trans, other_values):
-            return [["", "/old/path/human", ""]]
+            return [ParameterOption("", "/old/path/human", False)]
 
         parameter.options = Bunch(get_field_by_name_for_value=get_field_by_name_for_value, get_options=get_options)
         self.tool.set_params({"index_path": parameter})
diff --git a/test/unit/app/tools/test_select_parameters.py b/test/unit/app/tools/test_select_parameters.py
index 30b025137e2b..d2a05f687f6e 100644
--- a/test/unit/app/tools/test_select_parameters.py
+++ b/test/unit/app/tools/test_select_parameters.py
@@ -3,6 +3,7 @@
 import pytest
 
 from galaxy import model
+from galaxy.tools.parameters.options import ParameterOption
 from galaxy.tools.parameters.workflow_utils import RuntimeValue
 from .util import BaseParameterTestCase
 
@@ -41,15 +42,23 @@ def test_unvalidated_datasets(self):
 
     def test_filter_param_value(self):
         self.options_xml = """<options from_data_table="test_table"><filter type="param_value" ref="input_bam" column="0" /></options>"""
-        assert ("testname1", "testpath1", False) in self.param.get_options(self.trans, {"input_bam": "testname1"})
-        assert ("testname2", "testpath2", False) in self.param.get_options(self.trans, {"input_bam": "testname2"})
+        assert ParameterOption("testname1", "testpath1", False) in self.param.get_options(
+            self.trans, {"input_bam": "testname1"}
+        )
+        assert ParameterOption("testname2", "testpath2", False) in self.param.get_options(
+            self.trans, {"input_bam": "testname2"}
+        )
         assert len(self.param.get_options(self.trans, {"input_bam": "testname3"})) == 0
 
     def test_filter_param_value2(self):
         # Same test as above, but filtering on a different column.
         self.options_xml = """<options from_data_table="test_table"><filter type="param_value" ref="input_bam" column="1" /></options>"""
-        assert ("testname1", "testpath1", False) in self.param.get_options(self.trans, {"input_bam": "testpath1"})
-        assert ("testname2", "testpath2", False) in self.param.get_options(self.trans, {"input_bam": "testpath2"})
+        assert ParameterOption("testname1", "testpath1", False) in self.param.get_options(
+            self.trans, {"input_bam": "testpath1"}
+        )
+        assert ParameterOption("testname2", "testpath2", False) in self.param.get_options(
+            self.trans, {"input_bam": "testpath2"}
+        )
         assert len(self.param.get_options(self.trans, {"input_bam": "testpath3"})) == 0
 
     # TODO: Good deal of overlap here with TestDataToolParameter, refactor.