test: add service locator unit tests and changelog fragment

- 7 unit tests covering get_service, set_service, replacement lifecycle,
  close-on-clear_instance, multiple service types, and idempotent re-registration
- Changelog entry for the new service locator API

Author: pv-nvidia

source/isaaclab/changelog.d/service-locator.rst

@@ -0,0 +1,9 @@
+Added
+^^^^^
+
+* Added :class:`~isaaclab.sim.ServiceLocator` and exposed it as
+  :attr:`~isaaclab.sim.SimulationContext.services`.
+
+  Backend-specific caches can be registered and retrieved using subscript
+  syntax (``services[cls] = instance``, ``services[cls]``).  Services with
+  a ``close()`` method are automatically closed on ``clear_instance()``.

source/isaaclab/isaaclab/sim/service_locator.py

@@ -0,0 +1,100 @@
+# Copyright (c) 2022-2026, The Isaac Lab Project Developers (https://github.com/isaac-sim/IsaacLab/blob/main/CONTRIBUTORS.md).
+# All rights reserved.
+#
+# SPDX-License-Identifier: BSD-3-Clause
+
+"""Typed service locator for lifecycle-managed singletons."""
+
+from __future__ import annotations
+
+from typing import TypeVar
+
+_T = TypeVar("_T")
+
+
+def _try_close(service: object) -> None:
+    """Call close() on *service* if it exists and is callable."""
+    close = getattr(service, "close", None)
+    if callable(close):
+        close()
+
+
+class ServiceLocator:
+    """A typed service registry keyed by class, interface, or abstract base class.
+
+    Services are registered and retrieved using subscript syntax::
+
+        locator[FabricStageCache] = FabricStageCache(stage)
+        cache = locator[FabricStageCache]
+
+    Deleting a service calls ``close()`` on it if available::
+
+        del locator[FabricStageCache]
+
+    All registered services are closed and cleared via :meth:`close_all`.
+    """
+
+    def __init__(self) -> None:
+        self._services: dict[type, object] = {}
+
+    def __getitem__(self, cls: type[_T]) -> _T | None:
+        """Retrieve a service by its key class, or ``None`` if not registered."""
+        return self._services.get(cls)  # type: ignore[return-value]
+
+    def __setitem__(self, cls: type[_T], instance: _T) -> None:
+        """Register a service under the given key.
+
+        The key can be the concrete class of *instance*, a parent class,
+        or an abstract base class / protocol — allowing retrieval by
+        interface rather than implementation.
+
+        Does *not* close a previously registered service — the caller is
+        responsible for closing the old instance before replacing it.
+        Use ``del locator[cls]`` or :meth:`pop` to close and remove.
+        """
+        self._services[cls] = instance
+
+    def __delitem__(self, cls: type) -> None:
+        """Close and remove a service.
+
+        Calls ``close()`` on the instance if it has one, then removes it.
+
+        Raises:
+            KeyError: If no service is registered under *cls*.
+        """
+        instance = self._services.pop(cls)
+        _try_close(instance)
+
+    def __contains__(self, cls: type) -> bool:
+        """Check if a service is registered under *cls*."""
+        return cls in self._services
+
+    def pop(self, cls: type[_T]) -> _T | None:
+        """Remove and return a service without closing it.
+
+        Returns:
+            The previously registered instance, or ``None`` if not registered.
+        """
+        return self._services.pop(cls, None)  # type: ignore[return-value]
+
+    def close_all(self, caught_exceptions: list[Exception] | None) -> None:
+        """Close all registered services and clear the registry.
+
+        Calls ``close()`` on each service that has one.
+
+        Args:
+            caught_exceptions: If ``None``, the first exception raised by a
+                service's ``close()`` is propagated immediately (fail-fast).
+                If a list is provided, exceptions are appended to it and
+                closing continues for all remaining services (collect mode).
+                The registry is always cleared regardless of exceptions.
+        """
+        services = list(self._services.values())
+        self._services.clear()
+        for service in services:
+            try:
+                _try_close(service)
+            except Exception as e:
+                if caught_exceptions is None:
+                    raise
+                caught_exceptions.append(e)

source/isaaclab/isaaclab/sim/simulation_context.py

@@ -11,7 +11,7 @@
 import traceback
 from collections.abc import Iterator
 from contextlib import contextmanager
-from typing import TYPE_CHECKING, Any, TypeVar
+from typing import TYPE_CHECKING, Any
 
 import toml
 import torch
@@ -30,6 +30,7 @@
 )
 from isaaclab.renderers.render_context import RenderContext
 from isaaclab.scene.scene_data_provider import SceneDataProvider
+from isaaclab.sim.service_locator import ServiceLocator
 from isaaclab.sim.utils import create_new_stage
 from isaaclab.utils.string import clear_resolve_matching_names_cache
 from isaaclab.utils.version import has_kit
