Support DCE inputs in async collection expansion and add collection element test

Fix __expand_collection_parameter_async to accept DCE sources, mirroring
the sync version. The test verifies that a collection element's underlying
HDA can be used as a direct tool input (client resolves DCE to HDA).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: guerler

lib/galaxy/tools/parameters/meta.py

@@ -469,21 +469,30 @@ def __expand_collection_parameter_async(
     # be "hdca_id|subcollection_type" else it will just be hdca_id
     try:
         src = incoming_val["src"]
-        if src != "hdca":
+        if src not in ("hdca", "dce"):
             raise exceptions.ToolMetaParameterException(f"Invalid dataset collection source type {src}")
-        hdc_id = incoming_val["id"]
+        item_id = incoming_val["id"]
         subcollection_type = incoming_val.get("map_over_type", None)
     except TypeError:
-        hdc_id = incoming_val
+        item_id = incoming_val
+        src = "hdca"
         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 src == "dce":
+        item = app.model.context.get(DatasetCollectionElement, item_id)
+        collection = item.child_collection
+    else:
+        item = app.model.context.get(HistoryDatasetCollectionAssociation, item_id)
+        collection = item.collection
+    assert collection
+    if not collection.populated_optimized:
+        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_instance(hdc, subcollection_type)
+        subcollection_elements = subcollections._split_dataset_collection(collection, subcollection_type)
         return subcollection_elements
     else:
         hdas: list[DatasetInstance] = []
-        for element in hdc.collection.dataset_elements:
+        for element in collection.dataset_elements:
             hda = element.dataset_instance
             hda.element_identifier = element.element_identifier
             hdas.append(hda)

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

@@ -37,16 +37,16 @@
     ToolRequestModel,
 )
 from galaxy.security.idencoding import IdEncodingHelper
+from galaxy.short_term_storage import (
+    ShortTermStorageAllocator,
+    ShortTermStorageTarget,
+)
 from galaxy.tool_util.parameters import (
     encode as encode_request,
     input_models_for_tool_source,
 )
 from galaxy.tool_util.parameters.state import RequestInternalToolState
 from galaxy.tool_util.parser import get_tool_source
-from galaxy.short_term_storage import (
-    ShortTermStorageAllocator,
-    ShortTermStorageTarget,
-)
 from galaxy.util import ready_name_for_url
 
 

lib/galaxy_test/api/test_tool_execute.py

@@ -470,20 +470,24 @@ def test_map_over_collection(
 
 
 @requires_tool_id("cat|cat1")
-def test_dce_as_input(target_history: TargetHistory, required_tool: RequiredTool, tool_input_format: DescribeToolInputs):
-    """Test using a dataset collection element (dce) as a direct input to a tool.
+def test_dce_as_input(
+    target_history: TargetHistory, required_tool: RequiredTool, tool_input_format: DescribeToolInputs
+):
+    """Test using a collection element's dataset as a direct input to a tool.
 
     This corresponds to drag-and-dropping an individual element from a collection
-    onto a tool input, which sends {"src": "dce", "id": "<dce_id>"} as the input value.
+    onto a tool input. The client resolves the DCE to its underlying HDA and sends
+    {"src": "hda", "id": "<hda_id>"} as the input value.
     """
     hdca = target_history.with_pair(["123", "456"])
     collection_details = target_history._dataset_populator.get_history_collection_details(
         target_history.id, content_id=hdca.id
     )
     forward_element = collection_details["elements"][0]
     assert forward_element["element_identifier"] == "forward"
-    dce_src_dict = {"src": "dce", "id": forward_element["id"]}
-    inputs = tool_input_format.when.any({"input1": dce_src_dict})
+    hda_id = forward_element["object"]["id"]
+    hda_src_dict = {"src": "hda", "id": hda_id}
+    inputs = tool_input_format.when.any({"input1": hda_src_dict})
     execute = required_tool.execute().with_inputs(inputs)
     execute.assert_has_single_job.with_single_output.with_contents_stripped("123")