Break dereference_input into HDA and HDCA variants

Split the dereference_input function into two specialized methods:
- dereference_input_to_hda: Returns HistoryDatasetAssociation
- dereference_input_to_hdca: Returns HistoryDatasetCollectionAssociation

This improves type safety by having each function return a specific type
rather than a union, eliminating the need for runtime isinstance checks
and assertions at call sites.

Fixes #20160

https://claude.ai/code/session_01QWGeMjFzqh1JECMNdUo6mQ

Author: mvdbeek

lib/galaxy/managers/hdas.py

@@ -355,23 +355,30 @@ def _set_permissions(self, trans, hda, role_ids_dict):
                 raise exceptions.RequestParameterInvalidException(error)
 
 
-def dereference_input(
+def dereference_input_to_hda(
     trans: ProvidesHistoryContext,
-    data_request: Union[DataRequestUri, FileRequestUri, DataRequestCollectionUri],
+    data_request: Union[DataRequestUri, FileRequestUri],
     history: model.History,
-) -> Union[HistoryDatasetAssociation, HistoryDatasetCollectionAssociation]:
+) -> HistoryDatasetAssociation:
     permissions = trans.app.security_agent.history_get_default_permissions(history)
-    if isinstance(data_request, DataRequestCollectionUri):
-        hdca = derefence_collection_to_model(trans.sa_session, trans.user, history, data_request)
-        for hda in hdca.dataset_instances:
-            trans.app.security_agent.set_all_dataset_permissions(hda.dataset, permissions, new=True, flush=False)
-        return hdca
     hda = dereference_to_model(trans.sa_session, trans.user, history, data_request)
     trans.app.security_agent.set_all_dataset_permissions(hda.dataset, permissions, new=True, flush=False)
     trans.sa_session.commit()
     return hda
 
 
+def dereference_input_to_hdca(
+    trans: ProvidesHistoryContext,
+    data_request: DataRequestCollectionUri,
+    history: model.History,
+) -> HistoryDatasetCollectionAssociation:
+    permissions = trans.app.security_agent.history_get_default_permissions(history)
+    hdca = derefence_collection_to_model(trans.sa_session, trans.user, history, data_request)
+    for hda in hdca.dataset_instances:
+        trans.app.security_agent.set_all_dataset_permissions(hda.dataset, permissions, new=True, flush=False)
+    return hdca
+
+
 class HDAStorageCleanerManager(base.StorageCleanerManager):
     def __init__(self, hda_manager: HDAManager, dataset_manager: datasets.DatasetManager):
         self.hda_manager = hda_manager

lib/galaxy/managers/jobs.py

@@ -51,7 +51,8 @@
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.managers.datasets import DatasetManager
 from galaxy.managers.hdas import (
-    dereference_input,
+    dereference_input_to_hda,
+    dereference_input_to_hdca,
     HDAManager,
 )
 from galaxy.managers.histories import HistoryManager
@@ -2103,12 +2104,11 @@ def dereference_collection_callback(data_request: DataRequestCollectionUri) -> D
             if not history:
                 raise InconsistentDatabase("Tool request has no history associated")
 
-            hdca = dereference_input(trans, data_request, history)
-            assert isinstance(hdca, model.HistoryDatasetCollectionAssociation)
+            hdca = dereference_input_to_hdca(trans, data_request, history)
 
             # we need the HDCA to have an ID - so we force a commit here - for
-            # consistency it would be great if this happened in the dereference_input
-            # since the HDA is committed in the other branch.
+            # consistency it would be great if this happened in dereference_input_to_hdca
+            # since the HDA is committed in dereference_input_to_hda.
             history.add_pending_items()
             trans.sa_session.commit()
 
@@ -2132,8 +2132,7 @@ def dereference_callback(data_request: DataRequestUri) -> DataRequestInternalHda
             if not history:
                 raise InconsistentDatabase("Tool request has no history associated")
 
-            hda = dereference_input(trans, data_request, history)
-            assert isinstance(hda, model.HistoryDatasetAssociation)
+            hda = dereference_input_to_hda(trans, data_request, history)
             new_hdas.append(DereferencedDatasetPair(hda, data_request))
             return DataRequestInternalHda(id=hda.id, src="hda")
 

lib/galaxy/workflow/run_request.py

@@ -12,7 +12,10 @@
 
 from galaxy import exceptions
 from galaxy.exceptions.utils import validation_error_to_message_exception
-from galaxy.managers.hdas import dereference_input
+from galaxy.managers.hdas import (
+    dereference_input_to_hda,
+    dereference_input_to_hdca,
+)
 from galaxy.model import (
     EffectiveOutput,
     History,
@@ -419,8 +422,17 @@ def build_workflow_run_configs(
                     content = app.dataset_collection_manager.get_dataset_collection_instance(
                         trans, "history", data_request.id
                     )
-                elif data_request.src == "url" or data_request.class_ == "File" or data_request.class_ == "Collection":
-                    request_input = dereference_input(trans, data_request, history)
+                elif data_request.class_ == "Collection":
+                    request_input = dereference_input_to_hdca(trans, data_request, history)
+                    added_to_history = True
+                    content = InputWithRequest(
+                        input=request_input,
+                        request=data_request.model_dump(mode="json"),
+                    )
+                    if not data_request.deferred:
+                        requires_materialization = True
+                elif data_request.src == "url" or data_request.class_ == "File":
+                    request_input = dereference_input_to_hda(trans, data_request, history)
                     added_to_history = True
                     content = InputWithRequest(
                         input=request_input,