feat: add typed service locator to SimulationContext

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,98 @@
+# 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]`` to close and remove, or :meth:`pop` to
+        remove without closing.
+        """
+        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:
+        """Close all registered services and clear the registry.
+
+        Calls ``close()`` on each service that has one.  Exceptions are
+        always collected into *caught_exceptions* — closing continues for
+        all remaining services regardless of failures.
+
+        Args:
+            caught_exceptions: A list to which any exceptions raised by
+                service ``close()`` calls are appended.
+        """
+        services = list(self._services.values())
+        self._services.clear()
+        for service in services:
+            try:
+                _try_close(service)
+            except Exception as e:
+                caught_exceptions.append(e)

source/isaaclab/isaaclab/sim/simulation_context.py

@@ -30,6 +30,7 @@
 )
 from isaaclab.renderers.render_context import RenderContext
 from isaaclab.scene_data 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
@@ -214,6 +215,8 @@ def __init__(self, cfg: SimulationCfg | None = None):
             order=5,
         )
 
+        self._services = ServiceLocator()
+
         type(self)._instance = self  # Mark as valid singleton only after successful init
 
     def _apply_render_cfg_settings(self) -> None:
@@ -850,6 +853,22 @@ def get_setting(self, name: str) -> Any:
         """Get a setting value."""
         return self._settings_helper.get(name)
 
+    # ------------------------------------------------------------------
+    # Service locator
+    # ------------------------------------------------------------------
+
+    @property
+    def services(self) -> ServiceLocator:
+        """Typed service registry for backend-specific singletons.
+
+        Usage::
+
+            sim_context.services[FabricStageCache] = cache
+            cache = sim_context.services[FabricStageCache]
+            del sim_context.services[FabricStageCache]  # closes and removes
+        """
+        return self._services
+
     @classmethod
     def clear_instance(cls) -> None:
         """Clean up resources and clear the singleton instance."""
@@ -863,6 +882,10 @@ def clear_instance(cls) -> None:
                 viz.close()
             cls._instance._visualizers.clear()
 
+            # Close and drop all registered singleton services
+            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.
             stage_utils.close_stage()
@@ -876,6 +899,11 @@ def clear_instance(cls) -> None:
             gc.collect()
             logger.info("SimulationContext cleared")
 
+            if service_errors:
+                msg = f"SimulationContext.clear_instance(): {len(service_errors)} service(s) failed to close"
+                # TODO: Use ExceptionGroup when ruff target-version is bumped to py311+
+                raise RuntimeError(msg) from service_errors[0]
+
     @classmethod
     def clear_stage(cls) -> None:
         """Clear the current USD stage (preserving /World and PhysicsScene).

source/isaaclab/test/sim/test_service_locator.py

@@ -0,0 +1,169 @@
+# 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
+
+# -- Dummy service helpers --
+
+
+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")
+
+
+# -- Fixtures --
+
+
+@pytest.fixture
+def locator():
+    """Provide a fresh ServiceLocator for each test."""
+    return ServiceLocator()
+
+
+# -- Tests --
+
+
+def test_get_returns_none_when_unregistered(locator):
+    assert locator[_DummyServiceWithClose] is None
+
+
+def test_set_and_get(locator):
+    svc = _DummyServiceWithClose()
+    locator[_DummyServiceWithClose] = svc
+    assert locator[_DummyServiceWithClose] is svc
+
+
+def test_contains(locator):
+    assert _DummyServiceWithClose not in locator
+    locator[_DummyServiceWithClose] = _DummyServiceWithClose()
+    assert _DummyServiceWithClose in locator
+
+
+def test_del_closes_service(locator):
+    svc = _DummyServiceWithClose()
+    locator[_DummyServiceWithClose] = svc
+    del locator[_DummyServiceWithClose]
+    assert svc.closed
+    assert locator[_DummyServiceWithClose] is None
+
+
+def test_del_without_close_method(locator):
+    """del on a service without close() should not raise."""
+    locator[_DummyServiceWithoutClose] = _DummyServiceWithoutClose()
+    del locator[_DummyServiceWithoutClose]
+    assert locator[_DummyServiceWithoutClose] is None
+
+
+def test_del_with_non_callable_close_property(locator):
+    """del on a service where close is a property (not callable) should not raise."""
+    svc = _DummyServiceWithCloseProperty()
+    locator[_DummyServiceWithCloseProperty] = svc
+    del locator[_DummyServiceWithCloseProperty]
+    assert locator[_DummyServiceWithCloseProperty] is None
+
+
+def test_del_missing_raises_key_error(locator):
+    with pytest.raises(KeyError):
+        del locator[_DummyServiceWithClose]
+
+
+def test_pop_returns_without_closing(locator):
+    svc = _DummyServiceWithClose()
+    locator[_DummyServiceWithClose] = svc
+    popped = locator.pop(_DummyServiceWithClose)
+    assert popped is svc
+    assert not svc.closed
+    assert locator[_DummyServiceWithClose] is None
+
+
+def test_pop_missing_returns_none(locator):
+    assert locator.pop(_DummyServiceWithClose) is None
+
+
+def test_close_all(locator):
+    svc1 = _DummyServiceWithClose()
+    svc2 = _DummyServiceWithoutClose()
+    locator[_DummyServiceWithClose] = svc1
+    locator[_DummyServiceWithoutClose] = svc2
+    errors: list[Exception] = []
+    locator.close_all(caught_exceptions=errors)
+    assert svc1.closed
+    assert not errors
+    assert locator[_DummyServiceWithClose] is None
+    assert locator[_DummyServiceWithoutClose] is None
+
+
+def test_close_all_skips_non_callable_close(locator):
+    """close_all does not crash on services with non-callable close attribute."""
+    locator[_DummyServiceWithCloseProperty] = _DummyServiceWithCloseProperty()
+    errors: list[Exception] = []
+    locator.close_all(caught_exceptions=errors)
+    assert not errors
+    assert locator[_DummyServiceWithCloseProperty] is None
+
+
+def test_close_all_collects_exceptions(locator):
+    """Exceptions are collected and all services still get closed."""
+    svc_ok = _DummyServiceWithClose()
+    locator[_DummyServiceWithClose] = svc_ok
+    locator[_DummyServiceThatThrows] = _DummyServiceThatThrows()
+    errors: list[Exception] = []
+    locator.close_all(caught_exceptions=errors)
+    assert svc_ok.closed
+    assert len(errors) == 1
+    assert isinstance(errors[0], RuntimeError)
+    assert locator[_DummyServiceWithClose] is None
+    assert locator[_DummyServiceThatThrows] is None
+
+
+def test_multiple_service_types(locator):
+    svc1 = _DummyServiceWithClose()
+    svc2 = _DummyServiceWithoutClose()
+    locator[_DummyServiceWithClose] = svc1
+    locator[_DummyServiceWithoutClose] = svc2
+    assert locator[_DummyServiceWithClose] is svc1
+    assert locator[_DummyServiceWithoutClose] is svc2
+
+
+def test_base_class_key(locator):
+    """Can register under a base class and retrieve by it."""
+
+    class Base:
+        pass
+
+    class Impl(Base):
+        pass
+
+    impl = Impl()
+    locator[Base] = impl
+    assert locator[Base] is impl