@@ -38,8 +39,6 @@
 if TYPE_CHECKING:
     from isaaclab.cloner.clone_plan import ClonePlan
 
-_T = TypeVar("_T")
-
 from .simulation_cfg import SimulationCfg
 from .spawners import DomeLightCfg, GroundPlaneCfg
 
@@ -216,10 +215,7 @@ def __init__(self, cfg: SimulationCfg | None = None):
             order=5,
         )
 
-        # Singleton service registry — backend-specific caches register themselves
-        # here, keyed by their class.  Services with a ``close()`` method are closed
-        # when the SimulationContext is torn down via ``clear_instance()``.
-        self._services: dict[type, object] = {}
+        self._services = ServiceLocator()
 
         type(self)._instance = self  # Mark as valid singleton only after successful init
 
@@ -863,33 +859,17 @@ def get_setting(self, name: str) -> Any:
     # Service locator
     # ------------------------------------------------------------------
 
-    def get_service(self, cls: type[_T]) -> _T | None:
-        """Retrieve a registered singleton service by its class.
-
-        Args:
-            cls: The service class used as key.
-
-        Returns:
-            The registered instance, or ``None`` if not registered.
-        """
-        return self._services.get(cls)  # type: ignore[return-value]
-
-    def set_service(self, cls: type[_T], instance: _T) -> None:
-        """Register a singleton service, keyed by its class.
+    @property
+    def services(self) -> ServiceLocator:
+        """Typed service registry for backend-specific singletons.
 
-        Overwrites any previously registered instance for the same class.
-        If the old instance has a ``close()`` method it is called before
-        replacement.  Services are automatically closed and cleared when
-        :meth:`clear_instance` is called.
+        Usage::
 
