Refactor cloning around cfg-driven ClonePlan (#5528)

# Description
This PR a lot simplify cloner logic:
Refactors scene cloning so `InteractiveScene` builds a `ClonePlan`
directly from asset configuration, rewrites spawner configs to spawn
representative sources in their selected environment paths, and then
replicates directly from those sources to the remaining destinations.
This removes the previous template round trip and hard-deletes
`clone_from_template`.

This also updates the cloner API around `CloneCfg` and
`make_clone_plan`, adds explicit `spawn_paths` support for multi-asset
spawners, tightens rigid object collection spawning invariants, and
refreshes docs, tests, and changelog coverage for the new planning flow.

Fixes # N/A

Dependencies: none.

## Type of change

- New feature (non-breaking change which adds functionality)
- Breaking change (existing functionality will not work without user
modification)
- Documentation update

## Screenshots

N/A.

## Test plan

Focused tests were run individually while developing this branch:

- `source/isaaclab/test/scene/test_interactive_scene.py`
- `source/isaaclab/test/sim/test_cloner.py`
- `source/isaaclab/test/sim/test_spawn_wrappers.py`
- `source/isaaclab_physx/test/sim/test_cloner.py`
- `py_compile` checks for touched Python modules

## Checklist

- [x] I have read and understood the [contribution
guidelines](https://isaac-sim.github.io/IsaacLab/main/source/refs/contributing.html)
- [x] I have run the [`pre-commit` checks](https://pre-commit.com/) with
`./isaaclab.sh --format`
- [x] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] I have added a changelog fragment under
`source/<pkg>/changelog.d/` for every touched package (do **not** edit
`CHANGELOG.rst` or bump `extension.toml` — CI handles that)
- [x] I have added my name to the `CONTRIBUTORS.md` or my name already
exists there

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: ooctipus

docs/source/how-to/cloning.rst

@@ -56,7 +56,7 @@ This pattern applies to all simulation components:
      - :class:`~isaaclab_physx.scene_data_providers.PhysxSceneDataProvider`
      - :class:`~isaaclab_newton.scene_data_providers.NewtonSceneDataProvider`
    * - Cloner
-     - :func:`~isaaclab.cloner.clone_from_template`
+     - :func:`~isaaclab.cloner.usd_replicate`
      - :func:`~isaaclab_physx.cloner.physx_replicate`
      - :func:`~isaaclab_newton.cloner.newton_physics_replicate`
 

docs/source/overview/core-concepts/multi_backend_architecture.rst

@@ -0,0 +1,35 @@
+Added
+^^^^^
+
+* Added :class:`~isaaclab.cloner.ClonePlan` as the flat clone contract shared by
+  scene cloning, backend replication, and scene-data providers.
+* Added :meth:`~isaaclab.sim.SimulationContext.get_clone_plan` and
+  :meth:`~isaaclab.sim.SimulationContext.set_clone_plan` for publishing the
+  scene's clone plan.
+* Added :attr:`~isaaclab.scene.InteractiveScene.clone_plan` for consumers holding
+  a scene reference.
+
+Changed
+^^^^^^^
+
+* **Breaking:** Changed scene-data providers to build visualizer backend models
+  from :meth:`~isaaclab.sim.SimulationContext.get_clone_plan` instead of a
+  clone-time visualizer artifact. Use the published
+  :class:`~isaaclab.cloner.ClonePlan` for custom scene-data integrations.
+
+Removed
+^^^^^^^
+
+* **Breaking:** Removed
+  :attr:`~isaaclab.cloner.TemplateCloneCfg.visualizer_clone_fn`,
+  :func:`~isaaclab.cloner.resolve_visualizer_clone_fn`, and
+  :class:`~isaaclab.physics.scene_data_requirements.VisualizerPrebuiltArtifacts`.
+  Use the :class:`~isaaclab.cloner.ClonePlan` published through
+  :meth:`~isaaclab.sim.SimulationContext.get_clone_plan` instead.
+* **Breaking:** Removed
+  :meth:`~isaaclab.sim.SimulationContext.get_scene_data_visualizer_prebuilt_artifact`,
+  :meth:`~isaaclab.sim.SimulationContext.set_scene_data_visualizer_prebuilt_artifact`,
+  and
+  :meth:`~isaaclab.sim.SimulationContext.clear_scene_data_visualizer_prebuilt_artifact`.
+  Use :meth:`~isaaclab.sim.SimulationContext.get_clone_plan` /
+  :meth:`~isaaclab.sim.SimulationContext.set_clone_plan` instead.

source/isaaclab/changelog.d/clone-plan-visualizer-cleanup.minor.rst

@@ -0,0 +1,29 @@
+Added
+^^^^^
+
+* Added explicit ``spawn_paths`` support to multi-asset spawners so scene
+  planning can spawn representative heterogeneous sources directly.
+
+Changed
+^^^^^^^
+
+* **Breaking:** Changed :class:`~isaaclab.scene.InteractiveScene` to build clone
+  plans directly from asset configuration, spawn representative sources in their
+  selected environments, and replicate from those sources instead of spawning and
+  discovering prototypes under ``/World/template``.
+* **Breaking:** Replaced ``TemplateCloneCfg`` with
+  :class:`~isaaclab.cloner.CloneCfg` for clone execution settings.
+* **Breaking:** Changed :func:`~isaaclab.cloner.make_clone_plan` to return a
+  :class:`~isaaclab.cloner.ClonePlan` object directly.
+* **Breaking:** Changed clone plan publication to use
+  :meth:`~isaaclab.sim.SimulationContext.get_clone_plan` and
+  :meth:`~isaaclab.sim.SimulationContext.set_clone_plan` for the single scene
+  clone plan.
+
+Removed
+^^^^^^^
+
+* **Breaking:** Removed :func:`~isaaclab.cloner.clone_from_template`. Use
+  :func:`~isaaclab.cloner.make_clone_plan`,
+  :func:`~isaaclab.cloner.usd_replicate`, and backend physics replication
+  functions for direct cloning workflows.

source/isaaclab/changelog.d/octi-cloner_ordering.major.rst

@@ -4,11 +4,10 @@
 # SPDX-License-Identifier: BSD-3-Clause
 
 __all__ = [
+    "CloneCfg",
     "ClonePlan",
-    "TemplateCloneCfg",
     "random",
     "sequential",
-    "clone_from_template",
     "disabled_fabric_change_notifies",
     "filter_collisions",
     "grid_transforms",
@@ -17,10 +16,9 @@ __all__ = [
 ]
 
 from .clone_plan import ClonePlan
-from .cloner_cfg import TemplateCloneCfg
+from .cloner_cfg import CloneCfg
 from .cloner_strategies import random, sequential
 from .cloner_utils import (
-    clone_from_template,
     disabled_fabric_change_notifies,
     filter_collisions,
     grid_transforms,

source/isaaclab/isaaclab/cloner/__init__.pyi

@@ -5,35 +5,29 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 
 import torch
 
 
-@dataclass(frozen=True)
+@dataclass(frozen=True, eq=False)
 class ClonePlan:
-    """Per-group mapping from prototype prims to per-environment clones.
-
-    Produced by :func:`~isaaclab.cloner.clone_from_template` for each prototype group it
-    discovers under the template root. Lets downstream consumers (e.g. mesh samplers,
-    ray-cast sensors) read prototype geometry once and scatter to environments via
-    :attr:`clone_mask` instead of walking per-env USD paths.
-
-    Attributes are population-time invariants and the dataclass is frozen. Hash and
-    equality operate on :attr:`dest_template` only (the natural identity — it is the key
-    in :attr:`SimulationContext.get_clone_plans`); the mutable list/tensor fields are
-    excluded since ``torch.Tensor`` is not hashable and structural equality is rarely the
-    semantics consumers want.
+    """Flat cloning source of truth.
+
+    Produced by scene planning after representative source prims are assigned. The
+    three fields are the same flat replication contract consumed by USD, physics,
+    and downstream scene-data providers: each source path maps to the destination
+    template at the same index, and :attr:`clone_mask` selects the environments
+    populated from that source.
     """
 
-    dest_template: str
-    """Destination path template for this group, e.g. ``"/World/envs/env_{}/Object"``."""
+    sources: tuple[str, ...]
+    """Source prim paths used for replication."""
 
-    prototype_paths: list[str] = field(hash=False, compare=False)
-    """Prototype prim paths in this group, e.g.
-    ``["/World/template/Object/proto_asset_0", "/World/template/Object/proto_asset_1"]``."""
+    destinations: tuple[str, ...]
+    """Destination path templates, one per source path."""
 
-    clone_mask: torch.Tensor = field(hash=False, compare=False)
-    """Boolean tensor of shape ``[num_prototypes_in_group, num_envs]``;
+    clone_mask: torch.Tensor
+    """Boolean tensor of shape ``[len(sources), num_envs]``;
     ``clone_mask[i, j]`` is ``True`` iff env ``j`` was populated from
-    :attr:`prototype_paths` ``[i]``. Each column sums to exactly one."""
+    :attr:`sources` ``[i]``."""

source/isaaclab/isaaclab/cloner/clone_plan.py

@@ -11,52 +11,14 @@
 
 
 @configclass
-class TemplateCloneCfg:
-    """Configuration for template-based cloning.
+class CloneCfg:
+    """Configuration for environment replication.
 
-    This configuration is consumed by :func:`~isaaclab.scene.cloner.clone_from_template` to
-    replicate one or more "prototype" prims authored under a template root into multiple
-    per-environment destinations. It supports both USD-spec replication and PhysX replication
-    and allows choosing between random or round-robin prototype assignment across environments.
-
-    The cloning flow is:
-
-    1. Discover prototypes under :attr:`template_root` whose base name starts with
-        :attr:`template_prototype_identifier` (for example, ``proto_asset_0``, ``proto_asset_1``).
-    2. Build a per-prototype mapping to environments according to
-        :attr:`random_heterogeneous_cloning` (random) or modulo assignment (deterministic).
-    3. Stamp the selected prototypes to destinations derived from :attr:`clone_regex`.
-    4. Optionally perform PhysX replication for the same mapping.
-
-    Example
-    -------
-
-    .. code-block:: python
-
-        from isaaclab.cloner import TemplateCloneCfg, clone_from_template
-        from isaaclab.sim.utils.stage import get_current_stage
-
-        stage = get_current_stage()
-        cfg = TemplateCloneCfg(
-            num_clones=128,
-            template_root="/World/template",
-            template_prototype_identifier="proto_asset",
-            clone_regex="/World/envs/env_.*",
-            clone_usd=True,
-            clone_physics=True,
-            random_heterogeneous_cloning=False,  # use round-robin mapping
-            device="cpu",
-        )
-
-        clone_from_template(stage, num_clones=cfg.num_clones, template_clone_cfg=cfg)
+    The scene builds a :class:`~isaaclab.cloner.ClonePlan` directly from asset
+    configuration, spawns the representative source prims, and then uses this
+    configuration to dispatch USD and physics replication for that plan.
     """
 
-    template_root: str = "/World/template"
-    """Root path under which template prototypes are authored."""
-
-    template_prototype_identifier: str = "proto_asset"
-    """Name prefix used to identify prototype prims under :attr:`template_root`."""
-
     clone_regex: str = "/World/envs/env_.*"
     """Destination template for per-environment paths.
 

source/isaaclab/isaaclab/cloner/cloner_cfg.py

@@ -9,20 +9,13 @@
 import itertools
 import logging
 import math
-from collections.abc import Iterator
-from typing import TYPE_CHECKING
+from collections.abc import Iterator, Sequence
 
 import torch
 
 from pxr import Gf, Sdf, Usd, UsdGeom, UsdUtils, Vt
 
-import isaaclab.sim as sim_utils
-
 from . import _fabric_notices
-
-if TYPE_CHECKING:
-    from .cloner_cfg import TemplateCloneCfg
-
 from .clone_plan import ClonePlan
 
 logger = logging.getLogger(__name__)
@@ -105,118 +98,13 @@ def disabled_fabric_change_notifies(stage: Usd.Stage, *, restore: bool = True) -
             bindings.set_enable(fabric_id, True)
 
 
-def clone_from_template(
-    stage: Usd.Stage, num_clones: int, template_clone_cfg: TemplateCloneCfg
-) -> dict[str, ClonePlan]:
-    """Clone assets from a template root into per-environment destinations.
-
-    This utility discovers prototype prims under ``cfg.template_root`` whose names start with
-    ``cfg.template_prototype_identifier``, builds a per-prototype mapping across
-    ``num_clones`` environments (random or modulo), and then performs USD and/or PhysX replication
-    according to the flags in ``cfg``.
-
-    Args:
-        stage: The USD stage to author into.
-        num_clones: Number of environments to clone to (typically equals ``cfg.num_clones``).
-        template_clone_cfg: Configuration describing template location, destination pattern,
-            and replication/mapping behavior.
-
-    Returns:
-        Mapping from each group's destination template (e.g. ``"/World/envs/env_{}/Object"``)
-        to its :class:`ClonePlan`. Empty when no prototype groups are discovered.
-
-    Note:
-        This function suspends the Fabric USD notice listener for the duration of the call
-        and **leaves it disabled on return**. It is intended to be invoked from a scene-init
-        path that is followed by :meth:`isaaclab.sim.SimulationContext.reset`, whose Fabric
-        resync naturally recovers the listener state. Callers that bypass that reset
-        contract (ad-hoc tooling, unit tests on a bare stage) should re-enable Fabric
-        notices themselves or wrap the call in
-        :func:`disabled_fabric_change_notifies` with ``restore=True``.
-    """
-    cfg: TemplateCloneCfg = template_clone_cfg
-    plans: dict[str, ClonePlan] = {}
-    # Suspend Fabric's USD notice listener for the duration of bulk authoring. ``restore=False``
-    # because clone_from_template is only called at scene-init time, which is followed by
-    # ``SimulationContext.reset`` — that reset path does the Fabric resync naturally, and
-    # re-enabling here would trigger a redundant ``forceMinimalPopulate`` batch.
-    with disabled_fabric_change_notifies(stage, restore=False):
-        world_indices = torch.arange(num_clones, device=cfg.device)
-        clone_path_fmt = cfg.clone_regex.replace(".*", "{}")
-        prototype_id = cfg.template_prototype_identifier
-        prototypes = sim_utils.get_all_matching_child_prims(
-            cfg.template_root,
-            predicate=lambda prim: str(prim.GetPath()).split("/")[-1].startswith(prototype_id),
-        )
-        if len(prototypes) > 0:
-            # Canonicalize prototype-root order. Some simulation/visualization backends might apply order-dependent
-            # processing, so varying USD traversal or set iteration order can change outputs noticeably. Sorting here
-            # removes that nondeterminism at the source (group order feeds ``make_clone_plan`` and downstream
-            # replication), which matters for run-to-run reproducibility across IsaacLab's multi-backend stack.
-            prototype_roots = sorted({"/".join(str(prototype.GetPath()).split("/")[:-1]) for prototype in prototypes})
-
-            # discover prototypes per root then make a clone plan
-            src: list[list[str]] = []
-            dest: list[str] = []
-
-            for prototype_root in prototype_roots:
-                protos = sim_utils.find_matching_prim_paths(f"{prototype_root}/.*")
-                protos = [proto for proto in protos if proto.split("/")[-1].startswith(prototype_id)]
-                src.append(protos)
-                dest.append(prototype_root.replace(cfg.template_root, clone_path_fmt))
-
-            src_paths, dest_paths, clone_masking = make_clone_plan(
-                src, dest, num_clones, cfg.clone_strategy, cfg.device
-            )
-
-            # Per-group plans: slice ``clone_masking`` along the prototype axis using cumulative
-            # group sizes — each group's mask rows are contiguous in the ``[total_protos, num_envs]``
-            # tensor that ``make_clone_plan`` produced.
-            offsets = [0, *itertools.accumulate(len(g) for g in src)]
-            plans = {
-                d: ClonePlan(dest_template=d, prototype_paths=list(ps), clone_mask=clone_masking[lo:hi])
-                for ps, d, lo, hi in zip(src, dest, offsets, offsets[1:])
-            }
-
-            # Spawn the first instance of clones from prototypes, then deactivate the prototypes, those first
-            # instances will be served as sources for usd and physics replication.
-            proto_idx = clone_masking.to(torch.int32).argmax(dim=1)
-            proto_mask = torch.zeros_like(clone_masking)
-            proto_mask.scatter_(1, proto_idx.view(-1, 1).to(torch.long), clone_masking.any(dim=1, keepdim=True))
-            usd_replicate(stage, src_paths, dest_paths, world_indices, proto_mask)
-            stage.GetPrimAtPath(cfg.template_root).SetActive(False)
-            get_pos = lambda path: stage.GetPrimAtPath(path).GetAttribute("xformOp:translate").Get()  # noqa: E731
-            positions = torch.tensor([get_pos(clone_path_fmt.format(i)) for i in world_indices])
-            # Heterogeneous default: emit per-prototype (sources, destinations, mask) and trust
-            # env_0..N's existing xforms (proto-spawn above already placed them, so don't
-            # re-author). When every env happens to pick prototype 0, collapse below to a
-            # single env_0 → all-envs copy and re-author positions (the destination subtree
-            # replaces env_1..N's prior xform).
-            sources = [tpl.format(int(idx)) for tpl, idx in zip(dest_paths, proto_idx.tolist())]
-            usd_positions: torch.Tensor | None = None
-            if torch.all(proto_idx == 0):
-                sources = [clone_path_fmt.format(0)]
-                dest_paths = [clone_path_fmt]
-                clone_masking = clone_masking.new_ones(1, num_clones)
-                usd_positions = positions
-
-            if cfg.clone_physics and cfg.physics_clone_fn is not None:
-                cfg.physics_clone_fn(
-                    stage, sources, dest_paths, world_indices, clone_masking, positions=positions, device=cfg.device
-                )
-            if cfg.clone_usd:
-                usd_replicate(stage, sources, dest_paths, world_indices, clone_masking, positions=usd_positions)
-
-    return plans
-
-
 def make_clone_plan(
-    sources: list[list[str]],
-    destinations: list[str],
+    sources: Sequence[Sequence[str]],
+    destinations: Sequence[str],
     num_clones: int,
     clone_strategy: callable,
     device: str = "cpu",
-) -> tuple[list[str], list[str], torch.Tensor]:
+) -> ClonePlan:
     """Construct a cloning plan mapping prototype prims to per-environment destinations.
 
     The plan enumerates all combinations of prototypes, selects a combination per environment using ``clone_strategy``,
@@ -231,14 +119,20 @@ def make_clone_plan(
         device: Torch device for tensors in the plan. Defaults to ``"cpu"``.
 
     Returns:
-        tuple: ``(src, dest, masking)`` where ``src`` and ``dest`` are flattened lists of prototype and
-            destination paths, and ``masking`` is a ``[num_src, num_clones]`` boolean tensor with True
-            when source ``src[i]`` is used for clone ``j``.
+        A :class:`ClonePlan` whose ``sources`` and ``destinations`` are flattened per-source rows and
+        whose ``clone_mask`` is a ``[num_src, num_clones]`` boolean tensor.
     """
-    # 1) Flatten into src and dest lists
-    src = [p for group in sources for p in group]
-    dest = [dst for dst, group in zip(destinations, sources) for _ in group]
+    if len(sources) != len(destinations):
+        raise ValueError(f"Expected one destination per source group, got {len(destinations)} and {len(sources)}.")
+    if not sources:
+        raise ValueError("Expected at least one source group.")
     group_sizes = [len(group) for group in sources]
+    if any(size == 0 for size in group_sizes):
+        raise ValueError("Source groups must not be empty.")
+
+    # 1) Flatten into src and dest lists
+    src = tuple(p for group in sources for p in group)
+    dest = tuple(dst for dst, group in zip(destinations, sources) for _ in group)
 
     # 2) Enumerate all combinations of "one prototype per group"
     #    all_combos: list of tuples (g0_idx, g1_idx, ..., g_{G-1}_idx)
@@ -256,13 +150,13 @@ def make_clone_plan(
 
     masking = torch.zeros((sum(group_sizes), num_clones), dtype=torch.bool, device=device)
     masking[rows, cols] = True
-    return src, dest, masking
+    return ClonePlan(sources=src, destinations=dest, clone_mask=masking)
 
 
 def usd_replicate(
     stage: Usd.Stage,
-    sources: list[str],
-    destinations: list[str],
+    sources: Sequence[str],
+    destinations: Sequence[str],
     env_ids: torch.Tensor,
     mask: torch.Tensor | None = None,
     positions: torch.Tensor | None = None,