Tool Request API...

.github/workflows/framework_tools.yaml

@@ -28,6 +28,7 @@ jobs:
     strategy:
       matrix:
         python-version: ['3.9']
+        use-legacy-api: ['if_needed', 'always']
     services:
       postgres:
         image: postgres:17
@@ -61,7 +62,7 @@ jobs:
           path: 'galaxy root/.venv'
           key: gxy-venv-${{ runner.os }}-${{ steps.full-python-version.outputs.version }}-${{ hashFiles('galaxy root/requirements.txt') }}-framework
       - name: Run tests
-        run: ./run_tests.sh --coverage --framework-tools
+        run: GALAXY_TEST_USE_LEGACY_TOOL_API="${{ matrix.use-legacy-api }}" ./run_tests.sh --coverage --framework-tools
         working-directory: 'galaxy root'
       - uses: codecov/codecov-action@v5
         with:

lib/galaxy/app.py

@@ -688,6 +688,10 @@ class GalaxyManagerApplication(MinimalManagerApp, MinimalGalaxyApplication):
         self._register_singleton(Registry, self.datatypes_registry)
         galaxy.model.set_datatypes_registry(self.datatypes_registry)
         self.configure_sentry_client()
+        # Load dbkey / genome build manager
+        self._configure_genome_builds(data_table_name="__dbkeys__", load_old_style=True)
+        # Tool Data Tables
+        self._configure_tool_data_tables(from_shed_config=False)
 
         self._configure_tool_shed_registry()
         self._register_singleton(tool_shed_registry.Registry, self.tool_shed_registry)
@@ -766,10 +770,6 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT
         )
         self.api_keys_manager = self._register_singleton(ApiKeyManager)
 
-        # Setup lazy variables
-        self._configure_tool_data_tables(from_shed_config=False)
-        self._configure_genome_builds()
-
         # Genomes
         self.genomes = self._register_singleton(Genomes)
         # Data providers registry.

lib/galaxy/celery/tasks.py

@@ -31,6 +31,7 @@ from galaxy.managers.datasets import (
     DatasetManager,
 )
 from galaxy.managers.hdas import HDAManager
+from galaxy.managers.jobs import JobSubmitter
 from galaxy.managers.lddas import LDDAManager
 from galaxy.managers.markdown_util import generate_branded_pdf
 from galaxy.managers.model_stores import ModelStoreManager