-        Args:
-            cls: The service class used as key.
-            instance: The service instance to register.
+            sim_context.services[FabricStageCache] = cache
+            cache = sim_context.services[FabricStageCache]
+            del sim_context.services[FabricStageCache]  # closes and removes
         """
-        old = self._services.get(cls)
-        if old is not None and old is not instance and hasattr(old, "close"):
-            old.close()
-        self._services[cls] = instance
+        return self._services
 
     @classmethod
     def clear_instance(cls) -> None:
@@ -905,10 +885,8 @@ def clear_instance(cls) -> None:
             cls._instance._visualizers.clear()
 
             # Close and drop all registered singleton services
-            for service in cls._instance._services.values():
-                if hasattr(service, "close"):
-                    service.close()
-            cls._instance._services.clear()
+            service_errors: list[Exception] = []
+            cls._instance._services.close_all(caught_exceptions=service_errors)
 
             # Tear down the stage. We skip clear_stage() (prim-by-prim deletion) since
             # close_stage() + app shutdown destroy the entire stage at once.
@@ -920,6 +898,10 @@ def clear_instance(cls) -> None:
             # Clear instance
             cls._instance = None
 
+            if service_errors:
+                msg = f"SimulationContext.clear_instance(): {len(service_errors)} service(s) failed to close"
+                raise RuntimeError(msg) from service_errors[0]
+
             gc.collect()
             logger.info("SimulationContext cleared")
 

source/isaaclab/test/sim/test_service_locator.py

@@ -0,0 +1,164 @@
+# Copyright (c) 2022-2026, The Isaac Lab Project Developers (https://github.com/isaac-sim/IsaacLab/blob/main/CONTRIBUTORS.md).
+# All rights reserved.
+#
+# SPDX-License-Identifier: BSD-3-Clause
+
+"""Tests for ServiceLocator."""
+
+import pytest
+
+from isaaclab.sim.service_locator import ServiceLocator
+
+
+class _DummyServiceWithClose:
+    """Service with a callable close() method."""
+
+    def __init__(self):
+        self.closed = False
+
+    def close(self):
+        self.closed = True
+
+
+class _DummyServiceWithoutClose:
+    """Service without a close() method."""
+
+    pass
+
+
+class _DummyServiceWithCloseProperty:
+    """Service where 'close' exists but is not callable."""
+
+    close = 42  # attribute, not a method
+
+
+class _DummyServiceThatThrows:
+    """Service whose close() raises an exception."""
+
+    def close(self):
+        raise RuntimeError("close failed")
+
+
+class TestServiceLocator:
+    """Test ServiceLocator subscript API."""
+
+    def test_get_returns_none_when_unregistered(self):
+        loc = ServiceLocator()
+        assert loc[_DummyServiceWithClose] is None
+
+    def test_set_and_get(self):
+        loc = ServiceLocator()
+        svc = _DummyServiceWithClose()
+        loc[_DummyServiceWithClose] = svc
+        assert loc[_DummyServiceWithClose] is svc
+
+    def test_contains(self):
+        loc = ServiceLocator()
+        assert _DummyServiceWithClose not in loc
+        loc[_DummyServiceWithClose] = _DummyServiceWithClose()
+        assert _DummyServiceWithClose in loc
+
+    def test_del_closes_service(self):
+        loc = ServiceLocator()
+        svc = _DummyServiceWithClose()
+        loc[_DummyServiceWithClose] = svc
+        del loc[_DummyServiceWithClose]
+        assert svc.closed
+        assert loc[_DummyServiceWithClose] is None
+
+    def test_del_without_close_method(self):
+        """del on a service without close() should not raise."""
+        loc = ServiceLocator()
+        loc[_DummyServiceWithoutClose] = _DummyServiceWithoutClose()
+        del loc[_DummyServiceWithoutClose]
+        assert loc[_DummyServiceWithoutClose] is None
+
+    def test_del_with_non_callable_close_property(self):
+        """del on a service where close is a property (not callable) should not raise."""
+        loc = ServiceLocator()
+        svc = _DummyServiceWithCloseProperty()
+        loc[_DummyServiceWithCloseProperty] = svc
+        del loc[_DummyServiceWithCloseProperty]
+        assert loc[_DummyServiceWithCloseProperty] is None
+
+    def test_del_missing_raises_key_error(self):
+        loc = ServiceLocator()
+        with pytest.raises(KeyError):
+            del loc[_DummyServiceWithClose]
+
+    def test_pop_returns_without_closing(self):
+        loc = ServiceLocator()
+        svc = _DummyServiceWithClose()
+        loc[_DummyServiceWithClose] = svc
+        popped = loc.pop(_DummyServiceWithClose)
+        assert popped is svc
+        assert not svc.closed
+        assert loc[_DummyServiceWithClose] is None
+
+    def test_pop_missing_returns_none(self):
+        loc = ServiceLocator()
+        assert loc.pop(_DummyServiceWithClose) is None
+
+    def test_close_all(self):
+        loc = ServiceLocator()
+        svc1 = _DummyServiceWithClose()
+        svc2 = _DummyServiceWithoutClose()
+        loc[_DummyServiceWithClose] = svc1
+        loc[_DummyServiceWithoutClose] = svc2
+        loc.close_all(caught_exceptions=None)
+        assert svc1.closed
+        assert loc[_DummyServiceWithClose] is None
+        assert loc[_DummyServiceWithoutClose] is None
+
+    def test_close_all_skips_non_callable_close(self):
+        """close_all does not crash on services with non-callable close attribute."""
+        loc = ServiceLocator()
+        loc[_DummyServiceWithCloseProperty] = _DummyServiceWithCloseProperty()
+        loc.close_all(caught_exceptions=None)  # should not raise
+        assert loc[_DummyServiceWithCloseProperty] is None
+
+    def test_close_all_fail_fast_raises(self):
+        """With caught_exceptions=None, first failure propagates immediately."""
+        loc = ServiceLocator()
+        loc[_DummyServiceThatThrows] = _DummyServiceThatThrows()
+        with pytest.raises(RuntimeError, match="close failed"):
+            loc.close_all(caught_exceptions=None)
+        # Registry is cleared even on failure
+        assert loc[_DummyServiceThatThrows] is None
+
+    def test_close_all_collect_mode(self):
+        """With a list, exceptions are collected and all services are closed."""
+        loc = ServiceLocator()
+        svc_ok = _DummyServiceWithClose()
+        loc[_DummyServiceWithClose] = svc_ok
+        loc[_DummyServiceThatThrows] = _DummyServiceThatThrows()
+        errors: list[Exception] = []
+        loc.close_all(caught_exceptions=errors)
+        assert svc_ok.closed
+        assert len(errors) == 1
+        assert isinstance(errors[0], RuntimeError)
+        assert loc[_DummyServiceWithClose] is None
+        assert loc[_DummyServiceThatThrows] is None
+
+    def test_multiple_service_types(self):
+        loc = ServiceLocator()
+        svc1 = _DummyServiceWithClose()
+        svc2 = _DummyServiceWithoutClose()
+        loc[_DummyServiceWithClose] = svc1
+        loc[_DummyServiceWithoutClose] = svc2
+        assert loc[_DummyServiceWithClose] is svc1
+        assert loc[_DummyServiceWithoutClose] is svc2
+
+    def test_base_class_key(self):
+        """Can register under a base class and retrieve by it."""
+
+        class Base:
+            pass
+
+        class Impl(Base):
+            pass
+
+        loc = ServiceLocator()
+        impl = Impl()
+        loc[Base] = impl
+        assert loc[Base] is impl