Seed <repeat default=N> only when a request omits the repeat

lib/galaxy/tool_util/parameters/case.py

@@ -1,6 +1,7 @@
 import os
 from dataclasses import (
     dataclass,
+    field,
     replace,
 )
 from re import compile
@@ -291,6 +292,10 @@ class MergeContext:
     profile: str
     state_representation: Literal["test_case_xml", "test_case_json"]
     warnings: list[str]
+    # Names of raw test inputs already claimed by an exact structural match. Shared across the
+    # whole walk (like ``warnings``); consulted only by the repeat bare-name synthesis so a
+    # top-level input a fixed param already consumed can't also seed a nested repeat instance.
+    consumed_input_names: set[str] = field(default_factory=set)
 
     @property
     def inputs(self) -> ToolSourceTestInputs:
@@ -431,7 +436,9 @@ def _merge_into_state(
         if input_name not in state_at_level:
             state_at_level[input_name] = repeat_state_array
 
-        repeat_instance_inputs = _repeat_inputs_to_array(state_path, tool_input.parameters, context.inputs)
+        repeat_instance_inputs = _repeat_inputs_to_array(
+            state_path, tool_input.parameters, context.inputs, context.consumed_input_names
+        )
         if tool_input.min is not None:
             while len(repeat_instance_inputs) < tool_input.min:
                 repeat_instance_inputs.append([])
@@ -457,6 +464,10 @@ def _merge_into_state(
     else:
         test_input = context.input_for(state_path)
         if test_input is not None:
+            # An exact-path match means a fixed structural param owns this raw input; record it
+            # so the repeat bare-name synthesis does not re-consume it into a nested instance.
+            if test_input["name"] == state_path:
+                context.consumed_input_names.add(test_input["name"])
             input_value: Any
             if isinstance(tool_input, (DataCollectionParameterModel,)):
                 input_value = TestCollectionDef.from_dict(
@@ -505,7 +516,10 @@ def _merge_into_state(
 
 
 def _repeat_inputs_to_array(
-    state_path: str, parameters: list[ToolParameterT], inputs: ToolSourceTestInputs
+    state_path: str,
+    parameters: list[ToolParameterT],
+    inputs: ToolSourceTestInputs,
+    consumed_input_names: set[str] | None = None,
 ) -> list[ToolSourceTestInputs]:
     inputs_as_dict = _inputs_as_dict(inputs)
     repeat_instance_input_dicts = repeat_inputs_to_array(state_path, inputs_as_dict)
@@ -524,17 +538,50 @@ def _repeat_inputs_to_array(
     if repeat_instance_inputs:
         return repeat_instance_inputs
 
+    consumed = consumed_input_names or set()
     legacy_repeat_inputs: list[ToolSourceTestInputs] = []
     for parameter in parameters:
         parameter_name = parameter.name
-        matching_inputs = [input for input in inputs if input["name"] == parameter_name]
+        matching_inputs = [
+            input for input in inputs if input["name"] == parameter_name and input["name"] not in consumed
+        ]
         for i, input in enumerate(matching_inputs):
             while len(legacy_repeat_inputs) <= i:
                 legacy_repeat_inputs.append([])
             synthetic_input = cast(ToolSourceTestInput, dict(input))
             synthetic_input["name"] = f"{state_path}_{i}|{parameter_name}"
             legacy_repeat_inputs[i].append(synthetic_input)
-    return legacy_repeat_inputs
+    if legacy_repeat_inputs:
+        return legacy_repeat_inputs
+
+    # The repeat's instance params live inside a conditional/section and the legacy test names
+    # them by bare short name (e.g. deepmicro's parameter_set repeat wraps an rl_type conditional
+    # whose test supplies rl_type_choice/dm unqualified). Group those unconsumed matches into
+    # instances - one per occurrence count - so instance-scoped resolution re-attaches them.
+    return _nested_repeat_inputs_to_array(parameters, inputs, consumed)
+
+
+def _nested_repeat_inputs_to_array(
+    parameters: list[ToolParameterT], inputs: ToolSourceTestInputs, consumed: set[str]
+) -> list[ToolSourceTestInputs]:
+    leaf_names = _leaf_param_short_names(parameters)
+    matches_by_leaf: dict[str, list[ToolSourceTestInput]] = {}
+    for leaf_name in leaf_names:
+        matches = [input for input in inputs if input["name"] == leaf_name and input["name"] not in consumed]
+        if matches:
+            matches_by_leaf[leaf_name] = matches
+    if not matches_by_leaf:
+        return []
+    instance_count = max(len(matches) for matches in matches_by_leaf.values())
+    nested_repeat_inputs: list[ToolSourceTestInputs] = [[] for _ in range(instance_count)]
+    for leaf_name, matches in matches_by_leaf.items():
+        for i, input in enumerate(matches):
+            synthetic_input = cast(ToolSourceTestInput, dict(input))
+            # Keep the bare leaf name; the instance-scoped short-name fallback re-attaches it to
+            # its fully qualified path within the conditional/section.
+            synthetic_input["name"] = leaf_name
+            nested_repeat_inputs[i].append(synthetic_input)
+    return nested_repeat_inputs
 
 
 def _select_which_when(

test/unit/tool_util/test_parameter_test_cases.py

@@ -543,6 +543,83 @@ def test_omitted_conditional_discriminator_inferred_from_provided_params():
     dict_verify_each(tool_state.input_state, expectations)
 
 
+def test_toplevel_param_not_reconsumed_into_nested_repeat():
+    # A raw test input claimed by a fixed top-level param must not also be synthesized into a
+    # nested repeat that happens to declare a same-named param. Regression for the query_tabular
+    # async failure where the top-level `sqlquery` also seeded `addqueries|queries_0|sqlquery`,
+    # producing an extra `output1` and "Incorrect number of outputs - expected 1, found 2".
+    tool_source = raw_xml_tool_source("""
+<tool id="toplevel_vs_repeat" name="toplevel_vs_repeat" version="1.0.0" profile="22.01">
+    <command>echo</command>
+    <inputs>
+        <param name="sqlquery" type="text" value="" />
+        <section name="addqueries" title="Additional Queries">
+            <repeat name="queries" min="0" max="3">
+                <param name="sqlquery" type="text" value="" />
+            </repeat>
+        </section>
+    </inputs>
+    <outputs />
+    <tests>
+        <test>
+            <param name="sqlquery" value="SELECT 1" />
+        </test>
+    </tests>
+</tool>
+        """)
+    parsed_tool = parse_tool(tool_source)
+    test_case = tool_source.parse_tests_to_dict()["tests"][0]
+
+    tool_state = case_state(test_case, parsed_tool.inputs, tool_source.parse_profile()).tool_state
+
+    assert tool_state.input_state["sqlquery"] == "SELECT 1"
+    # The nested repeat must have no instances - the top-level sqlquery is not one of them.
+    assert tool_state.input_state["addqueries"]["queries"] == []
+
+
+def test_repeat_of_conditional_bare_names_synthesize_instance():
+    # A repeat whose instance params live inside a conditional may be specified by bare short
+    # name (discriminator + branch param) without any repeat/conditional wrappers. It must
+    # synthesize one instance, matching the synchronous tool API. Regression for the deepmicro
+    # async failure where `parameter_set` stayed empty and the output collection was empty.
+    tool_source = raw_xml_tool_source("""
+<tool id="repeat_of_conditional" name="repeat_of_conditional" version="1.0.0" profile="22.01">
+    <command>echo</command>
+    <inputs>
+        <repeat name="parameter_set" min="0">
+            <conditional name="rl_type">
+                <param name="rl_type_choice" type="select">
+                    <option value="--ae">AE</option>
+                    <option value="--pca">PCA</option>
+                </param>
+                <when value="--ae">
+                    <param name="dm" type="integer" value="0" />
+                </when>
+                <when value="--pca" />
+            </conditional>
+        </repeat>
+    </inputs>
+    <outputs />
+    <tests>
+        <test>
+            <param name="rl_type_choice" value="--ae" />
+            <param name="dm" value="40" />
+        </test>
+    </tests>
+</tool>
+        """)
+    parsed_tool = parse_tool(tool_source)
+    test_case = tool_source.parse_tests_to_dict()["tests"][0]
+
+    tool_state = case_state(test_case, parsed_tool.inputs, tool_source.parse_profile()).tool_state
+
+    expectations = [
+        (["parameter_set", 0, "rl_type", "rl_type_choice"], "--ae"),
+        (["parameter_set", 0, "rl_type", "dm"], 40),
+    ]
+    dict_verify_each(tool_state.input_state, expectations)
+
+
 def test_duplicate_identical_unqualified_test_param_is_tolerated():
     # A test may list the same unqualified conditional param twice (a common authoring
     # slip). When the duplicate values are identical it is tolerated - matching the