@@ -56,6 +57,7 @@ from galaxy.schema.tasks import (
     MaterializeDatasetInstanceTaskRequest,
     PrepareDatasetCollectionDownload,
     PurgeDatasetsTaskRequest,
+    QueueJobs,
     SetupHistoryExportJob,
     WriteHistoryContentTo,
     WriteHistoryTo,
@@ -72,8 +74,10 @@ log = get_logger(__name__)
 
 
 @lru_cache
-def cached_create_tool_from_representation(app: MinimalManagerApp, raw_tool_source: str):
-    return create_tool_from_representation(app=app, raw_tool_source=raw_tool_source, tool_source_class="XmlToolSource")
+def cached_create_tool_from_representation(app: MinimalManagerApp, raw_tool_source: str, tool_dir: str = ""):
+    return create_tool_from_representation(
+        app=app, raw_tool_source=raw_tool_source, tool_dir=tool_dir, tool_source_class="XmlToolSource"
+    )
 
 
 @galaxy_task(action="recalculate a user's disk usage")
@@ -328,6 +332,17 @@ def fetch_data(
     return abort_when_job_stops(_fetch_data, session=sa_session, job_id=job_id, setup_return=setup_return)
 
 
+@galaxy_task(action="queuing up submitted jobs")
+def queue_jobs(request: QueueJobs, app: MinimalManagerApp, job_submitter: JobSubmitter):
+    tool = cached_create_tool_from_representation(
+        app, request.tool_source.raw_tool_source, tool_dir=request.tool_source.tool_dir
+    )
+    job_submitter.queue_jobs(
+        tool,
+        request,
+    )
+
+
 @galaxy_task(ignore_result=True, action="setting up export history job")
 def export_history(
     model_store_manager: ModelStoreManager,

lib/galaxy/managers/jobs.py

@@ -1,6 +1,7 @@
 import json
 import logging
 from collections.abc import Iterable
+from dataclasses import dataclass
 from datetime import (
     date,
     datetime,
@@ -10,6 +11,7 @@ from typing import (
     Any,
     cast,
     Optional,
+    Tuple,
     TYPE_CHECKING,
     Union,
 )
@@ -36,6 +38,7 @@ from typing_extensions import TypedDict
 from galaxy import model
 from galaxy.exceptions import (
     ConfigDoesNotAllowException,
+    InconsistentDatabase,
     ItemAccessibilityException,
     ObjectNotFound,
     RequestParameterInvalidException,
@@ -48,14 +51,20 @@ from galaxy.job_metrics import (
 from galaxy.managers.collections import DatasetCollectionManager
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.managers.datasets import DatasetManager
-from galaxy.managers.hdas import HDAManager
+from galaxy.managers.hdas import (
+    dereference_input,
+    HDAManager,
+)
+from galaxy.managers.histories import HistoryManager
 from galaxy.managers.lddas import LDDAManager
+from galaxy.managers.users import UserManager
 from galaxy.model import (
     ImplicitCollectionJobs,
     ImplicitCollectionJobsJobAssociation,
     Job,
     JobMetricNumeric,
     JobParameter,
+    ToolRequest,
     User,
     Workflow,
     WorkflowInvocation,
@@ -72,8 +81,23 @@ from galaxy.schema.schema import (
     JobIndexQueryPayload,
     JobIndexSortByEnum,
 )
+from galaxy.schema.tasks import (
+    MaterializeDatasetInstanceTaskRequest,
+    QueueJobs,
+)
 from galaxy.security.idencoding import IdEncodingHelper
-from galaxy.structured_app import StructuredApp
+from galaxy.structured_app import (
+    MinimalManagerApp,
+    StructuredApp,
+)
+from galaxy.tool_util.parameters import (
+    DataRequestInternalHda,
+    DataRequestUri,
+    dereference,
+    RequestInternalDereferencedToolState,
+    RequestInternalToolState,
+)
+from galaxy.tools import Tool
 from galaxy.tools._types import (
     ToolStateDumpedToJsonInternalT,
     ToolStateJobInstancePopulatedT,
@@ -89,6 +113,7 @@ from galaxy.util.search import (
     parse_filters_structured,
     RawTextTerm,
 )
+from galaxy.work.context import WorkRequestContext
 
 if TYPE_CHECKING:
     from sqlalchemy.sql.expression import Select
@@ -155,6 +180,8 @@ class JobManager:
         workflow_id = payload.workflow_id
         invocation_id = payload.invocation_id
         implicit_collection_jobs_id = payload.implicit_collection_jobs_id
+        tool_request_id = payload.tool_request_id
+
         search = payload.search
         order_by = payload.order_by
 
@@ -171,6 +198,7 @@ class JobManager:
 
         def add_workflow_jobs():
             wfi_step = select(WorkflowInvocationStep)
+
             if workflow_id is not None:
                 wfi_step = (
                     wfi_step.join(WorkflowInvocation).join(Workflow).where(Workflow.stored_workflow_id == workflow_id)
@@ -185,6 +213,7 @@ class JobManager:
                 ImplicitCollectionJobsJobAssociation.implicit_collection_jobs_id
                 == wfi_step_sq.c.implicit_collection_jobs_id,
             )
+
             # Ensure the result is models, not tuples
             sq = stmt1.union(stmt2).subquery()
             # SQLite won't recognize Job.foo as a valid column for the ORDER BY clause due to the UNION clause, so we'll use the subquery `columns` collection (`sq.c`).
@@ -262,6 +291,9 @@ class JobManager:
         if history_id is not None:
             stmt = stmt.where(Job.history_id == history_id)
 
+        if tool_request_id is not None:
+            stmt = stmt.filter(model.Job.tool_request_id == tool_request_id)
+
         order_by_columns = Job
         if workflow_id or invocation_id:
             stmt, order_by_columns = add_workflow_jobs()
@@ -1910,3 +1942,107 @@ def get_jobs_to_check_at_startup(session: galaxy_scoped_session, track_jobs_in_d
 def get_job(session: galaxy_scoped_session, *where_clauses):
     stmt = select(Job).where(*where_clauses).limit(1)
     return session.scalars(stmt).first()
+
+
+@dataclass
+class DereferencedDatasetPair:
+    hda: model.HistoryDatasetAssociation
+    request: DataRequestUri
+
+
+class JobSubmitter:
+    def __init__(
+        self,
+        history_manager: HistoryManager,
+        user_manager: UserManager,
+        hda_manager: HDAManager,
+        app: MinimalManagerApp,
+    ):
+        self.history_manager = history_manager
+        self.user_manager = user_manager
+        self.hda_manager = hda_manager
+        self.app = app
+
+    def materialize_request_for(
+        self, trans: WorkRequestContext, hda: model.HistoryDatasetAssociation
+    ) -> MaterializeDatasetInstanceTaskRequest:
+        return MaterializeDatasetInstanceTaskRequest(
+            user=trans.async_request_user,
+            history_id=trans.history.id,
+            source="hda",
+            content=hda.id,
+        )
+
+    def dereference(
+        self, trans: WorkRequestContext, tool: Tool, request: QueueJobs, tool_request: ToolRequest
+    ) -> Tuple[RequestInternalDereferencedToolState, list[DereferencedDatasetPair]]:
+        new_hdas: list[DereferencedDatasetPair] = []
+
+        def dereference_callback(data_request: DataRequestUri) -> DataRequestInternalHda:
+            # a deferred dataset corresponding to request
+            history = tool_request.history
+            if not history:
+                raise InconsistentDatabase("Tool request has no history associated")
+
+            hda = dereference_input(trans, data_request, history)
+            if not isinstance(hda, model.HistoryDatasetAssociation):
+                raise RequestParameterInvalidException("Input dataset is not a history dataset association")
+
+            new_hdas.append(DereferencedDatasetPair(hda, data_request))
+            return DataRequestInternalHda(id=hda.id, src="hda")
+
+        tool_state = RequestInternalToolState(tool_request.request)
+        return dereference(tool_state, tool, dereference_callback), new_hdas
+
+    def queue_jobs(self, tool: Tool, request: QueueJobs) -> None:
+        tool_request: ToolRequest = self._tool_request(request.tool_request_id)
+        sa_session = self.app.model.context
+        try:
+            request_context = self._context(tool_request, request)
+            target_history = request_context.history
+            use_cached_jobs = request.use_cached_jobs
+            rerun_remap_job_id = request.rerun_remap_job_id
+            tool_state: RequestInternalDereferencedToolState
+            new_hdas: list[DereferencedDatasetPair]
+            tool_state, new_hdas = self.dereference(request_context, tool, request, tool_request)
+            to_materialize_list: list[DereferencedDatasetPair] = [p for p in new_hdas if not p.request.deferred]
+            for to_materialize in to_materialize_list:
+                materialize_request = self.materialize_request_for(request_context, to_materialize.hda)
+                # API dataset materialization is immutable and produces new datasets
+                # here we just created the datasets - lets just materialize them in place
+                # and avoid extra and confusing input copies
+                self.hda_manager.materialize(materialize_request, sa_session(), in_place=True)
+            tool.handle_input_async(
+                request_context,
+                tool_request,
+                tool_state,
+                history=target_history,
+                use_cached_job=use_cached_jobs,
+                rerun_remap_job_id=rerun_remap_job_id,
+            )
+            tool_request.state = ToolRequest.states.SUBMITTED
+            sa_session.add(tool_request)
+            sa_session.commit()
+        except Exception as e:
+            log.exception("Problem validating tool state after request created")
+            tool_request.state = ToolRequest.states.FAILED
+            tool_request.state_message = str(e)
+            sa_session.add(tool_request)
+            sa_session.commit()
+
+    def _context(self, tool_request: ToolRequest, request: QueueJobs) -> WorkRequestContext:
+        user = self.user_manager.by_id(request.user.user_id)
+        target_history = tool_request.history
+        trans = WorkRequestContext(
+            self.app,
+            user,
+            history=target_history,
+        )
+        return trans
+
+    def _tool_request(self, tool_request_id: int) -> ToolRequest:
+        sa_session = self.app.model.context
+        tool_request: ToolRequest = cast(ToolRequest, sa_session.query(ToolRequest).get(tool_request_id))
+        if tool_request is None:
+            raise Exception(f"Problem fetching request with ID {tool_request_id}")
+        return tool_request

lib/galaxy/model/__init__.py

@@ -1408,6 +1408,28 @@ class ToolRequest(Base, Dictifiable, RepresentById):
 
     tool_source: Mapped["ToolSource"] = relationship()
     history: Mapped[Optional["History"]] = relationship(back_populates="tool_requests")
+    jobs: Mapped[list["Job"]] = relationship(back_populates="tool_request", order_by=lambda: asc(Job.id))
+    implicit_collections: Mapped[list["ToolRequestImplicitCollectionAssociation"]] = relationship(
+        back_populates="tool_request"
+    )
+
+
+class ToolRequestImplicitCollectionAssociation(Base, Dictifiable, RepresentById):
+    __tablename__ = "tool_request_implicit_collection_association"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    tool_request_id: Mapped[int] = mapped_column(ForeignKey("tool_request.id", name="fk_trica_tri"), index=True)
+    dataset_collection_id: Mapped[int] = mapped_column(
+        ForeignKey("history_dataset_collection_association.id", name="fk_trica_dci"), index=True
+    )
+    output_name: Mapped[str] = mapped_column(String(255))
+
+    tool_request: Mapped["ToolRequest"] = relationship(back_populates="implicit_collections")
+    dataset_collection: Mapped["HistoryDatasetCollectionAssociation"] = relationship(
+        back_populates="tool_request_association", uselist=False
+    )
+
+    dict_collection_visible_keys = ["id", "tool_request_id", "dataset_collection_id", "output_name"]
 
 
 class UserDynamicToolAssociation(Base, Dictifiable, RepresentById):
@@ -1592,7 +1614,7 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable):
     tool_request_id: Mapped[Optional[int]] = mapped_column(ForeignKey("tool_request.id"), index=True)
 
     dynamic_tool: Mapped[Optional["DynamicTool"]] = relationship()
-    tool_request: Mapped[Optional["ToolRequest"]] = relationship()
+    tool_request: Mapped[Optional["ToolRequest"]] = relationship(back_populates="jobs")
     user: Mapped[Optional["User"]] = relationship()
     galaxy_session: Mapped[Optional["GalaxySession"]] = relationship()
     history: Mapped[Optional["History"]] = relationship(back_populates="jobs")
@@ -7546,6 +7568,9 @@ class HistoryDatasetCollectionAssociation(
         back_populates="dataset_collection",
     )
     creating_job_associations: Mapped[list["JobToOutputDatasetCollectionAssociation"]] = relationship(viewonly=True)
+    tool_request_association: Mapped[Optional["ToolRequestImplicitCollectionAssociation"]] = relationship(
+        back_populates="dataset_collection"
+    )
 
     dict_dbkeysandextensions_visible_keys = ["dbkeys", "extensions"]
     editable_keys = ("name", "deleted", "visible")

lib/galaxy/schema/jobs.py

@@ -79,6 +79,19 @@ class JobOutputAssociation(JobAssociation):
     )
 
 
+class JobOutputCollectionAssociation(Model):
+    name: str = Field(
+        default=...,
+        title="name",
+        description="Name of the job parameter.",
+    )
+    dataset_collection_instance: EncodedDataItemSourceId = Field(
+        default=...,
+        title="dataset_collection_instance",
+        description="Reference to the associated item.",
+    )
+
+
 class ReportJobErrorPayload(Model):
     dataset_id: DecodedDatabaseIdField = Field(
         default=...,

lib/galaxy/schema/schema.py

@@ -1668,6 +1668,7 @@ class JobIndexQueryPayload(Model):
     workflow_id: Optional[DecodedDatabaseIdField] = None
     invocation_id: Optional[DecodedDatabaseIdField] = None
     implicit_collection_jobs_id: Optional[DecodedDatabaseIdField] = None
+    tool_request_id: Optional[DecodedDatabaseIdField] = None
     order_by: JobIndexSortByEnum = JobIndexSortByEnum.update_time
     search: Optional[str] = None
     limit: int = 500
@@ -3982,6 +3983,22 @@ class ToolRequestModel(Model):
     state_message: Optional[str]
 
 
+class ToolRequestJobReference(Model):
+    src: Literal["job"]
+    id: EncodedDatabaseIdField
+
+
+class ToolRequestImplicitCollectionReference(Model):
+    src: Literal["hdca"]
+    id: EncodedDatabaseIdField
+    output_name: str
+
+
+class ToolRequestDetailedModel(ToolRequestModel):
+    jobs: list[ToolRequestJobReference] = Field(default=[])
+    implicit_collections: list[ToolRequestImplicitCollectionReference] = Field(default=[])
+
+
 class AsyncFile(Model):
     storage_request_id: UUID
     task: AsyncTaskResultSummary

lib/galaxy/schema/tasks.py

@@ -151,3 +151,16 @@ class TaskResult(Model):
         title="Result",
         description="The result message of the task. Empty if the task is still running. If the task failed, this will contain the exception message.",
     )
+
+
+class ToolSource(Model):
+    raw_tool_source: str
+    tool_dir: str
+
+
+class QueueJobs(Model):
+    tool_source: ToolSource
+    tool_request_id: int  # links to request ("incoming") and history
+    user: RequestUser  # TODO: test anonymous users through this submission path
+    use_cached_jobs: bool
+    rerun_remap_job_id: Optional[int]  # link to a job to rerun & remap

lib/galaxy/tool_util/parameters/__init__.py

@@ -1,3 +1,4 @@
+from galaxy.tool_util_models._types import is_optional
 from galaxy.tool_util_models.parameters import (
     BooleanParameterModel,
     ColorParameterModel,
@@ -141,6 +142,7 @@ __all__ = (
     "validate_workflow_step",
     "validate_workflow_step_linked",
     "validate_explicit_conditional_test_value",
+    "is_optional",
     "ToolState",
     "TestCaseToolState",
     "ToolParameterT",

lib/galaxy/tool_util/parameters/case.py

@@ -149,7 +149,11 @@ def legacy_from_string(parameter: ToolParameterT, value: Optional[Any], warnings
 
 
 def test_case_state(
-    test_dict: ToolSourceTest, tool_parameter_bundle: List[ToolParameterT], profile: str, validate: bool = True
+    test_dict: ToolSourceTest,
+    tool_parameter_bundle: List[ToolParameterT],
+    profile: str,
+    validate: bool = True,
+    name: Optional[str] = None,
 ) -> TestCaseStateAndWarnings:
     warnings: List[str] = []
     inputs: ToolSourceTestInputs = test_dict["inputs"]
@@ -165,7 +169,7 @@ def test_case_state(
 
     tool_state = TestCaseToolState(state)
     if validate:
-        tool_state.validate(tool_parameter_bundle)
+        tool_state.validate(tool_parameter_bundle, name=name)
         for input_name in unhandled_inputs:
             raise Exception(f"Invalid parameter name found {input_name}")
     return TestCaseStateAndWarnings(tool_state, warnings, unhandled_inputs)

lib/galaxy/tool_util/parameters/convert.py

@@ -23,6 +23,7 @@ from galaxy.tool_util_models.parameters import (
     FloatParameterModel,
     HiddenParameterModel,
     IntegerParameterModel,
+    TextParameterModel,
     ToolParameterBundle,
     ToolParameterT,
 )
@@ -79,11 +80,11 @@ def cwl_runtime_model(input_models: ToolParameterBundle):
 
 
 def decode(
-    external_state: RequestToolState, input_models: ToolParameterBundle, decode_id: Callable[[str], int]
+    external_state: RequestToolState, input_models: ToolParameterBundle, decode_id: Callable[[str], int], name_base: Optional[str] = None
 ) -> RequestInternalToolState:
     """Prepare an internal representation of tool state (request_internal) for storing in the database."""
 
-    external_state.validate(input_models)
+    external_state.validate(input_models, f"{name_base} (request model)")
     decode_callback = _decode_callback_for(decode_id)
     internal_state_dict = visit_input_values(
         input_models,
@@ -92,7 +93,7 @@ def decode(
     )
 
     internal_request_state = RequestInternalToolState(internal_state_dict)
-    internal_request_state.validate(input_models)
+    internal_request_state.validate(input_models, f"{name_base} (request internal model)")
     return internal_request_state
 
 
@@ -163,8 +164,7 @@ def dereference(
         if parameter.parameter_type == "gx_data":
             if value is None:
                 return VISITOR_NO_REPLACEMENT
-            if parameter.multiple:
-                assert isinstance(value, list), str(value)
+            if parameter.multiple and isinstance(value, list):
                 return list(map(derefrence_dict, value))
             else:
                 assert isinstance(value, dict), str(value)
@@ -328,6 +328,25 @@ def _fill_default_for(tool_state: Dict[str, Any], parameter: ToolParameterT) ->
             tool_state[parameter_name] = {}
         section_state = cast(Dict[str, Any], tool_state[parameter_name])
         _fill_defaults(section_state, parameter)
+    elif parameter.parameter_type == "gx_data_collection":
+        collection_parameter = parameter
+        if parameter_name not in tool_state and collection_parameter.optional:
+            tool_state[parameter_name] = None
+    elif parameter.parameter_type in ["gx_text"]:
+        text_parameter = cast(TextParameterModel, parameter)
+        if parameter_name not in tool_state:
+            if not text_parameter.optional:
+                # restore legacy behavior of allowing empty string implicit default
+                # for these non-optional inputs.
+                tool_state[parameter_name] = ""
+            else:
+                tool_state[parameter_name] = None
+        else:
+            # legacy behavior of converting explicit None into implicit null. We should introduce
+            # a layer somewhere to deal with this behavior further up the stack and clean up these models.
+            if not text_parameter.optional and tool_state[parameter_name] is None:
+                tool_state[parameter_name] = ""
+
 
 
 def _select_which_when(
@@ -354,17 +373,25 @@ def _encode_callback_for(encode_id: EncodeFunctionT) -> Callback:
         else:
             return src_dict
 
+    def encode_element(element: dict):
+        if element.get("__class__") == "Batch":
+            encoded = element.copy()
+            values = encoded.pop("values")
+            encoded["values"] = list(map(encode_src_dict, values))
+            return encoded
+        else:
+            return encode_src_dict(element)
+
     def encode_callback(parameter: ToolParameterT, value: Any):
         if parameter.parameter_type == "gx_data":
-            if parameter.multiple:
-                assert isinstance(value, list), str(value)
-                return list(map(encode_src_dict, value))
+            if parameter.multiple and isinstance(value, list):
+                return list(map(encode_element, value))
             else:
                 assert isinstance(value, dict), str(value)
-                return encode_src_dict(value)
+                return encode_element(value)
         elif parameter.parameter_type == "gx_data_collection":
             assert isinstance(value, dict), str(value)
-            return encode_src_dict(value)
+            return encode_element(value)
         else:
             return VISITOR_NO_REPLACEMENT
 
@@ -381,16 +408,24 @@ def _decode_callback_for(decode_id: DecodeFunctionT) -> Callback:
         else:
             return src_dict
 
+    def decode_element(element: dict):
+        if element.get("__class__") == "Batch":
+            decoded = element.copy()
+            values = decoded.pop("values")
+            decoded["values"] = list(map(decode_src_dict, values))
+            return decoded
+        else:
+            return decode_src_dict(element)
+
     def decode_callback(parameter: ToolParameterT, value: Any):
         if parameter.parameter_type == "gx_data":
             if value is None:
                 return VISITOR_NO_REPLACEMENT
-            if parameter.multiple:
-                assert isinstance(value, list), str(value)
-                return list(map(decode_src_dict, value))
+            if parameter.multiple and isinstance(value, list):
+                return list(map(decode_element, value))
             else:
                 assert isinstance(value, dict), str(value)
-                return decode_src_dict(value)
+                return decode_element(value)
         elif parameter.parameter_type == "gx_data_collection":
             if value is None:
                 return VISITOR_NO_REPLACEMENT

lib/galaxy/tool_util/parameters/factory.py

@@ -215,7 +215,10 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
             select_validators: List[SelectCompatiableValidators] = []
             for static_validator in static_validator_models:
                 if static_validator.type == "no_options":
-                    select_validators.append(static_validator)
+                    # test case test_tool_execute::test_select_optional_null_by_default verifies
+                    # these validators don't get applied effectively if the select is optional.
+                    if not optional:
+                        select_validators.append(static_validator)
             return SelectParameterModel(
                 type="select",
                 name=input_source.parse_name(),

lib/galaxy/tool_util/parser/interface.py

@@ -722,6 +722,8 @@ class TestCollectionDef:
         test_format_dict = JsonTestCollectionDefDict(**self._test_format_to_dict())
         if self.name:
             test_format_dict["name"] = self.name
+        if self.fields is not None:
+            test_format_dict["fields"] = self.fields
         return test_format_dict
 
     def to_dict(self) -> XmlTestCollectionDefDict:

lib/galaxy/tool_util/verify/_types.py

@@ -19,10 +19,15 @@ from galaxy.tool_util.parser.interface import (
     ToolSourceTestOutputs,
 )
 
-# inputs that have been processed with parse.py and expanded out
+# legacy inputs for working with POST /api/tools
+# + inputs that have been processed with parse.py and expanded out
 ExpandedToolInputs = Dict[str, Any]
-# ExpandedToolInputs where any model objects have been json-ified with to_dict()
+# + ExpandedToolInputs where any model objects have been json-ified with to_dict()
 ExpandedToolInputsJsonified = Dict[str, Any]
+
+# modern inputs for working with POST /api/jobs*
+RawTestToolRequest = Dict[str, Any]
+
 ExtraFileInfoDictT = Dict[str, Any]
 RequiredFileTuple = Tuple[str, ExtraFileInfoDictT]
 RequiredFilesT = List[RequiredFileTuple]
@@ -36,6 +41,8 @@ class ToolTestDescriptionDict(TypedDict):
     name: str
     test_index: int
     inputs: ExpandedToolInputsJsonified
+    request: NotRequired[Optional[Dict[str, Any]]]
+    request_schema: NotRequired[Optional[Dict[str, Any]]]
     outputs: ToolSourceTestOutputs
     output_collections: List[TestSourceTestOutputColllection]
     stdout: Optional[AssertionList]

lib/galaxy/tool_util/verify/interactor.py

@@ -36,6 +36,14 @@ from typing_extensions import (
 
 from galaxy import util
 from galaxy.tool_util.client.staging import StagingInterface
+from galaxy.tool_util.parameters import (
+    DataCollectionRequest,
+    DataRequestHda,
+    encode_test,
+    input_models_from_json,
+    TestCaseToolState,
+    ToolParameterBundle,
+)
 from galaxy.tool_util.parser.interface import (
     AssertionList,
     TestCollectionDef,
@@ -45,6 +53,10 @@ from galaxy.tool_util.parser.interface import (
     XmlTestCollectionDefDict,
 )
 from galaxy.tool_util.verify.test_data import TestDataResolver
+from galaxy.tool_util_models.tool_source import (
+    JsonTestCollectionDefDict,
+    JsonTestDatasetDefDict,
+)
 from galaxy.util import requests
 from galaxy.util.bunch import Bunch
 from galaxy.util.compression_utils import CompressedFile
@@ -56,6 +68,7 @@ from . import verify
 from ._types import (
     ExpandedToolInputs,
     ExpandedToolInputsJsonified,
+    RawTestToolRequest,
     RequiredDataTablesT,
     RequiredFilesT,
     RequiredLocFileT,
@@ -66,6 +79,9 @@ from .wait import wait_on
 
 log = getLogger(__name__)
 
+UseLegacyApiT = Literal["always", "never", "if_needed"]
+DEFAULT_USE_LEGACY_API: UseLegacyApiT = "always"
+
 # Off by default because it can pound the database pretty heavily
 # and result in sqlite errors on larger tests or larger numbers of
 # tests.
@@ -105,6 +121,8 @@ JobDataCallbackT = Callable[[JobDataT], None]
 
 class ValidToolTestDict(TypedDict):
     inputs: ExpandedToolInputs
+    request: NotRequired[Optional[RawTestToolRequest]]
+    request_schema: NotRequired[Optional[Dict[str, Any]]]
     outputs: ToolSourceTestOutputs
     output_collections: List[TestSourceTestOutputColllection]
     stdout: NotRequired[AssertionList]
@@ -155,7 +173,7 @@ def stage_data_in_history(
     tool_version=None,
     test_data_resolver: Optional[TestDataResolver] = None,
 ):
-    assert tool_id
+    assert tool_id, "Tool id not set"
 
     staging_interface = InteractorStagingInterface(galaxy_interactor, maxseconds=maxseconds, upload_async=UPLOAD_ASYNC)
     job = {}
@@ -268,6 +286,15 @@ class GalaxyInteractorApi:
         assert response.status_code == 200, f"Non 200 response from tool tests available API. [{response.content}]"
         return response.json()
 
+    def get_tool_inputs(self, tool_id: str, tool_version: Optional[str] = None) -> ToolParameterBundle:
+        url = f"tools/{tool_id}/inputs"
+        params = {"tool_version": tool_version} if tool_version else None
+        response = self._get(url, data=params)
+        assert response.status_code == 200, f"Non 200 response from tool inputs API. [{response.content}]"
+        raw_inputs_array = response.json()
+        tool_parameter_bundle = input_models_from_json(raw_inputs_array)
+        return tool_parameter_bundle
+
     def get_tool_tests(self, tool_id: str, tool_version: Optional[str] = None) -> List[ToolTestDescriptionDict]:
         url = f"tools/{tool_id}/test_data"
         params = {"tool_version": tool_version} if tool_version else None
@@ -398,9 +425,27 @@ class GalaxyInteractorApi:
     def wait_for_job(self, job_id: str, history_id: Optional[str] = None, maxseconds=DEFAULT_TOOL_TEST_WAIT) -> None:
         self.wait_for(lambda: self.__job_ready(job_id, history_id), maxseconds=maxseconds)
 
+    def wait_on_tool_request(self, tool_request_id: str):
+        def state():
+            state_response = self._get(f"tool_requests/{tool_request_id}/state")
+            state_response.raise_for_status()
+            return state_response.json()
+
+        def is_ready():
+            is_complete = state() in ["submitted", "failed"]
+            return True if is_complete else None
+
+        self.wait_for(is_ready, "waiting for tool request to submit")
+        return state() == "submitted"
+
+    def get_tool_request(self, tool_request_id: str):
+        response_raw = self._get(f"tool_requests/{tool_request_id}")
+        response_raw.raise_for_status()
+        return response_raw.json()
+
     def wait_for(self, func: Callable, what: str = "tool test run", **kwd) -> None:
         walltime_exceeded = int(kwd.get("maxseconds", DEFAULT_TOOL_TEST_WAIT))
-        wait_on(func, what, walltime_exceeded)
+        return wait_on(func, what, walltime_exceeded)
 
     def get_job_stdio(self, job_id: str) -> Dict[str, Any]:
         return self.__get_job_stdio(job_id).json()
@@ -632,39 +677,71 @@ class GalaxyInteractorApi:
         return location
 
     def run_tool(
-        self, testdef: "ToolTestDescription", history_id: str, resource_parameters: Optional[Dict[str, Any]] = None
+        self,
+        testdef: "ToolTestDescription",
+        history_id: str,
+        resource_parameters: Optional[Dict[str, Any]] = None,
+        use_legacy_api: UseLegacyApiT = DEFAULT_USE_LEGACY_API,
     ) -> RunToolResponse:
         # We need to handle the case where we've uploaded a valid compressed file since the upload
         # tool will have uncompressed it on the fly.
         resource_parameters = resource_parameters or {}
-        inputs_tree = testdef.inputs.copy()
-        for key, value in inputs_tree.items():
-            values = [value] if not isinstance(value, list) else value
-            new_values = []
-            for value in values:
-                if isinstance(value, TestCollectionDef):
-                    hdca_id = self._create_collection(history_id, value)
-                    new_values = [dict(src="hdca", id=hdca_id)]
-                elif value in self.uploads:
-                    new_values.append(self.uploads[value])
-                else:
-                    new_values.append(value)
-            inputs_tree[key] = new_values
+        request = testdef.request
+        request_schema = testdef.request_schema
+        submit_with_legacy_api = use_legacy_api == "always" or (use_legacy_api == "if_needed" and request is None)
+        if submit_with_legacy_api:
+            inputs_tree = testdef.inputs.copy()
+            for key, value in inputs_tree.items():
+                values = [value] if not isinstance(value, list) else value
+                new_values = []
+                for value in values:
+                    if isinstance(value, TestCollectionDef):
+                        hdca_id = self._create_collection(history_id, value)
+                        new_values = [dict(src="hdca", id=hdca_id)]
+                    elif value in self.uploads:
+                        new_values.append(self.uploads[value])
+                    else:
+                        new_values.append(value)
+                inputs_tree[key] = new_values
+
+            # HACK: Flatten single-value lists. Required when using expand_grouping
+            for key, value in inputs_tree.items():
+                if isinstance(value, list) and len(value) == 1:
+                    inputs_tree[key] = value[0]
+        else:
+            assert request is not None, "Request not set"
+            assert request_schema is not None, "Request schema not set"
+            parameters = request_schema["parameters"]
+
+            def adapt_datasets(test_input: JsonTestDatasetDefDict) -> DataRequestHda:
+                # if path is not set it might be a composite file with a path,
+                # e.g. composite_shapefile
+                test_input_path = test_input.get("path", "")
+                return DataRequestHda(**self.uploads[test_input_path])
+
+            def adapt_collections(test_input: JsonTestCollectionDefDict) -> DataCollectionRequest:
+                test_collection_def = TestCollectionDef.from_dict(test_input)
+                hdca_id = self._create_collection(history_id, test_collection_def)
+                return DataCollectionRequest(src="hdca", id=hdca_id)
+
+            test_case_state = TestCaseToolState(input_state=request)
+            inputs_tree = encode_test(
+                test_case_state, input_models_from_json(parameters), adapt_datasets, adapt_collections
+            ).input_state
 
         if resource_parameters:
             inputs_tree["__job_resource|__job_resource__select"] = "yes"
             for key, value in resource_parameters.items():
                 inputs_tree[f"__job_resource|{key}"] = value
 
-        # HACK: Flatten single-value lists. Required when using expand_grouping
-        for key, value in inputs_tree.items():
-            if isinstance(value, list) and len(value) == 1:
-                inputs_tree[key] = value[0]
-
         submit_response = None
         for _ in range(DEFAULT_TOOL_TEST_WAIT):
             submit_response = self.__submit_tool(
-                history_id, tool_id=testdef.tool_id, tool_input=inputs_tree, tool_version=testdef.tool_version
+                history_id,
+                tool_id=testdef.tool_id,
+                tool_input=inputs_tree,
+                tool_version=testdef.tool_version,
+                use_legacy_api=submit_with_legacy_api,
             )
             if _are_tool_inputs_not_ready(submit_response):
                 print("Tool inputs not ready yet")
@@ -673,10 +750,36 @@ class GalaxyInteractorApi:
             else:
                 break
         submit_response_object = ensure_tool_run_response_okay(submit_response, "execute tool", inputs_tree)
-        try:
+        if not submit_with_legacy_api:
+            tool_request_id = submit_response_object["tool_request_id"]
+            successful = self.wait_on_tool_request(tool_request_id)
+            if not successful:
+                request = self.get_tool_request(tool_request_id) or {}
+                raise RunToolException(
+                    f"Tool request failure - state {request.get('state')}, message: {request.get('state_message')}",
+                    inputs_tree,
+                )
+            job_refs = self.jobs_for_tool_request(tool_request_id)
+            outputs = OutputsDict()
+            output_collections = {}
+            if len(job_refs) != 1:
+                raise Exception(
+                    f"Found incorrect number of jobs for tool request - was expecting a single job {job_refs}"
+                )
+            assert len(job_refs) == 1, job_refs
+            job_id = job_refs[0]["id"]
+            jobs = [self.__get_job(job_id).json()]
+            job_outputs = self.job_outputs(job_id)
+            for job_output in job_outputs:
+                if "dataset" in job_output:
+                    outputs[job_output["name"]] = job_output["dataset"]
+                else:
+                    output_collections[job_output["name"]] = job_output["dataset_collection_instance"]
+        else:
             outputs = self.__dictify_outputs(submit_response_object)
             output_collections = self.__dictify_output_collections(submit_response_object)
             jobs = submit_response_object["jobs"]
+        try:
             return RunToolResponse(
                 inputs=inputs_tree,
                 outputs=outputs,
@@ -698,7 +801,9 @@ class GalaxyInteractorApi:
         )
         if collection_def.fields:
             create_payload["fields"] = collection_def.fields
-        return self._post("dataset_collections", data=create_payload, json=True).json()["id"]
+        create_response = self._post("dataset_collections", data=create_payload, json=True)
+        create_response.raise_for_status()
+        return create_response.json()["id"]
 
     def _element_identifiers(self, collection_def):
         element_identifiers = []
@@ -834,6 +939,16 @@ class GalaxyInteractorApi:
         dataset_json = self._get(f"histories/{history_id}/contents/{id}").json()
         return dataset_json
 
+    def jobs_for_tool_request(self, tool_request_id: str) -> List[Dict[str, Any]]:
+        job_list_response = self._get(f"tool_requests/{tool_request_id}")
+        job_list_response.raise_for_status()
+        return job_list_response.json()["jobs"]
+
+    def job_outputs(self, job_id: str) -> List[Dict[str, Any]]:
+        outputs = self._get(f"jobs/{job_id}/outputs")
+        outputs.raise_for_status()
+        return outputs.json()
+
     def __contents(self, history_id: str) -> List[Dict[str, Any]]:
         history_contents_response = self._get(f"histories/{history_id}/contents")
         history_contents_response.raise_for_status()
@@ -858,12 +973,25 @@ class GalaxyInteractorApi:
         extra_data: Optional[dict] = None,
         files: Optional[dict] = None,
         tool_version: Optional[str] = None,
+        use_legacy_api: bool = True,
     ):
         extra_data = extra_data or {}
-        data = dict(
-            history_id=history_id, tool_id=tool_id, inputs=dumps(tool_input), tool_version=tool_version, **extra_data
-        )
-        return self._post("tools", files=files, data=data)
+        if use_legacy_api:
+            data = dict(
+                history_id=history_id,
+                tool_id=tool_id,
+                inputs=dumps(tool_input),
+                tool_version=tool_version,
+                **extra_data,
+            )
+            return self._post("tools", files=files, data=data)
+        else:
+            assert files is None
+            data = dict(
+                history_id=history_id, tool_id=tool_id, inputs=tool_input, tool_version=tool_version, **extra_data
+            )
+            submit_tool_request_response = self._post("jobs", data=data, json=True)
+            return submit_tool_request_response
 
     def ensure_user_with_email(self, email, password=None):
         admin_key = self.master_api_key
@@ -1391,6 +1519,7 @@ def verify_tool(
     register_job_data: Optional[JobDataCallbackT] = None,
     test_index: int = 0,
     tool_version: Optional[str] = None,
+    use_legacy_api: UseLegacyApiT = DEFAULT_USE_LEGACY_API,
     quiet: bool = False,
     test_history: Optional[str] = None,
     no_history_cleanup: bool = False,
@@ -1408,11 +1537,7 @@ def verify_tool(
     if client_test_config is None:
         client_test_config = NullClientTestConfig()
     tool_test_dicts = _tool_test_dicts or galaxy_interactor.get_tool_tests(tool_id, tool_version=tool_version)
-    tool_test_dict = tool_test_dicts[test_index]
-    if "test_index" not in tool_test_dict:
-        tool_test_dict["test_index"] = test_index
-    if "tool_id" not in tool_test_dict:
-        tool_test_dict["tool_id"] = tool_id
+    tool_test_dict: ToolTestDescriptionDict = tool_test_dicts[test_index]
     if tool_version is None and "tool_version" in tool_test_dict:
         tool_version = tool_test_dict.get("tool_version")
 
@@ -1478,7 +1603,9 @@ def verify_tool(
             input_staging_exc_info = sys.exc_info()
             raise
         try:
-            tool_response = galaxy_interactor.run_tool(testdef, test_history, resource_parameters=resource_parameters)
+            tool_response = galaxy_interactor.run_tool(
+                testdef, test_history, resource_parameters=resource_parameters, use_legacy_api=use_legacy_api
+            )
             data_list, jobs, tool_inputs = tool_response.outputs, tool_response.jobs, tool_response.inputs
             data_collection_list = tool_response.output_collections
         except RunToolException as e:
@@ -1765,6 +1892,8 @@ def adapt_tool_source_dict(processed_dict: ToolTestDict) -> ToolTestDescriptionD
     expect_test_failure: bool = DEFAULT_EXPECT_TEST_FAILURE
     inputs: ExpandedToolInputsJsonified = {}
     maxseconds: Optional[int] = None
+    request: Optional[Dict[str, Any]] = None
+    request_schema: Optional[Dict[str, Any]] = None
 
     if not error_in_test_definition:
         processed_test_dict = cast(ValidToolTestDict, processed_dict)
@@ -1790,6 +1919,8 @@ def adapt_tool_source_dict(processed_dict: ToolTestDict) -> ToolTestDescriptionD
         expect_failure = processed_test_dict.get("expect_failure", DEFAULT_EXPECT_FAILURE)
         expect_test_failure = processed_test_dict.get("expect_test_failure", DEFAULT_EXPECT_TEST_FAILURE)
         inputs = processed_test_dict.get("inputs", {})
+        request = processed_test_dict.get("request", None)
+        request_schema = processed_test_dict.get("request_schema", None)
     else:
         invalid_test_dict = cast(InvalidToolTestDict, processed_dict)
         maxseconds = DEFAULT_TOOL_TEST_WAIT
@@ -1817,6 +1948,8 @@ def adapt_tool_source_dict(processed_dict: ToolTestDict) -> ToolTestDescriptionD
         expect_failure=expect_failure,
         expect_test_failure=expect_test_failure,
         inputs=inputs,
+        request=request,
+        request_schema=request_schema,
     )
 
 
@@ -1875,6 +2008,8 @@ class ToolTestDescription:
     expect_test_failure: bool
     exception: Optional[str]
     inputs: ExpandedToolInputs
+    request: Optional[Dict[str, Any]]
+    request_schema: Optional[Dict[str, Any]]
     outputs: ToolSourceTestOutputs
     output_collections: List[TestCollectionOutputDef]
     maxseconds: Optional[int]
@@ -1903,6 +2038,8 @@ class ToolTestDescription:
         self.expect_failure = json_dict.get("expect_failure", DEFAULT_EXPECT_FAILURE)
         self.expect_test_failure = json_dict.get("expect_test_failure", DEFAULT_EXPECT_TEST_FAILURE)
         self.inputs = expanded_inputs_from_json(json_dict.get("inputs", {}))
+        self.request = json_dict.get("request", None)
+        self.request_schema = json_dict.get("request_schema", None)
         self.tool_id = json_dict["tool_id"]
         self.tool_version = json_dict.get("tool_version")
         self.maxseconds = json_dict.get("maxseconds")
@@ -1934,6 +2071,8 @@ class ToolTestDescription:
             "required_files": self.required_files,
             "required_data_tables": self.required_data_tables,
             "required_loc_files": self.required_loc_files,
+            "request": self.request,
+            "request_schema": self.request_schema,
             "error": self.error,
             "exception": self.exception,
         }

lib/galaxy/tool_util/verify/parse.py

@@ -1,8 +1,10 @@
 import logging
 import os
 import traceback
+from dataclasses import dataclass
 from typing import (
     Any,
+    Dict,
     Iterable,
     List,
     Optional,
@@ -15,6 +17,8 @@ from packaging.version import Version
 from galaxy.tool_util.parameters import (
     input_models_for_tool_source,
     test_case_state as case_state,
+    TestCaseToolState,
+    ToolParameterBundleModel,
 )
 from galaxy.tool_util.parser.interface import (
     InputSource,
@@ -67,15 +71,18 @@ def parse_tool_test_descriptions(
     profile = tool_source.parse_profile()
     for i, raw_test_dict in enumerate(raw_tests_dict.get("tests", [])):
         validation_exception: Optional[Exception] = None
-        if validate_on_load:
+        request_and_schema: Optional[TestRequestAndSchema] = None
+        try:
             tool_parameter_bundle = input_models_for_tool_source(tool_source)
-            try:
-                case_state(raw_test_dict, tool_parameter_bundle.parameters, profile, validate=True)
-            except Exception as e:
-                # TOOD: restrict types of validation exceptions a bit probably?
-                validation_exception = e
+            validated_test_case = case_state(raw_test_dict, tool_parameter_bundle.parameters, profile, validate=True)
+            request_and_schema = TestRequestAndSchema(
+                validated_test_case.tool_state,
+                tool_parameter_bundle,
+            )
+        except Exception as e:
+            validation_exception = e
 
-        if validation_exception:
+        if validation_exception and validate_on_load:
             tool_id, tool_version = _tool_id_and_version(tool_source, tool_guid)
             test = ToolTestDescription.from_tool_source_dict(
                 InvalidToolTestDict(
@@ -91,13 +98,23 @@ def parse_tool_test_descriptions(
                 )
             )
         else:
-            test = _description_from_tool_source(tool_source, raw_test_dict, i, tool_guid)
+            test = _description_from_tool_source(tool_source, raw_test_dict, i, tool_guid, request_and_schema)
         tests.append(test)
     return tests
 
 
+@dataclass
+class TestRequestAndSchema:
+    request: TestCaseToolState
+    request_schema: ToolParameterBundleModel
+
+
 def _description_from_tool_source(
-    tool_source: ToolSource, raw_test_dict: ToolSourceTest, test_index: int, tool_guid: Optional[str]
+    tool_source: ToolSource,
+    raw_test_dict: ToolSourceTest,
+    test_index: int,
+    tool_guid: Optional[str],
+    request_and_schema: Optional[TestRequestAndSchema],
 ) -> ToolTestDescription:
     required_files: RequiredFilesT = []
     required_data_tables: RequiredDataTablesT = []
@@ -110,6 +127,12 @@ def _description_from_tool_source(
     if maxseconds is not None:
         maxseconds = int(maxseconds)
 
+    request: Optional[Dict[str, Any]] = None
+    request_schema: Optional[Dict[str, Any]] = None
+    if request_and_schema:
+        request = request_and_schema.request.input_state
+        request_schema = request_and_schema.request_schema.dict()
+
     tool_id, tool_version = _tool_id_and_version(tool_source, tool_guid)
     processed_test_dict: Union[ValidToolTestDict, InvalidToolTestDict]
     try:
@@ -124,6 +147,8 @@ def _description_from_tool_source(
         processed_test_dict = ValidToolTestDict(
             {
                 "inputs": processed_inputs,
+                "request": request,
+                "request_schema": request_schema,
                 "outputs": raw_test_dict["outputs"],
                 "output_collections": raw_test_dict["output_collections"],
                 "num_outputs": num_outputs,

lib/galaxy/tool_util_models/parameters.py

@@ -122,6 +122,7 @@ def allow_batching(job_template: DynamicModelInformation, batch_type: Optional[T
     class BatchRequest(StrictModel):
         meta_class: Literal["Batch"] = Field(..., alias="__class__")
         values: List[batch_type]  # type: ignore[valid-type]
+        linked: Optional[bool] = None  # maybe True instead?
 
     request_type = union_type([job_py_type, BatchRequest])
 
@@ -273,8 +274,17 @@ class TextParameterModel(BaseGalaxyToolParameterModelDefinition):
     def py_type(self) -> Type:
         return optional_if_needed(StrictStr, self.optional)
 
+    @property
+    def py_type_request(self) -> Type:
+        # such a hack but explicit nulls are always allowed in the API even for non-optional
+        # parameters - it becomes "" in the internal state.
+        return optional(StrictStr)
+
     def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
-        py_type = decorate_type_with_validators_if_needed(self.py_type, self.validators)
+        py_type = self.py_type
+        if state_representation in ["request", "request_internal", "request_internal_dereferenced"]:
+            py_type = self.py_type_request
+        py_type = decorate_type_with_validators_if_needed(py_type, self.validators)
         if state_representation == "workflow_step_linked":
             py_type = allow_connected_value(py_type)
         requires_value = self.request_requires_value
@@ -303,8 +313,15 @@ class IntegerParameterModel(BaseGalaxyToolParameterModelDefinition):
     def py_type(self) -> Type:
         return optional_if_needed(StrictInt, self.optional)
 
+    #@property
+    #def py_type_request(self) -> Type:
+    #    # ugh... we allow explicit nulls in the API even if the input is not optional
+    #    return optional_if_needed(StrictInt, True)
+
     def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
         py_type = self.py_type
+        # if state_representation == "request":
+        #    py_type = self.py_type_request
         validators = self.validators[:]
         if self.min is not None or self.max is not None:
             validators.append(InRangeParameterValidatorModel(min=self.min, max=self.max, implicit=True))
@@ -1138,7 +1155,7 @@ class GenomeBuildParameterModel(BaseGalaxyToolParameterModelDefinition):
         py_type: Type = StrictStr
         if self.multiple:
             py_type = list_type(py_type)
-        return optional_if_needed(py_type, self.optional)
+        return optional_if_needed(py_type, self.optional or self.multiple)
 
     def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
         requires_value = self.request_requires_value

lib/galaxy/tool_util_models/tool_source.py

@@ -216,6 +216,7 @@ BaseJsonTestCollectionDefCollectionElementDict = TypedDict(
         "class": Literal["Collection"],
         "collection_type": Optional[str],
         "elements": NotRequired[Optional[List[JsonTestCollectionDefElementDict]]],
+        "fields": NotRequired[Optional[List[FieldDict]]],
     },
 )
 
@@ -236,5 +237,6 @@ JsonTestCollectionDefDict = TypedDict(
         "collection_type": Optional[str],
         "elements": NotRequired[Optional[List[JsonTestCollectionDefElementDict]]],
         "name": NotRequired[Optional[str]],
+        "fields": NotRequired[Optional[List[FieldDict]]],
     },
 )

lib/galaxy/tools/__init__.py

@@ -56,6 +56,7 @@ from galaxy.model import (
     Job,
     JobToOutputDatasetAssociation,
     StoredWorkflow,
+    ToolRequest,
 )
 from galaxy.model.dataset_collections.matching import MatchingCollections
 from galaxy.schema.credentials import CredentialsContext
@@ -80,7 +81,10 @@ from galaxy.tool_util.ontologies.ontology_data import (
 )
 from galaxy.tool_util.output_checker import DETECTED_JOB_STATE
 from galaxy.tool_util.parameters import (
+    fill_static_defaults,
     input_models_for_pages,
+    JobInternalToolState,
+    RequestInternalDereferencedToolState,
     ToolParameterBundle,
 )
 from galaxy.tool_util.parser import (
@@ -138,12 +142,14 @@ from galaxy.tools.execution_helpers import ToolExecutionCache
 from galaxy.tools.imp_exp import JobImportHistoryArchiveWrapper
 from galaxy.tools.parameters import (
     check_param,
+    fill_dynamic_defaults,
     params_from_strings,
     params_to_incoming,
     params_to_json,
     params_to_json_internal,
     params_to_strings,
     populate_state,
+    populate_state_async,
     visit_input_values,
 )
 from galaxy.tools.parameters.basic import (
@@ -170,7 +176,10 @@ from galaxy.tools.parameters.grouping import (
     UploadDataset,
 )
 from galaxy.tools.parameters.input_translation import ToolInputTranslator
-from galaxy.tools.parameters.meta import expand_meta_parameters
+from galaxy.tools.parameters.meta import (
+    expand_meta_parameters,
+    expand_meta_parameters_async,
+)
 from galaxy.tools.parameters.populate_model import populate_model
 from galaxy.tools.parameters.workflow_utils import workflow_building_modes
 from galaxy.tools.parameters.wrapped_json import json_wrap
@@ -215,6 +224,7 @@ from ._types import (
     ToolRequestT,
     ToolStateDumpedToJsonInternalT,
     ToolStateDumpedToJsonT,
+    ToolStateJobInstanceExpansionT,
     ToolStateJobInstancePopulatedT,
     ToolStateJobInstanceT,
 )
@@ -225,7 +235,8 @@ from .execute import (
     DEFAULT_RERUN_REMAP_JOB_ID,
     DEFAULT_SET_OUTPUT_HID,
     DEFAULT_USE_CACHED_JOB,
-    execute as execute_job,
+    execute as execute_sync,
+    execute_async,
     ExecutionSlice,
     JobCallbackT,
     MappingParameters,
@@ -2060,6 +2071,64 @@ class Tool(UsesDictVisibleKeys, ToolParameterBundle):
         if self.check_values:
             visit_input_values(self.inputs, values, callback)
 
+    def expand_incoming_async(
+        self,
+        request_context: WorkRequestContext,
+        tool_request_internal_state: RequestInternalDereferencedToolState,
+        rerun_remap_job_id: Optional[int],
+    ) -> tuple[
+        list[ToolStateJobInstancePopulatedT],
+        list[ToolStateJobInstancePopulatedT],
+        Optional[MatchingCollections],
+        list[JobInternalToolState],
+    ]:
+        """The tool request API+tasks version of expand_incoming.
+
+        This is responsible for breaking the map over job requests into individual jobs for execution.
+        """
+        if self.input_translator:
+            raise exceptions.RequestParameterInvalidException(
+                "Failure executing tool request with id '%s' (cannot validate inputs from this type of data source tool - please POST to /api/tools).",
+                self.id,
+            )
+
+        set_dataset_matcher_factory(request_context, self)
+
+        expanded_incomings: list[ToolStateJobInstanceExpansionT]
+        job_tool_states: list[ToolStateJobInstanceT]
+        collection_info: Optional[MatchingCollections]
+        expanded_incomings, job_tool_states, collection_info = expand_meta_parameters_async(
+            request_context.app, self, tool_request_internal_state
+        )
+
+        self._ensure_expansion_is_valid(job_tool_states, rerun_remap_job_id)
+
+        # Process incoming data
+        validation_timer = self.app.execution_timer_factory.get_timer(
+            "internals.galaxy.tools.validation",
+            "Validated and populated state for tool request",
+        )
+        all_errors = []
+        all_params: list[ToolStateJobInstancePopulatedT] = []
+        internal_states: list[JobInternalToolState] = []
+        for expanded_incoming, job_tool_state in zip(expanded_incomings, job_tool_states):
+            expanded_incoming = fill_static_defaults(expanded_incoming, self, self.profile)
+            job_tool_state = fill_static_defaults(job_tool_state, self, self.profile)
+            params, errors = self._populate_async(request_context, expanded_incoming)
+            # params have had dynamic defaults requiring like dataset contents expanded out
+            # so we can use that backfill job_tool_state
+            fill_dynamic_defaults(request_context, self.inputs, job_tool_state, params)
+            internal_tool_state = JobInternalToolState(job_tool_state)
+            internal_tool_state.validate(self, f"{self.id} (job internal model)")
+
+            internal_states.append(internal_tool_state)
+            all_errors.append(errors)
+            all_params.append(params)
+        unset_dataset_matcher_factory(request_context)
+
+        log.info(validation_timer)
+        return all_params, all_errors, collection_info, internal_states
+
     def expand_incoming(
         self, request_context: WorkRequestContext, incoming: ToolRequestT, input_format: InputFormatT = "legacy"
     ) -> tuple[
@@ -2072,7 +2141,7 @@ class Tool(UsesDictVisibleKeys, ToolParameterBundle):
         set_dataset_matcher_factory(request_context, self)
         # Fixed set of input parameters may correspond to any number of jobs.
         # Expand these out to individual parameters for given jobs (tool executions).
-        expanded_incomings: list[ToolStateJobInstanceT]
+        expanded_incomings: list[ToolStateJobInstanceExpansionT]
         collection_info: Optional[MatchingCollections]
         expanded_incomings, collection_info = expand_meta_parameters(
             request_context, self, incoming, input_format=input_format
@@ -2098,7 +2167,9 @@ class Tool(UsesDictVisibleKeys, ToolParameterBundle):
         return all_params, all_errors, rerun_remap_job_id, collection_info
 
     def _ensure_expansion_is_valid(
-        self, expanded_incomings: list[ToolStateJobInstanceT], rerun_remap_job_id: Optional[int]
+        self,
+        expanded_incomings: Union[list[JobInternalToolState], list[ToolStateJobInstanceT]],
+        rerun_remap_job_id: Optional[int],
     ) -> None:
         """If the request corresponds to multiple jobs but this doesn't work with request configuration - raise an error.
 
@@ -2147,6 +2218,33 @@ class Tool(UsesDictVisibleKeys, ToolParameterBundle):
             self._handle_validate_input_hook(request_context, params, errors)
         return params, errors
 
+    def _populate_async(
+        self, request_context, expanded_incoming: ToolStateJobInstanceT
+    ) -> tuple[ToolStateJobInstancePopulatedT, ParameterValidationErrorsT]:
+        """Validate expanded parameters for a job to replace references with model objects.
+
+        So convert a ToolStateJobInstanceT to a ToolStateJobInstancePopulatedT.
+        """
+        params: ToolStateJobInstancePopulatedT = {}
+        errors: ParameterValidationErrorsT = {}
+        if self.input_translator:
+            self.input_translator.translate(expanded_incoming)
+        if not self.check_values:
+            # If `self.check_values` is false we don't do any checking or
+            # processing on input  This is used to pass raw values
+            # through to/from external sites.
+            params = expanded_incoming
+        else:
+            populate_state_async(
+                request_context,
+                self.inputs,
+                expanded_incoming,
+                params,
+                errors,
+            )
+            self._handle_validate_input_hook(request_context, params, errors)
+        return params, errors
+
     def _handle_validate_input_hook(
         self, request_context, params: ToolStateJobInstancePopulatedT, errors: ParameterValidationErrorsT
     ) -> None:
@@ -2183,6 +2281,39 @@ class Tool(UsesDictVisibleKeys, ToolParameterBundle):
                 completed_jobs[i] = None
         return completed_jobs
 
+    def handle_input_async(
+        self,
+        request_context: WorkRequestContext,
+        tool_request: ToolRequest,
+        tool_state: RequestInternalDereferencedToolState,
+        history: Optional[model.History] = None,
+        use_cached_job: bool = DEFAULT_USE_CACHED_JOB,
+        preferred_object_store_id: Optional[str] = DEFAULT_PREFERRED_OBJECT_STORE_ID,
+        rerun_remap_job_id: Optional[int] = None,
+        input_format: str = "legacy",
+    ):
+        """The tool request API+tasks version of handle_input."""
+        all_params, all_errors, collection_info, job_tool_states = self.expand_incoming_async(
+            request_context, tool_state, rerun_remap_job_id
+        )
+        self.handle_incoming_errors(all_errors)
+
+        mapping_params = MappingParameters(tool_request.request, all_params, tool_state, job_tool_states)
+        completed_jobs: dict[int, Optional[model.Job]] = self.completed_jobs(
+            request_context, use_cached_job, all_params
+        )
+        execute_async(
+            request_context,
+            self,
+            mapping_params,
+            request_context.history,
+            tool_request,
+            completed_jobs,
+            rerun_remap_job_id=rerun_remap_job_id,
+            preferred_object_store_id=preferred_object_store_id,
+            collection_info=collection_info,
+        )
+
     def handle_input(
         self,
         trans,
@@ -2207,11 +2338,11 @@ class Tool(UsesDictVisibleKeys, ToolParameterBundle):
         # If there were errors, we stay on the same page and display them
         self.handle_incoming_errors(all_errors)
 
-        mapping_params = MappingParameters(incoming, all_params)
+        mapping_params = MappingParameters(incoming, all_params, None, None)
         if use_cached_job:
             mapping_params.param_template["__use_cached_job__"] = use_cached_job
         completed_jobs: dict[int, Optional[Job]] = self.completed_jobs(trans, use_cached_job, all_params)
-        execution_tracker = execute_job(
+        execution_tracker = execute_sync(
             trans,
             self,
             mapping_params,

lib/galaxy/tools/_types.py

@@ -8,6 +8,8 @@ provide strong traditional typing semantics.
 +================================+============+=================================+============+===========+
 | ToolRequestT                   | request    | src dicts of encoded ids        | nope        |          |
 | ToolStateJobInstanceT          | a job      | src dicts of encoded ids        | nope        |          |
+| ToolStateJobInstanceExpansionT | a job      | a mix I think, things that were | nope        |          |
+|                                |            | expanded are objects            | nope        |          |
 | ToolStateJobInstancePopulatedT | a job      | model objs loaded from db       | check_param |          |
 | ToolStateDumpedToJsonT         | a job      | src dicts of encoded ids        | "           |          |
 |                                |            | (normalized into values attr)   | "           |          |
@@ -34,6 +36,10 @@ ToolRequestT = dict[str, Any]
 # been "checked" (check_param has not been called).
 ToolStateJobInstanceT = dict[str, Any]
 
+# After meta.expand_incoming stuff I think expanded parameters are in model object form but the other stuff is likely
+# still encoded IDs? None of this is verified though.
+ToolStateJobInstanceExpansionT = dict[str, Any]
+
 # Input dictionary for an individual job where objects are their model objects and parameters have been
 # "checked" (check_param has been called).
 ToolStateJobInstancePopulatedT = dict[str, Any]

lib/galaxy/tools/execute.py

@@ -22,12 +22,17 @@ from typing_extensions import TypeAlias
 
 from galaxy import model
 from galaxy.exceptions import ToolInputsNotOKException
+from galaxy.model import ToolRequest
 from galaxy.model.dataset_collections.matching import MatchingCollections
 from galaxy.model.dataset_collections.structure import (
     get_structure,
     tool_output_to_structure,
 )
 from galaxy.schema.credentials import CredentialsContext
+from galaxy.tool_util.parameters.state import (
+    JobInternalToolState,
+    RequestInternalDereferencedToolState,
+)
 from galaxy.tool_util.parser import ToolOutputCollectionPart
 from galaxy.tools.execution_helpers import (
     filter_output,
@@ -67,8 +72,58 @@ class PartialJobExecution(Exception):
 
 
 class MappingParameters(NamedTuple):
+    # the raw request - might correspond to multiple jobs
     param_template: ToolRequestT
+    # parameters corresponding to individual job
     param_combinations: list[ToolStateJobInstancePopulatedT]
+    # schema driven parameters
+    # model validated tool request - might correspond to multiple jobs
+    validated_param_template: Optional[RequestInternalDereferencedToolState] = None
+    # validated job parameters for individual jobs
+    validated_param_combinations: Optional[list[JobInternalToolState]] = None
+
+    def ensure_validated(self):
+        assert self.validated_param_template is not None
+        assert self.validated_param_combinations is not None
+
+
+def execute_async(
+    trans,
+    tool: "Tool",
+    mapping_params: MappingParameters,
+    history: model.History,
+    tool_request: ToolRequest,
+    completed_jobs: Optional[CompletedJobsT] = None,
+    rerun_remap_job_id: Optional[int] = None,
+    preferred_object_store_id: Optional[str] = None,
+    collection_info: Optional[MatchingCollections] = None,
+    workflow_invocation_uuid: Optional[str] = None,
+    invocation_step: Optional[model.WorkflowInvocationStep] = None,
+    max_num_jobs: Optional[int] = None,
+    job_callback: Optional[Callable] = None,
+    workflow_resource_parameters: Optional[dict[str, Any]] = None,
+    validate_outputs: bool = False,
+) -> "ExecutionTracker":
+    """The tool request/async version of execute."""
+    completed_jobs = completed_jobs or {}
+    mapping_params.ensure_validated()
+    return _execute(
+        trans,
+        tool,
+        mapping_params,
+        history,
+        tool_request,
+        rerun_remap_job_id,
+        preferred_object_store_id,
+        collection_info,
+        workflow_invocation_uuid,
+        invocation_step,
+        max_num_jobs,
+        job_callback,
+        completed_jobs,
+        workflow_resource_parameters,
+        validate_outputs,
+    )
 
 
 def execute(
@@ -87,12 +142,48 @@ def execute(
     completed_jobs: Optional[CompletedJobsT] = None,
     workflow_resource_parameters: Optional[WorkflowResourceParametersT] = None,
     validate_outputs: bool = False,
-):
+) -> "ExecutionTracker":
     """
     Execute a tool and return object containing summary (output data, number of
     failures, etc...).
     """
     completed_jobs = completed_jobs or {}
+    return _execute(
+        trans,
+        tool,
+        mapping_params,
+        history,
+        None,
+        rerun_remap_job_id,
+        preferred_object_store_id,
+        collection_info,
+        workflow_invocation_uuid,
+        invocation_step,
+        max_num_jobs,
+        job_callback,
+        completed_jobs,
+        workflow_resource_parameters,
+        validate_outputs,
+    )
+
+
+def _execute(
+    trans,
+    tool: "Tool",
+    mapping_params: MappingParameters,
+    history: model.History,
+    tool_request: Optional[ToolRequest],
+    rerun_remap_job_id: Optional[int],
+    preferred_object_store_id: Optional[str],
+    collection_info: Optional[MatchingCollections],
+    workflow_invocation_uuid: Optional[str],
+    invocation_step: Optional[model.WorkflowInvocationStep],
+    max_num_jobs: Optional[int],
+    job_callback: Optional[Callable],
+    completed_jobs: dict[int, Optional[model.Job]],
+    workflow_resource_parameters: Optional[dict[str, Any]],
+    validate_outputs: bool,
+) -> "ExecutionTracker":
     if max_num_jobs is not None:
         assert invocation_step is not None
     if rerun_remap_job_id:
@@ -117,10 +208,11 @@ def execute(
             "internals.galaxy.tools.execute.job_single", SINGLE_EXECUTION_SUCCESS_MESSAGE
         )
         params = execution_slice.param_combination
-        if "__data_manager_mode" in mapping_params.param_template:
-            params["__data_manager_mode"] = mapping_params.param_template["__data_manager_mode"]
-        if "__use_cached_job__" in mapping_params.param_template:
-            params["__use_cached_job__"] = mapping_params.param_template["__use_cached_job__"]
+        request_state = mapping_params.param_template
+        if "__data_manager_mode" in request_state:
+            params["__data_manager_mode"] = request_state["__data_manager_mode"]
+        if "__use_cached_job__" in request_state:
+            params["__use_cached_job__"] = request_state["__use_cached_job__"]
         if workflow_invocation_uuid:
             params["__workflow_invocation_uuid__"] = workflow_invocation_uuid
         elif "__workflow_invocation_uuid__" in params:
@@ -150,6 +242,8 @@ def execute(
             skip=skip,
         )
         if job:
+            if tool_request:
+                job.tool_request = tool_request
             log.debug(job_timer.to_str(tool_id=tool.id, job_id=job.id))
             execution_tracker.record_success(execution_slice, job, result)
             # associate dataset instances with the job that creates them
@@ -177,7 +271,7 @@ def execute(
             except ToolInputsNotOKException as e:
                 execution_tracker.record_error(e)
 
-    execution_tracker.ensure_implicit_collections_populated(history, mapping_params.param_template)
+    execution_tracker.ensure_implicit_collections_populated(history, mapping_params.param_template, tool_request)
     job_count = len(execution_tracker.param_combinations)
 
     jobs_executed = 0
@@ -190,7 +284,11 @@ def execute(
             has_remaining_jobs = True
             break
         else:
-            skip = execution_slice.param_combination.pop("__when_value__", None) is False
+            slice_params = execution_slice.param_combination
+            if isinstance(slice_params, JobInternalToolState):
+                slice_params = slice_params.input_state
+
+            skip = slice_params.pop("__when_value__", None) is False
             execute_single_job(execution_slice, completed_jobs[i], skip=skip)
             history = execution_slice.history or history
             jobs_executed += 1
@@ -444,15 +542,15 @@ class ExecutionTracker:
         mapped_output_structure = mapping_structure.multiply(output_structure)
         return mapped_output_structure
 
-    def ensure_implicit_collections_populated(self, history, params):
+    def ensure_implicit_collections_populated(self, history, params, tool_request: Optional[ToolRequest]):
         if not self.collection_info:
             return
 
         history = history or self.tool.get_default_history_by_trans(self.trans)
         # params = param_combinations[0] if param_combinations else mapping_params.param_template
-        self.precreate_output_collections(history, params)
+        self.precreate_output_collections(history, params, tool_request)
 
-    def precreate_output_collections(self, history, params):
+    def precreate_output_collections(self, history, params, tool_request: Optional[ToolRequest]):
         # params is just one sample tool param execution with parallelized
         # collection replaced with a specific dataset. Need to replace this
         # with the collection and wrap everything up so can evaluate output
@@ -497,6 +595,12 @@ class ExecutionTracker:
             )
             collection_instance.implicit_collection_jobs = implicit_collection_jobs
             collection_instances[output_name] = collection_instance
+            if tool_request:
+                assoc = model.ToolRequestImplicitCollectionAssociation()
+                assoc.tool_request = tool_request
+                assoc.dataset_collection = collection_instance
+                assoc.output_name = output_name
+                tool_request.implicit_collections.append(assoc)
             trans.sa_session.add(collection_instance)
         # Needed to flush the association created just above with
         # job.add_output_dataset_collection.
@@ -699,13 +803,13 @@ class WorkflowStepExecutionTracker(ExecutionTracker):
 
             yield ExecutionSlice(job_index, param_combination, dataset_collection_elements)
 
-    def ensure_implicit_collections_populated(self, history, params):
+    def ensure_implicit_collections_populated(self, history, params, tool_request: Optional[ToolRequest]):
         if not self.collection_info:
             return
 
         history = history or self.tool.get_default_history_by_trans(self.trans)
         if self.invocation_step.is_new:
-            self.precreate_output_collections(history, params)
+            self.precreate_output_collections(history, params, tool_request)
             for output_name, implicit_collection in self.implicit_collections.items():
                 self.invocation_step.add_output(output_name, implicit_collection)
         else:

lib/galaxy/tools/parameters/__init__.py

@@ -14,6 +14,10 @@ from typing import (
 from boltons.iterutils import remap
 
 from galaxy.exceptions import RequestParameterInvalidException
+from galaxy.model import (
+    HistoryDatasetAssociation,
+    HistoryDatasetCollectionAssociation,
+)
 from galaxy.util import unicodify
 from galaxy.util.expressions import ExpressionContext
 from galaxy.util.json import safe_loads
@@ -22,6 +26,7 @@ from .basic import (
     DataToolParameter,
     ParameterValueError,
     SelectToolParameter,
+    TextToolParameter,
     ToolParameter,
 )
 from .grouping import (
@@ -686,6 +691,204 @@ def _populate_state_legacy(
             state[input.name] = value
 
 
+def populate_state_async(
+    request_context,
+    inputs: ToolInputsT,
+    incoming: ToolStateJobInstanceT,
+    state: ToolStateJobInstancePopulatedT,
+    errors: ParameterValidationErrorsT,
+    context=None,
+):
+    context = ExpressionContext(state, context)
+    for input in inputs.values():
+        initial_value = input.get_initial_value(request_context, context)
+        input_name = input.name
+        state[input_name] = initial_value
+        group_state = state[input_name]
+        if input.type == "repeat":
+            repeat_input = cast(Repeat, input)
+            if (
+                len(incoming[repeat_input.name]) > repeat_input.max
+                or len(incoming[repeat_input.name]) < repeat_input.min
+            ):
+                errors[repeat_input.name] = "The number of repeat elements is outside the range specified by the tool."
+            else:
+                del group_state[:]
+                for rep in incoming[repeat_input.name]:
+                    new_state: ToolStateJobInstancePopulatedT = {}
+                    group_state.append(new_state)
+                    repeat_errors: ParameterValidationErrorsT = {}
+                    populate_state_async(
+                        request_context,
+                        repeat_input.inputs,
+                        rep,
+                        new_state,
+                        repeat_errors,
+                        context=context,
+                    )
+                    if repeat_errors:
+                        errors[repeat_input.name] = repeat_errors
+
+        elif input.type == "conditional":
+            conditional_input = cast(Conditional, input)
+            test_param = cast(ToolParameter, conditional_input.test_param)
+            test_param_value = incoming.get(conditional_input.name, {}).get(test_param.name)
+            value, error = check_param(request_context, test_param, test_param_value, context)
+            if error:
+                errors[test_param.name] = error
+            else:
+                try:
+                    current_case = conditional_input.get_current_case(value)
+                    group_state = state[conditional_input.name] = {}
+                    cast_errors: ParameterValidationErrorsT = {}
+                    populate_state_async(
+                        request_context,
+                        conditional_input.cases[current_case].inputs,
+                        cast(ToolStateJobInstanceT, incoming.get(conditional_input.name)),
+                        group_state,
+                        cast_errors,
+                        context=context,
+                    )
+                    if cast_errors:
+                        errors[conditional_input.name] = cast_errors
+                    group_state["__current_case__"] = current_case
+                except Exception:
+                    errors[test_param.name] = "The selected case is unavailable/invalid."
+            group_state[test_param.name] = value
+
+        elif input.type == "section":
+            section_input = cast(Section, input)
+            section_errors: ParameterValidationErrorsT = {}
+            populate_state_async(
+                request_context,
+                section_input.inputs,
+                cast(ToolStateJobInstanceT, incoming.get(section_input.name)),
+                group_state,
+                section_errors,
+                context=context,
+            )
+            if section_errors:
+                errors[section_input.name] = section_errors
+
+        elif input.type == "upload_dataset":
+            raise NotImplementedError
+
+        else:
+            assert isinstance(input, ToolParameter)
+            param_value = _get_incoming_value(incoming, input.name, state.get(input.name))
+            value, error = check_param(request_context, input, param_value, context, simple_errors=False)
+            if error:
+                errors[input.name] = error
+            state[input.name] = value
+
+            def to_internal_single(value):
+                if isinstance(value, HistoryDatasetCollectionAssociation):
+                    return {"src": "hdca", "id": value.id}
+                elif isinstance(value, HistoryDatasetAssociation):
+                    return {"src": "hda", "id": value.id}
+                else:
+                    # tests and such to confirm we need DCE, LDDA, etc...
+                    return value
+
+            def to_internal(value):
+                if isinstance(value, list):
+                    return [to_internal_single(v) for v in value]
+                else:
+                    return to_internal_single(value)
+
+            if input_name not in incoming:
+                if input.type == "data_column":
+                    if isinstance(value, str):
+                        incoming[input_name] = int(value)
+                    elif isinstance(value, list):
+                        incoming[input_name] = [int(v) for v in value]
+                    else:
+                        incoming[input_name] = value
+                elif input.type == "text":
+                    text_input = cast(TextToolParameter, input)
+                    # see behavior of tools in test_tools.py::test_null_to_text_tools
+                    # these parameters act as empty string in this context
+                    if value is None and not text_input.optional:
+                        incoming[input_name] = ""
+                    else:
+                        incoming[input_name] = value
+                else:
+                    incoming[input_name] = to_internal(value)
+
+
+def fill_dynamic_defaults(
+    request_context,
+    inputs: ToolInputsT,
+    job_tool_state: ToolStateJobInstanceT,
+    params: ToolStateJobInstancePopulatedT,
+    context=None,
+):
+    """
+    Expands incoming parameters with default values.
+    """
+    context = ExpressionContext(job_tool_state, job_tool_state)
+    for input in inputs.values():
+        if input.type == "repeat":
+            repeat_input = cast(Repeat, input)
+            repeat_name = repeat_input.name
+            for rep, rep_params in enumerate(job_tool_state[repeat_name]):
+                fill_dynamic_defaults(
+                    request_context,
+                    repeat_input.inputs,
+                    rep_params,
+                    params[repeat_name][rep],
+                    context=context,
+                )
+
+        elif input.type == "conditional":
+            conditional_input = cast(Conditional, input)
+            test_param = cast(ToolParameter, conditional_input.test_param)
+            test_param_value = job_tool_state.get(conditional_input.name, {}).get(test_param.name)
+            try:
+                current_case = conditional_input.get_current_case(test_param_value)
+                fill_dynamic_defaults(
+                    request_context,
+                    conditional_input.cases[current_case].inputs,
+                    cast(ToolStateJobInstanceT, job_tool_state.get(conditional_input.name)),
+                    cast(ToolStateJobInstancePopulatedT, params.get(conditional_input.name)),
+                    context=context,
+                )
+            except Exception:
+                raise Exception("The selected case is unavailable/invalid.")
+
+        elif input.type == "section":
+            section_input = cast(Section, input)
+            fill_dynamic_defaults(
+                request_context,
+                section_input.inputs,
+                cast(ToolStateJobInstanceT, job_tool_state.get(section_input.name)),
+                cast(ToolStateJobInstancePopulatedT, params.get(section_input.name)),
+                context=context,
+            )
+
+        elif input.type == "upload_dataset":
+            raise NotImplementedError
+
+        else:
+            if input.name not in job_tool_state and input.name in params:
+                if input.type == "data_column":
+                    if isinstance(params[input.name], str):
+                        job_tool_state[input.name] = int(params[input.name])
+                    elif isinstance(params[input.name], list):
+                        job_tool_state[input.name] = [int(v) for v in params[input.name]]
+                    else:
+                        job_tool_state[input.name] = params[input.name]
+                elif input.type == "data_collection":
+                    data_collection = params[input.name]
+                    if data_collection:
+                        job_tool_state[input.name] = {
+                            "src": "hdca",
+                            "id": data_collection.id,
+                        }
+                else:
+                    job_tool_state[input.name] = params[input.name]
+
+
 def _get_incoming_value(incoming, key, default):
     """
     Fetch value from incoming dict directly or check special nginx upload

lib/galaxy/tools/parameters/meta.py

@@ -15,13 +15,16 @@ from galaxy import (
 from galaxy.model import (
     DatasetCollectionElement,
     DatasetInstance,
+    HistoryDatasetAssociation,
     HistoryDatasetCollectionAssociation,
+    LibraryDatasetDatasetAssociation,
 )
 from galaxy.model.dataset_collections import (
     matching,
     subcollections,
 )
 from galaxy.model.dataset_collections.adapters import PromoteCollectionElementToCollectionAdapter
+from galaxy.tool_util.parameters import RequestInternalDereferencedToolState
 from galaxy.util.permutations import (
     build_combos,
     input_classification,
@@ -37,6 +40,7 @@ from .wrapped import process_key
 from .._types import (
     InputFormatT,
     ToolRequestT,
+    ToolStateDumpedToJsonInternalT,
     ToolStateJobInstanceT,
 )
 
@@ -333,6 +337,80 @@ def split_inputs_nested(inputs, nested_dict, classifier):
     return (single_inputs_nested, matched_multi_inputs, multiplied_multi_inputs)
 
 
+ExpandedAsyncT = tuple[
+    list[ToolStateJobInstanceT], list[ToolStateDumpedToJsonInternalT], Optional[matching.MatchingCollections]
+]
+
+
+def expand_meta_parameters_async(app, tool, incoming: RequestInternalDereferencedToolState) -> ExpandedAsyncT:
+    collections_to_match = matching.CollectionsToMatch()
+
+    def classifier_from_value(value, input_key):
+        if isinstance(value, dict) and "values" in value:
+            # Explicit meta wrapper for inputs...
+            is_batch = value.get("__class__", "Batch") == "Batch"
+            is_linked = value.get("linked", True)
+            if is_batch and is_linked:
+                classification = input_classification.MATCHED
+            elif is_batch:
+                classification = input_classification.MULTIPLIED
+            else:
+                classification = input_classification.SINGLE
+            if __collection_multirun_parameter(value):
+                collection_value = value["values"][0]
+                values = __expand_collection_parameter_async(
+                    app, input_key, collection_value, collections_to_match, linked=is_linked
+                )
+            else:
+                values = value["values"]
+        else:
+            classification = input_classification.SINGLE
+            values = value
+        return classification, values
+
+    # is there a way to make Pydantic ensure reordering isn't needed - model and serialize out the parameters maybe?
+    reordered_incoming = reorder_parameters(tool, incoming.input_state, incoming.input_state, True)
+    incoming_template = reordered_incoming
+
+    single_inputs, matched_multi_inputs, multiplied_multi_inputs = split_inputs_nested(
+        tool.inputs, incoming_template, classifier_from_value
+    )
+    expanded_incomings = build_combos(single_inputs, matched_multi_inputs, multiplied_multi_inputs, nested=True)
+    # those all have sa model objects from expansion to be used within for additional logic (maybe?)
+    # but we want to record just src and IDS in the job state object - so undo that
+    expanded_job_states = build_combos(
+        to_decoded_json(single_inputs),
+        to_decoded_json(matched_multi_inputs),
+        to_decoded_json(multiplied_multi_inputs),
+        nested=True,
+    )
+    if collections_to_match.has_collections():
+        collection_info = app.dataset_collection_manager.match_collections(collections_to_match)
+    else:
+        collection_info = None
+    return expanded_incomings, expanded_job_states, collection_info
+
+
+def to_decoded_json(has_objects):
+    if isinstance(has_objects, dict):
+        decoded_json = {}
+        for key, value in has_objects.items():
+            decoded_json[key] = to_decoded_json(value)
+        return decoded_json
+    elif isinstance(has_objects, list):
+        return [to_decoded_json(o) for o in has_objects]
+    elif isinstance(has_objects, DatasetCollectionElement):
+        return {"src": "dce", "id": has_objects.id}
+    elif isinstance(has_objects, HistoryDatasetAssociation):
+        return {"src": "hda", "id": has_objects.id}
+    elif isinstance(has_objects, HistoryDatasetCollectionAssociation):
+        return {"src": "hdca", "id": has_objects.id}
+    elif isinstance(has_objects, LibraryDatasetDatasetAssociation):
+        return {"src": "ldda", "id": has_objects.id}
+    else:
+        return has_objects
+
+
 CollectionExpansionListT = Union[
     list[Union[DatasetCollectionElement, PromoteCollectionElementToCollectionAdapter]], list[DatasetInstance]
 ]
@@ -371,7 +449,9 @@ def __expand_collection_parameter(
         raise exceptions.ToolInputsNotReadyException("An input collection is not populated.")
     collections_to_match.add(input_key, item, subcollection_type=subcollection_type, linked=linked)
     if subcollection_type is not None:
-        subcollection_elements = subcollections._split_dataset_collection(collection, subcollection_type)
+        subcollection_elements: list[Union[DatasetCollectionElement, PromoteCollectionElementToCollectionAdapter]] = (
+            subcollections._split_dataset_collection(collection, subcollection_type)
+        )
         return subcollection_elements
     else:
         hdas: list[DatasetInstance] = []
@@ -382,8 +462,36 @@ def __expand_collection_parameter(
         return hdas
 
 
+def __expand_collection_parameter_async(
+    app, input_key, incoming_val, collections_to_match: "matching.CollectionsToMatch", linked=False
+) -> CollectionExpansionListT:
+    # If subcollection multirun of data_collection param - value will
+    # be "hdca_id|subcollection_type" else it will just be hdca_id
+    try:
+        src = incoming_val["src"]
+        if src != "hdca":
+            raise exceptions.ToolMetaParameterException(f"Invalid dataset collection source type {src}")
+        hdc_id = incoming_val["id"]
+        subcollection_type = incoming_val.get("map_over_type", None)
+    except TypeError:
+        hdc_id = incoming_val
+        subcollection_type = None
+    hdc = app.model.context.get(HistoryDatasetCollectionAssociation, hdc_id)
+    collections_to_match.add(input_key, hdc, subcollection_type=subcollection_type, linked=linked)
+    if subcollection_type is not None:
+        subcollection_elements = subcollections.split_dataset_collection_instance(hdc, subcollection_type)
+        return subcollection_elements
+    else:
+        hdas: list[DatasetInstance] = []
+        for element in hdc.collection.dataset_elements:
+            hda = element.dataset_instance
+            hda.element_identifier = element.element_identifier
+            hdas.append(hda)
+        return hdas
+
+
 def __collection_multirun_parameter(value: dict[str, Any]) -> bool:
-    is_batch = value.get("batch", False)
+    is_batch = value.get("batch", False) or value.get("__class__", None) == "Batch"
     if not is_batch:
         return False
 

lib/galaxy/webapps/galaxy/api/histories.py

@@ -60,6 +60,7 @@ from galaxy.schema.schema import (
     ShareWithPayload,
     SharingStatus,
     StoreExportPayload,
+    ToolRequestModel,
     UpdateHistoryPayload,
     WriteStoreToPayload,
 )
@@ -373,6 +374,17 @@ class FastAPIHistories:
     ) -> list[Any]:
         return self.service.citations(trans, history_id)
 
+    @router.get(
+        "/api/histories/{history_id}/tool_requests",
+        summary="Return all the tool requests for the tools submitted to this history.",
+    )
+    def tool_requests(
+        self,
+        history_id: HistoryIDPathParam,
+        trans: ProvidesHistoryContext = DependsOnTrans,
+    ) -> list[ToolRequestModel]:
+        return self.service.tool_requests(trans, history_id)
+
     @router.post(
         "/api/histories",
         summary="Creates a new history.",

lib/galaxy/webapps/galaxy/api/jobs.py

@@ -46,6 +46,7 @@ from galaxy.schema.jobs import (
     JobInputAssociation,
     JobInputSummary,
     JobOutputAssociation,
+    JobOutputCollectionAssociation,
     ReportJobErrorPayload,
     SearchJobsPayload,
 )
@@ -70,11 +71,14 @@ from galaxy.webapps.galaxy.api import (
 )
 from galaxy.webapps.galaxy.api.common import query_parameter_as_list
 from galaxy.webapps.galaxy.services.jobs import (
+    JobCreateResponse,
     JobIndexPayload,
     JobIndexViewEnum,
+    JobRequest,
     JobsService,
 )
 from galaxy.work.context import proxy_work_context_for_history
+from .tools import validate_not_protected
 
 log = logging.getLogger(__name__)
 
@@ -158,6 +162,12 @@ ImplicitCollectionJobsIdQueryParam: Optional[DecodedDatabaseIdField] = Query(
     description="Limit listing of jobs to those that match the specified implicit collection job ID. If none, jobs from any implicit collection execution (or from no implicit collection execution) may be returned.",
 )
 
+ToolRequestIdQueryParam: Optional[DecodedDatabaseIdField] = Query(
+    default=None,
+    title="Tool Request ID",
+    description="Limit listing of jobs to those that were created from the supplied tool request ID. If none, jobs from any tool request (or from no workflows) may be returned.",
+)
+
 SortByQueryParam: JobIndexSortByEnum = Query(
     default=JobIndexSortByEnum.update_time,
     title="Sort By",
@@ -261,6 +271,13 @@ class ShowFullJobResponse(EncodedJobDetails):
 class FastAPIJobs:
     service: JobsService = depends(JobsService)
 
+    @router.post("/api/jobs")
+    def create(
+        self, trans: ProvidesHistoryContext = DependsOnTrans, job_request: JobRequest = Body(...)
+    ) -> JobCreateResponse:
+        validate_not_protected(job_request.tool_id)
+        return self.service.create(trans, job_request)
+
     @router.get("/api/jobs")
     def index(
         self,
@@ -277,6 +294,7 @@ class FastAPIJobs:
         workflow_id: Optional[DecodedDatabaseIdField] = WorkflowIdQueryParam,
         invocation_id: Optional[DecodedDatabaseIdField] = InvocationIdQueryParam,
         implicit_collection_jobs_id: Optional[DecodedDatabaseIdField] = ImplicitCollectionJobsIdQueryParam,
+        tool_request_id: Optional[DecodedDatabaseIdField] = ToolRequestIdQueryParam,
         order_by: JobIndexSortByEnum = SortByQueryParam,
         search: Optional[str] = SearchQueryParam,
         limit: int = LimitQueryParam,
@@ -295,6 +313,7 @@ class FastAPIJobs:
             workflow_id=workflow_id,
             invocation_id=invocation_id,
             implicit_collection_jobs_id=implicit_collection_jobs_id,
+            tool_request_id=tool_request_id,
             order_by=order_by,
             search=search,
             limit=limit,
@@ -415,12 +434,14 @@ class FastAPIJobs:
         self,
         job_id: JobIdPathParam,
         trans: ProvidesUserContext = DependsOnTrans,
-    ) -> list[JobOutputAssociation]:
+    ) -> list[Union[JobOutputAssociation, JobOutputCollectionAssociation]]:
         job = self.service.get_job(trans=trans, job_id=job_id)
         associations = self.service.dictify_associations(trans, job.output_datasets, job.output_library_datasets)
-        output_associations = []
+        output_associations: list[Union[JobOutputAssociation, JobOutputCollectionAssociation]] = []
         for association in associations:
             output_associations.append(JobOutputAssociation(name=association.name, dataset=association.dataset))
+
+        output_associations.extend(self.service.dictify_output_collection_associations(trans, job))
         return output_associations
 
     @router.get(

lib/galaxy/webapps/galaxy/api/tools.py

@@ -14,6 +14,7 @@ from typing import (
 from fastapi import (
     Body,
     Depends,
+    Path,
     Query,
     Request,
     Response,
@@ -38,6 +39,8 @@ from galaxy.managers.context import (
 from galaxy.managers.hdas import HDAManager
 from galaxy.managers.histories import HistoryManager
 from galaxy.managers.landing import LandingRequestManager
+from galaxy.managers.tools import ToolRunReference
+from galaxy.model import ToolRequest
 from galaxy.model.dataset_collections.workbook_util import workbook_to_bytes
 from galaxy.schema.fetch_data import (
     CreateDataLandingPayload,
@@ -45,11 +48,14 @@ from galaxy.schema.fetch_data import (
     FetchDataFormPayload,
     FetchDataPayload,
 )
+from galaxy.schema.fields import DecodedDatabaseIdField
 from galaxy.schema.schema import (
     ClaimLandingPayload,
     CreateToolLandingRequestPayload,
     ToolLandingRequest,
+    ToolRequestDetailedModel,
 )
+from galaxy.tool_util.parameters import ToolParameterT
 from galaxy.tool_util.verify import ToolTestDescriptionDict
 from galaxy.tool_util_models import UserToolSource
 from galaxy.tools.evaluation import global_tool_errors
@@ -76,6 +82,7 @@ from galaxy.web import (
 from galaxy.webapps.base.controller import UsesVisualizationMixin
 from galaxy.webapps.base.webapp import GalaxyWebTransaction
 from galaxy.webapps.galaxy.api.common import serve_workbook
+from galaxy.webapps.galaxy.services.base import tool_request_detailed_to_model
 from galaxy.webapps.galaxy.services.tools import ToolsService
 from . import (
     APIContentTypeRoute,
@@ -130,6 +137,13 @@ router = Router(tags=["tools"])
 
 FetchDataForm = as_form(FetchDataFormPayload)
 
+ToolIDPathParam: str = Path(
+    ...,
+    title="Tool ID",
+    description="The tool ID for the lineage stored in Galaxy's toolbox.",
+)
+ToolVersionQueryParam: Optional[str] = Query(default=None, title="Tool Version", description="")
+
 
 async def get_files(request: Request, files: Optional[list[UploadFile]] = None):
     # FastAPI's UploadFile is a very light wrapper around starlette's UploadFile
@@ -269,6 +283,57 @@ class FetchTools:
         tool_landing_request = self.service.data_landing_to_tool_landing(trans, data_landing_request)
         return self.landing_manager.create_tool_landing_request(tool_landing_request)
 
+    @router.get(
+        "/api/tool_requests/{id}",
+        summary="Get tool request state.",
+    )
+    def get_tool_request(
+        self,
+        id: DecodedDatabaseIdField,
+        trans: ProvidesHistoryContext = DependsOnTrans,
+    ) -> ToolRequestDetailedModel:
+        tool_request = self._get_tool_request_or_raise_not_found(trans, id)
+        return tool_request_detailed_to_model(tool_request)
+
+    @router.get(
+        "/api/tool_requests/{id}/state",
+        summary="Get tool request state.",
+    )
+    def tool_request_state(
+        self,
+        id: DecodedDatabaseIdField,
+        trans: ProvidesHistoryContext = DependsOnTrans,
+    ) -> str:
+        tool_request = self._get_tool_request_or_raise_not_found(trans, id)
+        state = tool_request.state
+        if not state:
+            raise exceptions.InconsistentDatabase()
+        return state
+
+    @router.get(
+        "/api/tools/{tool_id}/inputs",
+        summary="Get tool inputs.",
+    )
+    def tool_inputs(
+        self,
+        tool_id: str = ToolIDPathParam,
+        tool_version: Optional[str] = ToolVersionQueryParam,
+        trans: ProvidesHistoryContext = DependsOnTrans,
+    ) -> list[ToolParameterT]:
+        tool_run_ref = ToolRunReference(tool_id=tool_id, tool_version=tool_version, tool_uuid=None)
+        return self.service.inputs(trans, tool_run_ref)
+
+    def _get_tool_request_or_raise_not_found(
+        self, trans: ProvidesHistoryContext, id: DecodedDatabaseIdField
+    ) -> ToolRequest:
+        tool_request: Optional[ToolRequest] = cast(
+            Optional[ToolRequest], trans.app.model.context.query(ToolRequest).get(id)
+        )
+        if tool_request is None:
+            raise exceptions.ObjectNotFound()
+        assert tool_request
+        return tool_request
+
     @router.post("/api/tool_landings", public=True, allow_cors=True)
     def create_landing(
         self,
@@ -299,7 +364,6 @@ class FetchTools:
     ) -> ToolLandingRequest:
         return self.landing_manager.get_tool_landing_request(trans, uuid)
 
-
 class ToolsController(BaseGalaxyAPIController, UsesVisualizationMixin):
     """
     RESTful controller for interactions with tools.
@@ -787,10 +851,7 @@ class ToolsController(BaseGalaxyAPIController, UsesVisualizationMixin):
         """
         tool_id = payload.get("tool_id")
         tool_uuid = payload.get("tool_uuid")
-        if tool_id in PROTECTED_TOOLS:
-            raise exceptions.RequestParameterInvalidException(
-                f"Cannot execute tool [{tool_id}] directly, must use alternative endpoint."
-            )
+        validate_not_protected(tool_id)
         if tool_id is None and tool_uuid is None:
             raise exceptions.RequestParameterInvalidException("Must specify a valid tool_id to use this endpoint.")
         __tags = payload.get("__tags", [])
@@ -799,6 +860,13 @@ class ToolsController(BaseGalaxyAPIController, UsesVisualizationMixin):
         return self.service._create(trans, payload, **kwd)
 
 
+def validate_not_protected(tool_id: Optional[str]):
+    if tool_id in PROTECTED_TOOLS:
+        raise exceptions.RequestParameterInvalidException(
+            f"Cannot execute tool [{tool_id}] directly, must use alternative endpoint."
+        )
+
+
 def _kwd_or_payload(kwd: dict[str, Any]) -> dict[str, Any]:
     if "payload" in kwd:
         kwd = cast(dict[str, Any], kwd.get("payload"))

lib/galaxy/webapps/galaxy/services/base.py

@@ -22,13 +22,20 @@ from galaxy.managers.base import (
 )
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.managers.model_stores import create_objects_from_store
-from galaxy.model import User
+from galaxy.model import (
+    ToolRequest,
+    User,
+)
 from galaxy.model.store import (
     get_export_store_factory,
     ModelExportStore,
 )
 from galaxy.schema.fields import EncodedDatabaseIdField
-from galaxy.schema.schema import AsyncTaskResultSummary
+from galaxy.schema.schema import (
+    AsyncTaskResultSummary,
+    ToolRequestDetailedModel,
+    ToolRequestModel,
+)
 from galaxy.security.idencoding import IdEncodingHelper
 from galaxy.short_term_storage import (
     ShortTermStorageAllocator,
@@ -192,3 +199,31 @@ def async_task_summary(async_result: AsyncResult) -> AsyncTaskResultSummary:
         name=name,
         queue=queue,
     )
+
+
+def tool_request_to_model(tool_request: ToolRequest) -> ToolRequestModel:
+    as_dict = {
+        "id": tool_request.id,
+        "request": tool_request.request,
+        "state": tool_request.state,
+        "state_message": tool_request.state_message,
+    }
+    return ToolRequestModel.model_validate(as_dict)
+
+
+def tool_request_detailed_to_model(tool_request: ToolRequest) -> ToolRequestDetailedModel:
+    jobs = [{"src": "job", "id": job.id} for job in tool_request.jobs]
+    implicit_collections = [
+        {"src": "hdca", "id": assoc.dataset_collection.id, "output_name": assoc.output_name}
+        for assoc in tool_request.implicit_collections
+    ]
+    as_dict = {
+        "id": tool_request.id,
+        "request": tool_request.request,
+        "state": tool_request.state,
+        "state_message": tool_request.state_message,
+        "jobs": jobs,
+        "implicit_collections": implicit_collections,
+    }
+    model = ToolRequestDetailedModel.model_validate(as_dict)
+    return model

lib/galaxy/webapps/galaxy/services/histories.py

@@ -68,6 +68,7 @@ from galaxy.schema.schema import (
     ShareHistoryWithStatus,
     ShareWithPayload,
     StoreExportPayload,
+    ToolRequestModel,
     WriteStoreToPayload,
 )
 from galaxy.schema.tasks import (
@@ -85,6 +86,7 @@ from galaxy.webapps.galaxy.services.base import (
     model_store_storage_target,
     ServesExportStores,
     ServiceBase,
+    tool_request_to_model,
 )
 from galaxy.webapps.galaxy.services.notifications import NotificationService
 from galaxy.webapps.galaxy.services.sharable import ShareableService
@@ -536,6 +538,13 @@ class HistoriesService(ServiceBase, ConsumesModelStores, ServesExportStores):
         ]
         return rval
 
+    def tool_requests(
+        self, trans: ProvidesHistoryContext, history_id: DecodedDatabaseIdField
+    ) -> list[ToolRequestModel]:
+        history = self.manager.get_accessible(history_id, trans.user, current_history=trans.history)
+        tool_requests = history.tool_requests
+        return [tool_request_to_model(tr) for tr in tool_requests]
+
     def citations(self, trans: ProvidesHistoryContext, history_id: DecodedDatabaseIdField):
         """
         Return all the references for the tools used to produce the datasets in

lib/galaxy/webapps/galaxy/services/jobs.py

@@ -1,3 +1,4 @@
+import logging
 from enum import Enum
 from typing import (
     Any,
@@ -6,24 +7,60 @@ from typing import (
     Union,
 )
 
+from pydantic import (
+    BaseModel,
+    Field,
+)
+
 from galaxy import (
     exceptions,
     model,
 )
+from galaxy.celery.tasks import queue_jobs
 from galaxy.managers import hdas
 from galaxy.managers.base import security_check
-from galaxy.managers.context import ProvidesUserContext
+from galaxy.managers.context import (
+    ProvidesHistoryContext,
+    ProvidesUserContext,
+)
+from galaxy.managers.histories import HistoryManager
 from galaxy.managers.jobs import (
     JobManager,
     JobSearch,
     view_show_job,
 )
-from galaxy.model import Job
-from galaxy.schema.fields import DecodedDatabaseIdField
-from galaxy.schema.jobs import JobAssociation
-from galaxy.schema.schema import JobIndexQueryPayload
+from galaxy.managers.tools import ToolRunReference
+from galaxy.model import (
+    Job,
+    ToolRequest,
+    ToolSource as ToolSourceModel,
+)
+from galaxy.schema.fields import (
+    DecodedDatabaseIdField,
+    EncodedDatabaseIdField,
+)
+from galaxy.schema.jobs import (
+    JobAssociation,
+    JobOutputCollectionAssociation,
+)
+from galaxy.schema.schema import (
+    AsyncTaskResultSummary,
+    JobIndexQueryPayload,
+)
+from galaxy.schema.tasks import (
+    QueueJobs,
+    ToolSource,
+)
 from galaxy.security.idencoding import IdEncodingHelper
-from galaxy.webapps.galaxy.services.base import ServiceBase
+from galaxy.tool_util.parameters import (
+    decode,
+    RequestToolState,
+)
+from galaxy.webapps.galaxy.services.base import (
+    async_task_summary,
+    ServiceBase,
+)
+from .tools import validate_tool_for_running
 
 if TYPE_CHECKING:
     from galaxy.model import (
@@ -31,6 +68,26 @@ if TYPE_CHECKING:
         LibraryDatasetDatasetAssociation,
     )
 
+log = logging.getLogger(__name__)
+
+
+class JobRequest(BaseModel):
+    tool_id: Optional[str] = Field(default=None, title="tool_id", description="TODO")
+    tool_uuid: Optional[str] = Field(default=None, title="tool_uuid", description="TODO")
+    tool_version: Optional[str] = Field(default=None, title="tool_version", description="TODO")
+    history_id: Optional[DecodedDatabaseIdField] = Field(default=None, title="history_id", description="TODO")
+    inputs: Optional[dict[str, Any]] = Field(default_factory=lambda: {}, title="Inputs", description="TODO")
+    use_cached_jobs: Optional[bool] = Field(default=None, title="use_cached_jobs")
+    rerun_remap_job_id: Optional[DecodedDatabaseIdField] = Field(
+        default=None, title="rerun_remap_job_id", description="TODO"
+    )
+    send_email_notification: bool = Field(default=False, title="Send Email Notification", description="TODO")
+
+
+class JobCreateResponse(BaseModel):
+    tool_request_id: EncodedDatabaseIdField
+    task_result: AsyncTaskResultSummary
+
 
 class JobIndexViewEnum(str, Enum):
     collection = "collection"
@@ -45,6 +102,7 @@ class JobsService(ServiceBase):
     job_manager: JobManager
     job_search: JobSearch
     hda_manager: hdas.HDAManager
+    history_manager: HistoryManager
 
     def __init__(
         self,
@@ -52,11 +110,13 @@ class JobsService(ServiceBase):
         job_manager: JobManager,
         job_search: JobSearch,
         hda_manager: hdas.HDAManager,
+        history_manager: HistoryManager,
     ):
         super().__init__(security=security)
         self.job_manager = job_manager
         self.job_search = job_search
         self.hda_manager = hda_manager
+        self.history_manager = history_manager
 
     def show(
         self,
@@ -158,3 +218,61 @@ class JobsService(ServiceBase):
             else:
                 dataset_dict = {"src": "ldda", "id": dataset.id}
         return JobAssociation(name=job_dataset_association.name, dataset=dataset_dict)
+
+    def dictify_output_collection_associations(self, trans, job: model.Job) -> list[JobOutputCollectionAssociation]:
+        output_associations: list[JobOutputCollectionAssociation] = []
+        for job_output_collection_association in job.output_dataset_collection_instances:
+            ref_dict = {"src": "hdca", "id": job_output_collection_association.dataset_collection_id}
+            output_associations.append(
+                JobOutputCollectionAssociation(
+                    name=job_output_collection_association.name,
+                    dataset_collection_instance=ref_dict,
+                )
+            )
+        return output_associations
+
+    def create(self, trans: ProvidesHistoryContext, job_request: JobRequest) -> JobCreateResponse:
+        tool_run_reference = ToolRunReference(job_request.tool_id, job_request.tool_uuid, job_request.tool_version)
+        tool = validate_tool_for_running(trans, tool_run_reference)
+        history_id = job_request.history_id
+        target_history = None
+        if history_id is not None:
+            target_history = self.history_manager.get_owned(history_id, trans.user, current_history=trans.history)
+        inputs = job_request.inputs
+        request_state = RequestToolState(inputs or {})
+        request_state.validate(tool, f"{tool.id} (request model)")
+        request_internal_state = decode(request_state, tool, trans.security.decode_id)
+        tool_request = ToolRequest()
+        # TODO: hash and such...
+        tool_source_model = ToolSourceModel(
+            source=[p.model_dump() for p in tool.parameters],
+            hash="TODO",
+        )
+        tool_request.request = request_internal_state.input_state
+        tool_request.tool_source = tool_source_model
+        tool_request.state = ToolRequest.states.NEW
+        tool_request.history = target_history
+        sa_session = trans.sa_session
+        sa_session.add(tool_source_model)
+        sa_session.add(tool_request)
+        sa_session.commit()
+        tool_request_id = tool_request.id
+        tool_source = ToolSource(
+            raw_tool_source=tool.tool_source.to_string(),
+            tool_dir=tool.tool_dir,
+        )
+        task_request = QueueJobs(
+            user=trans.async_request_user,
+            history_id=target_history and target_history.id,
+            tool_source=tool_source,
+            tool_request_id=tool_request_id,
+            use_cached_jobs=job_request.use_cached_jobs or False,
+            rerun_remap_job_id=job_request.rerun_remap_job_id,
+        )
+        result = queue_jobs.delay(request=task_request)
+        return JobCreateResponse(
+            **{
+                "tool_request_id": tool_request_id,
+                "task_result": async_task_summary(result),
+            }
+        )

lib/galaxy/webapps/galaxy/services/tools.py

@@ -6,7 +6,6 @@ from json import dumps
 from typing import (
     Any,
     cast,
-    Literal,
     Optional,
     Union,
 )
@@ -51,6 +50,7 @@ from galaxy.schema.fetch_data import (
 )
 from galaxy.schema.schema import CreateToolLandingRequestPayload
 from galaxy.security.idencoding import IdEncodingHelper
+from galaxy.tool_util.parameters import ToolParameterT
 from galaxy.tool_util_models.parameters import (
     CollectionElementCollectionRequestUri,
     CollectionElementDataRequestUri,
@@ -59,6 +59,7 @@ from galaxy.tool_util_models.parameters import (
     FileRequestUri,
 )
 from galaxy.tools import Tool
+from galaxy.tools._types import InputFormatT
 from galaxy.tools.search import ToolBoxSearch
 from galaxy.util.path import safe_contains
 from galaxy.webapps.galaxy.services._fetch_util import validate_and_normalize_targets
@@ -236,6 +237,14 @@ class ToolsService(ServiceBase):
             public=data_landing_payload.public,
         )
 
+    def inputs(
+        self,
+        trans: ProvidesHistoryContext,
+        tool_ref: ToolRunReference,
+    ) -> list[ToolParameterT]:
+        tool = get_tool(trans, tool_ref)
+        return tool.parameters
+
     def create_fetch(
         self,
         trans: ProvidesHistoryContext,
@@ -328,7 +337,7 @@ class ToolsService(ServiceBase):
         input_format = str(payload.get("input_format", "legacy"))
         if input_format not in ("legacy", "21.01"):
             raise exceptions.RequestParameterInvalidException(f"input_format invalid {input_format}")
-        input_format = cast(Literal["legacy", "21.01"], input_format)
+        input_format = cast(InputFormatT, input_format)
         if "data_manager_mode" in payload:
             incoming["__data_manager_mode"] = payload["data_manager_mode"]
         vars = tool.handle_input(

lib/galaxy_test/api/conftest.py

@@ -147,7 +147,7 @@ def required_tool(dataset_populator: DatasetPopulator, history_id: str, required
     return tool
 
 
-@pytest.fixture(params=["legacy", "21.01"])
+@pytest.fixture(params=["legacy", "21.01", "request"])
 def tool_input_format(request) -> Iterator[DescribeToolInputs]:
     yield DescribeToolInputs(request.param)
 

lib/galaxy_test/api/test_tool_execute.py

@@ -27,16 +27,25 @@ def test_multidata_param(
 ):
     hda1 = target_history.with_dataset("1\t2\t3").src_dict
     hda2 = target_history.with_dataset("4\t5\t6").src_dict
-    inputs = tool_input_format.when.flat(
-        {
-            "f1": {"batch": False, "values": [hda1, hda2]},
-            "f2": {"batch": False, "values": [hda2, hda1]},
-        }
-    ).when.nested(
-        {
-            "f1": {"batch": False, "values": [hda1, hda2]},
-            "f2": {"batch": False, "values": [hda2, hda1]},
-        }
+    inputs = (
+        tool_input_format.when.flat(
+            {
+                "f1": {"batch": False, "values": [hda1, hda2]},
+                "f2": {"batch": False, "values": [hda2, hda1]},
+            }
+        )
+        .when.nested(
+            {
+                "f1": {"batch": False, "values": [hda1, hda2]},
+                "f2": {"batch": False, "values": [hda2, hda1]},
+            }
+        )
+        .when.request(
+            {
+                "f1": [hda1, hda2],
+                "f2": [hda2, hda1],
+            }
+        )
     )
     execution = required_tool.execute.with_inputs(inputs)
     execution.assert_has_job(0).with_output("out1").with_contents("1\t2\t3\n4\t5\t6\n")
@@ -139,18 +148,29 @@ def test_map_over_with_output_format_actions(
 ):
     hdca = target_history.with_pair()
     for use_action in ["do", "dont"]:
-        inputs = tool_input_format.when.flat(
-            {
-                "input_cond|dispatch": use_action,
-                "input_cond|input": {"batch": True, "values": [hdca.src_dict]},
-            }
-        ).when.nested(
-            {
-                "input_cond": {
-                    "dispatch": use_action,
-                    "input": {"batch": True, "values": [hdca.src_dict]},
+        inputs = (
+            tool_input_format.when.flat(
+                {
+                    "input_cond|dispatch": use_action,
+                    "input_cond|input": {"batch": True, "values": [hdca.src_dict]},
                 }
-            }
+            )
+            .when.nested(
+                {
+                    "input_cond": {
+                        "dispatch": use_action,
+                        "input": {"batch": True, "values": [hdca.src_dict]},
+                    }
+                }
+            )
+            .when.request(
+                {
+                    "input_cond": {
+                        "dispatch": use_action,
+                        "input": {"__class__": "Batch", "values": [hdca.src_dict]},
+                    }
+                }
+            )
         )
         execute = required_tool.execute.with_inputs(inputs)
         execute.assert_has_n_jobs(2).assert_creates_n_implicit_collections(1)
@@ -256,20 +276,33 @@ def test_multi_run_in_repeat(
     multi_run_in_repeat_datasets: MultiRunInRepeatFixtures,
     tool_input_format: DescribeToolInputs,
 ):
-    inputs = tool_input_format.when.flat(
-        {
-            "input1": {"batch": False, "values": [multi_run_in_repeat_datasets.common_dataset]},
-            "queries_0|input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets},
-        }
-    ).when.nested(
-        {
-            "input1": {"batch": False, "values": [multi_run_in_repeat_datasets.common_dataset]},
-            "queries": [
-                {
-                    "input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets},
-                }
-            ],
-        }
+    inputs = (
+        tool_input_format.when.flat(
+            {
+                "input1": {"batch": False, "values": [multi_run_in_repeat_datasets.common_dataset]},
+                "queries_0|input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets},
+            }
+        )
+        .when.nested(
+            {
+                "input1": {"batch": False, "values": [multi_run_in_repeat_datasets.common_dataset]},
+                "queries": [
+                    {
+                        "input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets},
+                    }
+                ],
+            }
+        )
+        .when.request(
+            {
+                "input1": multi_run_in_repeat_datasets.common_dataset,
+                "queries": [
+                    {
+                        "input2": {"__class__": "Batch", "values": multi_run_in_repeat_datasets.repeat_datasets},
+                    }
+                ],
+            }
+        )
     )
     execute = required_tool.execute.with_inputs(inputs)
     _check_multi_run_in_repeat(execute)
@@ -282,20 +315,33 @@ def test_multi_run_in_repeat_mismatch(
     tool_input_format: DescribeToolInputs,
 ):
     """Same test as above but without the batch wrapper around the common dataset shared between multirun."""
-    inputs = tool_input_format.when.flat(
-        {
-            "input1": multi_run_in_repeat_datasets.common_dataset,
-            "queries_0|input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets},
-        }
-    ).when.nested(
-        {
-            "input1": multi_run_in_repeat_datasets.common_dataset,
-            "queries": [
-                {
-                    "input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets},
-                }
-            ],
-        }
+    inputs = (
+        tool_input_format.when.flat(
+            {
+                "input1": multi_run_in_repeat_datasets.common_dataset,
+                "queries_0|input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets},
+            }
+        )
+        .when.nested(
+            {
+                "input1": multi_run_in_repeat_datasets.common_dataset,
+                "queries": [
+                    {
+                        "input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets},
+                    }
+                ],
+            }
+        )
+        .when.request(
+            {
+                "input1": multi_run_in_repeat_datasets.common_dataset,
+                "queries": [
+                    {
+                        "input2": {"__class__": "Batch", "values": multi_run_in_repeat_datasets.repeat_datasets},
+                    }
+                ],
+            }
+        )
     )
     execute = required_tool.execute.with_inputs(inputs)
     _check_multi_run_in_repeat(execute)
@@ -328,18 +374,31 @@ def test_multirun_on_multiple_inputs(
     two_multi_run_datasets: TwoMultiRunsFixture,
     tool_input_format: DescribeToolInputs,
 ):
-    inputs = tool_input_format.when.flat(
-        {
-            "input1": {"batch": True, "values": two_multi_run_datasets.first_two_datasets},
-            "queries_0|input2": {"batch": True, "values": two_multi_run_datasets.second_two_datasets},
-        }
-    ).when.nested(
-        {
-            "input1": {"batch": True, "values": two_multi_run_datasets.first_two_datasets},
-            "queries": [
-                {"input2": {"batch": True, "values": two_multi_run_datasets.second_two_datasets}},
-            ],
-        }
+    inputs = (
+        tool_input_format.when.flat(
+            {
+                "input1": {"batch": True, "values": two_multi_run_datasets.first_two_datasets},
+                "queries_0|input2": {"batch": True, "values": two_multi_run_datasets.second_two_datasets},
+            }
+        )
+        .when.nested(
+            {
+                "input1": {"batch": True, "values": two_multi_run_datasets.first_two_datasets},
+                "queries": [
+                    {"input2": {"batch": True, "values": two_multi_run_datasets.second_two_datasets}},
+                ],
+            }
+        )
+        .when.request(
+            {
+                "input1": {"__class__": "Batch", "values": two_multi_run_datasets.first_two_datasets},
+                "queries": [
+                    {
+                        "input2": {"__class__": "Batch", "values": two_multi_run_datasets.second_two_datasets},
+                    }
+                ],
+            }
+        )
     )
     execute = required_tool.execute.with_inputs(inputs)
     execute.assert_has_n_jobs(2)
@@ -353,18 +412,39 @@ def test_multirun_on_multiple_inputs_unlinked(
     two_multi_run_datasets: TwoMultiRunsFixture,
     tool_input_format: DescribeToolInputs,
 ):
-    inputs = tool_input_format.when.flat(
-        {
-            "input1": {"batch": True, "linked": False, "values": two_multi_run_datasets.first_two_datasets},
-            "queries_0|input2": {"batch": True, "linked": False, "values": two_multi_run_datasets.second_two_datasets},
-        }
-    ).when.nested(
-        {
-            "input1": {"batch": True, "linked": False, "values": two_multi_run_datasets.first_two_datasets},
-            "queries": [
-                {"input2": {"batch": True, "linked": False, "values": two_multi_run_datasets.second_two_datasets}},
-            ],
-        }
+    inputs = (
+        tool_input_format.when.flat(
+            {
+                "input1": {"batch": True, "linked": False, "values": two_multi_run_datasets.first_two_datasets},
+                "queries_0|input2": {
+                    "batch": True,
+                    "linked": False,
+                    "values": two_multi_run_datasets.second_two_datasets,
+                },
+            }
+        )
+        .when.nested(
+            {
+                "input1": {"batch": True, "linked": False, "values": two_multi_run_datasets.first_two_datasets},
+                "queries": [
+                    {"input2": {"batch": True, "linked": False, "values": two_multi_run_datasets.second_two_datasets}},
+                ],
+            }
+        )
+        .when.request(
+            {
+                "input1": {"__class__": "Batch", "values": two_multi_run_datasets.first_two_datasets, "linked": False},
+                "queries": [
+                    {
+                        "input2": {
+                            "__class__": "Batch",
+                            "values": two_multi_run_datasets.second_two_datasets,
+                            "linked": False,
+                        },
+                    }
+                ],
+            }
+        )
     )
     execute = required_tool.execute.with_inputs(inputs)
     execute.assert_has_n_jobs(4)
@@ -379,7 +459,9 @@ def test_map_over_collection(
     target_history: TargetHistory, required_tool: RequiredTool, tool_input_format: DescribeToolInputs
 ):
     hdca = target_history.with_pair(["123", "456"])
-    inputs = tool_input_format.when.any({"input1": {"batch": True, "values": [hdca.src_dict]}})
+    legacy = {"input1": {"batch": True, "values": [hdca.src_dict]}}
+    request = {"input1": {"__class__": "Batch", "values": [hdca.src_dict]}}
+    inputs = tool_input_format.when.flat(legacy).when.nested(legacy).when.request(request)
     execute = required_tool.execute.with_inputs(inputs)
     execute.assert_has_n_jobs(2).assert_creates_n_implicit_collections(1)
     output_collection = execute.assert_creates_implicit_collection(0)
@@ -544,7 +626,11 @@ def test_select_on_null_errors(required_tools: list[RequiredTool], tool_input_fo
     # is passed, an error (rightfully) occurs. This test verifies that.
     null_parameter = tool_input_format.when.any({"parameter": None})
     for required_tool in required_tools:
-        required_tool.execute.with_inputs(null_parameter).assert_fails.with_error_containing("an invalid option")
+        fails = required_tool.execute.with_inputs(null_parameter).assert_fails
+        if tool_input_format.is_request:
+            fails.with_error_containing("Input should be")
+        else:
+            fails.with_error_containing("an invalid option")
 
 
 @requires_tool_id("gx_select_dynamic_empty")
@@ -555,7 +641,11 @@ def test_select_empty_causes_error_regardless(
     # despite selects otherwise selecting defaults - nothing can be done if the select option list is empty
     empty = tool_input_format.when.any({})
     for required_tool in required_tools:
-        required_tool.execute.with_inputs(empty).assert_fails.with_error_containing("an invalid option")
+        failure = required_tool.execute.with_inputs(empty).assert_fails
+        if tool_input_format.is_request:
+            failure.with_error_containing("validation error")
+        else:
+            failure.with_error_containing("an invalid option")
 
 
 @requires_tool_id("gx_select_optional")

lib/galaxy_test/api/test_tool_execution.py

@@ -0,0 +1,134 @@
+""" """
+
+from typing import (
+    Any,
+    Dict,
+)
+
+import requests
+
+from galaxy_test.base.api_asserts import assert_status_code_is_ok
+from galaxy_test.base.populators import (
+    DatasetPopulator,
+    skip_without_tool,
+)
+from ._framework import ApiTestCase
+
+
+class TestToolExecution(ApiTestCase):
+    dataset_populator: DatasetPopulator
+
+    def setUp(self):
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    @skip_without_tool("gx_int")
+    def test_validation(self):
+        with self.dataset_populator.test_history() as history_id:
+            self._assert_request_validates("gx_int", history_id, {"parameter": 5})
+            self._assert_request_invalid("gx_int", history_id, {"parameter": None})
+            self._assert_request_invalid("gx_int", history_id, {"parameter": "5"})
+
+    @skip_without_tool("gx_int")
+    def test_execution(self):
+        with self.dataset_populator.test_history() as history_id:
+            response = self._run("gx_int", history_id, {"parameter": 5})
+            assert_status_code_is_ok(response)
+            response_json = response.json()
+            tool_request_id = response_json.get("tool_request_id")
+            task_result = response_json["task_result"]
+            history_tool_requests = self.dataset_populator.get_history_tool_requests(history_id)
+            assert tool_request_id in [tr["id"] for tr in history_tool_requests]
+            self.dataset_populator.wait_on_task_object(task_result)
+            state = self.dataset_populator.wait_on_tool_request(tool_request_id)
+            assert state
+            jobs = self.galaxy_interactor.jobs_for_tool_request(tool_request_id)
+            self.dataset_populator.wait_for_jobs(jobs, assert_ok=True)
+
+    @skip_without_tool("gx_data")
+    def test_execution_with_src_urls(self):
+        with self.dataset_populator.test_history() as history_id:
+            response = self._run(
+                "gx_data",
+                history_id,
+                {
+                    "parameter": {
+                        "src": "url",
+                        "url": "https://raw.githubusercontent.com/galaxyproject/planemo/7be1bf5b3971a43eaa73f483125bfb8cabf1c440/tests/data/hello.txt",
+                        "ext": "txt",
+                    }
+                },
+            )
+            assert_status_code_is_ok(response)
+            response_json = response.json()
+            tool_request_id = response_json.get("tool_request_id")
+            task_result = response_json["task_result"]
+            self.dataset_populator.wait_on_task_object(task_result)
+            state = self.dataset_populator.wait_on_tool_request(tool_request_id)
+            assert state, str(self.dataset_populator.get_tool_request(tool_request_id))
+            jobs = self.galaxy_interactor.jobs_for_tool_request(tool_request_id)
+            self.dataset_populator.wait_for_jobs(jobs, assert_ok=True)
+            if len(jobs) != 1:
+                raise Exception(f"Found incorrect number of jobs for tool request - was expecting a single job {jobs}")
+            assert len(jobs) == 1, jobs
+            job_id = jobs[0]["id"]
+            job_outputs = self.galaxy_interactor.job_outputs(job_id)
+            assert len(job_outputs) == 1
+            job_output = job_outputs[0]
+            assert job_output["name"] == "output"
+            content = self.dataset_populator.get_history_dataset_content(history_id, dataset=job_output["dataset"])
+            assert content == "Hello World!"
+
+            # verify input was not left deferred and materialized before the job started
+            input_dataset_details = self.dataset_populator.get_history_dataset_details(history_id, hid=1)
+            assert input_dataset_details["state"] == "ok", input_dataset_details
+
+    @skip_without_tool("gx_data")
+    def test_execution_with_deferred_src_urls(self):
+        with self.dataset_populator.test_history() as history_id:
+            response = self._run(
+                "gx_data",
+                history_id,
+                {
+                    "parameter": {
+                        "src": "url",
+                        "url": "https://raw.githubusercontent.com/galaxyproject/planemo/7be1bf5b3971a43eaa73f483125bfb8cabf1c440/tests/data/hello.txt",
+                        "ext": "txt",
+                        "deferred": True,
+                    }
+                },
+            )
+            assert_status_code_is_ok(response)
+            response_json = response.json()
+            tool_request_id = response_json.get("tool_request_id")
+            task_result = response_json["task_result"]
+            self.dataset_populator.wait_on_task_object(task_result)
+            state = self.dataset_populator.wait_on_tool_request(tool_request_id)
+            assert state, str(self.dataset_populator.get_tool_request(tool_request_id))
+            jobs = self.galaxy_interactor.jobs_for_tool_request(tool_request_id)
+            self.dataset_populator.wait_for_jobs(jobs, assert_ok=True)
+            if len(jobs) != 1:
+                raise Exception(f"Found incorrect number of jobs for tool request - was expecting a single job {jobs}")
+            assert len(jobs) == 1, jobs
+            job_id = jobs[0]["id"]
+            job_outputs = self.galaxy_interactor.job_outputs(job_id)
+            assert len(job_outputs) == 1
+            job_output = job_outputs[0]
+            assert job_output["name"] == "output"
+            content = self.dataset_populator.get_history_dataset_content(history_id, dataset=job_output["dataset"])
+            assert content == "Hello World!"
+
+            # verify input was left deferred and infer must have been materialized just for the job
+            input_dataset_details = self.dataset_populator.get_history_dataset_details(history_id, hid=1)
+            assert input_dataset_details["state"] == "deferred", input_dataset_details
+
+    def _assert_request_validates(self, tool_id: str, history_id: str, inputs: Dict[str, Any]):
+        response = self._run(tool_id, history_id, inputs)
+        assert response.status_code == 200
+
+    def _assert_request_invalid(self, tool_id: str, history_id: str, inputs: Dict[str, Any]):
+        response = self._run(tool_id, history_id, inputs)
+        assert response.status_code == 400
+
+    def _run(self, tool_id: str, history_id: str, inputs: Dict[str, Any]) -> requests.Response:
+        return self.dataset_populator.tool_request_raw(tool_id, inputs, history_id)

lib/galaxy_test/base/populators.py

@@ -174,6 +174,7 @@ TOOL_WITH_SHELL_COMMAND = {
 DEFAULT_TIMEOUT = 60  # Secs to wait for state to turn ok
 
 SKIP_FLAKEY_TESTS_ON_ERROR = os.environ.get("GALAXY_TEST_SKIP_FLAKEY_TESTS_ON_ERROR", None)
+INPUT_FORMAT_T = Literal["legacy", "21.01", "request"]
 
 PRIVATE_ROLE_TYPE = "private"
 
@@ -1210,6 +1211,15 @@ class BaseDatasetPopulator(BasePopulator):
         payload = self.run_tool_payload(tool_id, inputs, history_id, **kwds)
         return self.tools_post(payload)
 
+    def tool_request_raw(self, tool_id: str, inputs: dict[str, Any], history_id: str) -> Response:
+        payload = {
+            "tool_id": tool_id,
+            "history_id": history_id,
+            "inputs": inputs,
+        }
+        response = self._post("jobs", data=payload, json=True)
+        return response
+
     def run_tool(self, tool_id: str, inputs: dict, history_id: str, **kwds):
         tool_response = self.run_tool_raw(tool_id, inputs, history_id, **kwds)
         api_asserts.assert_status_code_is(tool_response, 200)
@@ -1752,8 +1762,38 @@ class BaseDatasetPopulator(BasePopulator):
         wait_on(is_ready, "waiting for download to become ready")
         assert is_ready()
 
+    def wait_on_tool_request(self, tool_request_id: str):
+        # should this to defer to interactor's copy of this method?
+
+        def state():
+            state_response = self._get(f"tool_requests/{tool_request_id}/state")
+            state_response.raise_for_status()
+            return state_response.json()
+
+        def is_ready():
+            is_complete = state() in ["submitted", "failed"]
+            return True if is_complete else None
+
+        wait_on(is_ready, "waiting for tool request to submit")
+        return state() == "submitted"
+
+    def get_tool_request(self, tool_request_id: str) -> dict[str, Any]:
+        response = self._get(f"tool_requests/{tool_request_id}")
+        api_asserts.assert_status_code_is_ok(response)
+        return response.json()
+
+    def get_history_tool_requests(self, history_id: str) -> list[dict[str, Any]]:
+        response = self._get(f"histories/{history_id}/tool_requests")
+        api_asserts.assert_status_code_is_ok(response)
+        return response.json()
+
     def wait_on_task(self, async_task_response: Response):
-        task_id = async_task_response.json()["id"]
+        response_json = async_task_response.json()
+        self.wait_on_task_object(response_json)
+
+    def wait_on_task_object(self, async_task_json: dict[str, Any]):
+        assert "id" in async_task_json, f"Task response {async_task_json} does not contain expected 'id' field."
+        task_id = async_task_json["id"]
         return self.wait_on_task_id(task_id)
 
     def wait_on_task_id(self, task_id: str):
@@ -3969,8 +4009,9 @@ class DescribeJob:
 
 
 class DescribeFailure:
-    def __init__(self, response: Response):
+    def __init__(self, response: Response, tool_request: Optional[dict[str, Any]] = None):
         self._response = response
+        self._tool_request = tool_request
 
     def __call__(self) -> Self:
         return self
@@ -3980,7 +4021,14 @@ class DescribeFailure:
         return self
 
     def with_error_containing(self, message: str) -> Self:
-        assert message in self._response.text
+        actual_text = self._response.text
+        if message not in actual_text:
+            if self._tool_request:
+                state_message = self._tool_request["state_message"]
+                if message not in state_message:
+                    raise AssertionError(f"'{message}' not found in '{state_message}'")
+            else:
+                raise AssertionError(f"'{message}' not found in '{actual_text}'")
         return self
 
 
@@ -4000,10 +4048,10 @@ class RequiredTool:
 
 
 class DescribeToolInputs:
-    _input_format: str = "legacy"
+    _input_format: INPUT_FORMAT_T = "legacy"
     _inputs: Optional[dict[str, Any]]
 
-    def __init__(self, input_format: str):
+    def __init__(self, input_format: INPUT_FORMAT_T):
         self._input_format = input_format
         self._inputs = None
 
@@ -4017,10 +4065,19 @@ class DescribeToolInputs:
         return self
 
     def nested(self, inputs: dict[str, Any]) -> Self:
-        if self._input_format == "21.01":
+        if self._input_format in ["21.01", "request"]:
             self._inputs = inputs
         return self
 
+    def request(self, inputs: dict[str, Any]) -> Self:
+        if self._input_format in ["request"]:
+            self._inputs = inputs
+        return self
+
+    @property
+    def is_request(self) -> bool:
+        return self._input_format == "request"
+
     # aliases for self to create silly little English sentense... inputs.when.flat().when.legacy()
     @property
     def when(self) -> Self:
@@ -4030,8 +4087,9 @@ class DescribeToolInputs:
 class DescribeToolExecution:
     _history_id: Optional[str] = None
     _execute_response: Optional[Response] = None
-    _input_format: Optional[str] = None
+    _input_format: Optional[INPUT_FORMAT_T] = None
     _inputs: dict[str, Any]
+    _tool_request_id: Optional[str] = None  # if input_format == "request" request ID
 
     def __init__(self, dataset_populator: BaseDatasetPopulator, tool_id: str):
         self._dataset_populator = dataset_populator
@@ -4059,14 +4117,27 @@ class DescribeToolExecution:
         self._input_format = "21.01"
         return self
 
+    def with_request(self, inputs: dict[str, Any]) -> Self:
+        self._inputs = inputs
+        self._input_format = "request"
+        return self
+
     def _execute(self):
         kwds = {}
         if self._input_format is not None:
             kwds["input_format"] = self._input_format
         history_id = self._ensure_history_id
-        self._execute_response = self._dataset_populator.run_tool_raw(
-            self._tool_id, self._inputs, history_id, assert_ok=False, **kwds
-        )
+        if self._input_format == "request":
+            execute_response = self._dataset_populator.tool_request_raw(self._tool_id, self._inputs, history_id)
+            if execute_response.status_code == 200:
+                response_json = execute_response.json()
+                tool_request_id = response_json.get("tool_request_id")
+                self._dataset_populator.wait_on_tool_request(tool_request_id)
+            self._execute_response = execute_response
+        else:
+            self._execute_response = self._dataset_populator.run_tool_raw(
+                self._tool_id, self._inputs, history_id, assert_ok=False, **kwds
+            )
 
     @property
     def _ensure_history_id(self) -> str:
@@ -4084,25 +4155,58 @@ class DescribeToolExecution:
         execute_response = self._execute_response
         assert execute_response is not None
         api_asserts.assert_status_code_is_ok(execute_response)
+        if self._input_format == "request":
+            response_json = execute_response.json()
+            tool_request_id = response_json.get("tool_request_id")
+            task_result = response_json["task_result"]
+            self._dataset_populator.wait_on_task_object(task_result)
+            self._tool_request_id = tool_request_id
+
         return execute_response.json()
 
+    @property
+    def _jobs(self) -> list[dict[str, Any]]:
+        if self._input_format == "request":
+            tool_request_id = self._tool_request_id
+            assert tool_request_id, "request not exected"
+            jobs = self._dataset_populator.galaxy_interactor.jobs_for_tool_request(tool_request_id)
+        else:
+            response = self._assert_executed_ok()
+            jobs = response["jobs"]
+        return jobs
+
     def assert_has_n_jobs(self, n: int) -> Self:
-        response = self._assert_executed_ok()
-        jobs = response["jobs"]
-        if len(jobs) != n:
-            raise AssertionError(f"Expected tool execution to produce {n} jobs but it produced {len(jobs)}")
+        self._assert_executed_ok()
+        jobs = self._jobs
+        num_jobs = len(jobs)
+        if num_jobs != n:
+            raise AssertionError(f"Expected tool execution to produce {n} jobs but it produced {num_jobs}")
         return self
 
-    def assert_creates_n_implicit_collections(self, n: int) -> Self:
+    @property
+    def _tool_request(self):
+        tool_request_id = self._tool_request_id
+        assert tool_request_id, "No tool request set - was tool executed as a request"
+        return self._dataset_populator.get_tool_request(tool_request_id)
+
+    @property
+    def _implicit_collections(self) -> list:
         response = self._assert_executed_ok()
-        collections = response["implicit_collections"]
+        if self._input_format == "request":
+            tool_request = self._tool_request
+            collections = tool_request["implicit_collections"]
+        else:
+            collections = response["implicit_collections"]
+        return collections
+
+    def assert_creates_n_implicit_collections(self, n: int) -> Self:
+        collections = self._implicit_collections
         if len(collections) != n:
             raise AssertionError(f"Expected tool execution to produce {n} implicit but it produced {len(collections)}")
         return self
 
     def assert_creates_implicit_collection(self, index: Union[str, int]) -> "DescribeToolExecutionOutputCollection":
-        response = self._assert_executed_ok()
-        collections = response["implicit_collections"]
+        collections = self._implicit_collections
         assert isinstance(index, int)  # TODO: implement and then prefer str.
         history_id = self._ensure_history_id
         return DescribeToolExecutionOutputCollection(self._dataset_populator, history_id, collections[index]["id"])
@@ -4112,8 +4216,8 @@ class DescribeToolExecution:
         return self.assert_has_n_jobs(1).assert_has_job(0)
 
     def assert_has_job(self, job_index: int = 0) -> DescribeJob:
-        response = self._assert_executed_ok()
-        job = response["jobs"][job_index]
+        self._assert_executed_ok()
+        job = self._jobs[job_index]
         history_id = self._ensure_history_id
         return DescribeJob(self._dataset_populator, history_id, job["id"])
 
@@ -4122,15 +4226,18 @@ class DescribeToolExecution:
         self._ensure_executed()
         execute_response = self._execute_response
         assert execute_response is not None
+        tool_request = None
         if execute_response.status_code != 200:
             return DescribeFailure(execute_response)
         else:
-            response = self._assert_executed_ok()
-            jobs = response["jobs"]
+            self._assert_executed_ok()
+            jobs = self._jobs
             for job in jobs:
                 final_state = self._dataset_populator.wait_for_job(job["id"])
                 assert final_state == "error"
-            return DescribeFailure(execute_response)
+            if self._tool_request_id:
+                tool_request = self._tool_request
+            return DescribeFailure(execute_response, tool_request)
 
     # alternative assert_ syntax for cases where it reads better.
     @property

test/functional/test_toolbox_pytest.py

@@ -1,10 +1,15 @@
 import os
 from typing import (
+    cast,
     NamedTuple,
 )
 
 import pytest
 
+from galaxy.tool_util.verify.interactor import (
+    DEFAULT_USE_LEGACY_API,
+    UseLegacyApiT,
+)
 from galaxy_test.api._framework import ApiTestCase
 from galaxy_test.driver.driver_util import GalaxyTestDriver
 
@@ -60,4 +65,7 @@ class TestFrameworkTools(ApiTestCase):
 
     @pytest.mark.parametrize("testcase", cases(), ids=idfn)
     def test_tool(self, testcase: ToolTest):
-        self._test_driver.run_tool_test(testcase.tool_id, testcase.test_index, tool_version=testcase.tool_version)
+        use_legacy_api = cast(UseLegacyApiT, os.environ.get("GALAXY_TEST_USE_LEGACY_TOOL_API", DEFAULT_USE_LEGACY_API))
+        self._test_driver.run_tool_test(
+            testcase.tool_id, testcase.test_index, tool_version=testcase.tool_version, use_legacy_api=use_legacy_api
+        )

test/functional/tools/parameters/gx_conditional_select_dynamic.xml

@@ -0,0 +1,52 @@
+<tool id="gx_conditional_select_dynamic" name="gx_conditional_select_dynamic" version="1.0.0">
+    <command><![CDATA[
+echo '$conditional_parameter.parameter' >> '$output'
+    ]]></command>
+    <code file="gx_select_dynamic_options.py"/>
+    <inputs>
+        <conditional name="conditional_parameter">
+            <param name="case_parameter" type="select">
+                <option value="use_select">Use Select</option>
+                <option value="no_select">No Select</option>
+            </param>
+            <when value="use_select">
+                <param name="ref_parameter" type="data" ext="txt" >
+                </param>
+                <param name="parameter" type="select" dynamic_options="every_other_word(ref_parameter.dataset.get_file_name())">
+                </param>
+            </when>
+            <when value="no_select">
+            </when>
+        </conditional>
+    </inputs>
+    <outputs>
+        <data name="output" format="txt" />
+    </outputs>
+    <tests>
+        <test>
+            <conditional name="conditional_parameter">
+                <param name="case_parameter" value="use_select" />
+                <param name="ref_parameter" value="simple_line.txt" />
+                <param name="parameter" value="of" />
+            </conditional>
+            <output name="output">
+                <assert_contents>
+                    <has_line line="of" />
+                </assert_contents>
+            </output>
+        </test>
+        <test>
+            <!-- Selects implicitly default to the first option as the default option.
+            -->
+            <conditional name="conditional_parameter">
+                <param name="case_parameter" value="use_select" />
+                <param name="ref_parameter" value="simple_line.txt" />
+            </conditional>
+            <output name="output">
+                <assert_contents>
+                    <has_line line="This" />
+                </assert_contents>
+            </output>
+        </test>
+    </tests>
+</tool>

test/functional/tools/parameters/gx_repeat_select_dynamic.xml

@@ -0,0 +1,42 @@
+<tool id="gx_repeat_select_dynamic" name="gx_repeat_select_dynamic" version="1.0.0">
+    <command><![CDATA[
+echo '$repeat_parameter[0].parameter' >> '$output'
+    ]]></command>
+    <code file="gx_select_dynamic_options.py"/>
+    <inputs>
+        <repeat name="repeat_parameter">
+            <param name="ref_parameter" type="data" ext="txt" >
+            </param>
+            <param name="parameter" type="select" dynamic_options="every_other_word(ref_parameter.dataset.get_file_name())">
+            </param>
+        </repeat>
+    </inputs>
+    <outputs>
+        <data name="output" format="txt" />
+    </outputs>
+    <tests>
+        <test>
+            <repeat name="repeat_parameter">
+                <param name="ref_parameter" value="simple_line.txt" />
+                <param name="parameter" value="of" />
+            </repeat>
+            <output name="output">
+                <assert_contents>
+                    <has_line line="of" />
+                </assert_contents>
+            </output>
+        </test>
+        <test>
+            <!-- Selects implicitly default to the first option as the default option.
+            -->
+            <repeat name="repeat_parameter">
+                <param name="ref_parameter" value="simple_line.txt" />
+            </repeat>
+            <output name="output">
+                <assert_contents>
+                    <has_line line="This" />
+                </assert_contents>
+            </output>
+        </test>
+    </tests>
+</tool>

test/functional/tools/parameters/gx_section_select_dynamic.xml

@@ -0,0 +1,42 @@
+<tool id="gx_section_select_dynamic" name="gx_section_select_dynamic" version="1.0.0">
+    <command><![CDATA[
+echo '$section_parameter.parameter' >> '$output'
+    ]]></command>
+    <code file="gx_select_dynamic_options.py"/>
+    <inputs>
+        <section name="section_parameter">
+            <param name="ref_parameter" type="data" ext="txt" >
+            </param>
+            <param name="parameter" type="select" dynamic_options="every_other_word(ref_parameter.dataset.get_file_name())">
+            </param>
+        </section>
+    </inputs>
+    <outputs>
+        <data name="output" format="txt" />
+    </outputs>
+    <tests>
+        <test>
+            <section name="section_parameter">
+                <param name="ref_parameter" value="simple_line.txt" />
+                <param name="parameter" value="of" />
+            </section>
+            <output name="output">
+                <assert_contents>
+                    <has_line line="of" />
+                </assert_contents>
+            </output>
+        </test>
+        <test>
+            <!-- Selects implicitly default to the first option as the default option.
+            -->
+            <section name="section_parameter">
+                <param name="ref_parameter" value="simple_line.txt" />
+            </section>
+            <output name="output">
+                <assert_contents>
+                    <has_line line="This" />
+                </assert_contents>
+            </output>
+        </test>
+    </tests>
+</tool>

test/unit/tool_util/parameter_specification.yml

@@ -263,8 +263,9 @@ gx_text:
     - parameter: moocow
     - parameter: 'some spaces'
     - parameter: ''
+    # The legacy API behavior is to make these non-null request parameters into implicit
+    # "" values for text parameters.
     - {}
-    # need to explicitly mark these as non-optional
     - parameter: null
   request_invalid: &gx_text_request_invalid
     - parameter: 5
@@ -627,10 +628,16 @@ gx_genomebuild_optional:
 gx_genomebuild_multiple:
   request_valid:
   - parameter: ["hg18", hg19"]
+  # I don't love this but this is how the legacy API works and the tool parameters
+  # wrappers work this way also. Not sure how I would change this.
+  - parameter: null
   request_invalid:
   - parameter: 6
-  # DIVERGES_FROM_CURRENT_API
+  job_internal_valid:
+  - parameter: ["hg18", hg19"]
   - parameter: null
+  job_internal_invalid:
+  - {}
 
 gx_directory_uri:
   request_valid:

test/unit/tool_util/test_parameter_convert.py

@@ -43,7 +43,19 @@ def test_decode_data():
     assert decoded_state.input_state["parameter"]["id"] == EXAMPLE_ID_1
 
 
-def test_encode_collection():
+def test_decode_data_batch():
+    tool_source = tool_source_for("parameters/gx_data")
+    bundle = input_models_for_tool_source(tool_source)
+    request_state = RequestToolState(
+        {"parameter": {"__class__": "Batch", "values": [{"src": "hda", "id": EXAMPLE_ID_1_ENCODED}]}}
+    )
+    request_state.validate(bundle)
+    decoded_state = decode(request_state, bundle, _fake_decode)
+    assert decoded_state.input_state["parameter"]["values"][0]["src"] == "hda"
+    assert decoded_state.input_state["parameter"]["values"][0]["id"] == EXAMPLE_ID_1
+
+
+def test_decode_collection():
     tool_source = tool_source_for("parameters/gx_data_collection")
     bundle = input_models_for_tool_source(tool_source)
     request_state = RequestToolState({"parameter": {"src": "hdca", "id": EXAMPLE_ID_1_ENCODED}})
@@ -119,6 +131,22 @@ def test_landing_encode_data():
     assert encoded_state.input_state["parameter"]["id"] == EXAMPLE_ID_1_ENCODED
 
 
+def test_landing_encode_data_batch():
+    tool_source = tool_source_for("parameters/gx_data")
+    bundle = input_models_for_tool_source(tool_source)
+    request_state = LandingRequestToolState(
+        {"parameter": {"__class__": "Batch", "values": [{"src": "hda", "id": EXAMPLE_ID_1_ENCODED}]}}
+    )
+    request_state.validate(bundle)
+    decoded_state = landing_decode(request_state, bundle, _fake_decode)
+    assert decoded_state.input_state["parameter"]["values"][0]["src"] == "hda"
+    assert decoded_state.input_state["parameter"]["values"][0]["id"] == EXAMPLE_ID_1
+
+    encoded_state = landing_encode(decoded_state, bundle, _fake_encode)
+    assert encoded_state.input_state["parameter"]["values"][0]["src"] == "hda"
+    assert encoded_state.input_state["parameter"]["values"][0]["id"] == EXAMPLE_ID_1_ENCODED
+
+
 def test_dereference():
     tool_source = tool_source_for("parameters/gx_data")
     bundle = input_models_for_tool_source(tool_source)

test/unit/tool_util/test_parameter_test_cases.py

@@ -288,13 +288,15 @@ def test_convert_to_requests():
 
 def _validate_path(tool_path: str):
     tool_source = get_tool_source(tool_path)
+    tool_id = tool_source.parse_id()
+    model_name = f"{tool_id} (test case model)"
     parsed_tool = parse_tool(tool_source)
     profile = tool_source.parse_profile()
     test_cases: List[ToolSourceTest] = tool_source.parse_tests_to_dict()["tests"]
     for test_case in test_cases:
         if test_case.get("expect_failure"):
             continue
-        test_case_state_and_warnings = case_state(test_case, parsed_tool.inputs, profile)
+        test_case_state_and_warnings = case_state(test_case, parsed_tool.inputs, profile, name=model_name)
         tool_state = test_case_state_and_warnings.tool_state
         assert tool_state.state_representation == "test_case_xml"