Fix clone-plan source resolution for nested destination templates

source/isaaclab/changelog.d/fix-cloner_resolve_plan_source.rst

@@ -0,0 +1,10 @@
+Fixed
+^^^^^
+
+* Fixed :func:`~isaaclab.cloner.cloner_utils.resolve_clone_plan_source` raising a
+  ``ValueError`` when a path expression was owned by nested clone-plan destination
+  templates (e.g. a camera cloned under a robot at
+  ``/World/envs/env_{}/Robot/ee_link/palm_link/Camera``). It now selects the most
+  specific (longest-matching) template, mirroring
+  :func:`~isaaclab.cloner.cloner_utils.iter_clone_plan_matches`, and only raises when
+  a path is owned by multiple distinct, equally specific templates.

source/isaaclab/isaaclab/cloner/cloner_utils.py

@@ -74,31 +74,35 @@ def resolve_clone_plan_source(path_expr: str, plan: ClonePlan) -> tuple[str, str
         mounted at the env root rather than under a planned asset).
 
     Raises:
-        ValueError: When ``path_expr``'s matching rows span multiple distinct
-            destination templates.
+        ValueError: When ``path_expr`` is owned by multiple distinct, equally
+            specific destination templates (a genuine ambiguity). Nested
+            templates do not conflict: the most specific (longest-matching) one
+            wins, mirroring :func:`iter_clone_plan_matches`.
         NotImplementedError: When the union of matching rows' clone masks does not
             cover every env (partial-env heterogeneous coverage is unsupported).
     """
-    matching_template: str | None = None
-    matching_rows: list[int] = []
-    matching_suffix: str | None = None
+    # Collect every template that owns ``path_expr`` together with the suffix below it.
+    # A shorter suffix means a longer matched prefix, i.e. a more specific (nearer) owner.
+    candidates: list[tuple[str, str, int]] = []
     for source_index, destination_template in enumerate(plan.destinations):
         if "{}" not in destination_template:
             continue
         suffix = get_suffix(path_expr, destination_template)
         if suffix is None:
             continue
-        if matching_template is None:
-            matching_template = destination_template
-            matching_suffix = suffix
-        elif destination_template != matching_template:
-            raise ValueError(
-                f"path_expr {path_expr!r}: matches multiple destination templates"
-                f" {matching_template!r} and {destination_template!r}."
-            )
-        matching_rows.append(source_index)
-    if matching_template is None:
+        candidates.append((destination_template, suffix, source_index))
+    if not candidates:
         return None
+
+    # The nearest owner is the one with the shortest suffix. Distinct templates that tie
+    # at this minimal suffix length are a genuine ambiguity that callers cannot resolve.
+    min_suffix_len = min(len(suffix) for _, suffix, _ in candidates)
+    owning_templates = {template for template, suffix, _ in candidates if len(suffix) == min_suffix_len}
+    if len(owning_templates) > 1:
+        raise ValueError(f"path_expr {path_expr!r}: matches multiple destination templates {sorted(owning_templates)}.")
+    matching_template = next(iter(owning_templates))
+    matching_rows = [index for template, _, index in candidates if template == matching_template]
+    matching_suffix = next(suffix for template, suffix, _ in candidates if template == matching_template)
     if not plan.clone_mask[matching_rows].any(dim=0).all():
         raise NotImplementedError(
             f"path_expr {path_expr!r}: partial-env heterogeneous coverage is unsupported;"

source/isaaclab/test/sim/test_cloner.py

@@ -21,7 +21,7 @@
 
 import isaaclab.sim as sim_utils
 from isaaclab.cloner import ClonePlan, make_clone_plan, sequential, usd_replicate
-from isaaclab.cloner.cloner_utils import iter_clone_plan_matches
+from isaaclab.cloner.cloner_utils import iter_clone_plan_matches, resolve_clone_plan_source
 from isaaclab.sim import build_simulation_context
 
 pytestmark = pytest.mark.isaacsim_ci
@@ -336,3 +336,67 @@ def test_iter_clone_plan_matches(sim):
             (0, 1),
         )
     ]
+
+
+def test_resolve_clone_plan_source_nested_templates_pick_most_specific(sim):
+    """A path owned by both an ancestor and a descendant template resolves to the descendant."""
+    plan = ClonePlan(
+        sources=("/World/envs/env_0/Robot", "/World/envs/env_0/Robot/ee_link/palm_link/Camera"),
+        destinations=("/World/envs/env_{}/Robot", "/World/envs/env_{}/Robot/ee_link/palm_link/Camera"),
+        clone_mask=torch.tensor([[True, True], [True, True]], device=sim.cfg.device),
+    )
+
+    # The camera path matches both templates; the more specific (longer-matching) one wins.
+    resolved = resolve_clone_plan_source(plan=plan, path_expr="/World/envs/env_0/Robot/ee_link/palm_link/Camera")
+
+    assert resolved == (
+        "/World/envs/env_0/Robot/ee_link/palm_link/Camera",
+        "/World/envs/env_*/Robot/ee_link/palm_link/Camera",
+        "",
+    )
+
+    # A path that only the ancestor template owns still resolves against it with its suffix.
+    resolved = resolve_clone_plan_source(plan=plan, path_expr="/World/envs/env_0/Robot/base")
+
+    assert resolved == ("/World/envs/env_0/Robot", "/World/envs/env_*/Robot", "/base")
+
+
+def test_resolve_clone_plan_source_ambiguous_templates_raise(sim):
+    """Two distinct, equally specific templates owning a path remain a genuine ambiguity."""
+    plan = ClonePlan(
+        sources=("/World/envs/env_0/Robot", "/World/envs/env_0/Robot"),
+        destinations=("/World/envs/{}/Robot", "/World/{}/env_0/Robot"),
+        clone_mask=torch.tensor([[True, True], [True, True]], device=sim.cfg.device),
+    )
+
+    with pytest.raises(ValueError, match="matches multiple destination templates"):
+        resolve_clone_plan_source(plan=plan, path_expr="/World/envs/env_0/Robot")
+
+
+def test_resolve_clone_plan_source_merges_same_template_rows(sim):
+    """Heterogeneous source rows sharing one template OR-merge their masks for the coverage check."""
+    # One logical asset cloned from two source variants onto the same destination template.
+    # Neither row alone covers all envs; row 0 -> envs (0, 2), row 1 -> envs (1, 3).
+    plan = ClonePlan(
+        sources=("/World/envs/env_0/Object", "/World/envs/env_1/Object"),
+        destinations=("/World/envs/env_{}/Object", "/World/envs/env_{}/Object"),
+        clone_mask=torch.tensor([[True, False, True, False], [False, True, False, True]], device=sim.cfg.device),
+    )
+
+    # The union of both rows covers every env, so resolution succeeds and reports the first row's source.
+    resolved = resolve_clone_plan_source(plan=plan, path_expr="/World/envs/env_.*/Object/Body/Camera")
+
+    assert resolved == ("/World/envs/env_0/Object", "/World/envs/env_*/Object", "/Body/Camera")
+
+
+def test_resolve_clone_plan_source_partial_coverage_raises(sim):
+    """When the matching rows' merged mask misses an env, partial coverage is rejected."""
+    # Row 0 -> envs (0, 2), row 1 -> env (1); env 3 is covered by neither row.
+    plan = ClonePlan(
+        sources=("/World/envs/env_0/Object", "/World/envs/env_1/Object"),
+        destinations=("/World/envs/env_{}/Object", "/World/envs/env_{}/Object"),
+        clone_mask=torch.tensor([[True, False, True, False], [False, True, False, False]], device=sim.cfg.device),
+    )
+
+    with pytest.raises(NotImplementedError, match="partial-env heterogeneous coverage"):
+        resolve_clone_plan_source(plan=plan, path_expr="/World/envs/env_.*/Object/Body/Camera")