diff --git a/README.md b/README.md index 8fdc0d6c7e..ad3a0f51bf 100644 --- a/README.md +++ b/README.md @@ -69,6 +69,7 @@ Our [documentation page](https://nvidia.github.io/IsaacCapture) provides everyth - [Architecture](https://nvidia.github.io/IsaacCapture/main/overview/architecture.html) - [Quick installation steps](https://nvidia.github.io/IsaacCapture/main/getting_started/quick_start.html) - [How to build from source](https://nvidia.github.io/IsaacCapture/main/getting_started/build_from_source.html) +- [OVXR installation](docs/source/getting_started/ovxr.rst) ### Check a Device Integration @@ -86,3 +87,17 @@ To get started, please refer to Isaac Lab's [Installation](https://isaac-sim.git The default license is [Apache-2.0](LICENSE.md). EgoPost is licensed under [OpenMDW-1.1](LICENSES/LicenseRef-OpenMDW-1.1.txt). Third-party materials may carry their own licenses. + +### Separately retrieved optional materials + +The optional `isaaccapture[ovxr]` extra retrieves OVXR and its OVRTX/OVStage +dependencies. +These packages are downloaded separately, rather than bundled in the Isaac Capture +wheel. Their licenses apply independently of Isaac Capture's Apache-2.0 license. +OVXR, OVRTX and OVStage are NVIDIA proprietary software. +See [package sources and license details](docs/source/getting_started/ovxr.rst#optional-packages-and-licenses). +Review the applicable package licenses and terms before installing or using these +optional dependencies. + +Existing CloudXR native libraries are bundled in Isaac Capture wheels and remain subject +to their separate NVIDIA license and EULA acceptance. diff --git a/docs/source/getting_started/ovxr.rst b/docs/source/getting_started/ovxr.rst new file mode 100644 index 0000000000..2b6f1de545 --- /dev/null +++ b/docs/source/getting_started/ovxr.rst @@ -0,0 +1,127 @@ +.. SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +.. SPDX-License-Identifier: Apache-2.0 + +OVXR installation +================= + +OVXR provides optional XR presentation for applications that use Isaac Capture +for teleoperation input. Select the ``ovxr`` extra when installing Isaac Capture +to include OVXR and its rendering dependencies. + +Install XR support +------------------ + +Follow :doc:`/development/build_from_source/index` to build Isaac Capture and +select a Python environment matching the generated wheel. Include ``ovxr`` in +the installation requirement: + +.. code-block:: bash + + uv pip install "isaaccapture[ovxr]==X.Y+local" --find-links=./install/wheels/ + uv pip check + +Replace ``X.Y`` with the version of your source build, as described in the build +guide. Base installations and imports do not require OVXR. + +OVXR declares its compatible OVRTX and OVStage versions in its package metadata. +The consuming application's dependency constraints must accept those versions. +See the package pages below for supported Python versions and platforms. + +Optional packages and licenses +------------------------------ + +The ``ovxr`` extra installs OVXR and its OVRTX/OVStage dependencies separately +from the Isaac Capture wheel. These packages retain their own licenses: + +* `OVXR `__, + `OVRTX `__ and + `OVStage `__ are NVIDIA proprietary software. + Review the license included with each distribution before use. + +CloudXR native libraries are bundled in Isaac Capture wheels and remain subject +to their separate NVIDIA license and CloudXR EULA acceptance. + +Use from another application +---------------------------- + +``isaaccapture.ovxr_session_manager.ovxr_session`` provides ``OvxrSession`` +and ``OvxrSessionConfig``. Implement ``OvxrApplicationInterface`` to supply the +application's renderer, Stage and ordinal ledger in an ``OvxrSessionBinding``. +``camera_paths`` is required and must contain both stereo cameras. Set +``up_axis="Y"`` for a Y-up Stage; the default is ``"Z"``. + +For a uv-managed application, declare Isaac Capture with its ``ovxr`` extra and +select the source-built wheel in the application's project configuration: + +.. code-block:: toml + + [project.optional-dependencies] + ovxr = ["isaaccapture[ovxr]"] + + [tool.uv.sources] + isaaccapture = { path = "path/to/your/local-wheel.whl", extra = "ovxr" } + +Replace the wheel path with the file produced by your build. Refresh the +application's lock file after changing the wheel or dependency selection, then +run with ``uv run --extra ovxr``. Installing packages with ``uv pip install`` +alone does not update the application's lock file. + +Application-specific OVXR versions +---------------------------------- + +Isaac Capture pins the OVXR version used for compatibility testing. uv respects +that pin and reports a conflict if the application requires incompatible OVRTX +or OVStage versions. + +To validate another OVXR build with the same Isaac Capture release, select an +OVXR release whose declared dependencies match the application's OVRTX/OVStage +requirements. Add an override in the application's root ``pyproject.toml``: + +.. code-block:: toml + + [tool.uv] + override-dependencies = ["ovxr=="] + +Replace ```` with the selected release. If the application +already has an ``override-dependencies`` list, append this entry to that list. +Then refresh the lock file and synchronize the environment using the ``ovxr`` +extra from the example above: + +.. code-block:: bash + + uv lock --upgrade-package ovxr + uv sync --extra ovxr + +An `override `__ +changes dependency resolution; it does not establish compatibility with Isaac +Capture's OVXR integration. Validate the application's XR workflow before +relying on the selected build. + +Applications using different provider versions need separate project +environments and lock files. Each can install the same Isaac Capture wheel +with its own OVXR/OVRTX/OVStage combination. + + +Session lifetime +---------------- + +Use ``OvxrSession`` as a context manager, or call ``close()`` in a ``finally`` +block. Stop application steps and input borrowers before closing. Startup failure +before handles are borrowed retires its resources on the owner thread; successful +startup requires explicit closure. Do not rely on interpreter exit or object +finalizers to stop native rendering. + +The session schedules work; its frame pump owns native frame processing and +retirement, and its runtime owner stops only the CloudXR process it launched. +Applications may supply ``runtime_launcher_factory`` for a compatible launcher; +it receives the environment path and ``accept_eula`` and returns an object with +``stop()``. Externally managed runtimes remain externally owned. + +``OvxrScenePublication`` and ``OvxrResourceOwner`` describe the two parts of the +application contract. ``OvxrApplicationInterface`` combines them, so existing +adapters can continue implementing the combined interface. + +If native context destruction fails without proving the context was consumed, +closure reports the failure and retains its borrowed resources and runtime. +Do not treat this as successful retirement or start a replacement session over +those resources. There is no automatic retry of uncertain native destruction. diff --git a/docs/source/index.rst b/docs/source/index.rst index 038f2e8d78..b304b65ded 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -54,6 +54,7 @@ Table of Contents :caption: Isaac Teleop teleop/quick_start + getting_started/ovxr teleop/teleop_session teleop/control_state_machine diff --git a/docs/source/references/build.rst b/docs/source/references/build.rst index f9b281c7dd..acf6140a30 100644 --- a/docs/source/references/build.rst +++ b/docs/source/references/build.rst @@ -34,6 +34,8 @@ The layout below reflects the actual ``CMakeLists.txt`` hierarchy (root ``CMakeL │ ├── teleop_session_manager/ # Teleop session manager (Python) │ ├── cloudxr/ # CloudXR runtime helper (Python) │ └── python/ # Python wheel packaging (when BUILD_PYTHON_BINDINGS=ON) + ├── src/python/isaaccapture/ + │ └── ovxr_session_manager/ # OVXR session ownership and presentation ├── src/plugins/ # Built when BUILD_PLUGINS=ON │ ├── plugin_utils/ │ ├── controller_synthetic_hands/ @@ -41,6 +43,8 @@ The layout below reflects the actual ``CMakeLists.txt`` hierarchy (root ``CMakeL │ ├── manus/ │ ├── oak/ # When BUILD_PLUGIN_OAK_CAMERA=ON | └── haptikos/ + ├── tests/python/core/ + │ └── ovxr_session_manager/ # OVXR session and presentation tests └── examples/ ├── oxr/ # OpenXR examples (C++ and Python) ├── retargeting/ diff --git a/docs/source/teleop/teleop_session.rst b/docs/source/teleop/teleop_session.rst index b48d411c47..cf97ef46a1 100644 --- a/docs/source/teleop/teleop_session.rst +++ b/docs/source/teleop/teleop_session.rst @@ -24,6 +24,15 @@ manage them directly: #. Runs retargeting pipeline with automatic updates #. Handles cleanup via RAII +If resource cleanup fails during entry rollback or exit, ``cleanup_failed`` +becomes true. Hosts lending OpenXR handles must retain their renderer/session +while this is true. If stopping the retarget worker was interrupted or failed, +calling ``__exit__`` again retries that stop before releasing any resources. +Successful retry clears the failure flag. Failures while unwinding the resource +stack remain terminal: later entry, stepping and exit raise the stored error. +An entry failure with successful rollback still permits reuse; the original +setup error remains primary in either case. + Quick Start ----------- @@ -172,8 +181,13 @@ Per application frame in pipelined mode, ``step()`` submits the current request and returns the latest completed ``RetargeterIO``. On a normal hit this is often the previous application frame's output. If that returned output came from request N-1, then ``session.last_context`` also comes from N-1: tracker data, -control messages, execution events, outputs, and context stay self-consistent -for the completed frame that is returned. Use ``mode="sync"`` when the +control messages, execution state, outputs, and context stay self-consistent +for the completed frame that is returned. Reset is an exception: a reset edge +from replaced pending work or an earlier completed frame remains latched until +a successful ``step()`` return delivers it through ``last_context``. Repeated +undelivered reset pulses collapse to one idempotent reset. Reading the latest +worker result does not consume that edge, and a failed output snapshot does +not acknowledge it. This is not command-history replay. Use ``mode="sync"`` when the application needs exact current-frame sample/compute/return behavior. The return type is unchanged; inspect ``session.last_step_info`` for age and diff --git a/pyproject.toml b/pyproject.toml index ceec1e4ddb..c88d5779ec 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -50,6 +50,9 @@ dev = [ ui = [ "imgui[glfw]>=2.0.0", ] +ovxr = [ + "ovxr==0.1.0.5842", +] retargeters = [ "dex-retargeting>=0.4.6,<0.6.0", "scipy>=1.15.0", diff --git a/src/compat/pyproject.toml b/src/compat/pyproject.toml index 25f6433c6b..9bdb3894f0 100644 --- a/src/compat/pyproject.toml +++ b/src/compat/pyproject.toml @@ -22,6 +22,7 @@ Homepage = "https://github.com/NVIDIA/IsaacCapture" [project.optional-dependencies] dev = ["isaaccapture[dev]>=1.6.dev0,<1.9"] ui = ["isaaccapture[ui]>=1.6.dev0,<1.9"] +ovxr = ["isaaccapture[ovxr]>=1.6.dev0,<1.9"] retargeters = ["isaaccapture[retargeters]>=1.6.dev0,<1.9"] retargeters-lite = ["isaaccapture[retargeters-lite]>=1.6.dev0,<1.9"] grounding = ["isaaccapture[grounding]>=1.6.dev0,<1.9"] diff --git a/src/core/python/CMakeLists.txt b/src/core/python/CMakeLists.txt index dea5a3baff..b2edfa611f 100644 --- a/src/core/python/CMakeLists.txt +++ b/src/core/python/CMakeLists.txt @@ -77,6 +77,7 @@ add_custom_target(python_package ALL COMMAND ${CMAKE_COMMAND} -E copy "${CMAKE_CURRENT_SOURCE_DIR}/requirements.txt" "${CMAKE_BINARY_DIR}/python_package/$/" COMMAND ${CMAKE_COMMAND} -E copy "${CMAKE_CURRENT_SOURCE_DIR}/requirements-dev.txt" "${CMAKE_BINARY_DIR}/python_package/$/" COMMAND ${CMAKE_COMMAND} -E copy "${CMAKE_CURRENT_SOURCE_DIR}/requirements-ui.txt" "${CMAKE_BINARY_DIR}/python_package/$/" + COMMAND ${CMAKE_COMMAND} -E copy "${CMAKE_CURRENT_SOURCE_DIR}/requirements-ovxr.txt" "${CMAKE_BINARY_DIR}/python_package/$/" COMMAND ${CMAKE_COMMAND} -E copy "${CMAKE_CURRENT_SOURCE_DIR}/requirements-cloudxr.txt" "${CMAKE_BINARY_DIR}/python_package/$/" COMMAND ${CMAKE_COMMAND} -E copy "${CMAKE_CURRENT_SOURCE_DIR}/requirements-retargeters.txt" "${CMAKE_BINARY_DIR}/python_package/$/" COMMAND ${CMAKE_COMMAND} -E copy "${CMAKE_CURRENT_SOURCE_DIR}/requirements-retargeters-lite.txt" "${CMAKE_BINARY_DIR}/python_package/$/" diff --git a/src/core/python/pyproject.toml.in b/src/core/python/pyproject.toml.in index e9ced4db91..05eec7faf9 100644 --- a/src/core/python/pyproject.toml.in +++ b/src/core/python/pyproject.toml.in @@ -31,6 +31,7 @@ dynamic = ["dependencies", "optional-dependencies"] dependencies = {file = ["requirements.txt"]} optional-dependencies.dev = {file = ["requirements-dev.txt"]} optional-dependencies.ui = {file = ["requirements-ui.txt"]} +optional-dependencies.ovxr = {file = ["requirements-ovxr.txt"]} optional-dependencies.retargeters = {file = ["requirements-retargeters.txt"]} optional-dependencies.retargeters-lite = {file = ["requirements-retargeters-lite.txt"]} optional-dependencies.grounding = {file = ["requirements-grounding.txt"]} diff --git a/src/core/python/requirements-ovxr.txt b/src/core/python/requirements-ovxr.txt new file mode 100644 index 0000000000..735767ae6b --- /dev/null +++ b/src/core/python/requirements-ovxr.txt @@ -0,0 +1,5 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# Application overrides: docs/source/getting_started/ovxr.rst. +ovxr==0.1.0.5842 diff --git a/src/python/isaaccapture/__init__.py b/src/python/isaaccapture/__init__.py index e9b3c07edb..4a63b9a2d9 100644 --- a/src/python/isaaccapture/__init__.py +++ b/src/python/isaaccapture/__init__.py @@ -60,6 +60,7 @@ def _reserve_closed_standard_fds() -> list[int]: deviceio_trackers, logging_config, oxr, + ovxr_session_manager, plugin_manager, schema, teleop_session_manager, @@ -77,6 +78,7 @@ def _reserve_closed_standard_fds() -> list[int]: "deviceio_session", "deviceio", "oxr", + "ovxr_session_manager", "plugin_manager", "schema", "teleop_session_manager", diff --git a/src/python/isaaccapture/ovxr_session_manager/__init__.py b/src/python/isaaccapture/ovxr_session_manager/__init__.py new file mode 100644 index 0000000000..8528115853 --- /dev/null +++ b/src/python/isaaccapture/ovxr_session_manager/__init__.py @@ -0,0 +1,24 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""OVXR presentation over application-owned rendering resources.""" + +from .ovxr_session import ( + OvxrSessionBinding, + OvxrSessionConfig, + OvxrApplicationInterface, + OvxrSession, + StageOrdinalLedger, +) + +__all__ = [ + "OvxrSessionBinding", + "OvxrSessionConfig", + "OvxrApplicationInterface", + "OvxrSession", + "StageOrdinalLedger", +] + +from .interfaces import OvxrScenePublication, OvxrResourceOwner + +__all__ += ["OvxrScenePublication", "OvxrResourceOwner"] diff --git a/src/python/isaaccapture/ovxr_session_manager/frame_pump.py b/src/python/isaaccapture/ovxr_session_manager/frame_pump.py new file mode 100644 index 0000000000..f13eba25bc --- /dev/null +++ b/src/python/isaaccapture/ovxr_session_manager/frame_pump.py @@ -0,0 +1,327 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Owner-thread OVXR frame processing and native resource retirement.""" + +from __future__ import annotations +import logging +import threading +from collections.abc import Callable, Iterable +from .interfaces import OvxrApplicationInterface, OvxrSessionBinding, OvxrSessionConfig +from .runtime_owner import CloudXRRuntimeOwner + +_LOGGER = logging.getLogger(__name__) + + +def _import_ovxr(): + import ovxr + + return ovxr + + +class OvxrFramePump: + """Own native state; the session supplies scheduling and shutdown signals.""" + + def __init__( + self, + host: OvxrApplicationInterface, + config: OvxrSessionConfig, + runtime: CloudXRRuntimeOwner, + *, + shutdown_requested: Callable[[], bool], + request_shutdown: Callable[[], None], + required_openxr_extensions: Iterable[str], + max_blank_frames: int, + ) -> None: + self._host = host + self._config = config + self._runtime = runtime + self._shutdown_requested = shutdown_requested + self.request_shutdown = request_shutdown + self._required_openxr_extensions = tuple(required_openxr_extensions) + self._max_blank_frames = max_blank_frames + self._context = None + self._binding: OvxrSessionBinding | None = None + self._pending_plan = None + self._context_lifecycle_lock = threading.Lock() + self._last_ordinal: int | None = None + self._first_ordinal_validated = False + + def attach(self, capability: object) -> None: + self._binding = self._host.attach_on_render_thread(capability) + + @staticmethod + def validate_provider() -> None: + ovxr = _import_ovxr() + required_phase_methods = ( + "begin_frame", + "bind_frame", + "submit_frame", + "cancel_frame", + "get_vulkan_handles", + "set_frame_render_products", + ) + missing = [ + method + for method in required_phase_methods + if not callable(getattr(ovxr.Context, method, None)) + ] + if missing: + raise RuntimeError( + f"OVXR session is missing required Context methods: {', '.join(missing)}" + ) + + def create_context(self) -> None: + assert self._binding is not None + self._runtime.start() + if self._shutdown_requested(): + return + ovxr = _import_ovxr() + extensions = list( + dict.fromkeys( + ( + *self._config.extra_openxr_extensions, + *self._required_openxr_extensions, + ) + ) + ) + with self._host.publication_guard(): + desc = ovxr.ContextDesc( + ovrtx_renderer=self._binding.renderer, + ovstage=self._binding.stage, + application_name=self._config.application_name, + display_mode=ovxr.DisplayMode.STEREO, + session_mode={"ar": ovxr.SessionMode.AR, "vr": ovxr.SessionMode.VR}[ + self._config.session_mode + ], + render_products=list(self._config.render_products), + camera_paths=list(self._config.camera_paths), + create_render_products=self._config.create_render_products, + tracking_anchor_path=self._config.tracking_anchor_path, + floor_aligned_anchor=self._config.floor_aligned_anchor, + up_axis=getattr(ovxr.StageUpAxis, self._config.up_axis), + meters_per_unit=self._config.meters_per_unit, + depth_near_z=self._config.depth_near_z, + depth_far_z=self._config.depth_far_z, + scene_info_policy=ovxr.SceneInfoPolicy.AUTO, + extra_openxr_extensions=extensions, + host_wait_render_output=self._config.host_wait_render_output, + ordinal_mode=ovxr.StageOrdinalMode.APPLICATION, + ) + with self._host.owner_lane(): + self._context = ovxr.Context(desc) + + def request_exit(self) -> None: + """Wake an in-flight OpenXR frame wait without stopping CloudXR.""" + # Startup failure may already be retiring the context on its owner. + # Do not block the caller before its bounded worker join. + if not self._context_lifecycle_lock.acquire(blocking=False): + return + try: + context = self._context + if context is not None: + context.request_exit() + finally: + self._context_lifecycle_lock.release() + + def bootstrap(self) -> None: + ovxr = _import_ovxr() + handles = self.external_handles() + for _ in range(self._max_blank_frames): + if handles is not None: + return + result, _frame = self._context.submit_openxr_blank_frame() + if result not in ( + ovxr.Result.SUCCESS, + ovxr.Result.TIMEOUT, + ovxr.Result.NOT_READY, + ): + raise RuntimeError( + f"OVXR blank-frame bootstrap failed: {ovxr.result_name(result)}" + ) + handles = self.external_handles() + + def external_handles(self): + handles = self._context.get_openxr_handles() + if handles is None: + return None + from isaaccapture.oxr import OpenXRSessionHandles + + return OpenXRSessionHandles( + handles.xr_instance, + handles.xr_session, + handles.xr_space, + handles.xr_get_instance_proc_addr, + ) + + def render(self, ordinal: int) -> bool: + """Return whether OpenXR began a frame and supplied its pacing.""" + ovxr = _import_ovxr() + result, plan = self._context.begin_frame() + self._last_ordinal = ordinal + if result not in ( + ovxr.Result.SUCCESS, + ovxr.Result.TIMEOUT, + ovxr.Result.NOT_READY, + ): + raise RuntimeError( + f"OVXR begin_frame failed: {ovxr.result_name(result)}: {self._context.last_error}" + ) + if result != ovxr.Result.SUCCESS: + return False + active_plan = plan + self._pending_plan = plan + frame_error: BaseException | None = None + try: + with self._host.publication_guard(): + if not self._first_ordinal_validated: + self._host.validate_first_published_scene(ordinal) + self._first_ordinal_validated = True + if self._binding.prepare_frame_scene is not None: + self._binding.prepare_frame_scene() + frame_ordinal = self._binding.ordinal_ledger.reserve() + try: + result, sealed = self._context.bind_frame( + active_plan, 0.0, stage_ordinal=frame_ordinal + ) + except BaseException as bind_error: + try: + self._binding.ordinal_ledger.resolve_external( + frame_ordinal, expected_published=None + ) + except BaseException as reconciliation_error: + raise bind_error from reconciliation_error + raise + self._binding.ordinal_ledger.resolve_external( + frame_ordinal, + expected_published=sealed, + ) + if result not in ( + ovxr.Result.SUCCESS, + ovxr.Result.TIMEOUT, + ovxr.Result.NOT_READY, + ): + raise RuntimeError( + f"OVXR bind_frame failed: {ovxr.result_name(result)}: {self._context.last_error}" + ) + if result != ovxr.Result.SUCCESS: + return True + result, _frame = self._context.submit_frame(active_plan) + if result not in ( + ovxr.Result.SUCCESS, + ovxr.Result.TIMEOUT, + ovxr.Result.NOT_READY, + ): + raise RuntimeError( + f"OVXR submit_frame failed: {ovxr.result_name(result)}: {self._context.last_error}" + ) + active_plan = None + self._pending_plan = None + except BaseException as error: + frame_error = error + raise + finally: + if active_plan is not None: + try: + self._context.cancel_frame(active_plan) + self._pending_plan = None + except BaseException as cleanup_error: + if frame_error is None: + raise + frame_error.add_note(f"Frame retirement failed: {cleanup_error}") + return True + + def render_reused(self) -> bool: + if self._last_ordinal is None: + return False + return self.render(self._last_ordinal) + + def close(self) -> None: + first_error: BaseException | None = None + try: + with self._host.publication_guard(): + self._host.close_publication() + except BaseException as exc: + first_error = exc + if self._pending_plan is not None: + try: + self._context.cancel_frame(self._pending_plan) + except BaseException as exc: + # Native destroy remains responsible for the pending frame. + if first_error is None: + first_error = exc + else: + self._pending_plan = None + context_retired = self._context is None + if self._context is not None: + assert self._binding is not None + with self._context_lifecycle_lock: + try: + # The close caller normally requested this already to wake + # xrWaitFrame. Repeat idempotently on the owner thread so + # worker-failure teardown follows the same graceful path. + try: + self._context.request_exit() + except BaseException as exc: + if first_error is None: + first_error = exc + try: + with self._host.publication_guard(): + ordinal = self._binding.ordinal_ledger.reserve() + try: + with self._host.owner_lane(): + try: + self._context.destroy(stage_ordinal=ordinal) + except _import_ovxr().OvxrError: + # OVXR raises its result error + # only after native destroy consumed the context. + # Pre-call/FFI errors still quarantine borrowed owners. + context_retired = True + raise + else: + context_retired = True + except BaseException as destroy_error: + try: + self._binding.ordinal_ledger.resolve_external( + ordinal, expected_published=None + ) + except BaseException as reconciliation_error: + raise destroy_error from reconciliation_error + raise + self._binding.ordinal_ledger.resolve_external( + ordinal, expected_published=None + ) + except BaseException as exc: + if first_error is None: + first_error = exc + elif exc is not first_error: + first_error.add_note(f"OVXR retirement failed: {exc}") + finally: + if context_retired: + self._context = None + self._pending_plan = None + if not context_retired: + # A pre-call/FFI failure does not prove native consumption. Keep its + # renderer, Stage and runtime alive; never retry an ambiguous destroy. + assert first_error is not None + raise first_error + for close in ( + self._runtime.close, + self._host.close_on_render_thread, + self._host.close_environment_on_render_thread, + ): + try: + close() + except BaseException as exc: + # Finish independent cleanup, then propagate the first failure. + if first_error is None: + first_error = exc + elif exc is not first_error: + first_error.add_note(f"Dependent cleanup failed: {exc}") + if first_error is not None: + raise first_error + + def retry_runtime_stop(self) -> None: + """Retry service cleanup only after native borrowers have retired.""" + if self._context is None: + self._runtime.close() diff --git a/src/python/isaaccapture/ovxr_session_manager/interfaces.py b/src/python/isaaccapture/ovxr_session_manager/interfaces.py new file mode 100644 index 0000000000..48783edbbe --- /dev/null +++ b/src/python/isaaccapture/ovxr_session_manager/interfaces.py @@ -0,0 +1,143 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Application publication, resource ownership and frame-consumer contracts.""" + +from __future__ import annotations +import math +from abc import abstractmethod +from collections.abc import Callable, Hashable +from contextlib import AbstractContextManager +from dataclasses import dataclass +from typing import Any, Literal, Protocol + + +class OvxrScenePublication(Protocol): + """Application-thread publication and scene reads.""" + + @abstractmethod + def publication_guard(self) -> AbstractContextManager[None]: + raise NotImplementedError + + @abstractmethod + def publish_complete_scene(self, publication_id: Hashable) -> int: + raise NotImplementedError + + @abstractmethod + def validate_first_published_scene(self, ordinal: int) -> None: + raise NotImplementedError + + @abstractmethod + def get_target_frame_T_world(self, prim_path: str) -> Any: + raise NotImplementedError + + @abstractmethod + def close_publication(self) -> None: + raise NotImplementedError + + +class OvxrResourceOwner(Protocol): + """Transfer and retirement of application-owned rendering resources.""" + + @abstractmethod + def offer_render_owner_transfer(self) -> object: + raise NotImplementedError + + @abstractmethod + def attach_on_render_thread(self, capability: object) -> OvxrSessionBinding: + raise NotImplementedError + + @abstractmethod + def owner_lane(self) -> AbstractContextManager[None]: + raise NotImplementedError + + @abstractmethod + def close_on_render_thread(self) -> None: + raise NotImplementedError + + @abstractmethod + def close_environment_on_render_thread(self) -> None: + raise NotImplementedError + + @abstractmethod + def abort(self) -> None: + raise NotImplementedError + + +class OvxrApplicationInterface(OvxrScenePublication, OvxrResourceOwner, Protocol): + """Combined application contract accepted by OvxrSession.""" + + +class StageOrdinalLedger(Protocol): + """Application-owned reservations for explicit OVXR Stage writes.""" + + @abstractmethod + def reserve(self) -> int: + raise NotImplementedError + + @abstractmethod + def resolve_external( + self, ordinal: int, *, expected_published: bool | None + ) -> bool: + raise NotImplementedError + + @abstractmethod + def abandon(self, ordinal: int) -> None: + raise NotImplementedError + + +@dataclass(frozen=True) +class OvxrSessionBinding: + """Exact application-owned objects borrowed by one OVXR context.""" + + renderer: object + stage: object + ordinal_ledger: StageOrdinalLedger + prepare_frame_scene: Callable[[], None] | None = None + + def __post_init__(self) -> None: + if self.renderer is None or self.stage is None or self.ordinal_ledger is None: + raise ValueError( + "OVXR session requires renderer, Stage, and ordinal ledger" + ) + + +@dataclass +class OvxrSessionConfig: + """Configure an application-owned renderer/Stage XR session.""" + + camera_paths: tuple[str, ...] + application_name: str = "isaacteleop-ovxr-session" + session_mode: Literal["ar", "vr"] = "vr" + cloudxr_env_file: str | None = None + auto_launch_cloudxr: bool = True + accept_eula: bool = False + render_products: tuple[str, ...] = () + create_render_products: bool = True + up_axis: Literal["Y", "Z"] = "Z" + meters_per_unit: float = 1.0 + depth_near_z: float = 0.0 + depth_far_z: float = 0.0 + tracking_anchor_path: str = "" + floor_aligned_anchor: bool | None = None + host_wait_render_output: bool = False + extra_openxr_extensions: tuple[str, ...] = () + + def __post_init__(self) -> None: + if self.up_axis not in ("Y", "Z"): + raise ValueError("up_axis must be 'Y' or 'Z'") + if self.session_mode not in ("ar", "vr"): + raise ValueError("session_mode must be 'ar' or 'vr'") + if not math.isfinite(self.meters_per_unit) or self.meters_per_unit <= 0: + raise ValueError("meters_per_unit must be finite and positive") + if (self.depth_near_z, self.depth_far_z) != (0.0, 0.0) and ( + not math.isfinite(self.depth_near_z) + or not math.isfinite(self.depth_far_z) + or self.depth_near_z <= 0.0 + or self.depth_far_z <= self.depth_near_z + ): + raise ValueError( + "depth_near_z and depth_far_z must both be zero or finite with 0 < near < far" + ) + if len(self.camera_paths) != 2 or any(not path for path in self.camera_paths): + raise ValueError("OVXR stereo requires exactly two non-empty camera paths") diff --git a/src/python/isaaccapture/ovxr_session_manager/ovxr_session.py b/src/python/isaaccapture/ovxr_session_manager/ovxr_session.py new file mode 100644 index 0000000000..efd567ae44 --- /dev/null +++ b/src/python/isaaccapture/ovxr_session_manager/ovxr_session.py @@ -0,0 +1,390 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Public scheduling and lifetime facade for an application-owned OVXR scene.""" + +from __future__ import annotations +import logging +import math +import threading +from enum import Enum, auto +from collections.abc import Callable, Hashable, Iterable +from typing import Any +from .frame_pump import OvxrFramePump +from .runtime_owner import CloudXRRuntime, CloudXRRuntimeOwner +from .interfaces import ( + OvxrApplicationInterface as OvxrApplicationInterface, + StageOrdinalLedger as StageOrdinalLedger, + OvxrSessionBinding as OvxrSessionBinding, + OvxrSessionConfig as OvxrSessionConfig, + OvxrScenePublication as OvxrScenePublication, + OvxrResourceOwner as OvxrResourceOwner, +) + +_WORKER_SHUTDOWN_TIMEOUT_SECONDS = 30.0 +_LOGGER = logging.getLogger(__name__) + + +class _Phase(Enum): + NEW = auto() + STARTING = auto() + READY = auto() + CLOSED = auto() + + +class OvxrSession: + """Schedule one frame owner and close it after application borrowers stop.""" + + def __init__( + self, + host: OvxrApplicationInterface, + config: OvxrSessionConfig, + *, + required_openxr_extensions: Iterable[str] = (), + max_blank_frames: int = 1, + runtime_launcher_factory: Callable[..., CloudXRRuntime] | None = None, + ) -> None: + if max_blank_frames < 0: + raise ValueError("max_blank_frames must be non-negative") + self._host = host + self._worker: threading.Thread | None = None + self._worker_native_thread_id: int | None = None + self._worker_condition = threading.Condition() + self._worker_bootstrap_ready = False + self._worker_error: BaseException | None = None + self._shutdown_requested = threading.Event() + self._borrowers_quiesced = threading.Event() + self._retirement_complete = False + self._worker_pending_ordinal: int | None = None + self._worker_publication_requested = True + self._worker_last_published_ordinal: int | None = None + self._worker_handles = None + self._phase = _Phase.NEW + self._borrowers_exposed = False + self._pump = OvxrFramePump( + host, + config, + CloudXRRuntimeOwner(config, runtime_launcher_factory), + shutdown_requested=lambda: self.shutdown_requested, + request_shutdown=self.request_shutdown, + required_openxr_extensions=required_openxr_extensions, + max_blank_frames=max_blank_frames, + ) + + def start(self, timeout_s: float = 30.0) -> bool: + """Transfer the borrowed pair and return whether XR startup completed. + + ``timeout_s`` must be finite, positive and within ``threading.TIMEOUT_MAX``. + It bounds host attachment readiness, not the subsequent peer-gated XR wait. + Expiry and interruption request cancellation; an active attachment must finish + before the owner thread can retire its borrowed resources. Once start + succeeds or input handles are borrowed, close() is required; use a context + manager or try/finally. Failed startup without borrowers retires itself. + """ + self._require_open() + if self._worker is not None: + raise RuntimeError("OVXR session is already started") + if not math.isfinite(timeout_s) or not 0 < timeout_s <= threading.TIMEOUT_MAX: + raise ValueError( + "OVXR bootstrap timeout must be finite, positive and within threading.TIMEOUT_MAX" + ) + self._pump.validate_provider() + capability = self._host.offer_render_owner_transfer() + + self._phase = _Phase.STARTING + + def run() -> None: + try: + self._worker_native_thread_id = threading.get_native_id() + self._pump.attach(capability) + with self._worker_condition: + self._worker_bootstrap_ready = True + self._worker_condition.notify_all() + if self._shutdown_requested.is_set(): + return + self._pump.create_context() + with self._worker_condition: + if self._shutdown_requested.is_set(): + return + self._pump.bootstrap() + with self._worker_condition: + if self._shutdown_requested.is_set(): + return + self._worker_handles = self._pump.external_handles() + self._phase = _Phase.READY + self._worker_condition.notify_all() + while True: + with self._worker_condition: + if self._shutdown_requested.is_set(): + break + pending_ordinal = self._worker_pending_ordinal + self._worker_pending_ordinal = None + if pending_ordinal is not None: + self._worker_publication_requested = True + self._worker_condition.notify_all() + if pending_ordinal is not None: + frame_began = self._pump.render(pending_ordinal) + else: + frame_began = self._pump.render_reused() + # A peer can become ready after the bounded blank-frame + # bootstrap. Publish its input handles from the owning + # render thread once the session has actually appeared. + if self._worker_handles is None: + handles = self._pump.external_handles() + if handles is not None: + with self._worker_condition: + self._worker_handles = handles + self._worker_condition.notify_all() + # A begun frame is paced by OpenXR even when simulation is + # paused. Back off only without a scene or a ready session; + # sleeping after a reused frame misses runtime deadlines. + if not frame_began: + with self._worker_condition: + if ( + not self._shutdown_requested.is_set() + and self._worker_pending_ordinal is None + ): + self._worker_condition.wait(0.01) + except BaseException as exc: + # Thread failures cannot reach the caller automatically. Preserve + # interrupts too, for caller-side re-raising and cleanup. + self._record_worker_failure(exc) + finally: + # Once handles are exposed, only the host can acknowledge that + # its application step and DeviceIO borrowers have stopped. + with self._worker_condition: + borrowers_exposed = self._borrowers_exposed + if borrowers_exposed: + self._borrowers_quiesced.wait() + try: + self._pump.close() + self._retirement_complete = True + except BaseException as exc: + # Failed consumer retirement retains its borrowed owners. + # Report it to close() even when rendering itself succeeded. + self._record_worker_failure(exc) + finally: + with self._worker_condition: + self._phase = _Phase.CLOSED + self._worker_condition.notify_all() + + try: + self._worker = threading.Thread( + target=run, name="isaacteleop-ovxr-render", daemon=False + ) + self._worker.start() + except Exception: + # No owner thread accepted the borrowed pair; rollback stays on the application thread. + self._worker = None + self._phase = _Phase.CLOSED + try: + self._host.abort() + except Exception: + _LOGGER.exception( + "Cleanup after failed render-worker start also failed" + ) + raise + with self._worker_condition: + try: + bootstrap_ready = self._worker_condition.wait_for( + lambda: ( + self._worker_bootstrap_ready + or self._worker_error is not None + or self._shutdown_requested.is_set() + ), + timeout_s, + ) + if not bootstrap_ready: + raise TimeoutError( + "OVXR session worker bootstrap did not become ready" + ) + self._raise_worker_error() + self._worker_condition.wait_for( + lambda: ( + self._phase is _Phase.READY + or self._worker_error is not None + or self._shutdown_requested.is_set() + ) + ) + self._raise_worker_error() + if self._phase is not _Phase.READY or self._shutdown_requested.is_set(): + return False + self._borrowers_exposed = True + return True + except BaseException: + self._request_worker_shutdown() + raise + + def request_render( + self, publication_id: Hashable, *, force: bool = False + ) -> int | None: + """Publish the newest host generation when the render worker requests it. + + ``force`` preserves discontinuous reset/lifecycle generations even when + an ordinary render demand is already satisfied. + + Return the published ordinal on admission, or ``None`` if demand is + satisfied or shutdown has been requested. The application must retry + its latest dirty generation while running, including while paused; + the worker cannot read mutable host state. Worker failures still raise. + """ + with self._worker_condition: + self._raise_worker_error() + self._require_open() + # UI/signal shutdown can arrive during the current application step. + # Decline its late publication, but never hide a latched worker error. + if self._shutdown_requested.is_set(): + return None + if self._worker is None: + raise RuntimeError("OVXR session is not accepting requests") + if ( + not force + and self._worker_last_published_ordinal is not None + and not self._worker_publication_requested + ): + return None + self._worker_publication_requested = False + + try: + with self._host.publication_guard(): + ordinal = self._host.publish_complete_scene(publication_id) + self._worker_last_published_ordinal = ordinal + with self._worker_condition: + if self._shutdown_requested.is_set(): + self._raise_worker_error() + return None + self._worker_pending_ordinal = ordinal + self._worker_condition.notify_all() + self._raise_worker_error() + return ordinal + except BaseException: + with self._worker_condition: + if not self._shutdown_requested.is_set(): + self._worker_publication_requested = True + self._worker_condition.notify_all() + raise + + def external_oxr_handles(self): + """Return typed handles after peer-gated context startup, otherwise ``None``.""" + with self._worker_condition: + self._raise_worker_error() + self._require_open() + if self._shutdown_requested.is_set(): + return None + if self._worker_handles is not None: + self._borrowers_exposed = True + return self._worker_handles + + def get_target_frame_T_world(self, prim_path: str) -> Any: + """Delegate a current target transform read to the simulation host.""" + self._require_open() + return self._host.get_target_frame_T_world(prim_path) + + def close(self) -> None: + """Acknowledge stopped app/DeviceIO borrowers, then drain the owner thread. + + The host must finish its current application step and stop every input + borrower before calling this method, including after worker failure. + A terminal worker is not proof of successful resource retirement; inspect + ``retirement_complete`` and preserve any raised first error. After + requesting exit, wait up to 30 seconds for the worker; a timeout leaves + its borrowed resources and runtime alive for the worker to retire. + """ + if self._phase is _Phase.CLOSED: + # Service.stop supports retry; native destruction does not promise it. + try: + self._pump.retry_runtime_stop() + except Exception as error: + if self._worker_error is not None: + raise self._worker_error from error + raise + self._raise_worker_error() + return + if self._worker is None: + self._host.abort() + self._phase = _Phase.CLOSED + self._retirement_complete = True + return + self._request_worker_shutdown() + wake_error: BaseException | None = None + try: + self._pump.request_exit() + except BaseException as exc: + wake_error = exc + self._borrowers_quiesced.set() + # Slow retirement is not proof of a blocked OpenXR call. Keep the + # runtime alive until its borrowers retire, even when this wait expires. + self._worker.join(timeout=_WORKER_SHUTDOWN_TIMEOUT_SECONDS) + if self._worker.is_alive(): + raise TimeoutError("OVXR session worker did not stop") from wake_error + worker_error = self._worker_error + if worker_error is not None: + if wake_error is not None and worker_error is not wake_error: + raise worker_error from wake_error + raise worker_error + if wake_error is not None: + raise wake_error + + @property + def owns_environment(self) -> bool: + """Whether the worker accepted host-environment teardown ownership.""" + return self._worker is not None + + @property + def shutdown_requested(self) -> bool: + """Whether UI close or explicit teardown requested application shutdown.""" + return self._shutdown_requested.is_set() + + def request_shutdown(self) -> None: + """Ask the embedding application to leave its active simulation loop.""" + self._request_worker_shutdown() + + def wait_for_shutdown(self, timeout_s: float) -> bool: + """Wait interruptibly on the same event used by signals and mirror close.""" + return self._shutdown_requested.wait(timeout_s) + + @property + def retirement_complete(self) -> bool: + """Whether all owned resources retired successfully, not merely stopped.""" + with self._worker_condition: + return self._retirement_complete + + @property + def render_worker_bootstrap_ready(self) -> bool: + """Whether the borrowed-pair owner transfer completed.""" + with self._worker_condition: + return self._worker_bootstrap_ready + + @property + def render_worker_native_thread_id(self) -> int | None: + """Native ID of the sole OVXR owner thread, when started.""" + return self._worker_native_thread_id + + def _request_worker_shutdown(self) -> None: + with self._worker_condition: + if self._phase is _Phase.CLOSED or self._shutdown_requested.is_set(): + return + self._shutdown_requested.set() + self._worker_condition.notify_all() + + def _record_worker_failure(self, error: BaseException) -> None: + self._shutdown_requested.set() + with self._worker_condition: + if self._worker_error is None: + self._worker_error = error + self._worker_condition.notify_all() + + def _raise_worker_error(self) -> None: + if self._worker_error is not None: + raise self._worker_error + + def _require_open(self) -> None: + if self._phase is _Phase.CLOSED: + raise RuntimeError("OVXR session is closed") + + def __enter__(self) -> OvxrSession: + self._require_open() + return self + + def __exit__(self, *_exc: object) -> None: + self.close() diff --git a/src/python/isaaccapture/ovxr_session_manager/runtime_owner.py b/src/python/isaaccapture/ovxr_session_manager/runtime_owner.py new file mode 100644 index 0000000000..5446067fef --- /dev/null +++ b/src/python/isaaccapture/ovxr_session_manager/runtime_owner.py @@ -0,0 +1,51 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Lifetime of the CloudXR runtime launched by an OVXR session.""" + +from __future__ import annotations +from collections.abc import Callable +from pathlib import Path +from typing import Protocol +from .interfaces import OvxrSessionConfig + + +class CloudXRRuntime(Protocol): + """Owned runtime returned by a session's launcher factory.""" + + def stop(self) -> None: ... + + +def _create_cloudxr_launcher(env_config: str, *, accept_eula: bool = False): + from isaaccapture.cloudxr import CloudXRLauncher + + return CloudXRLauncher( + install_dir=str(Path.home() / ".cloudxr"), + env_config=env_config, + accept_eula=accept_eula, + run_embedded=True, + ) + + +class CloudXRRuntimeOwner: + """Stop only the runtime this owner launched; preserve failed stops for retry.""" + + def __init__( + self, + config: OvxrSessionConfig, + factory: Callable[..., CloudXRRuntime] | None = None, + ) -> None: + self._config = config + self._factory = factory or _create_cloudxr_launcher + self._launcher: CloudXRRuntime | None = None + + def start(self) -> None: + if self._config.cloudxr_env_file and self._config.auto_launch_cloudxr: + self._launcher = self._factory( + self._config.cloudxr_env_file, accept_eula=self._config.accept_eula + ) + + def close(self) -> None: + if self._launcher is not None: + self._launcher.stop() + self._launcher = None diff --git a/src/python/isaaccapture/teleop_session_manager/async_retarget_runner.py b/src/python/isaaccapture/teleop_session_manager/async_retarget_runner.py index afbe580553..e04fda55dc 100644 --- a/src/python/isaaccapture/teleop_session_manager/async_retarget_runner.py +++ b/src/python/isaaccapture/teleop_session_manager/async_retarget_runner.py @@ -7,7 +7,7 @@ import copy from collections import deque -from dataclasses import dataclass +from dataclasses import dataclass, replace import threading import time from collections.abc import Callable @@ -38,6 +38,7 @@ class StepRequest: graph_time: GraphTime | None execution_events: ExecutionEvents | None submitted_time_s: float + reset_requested: bool = False @dataclass(frozen=True) @@ -51,6 +52,7 @@ class RetargetFrame: started_time_s: float completed_time_s: float compute_duration_s: float + reset_generation: int = 0 class AsyncRetargetWorkerError(RuntimeError): @@ -152,6 +154,8 @@ def __init__(self, step_fn: StepExecutor, cfg: RetargetingExecutionConfig): self._cond = threading.Condition() self._pending: StepRequest | None = None self._published: RetargetFrame | None = None + self._reset_generation = 0 + self._acknowledged_reset_generation = 0 self._dropped_submissions = 0 self._last_submission_time_s: float | None = None self._submission_count = 0 @@ -277,6 +281,13 @@ def submit(self, request: StepRequest) -> int: if self._pending is not None: dropped = 1 self._dropped_submissions += 1 + if self._pending.reset_requested or ( + self._pending.execution_events is not None + and self._pending.execution_events.reset + ): + # Reset is an idempotent edge, not replaceable latest state. + # Do not override the new request's live control decoding. + request = replace(request, reset_requested=True) self._pending = request self._record_submission_locked(request.submitted_time_s) self._cond.notify_all() @@ -290,7 +301,31 @@ def latest(self) -> RetargetFrame | None: with self._cond: self._raise_if_failed_locked() - return self._published + if self._published is None: + return None + context = snapshot_compute_context(self._published.context) + context.execution_events.reset = ( + self._reset_generation > self._acknowledged_reset_generation + ) + return replace( + self._published, + context=context, + reset_generation=self._reset_generation, + ) + + def acknowledge_reset(self, generation: int) -> None: + """Acknowledge delivered resets, without clearing a concurrently newer edge. + + Latest state may replace completed outputs. Pending reset pulses collapse + to one idempotent reset, but remain visible until the application has + successfully copied the returned frame. Observers do not acknowledge it. + """ + with self._cond: + if not 0 <= generation <= self._reset_generation: + raise ValueError("Unknown reset generation") + self._acknowledged_reset_generation = max( + self._acknowledged_reset_generation, generation + ) def wait_for_frame( self, @@ -355,6 +390,8 @@ def _run(self) -> None: with self._cond: self._record_compute_duration_locked(frame.compute_duration_s) + if frame.context.execution_events.reset: + self._reset_generation += 1 self._published = frame self._cond.notify_all() diff --git a/src/python/isaaccapture/teleop_session_manager/config.py b/src/python/isaaccapture/teleop_session_manager/config.py index 8307a43af9..3f018e4ecd 100644 --- a/src/python/isaaccapture/teleop_session_manager/config.py +++ b/src/python/isaaccapture/teleop_session_manager/config.py @@ -263,6 +263,8 @@ class RetargetingExecutionConfig: Pipelined mode keeps the public ``TeleopSession.step()`` return type as a normal ``RetargeterIO`` dict, but returns the latest completed retarget frame while submitting the current step request to a background worker. + Undelivered reset edges stay latched until a successful ``step()`` return; + other execution state follows the returned completed frame. """ mode: RetargetingExecutionMode | str = RetargetingExecutionMode.SYNC diff --git a/src/python/isaaccapture/teleop_session_manager/teleop_session.py b/src/python/isaaccapture/teleop_session_manager/teleop_session.py index f9c019ac7d..985319e867 100644 --- a/src/python/isaaccapture/teleop_session_manager/teleop_session.py +++ b/src/python/isaaccapture/teleop_session_manager/teleop_session.py @@ -231,12 +231,18 @@ def __init__(self, config: TeleopSessionConfig): self._active_retargeting_execution_mode: RetargetingExecutionMode | None = None # True from the start of __enter__ until __exit__ or rollback finishes. self._in_context: bool = False + self._cleanup_error: BaseException | None = None # Discover sources and external leaves from pipeline self._discover_sources() self._controller_status_trackers = self._find_controller_status_trackers() # Vendor selection is a live-only concern; reject it up front in replay. self._reject_vendor_selection_in_replay() + @property + def cleanup_failed(self) -> bool: + """Whether resource cleanup failed; reuse and borrowed-owner release are unsafe.""" + return self._cleanup_error is not None + @property def oxr_session(self) -> oxr.OpenXRSession | None: """The internal OpenXR session, or ``None`` when using external handles or after the context manager exits (read-only).""" @@ -494,7 +500,10 @@ def step( RuntimeError: If a critical DeviceIO/tracker/runtime failure occurs while updating the session. This is a fatal condition; the application is expected to terminate rather than continue. + BaseException: The stored failure if earlier resource cleanup failed. """ + if self._cleanup_error is not None: + raise self._cleanup_error if self.frame_count % 60 == 0: self._check_plugin_health() @@ -619,7 +628,8 @@ def _execute_step_request( real_time_ns=graph_time.real_time_ns, ), execution_events=ExecutionEvents( - reset=bool(execution_events.reset), + # Merge the retained reset only after decoding current controls. + reset=bool(execution_events.reset or request.reset_requested), execution_state=ExecutionState(execution_events.execution_state), ), ) @@ -827,12 +837,19 @@ def _return_pipelined_frame( """Publish metadata/context for a pipelined result and finish the step. ``last_context`` must always describe the output returned from this - call, not the request just submitted. The returned output is copied so + call, not the request just submitted; its reset flag also delivers any + unacknowledged reset edge from earlier coalesced work. The output is copied so user mutation cannot corrupt the cached latest frame; uncopyable outputs should implement ``create_snapshot()`` or use sync mode. """ returned_outputs = snapshot_retargeter_io(frame.outputs) - returned_context = snapshot_compute_context(frame.context) + # latest() already gives this caller an isolated context. The seed is + # still shared with the runner and needs its own copy. + returned_context = ( + snapshot_compute_context(frame.context) + if ran_synchronously + else frame.context + ) now = time.monotonic() returned_age_frames = submitted_frame_id - frame.frame_id frame_deadline_miss = returned_age_frames > 1 @@ -848,6 +865,8 @@ def _return_pipelined_frame( ran_synchronously=ran_synchronously, frame_deadline_miss=frame_deadline_miss, ) + if self._async_runner is not None: + self._async_runner.acknowledge_reset(frame.reset_generation) self.frame_count += 1 return returned_outputs @@ -1077,7 +1096,10 @@ def __enter__(self): Raises: RuntimeError: If the session is already active. + BaseException: The stored failure if earlier resource cleanup failed. """ + if self._cleanup_error is not None: + raise self._cleanup_error # Detect nested context-manager use; this does not make the session thread-safe. if self._in_context: raise RuntimeError("TeleopSession is already active") @@ -1096,13 +1118,22 @@ def __enter__(self): self._exit_stack = stack.pop_all() except BaseException as error: try: - # Give resources the setup error, but do not let one suppress it. stack.__exit__(type(error), error, error.__traceback__) - except BaseException: + except BaseException as cleanup_error: + # An exhausted ExitStack does not prove that native owners drained. + self._cleanup_error = error + add_note = getattr(error, "add_note", None) + if callable(add_note): + add_note( + f"TeleopSession setup rollback also failed: {cleanup_error!r}" + ) logger.exception("Failed to roll back TeleopSession setup") finally: self._status_monitor.mark_startup_failed(error) self._oxr_session = None + self.deviceio_session = None + self.plugin_managers = [] + self.plugin_contexts = [] self._in_context = False raise @@ -1400,14 +1431,25 @@ def _start_configured_plugins(self, stack: ExitStack) -> None: ) def __exit__(self, exc_type, exc_val, exc_tb): - """Exit the context - cleanup resources.""" + """Clean up resources, latching failures that prohibit reuse or borrowed-owner release.""" + if self._cleanup_error is not None and not ( + self._in_context and self._async_runner is not None + ): + raise self._cleanup_error if not self._in_context: return False runner_error = None if self._async_runner is not None: runner = self._async_runner - runner.stop() + try: + runner.stop() + except BaseException as cleanup_error: + self._cleanup_error = cleanup_error + raise + # A failed/interrupted join retains the stack. Retry is safe only + # through stop(), which must finish before any borrower is released. + self._cleanup_error = None try: runner.raise_if_failed(only_unreported=True) except BaseException as err: @@ -1429,6 +1471,9 @@ def __exit__(self, exc_type, exc_val, exc_tb): # session and stopping each plugin both reach the same native code. with capture_native_output(): self._exit_stack.__exit__(exc_type, exc_val, exc_tb) + except BaseException as cleanup_error: + self._cleanup_error = cleanup_error + raise finally: self._status_monitor.mark_stopped() # The ExitStack above closes the OpenXR session; drop our reference so the diff --git a/tests/python/core/CMakeLists.txt b/tests/python/core/CMakeLists.txt index c96d5abc4c..e9ff6b6993 100644 --- a/tests/python/core/CMakeLists.txt +++ b/tests/python/core/CMakeLists.txt @@ -4,6 +4,7 @@ add_subdirectory(schema) add_subdirectory(retargeting_engine) add_subdirectory(teleop_session_manager) +add_subdirectory(ovxr_session_manager) if(BUILD_PYTHON_BINDINGS) add_subdirectory(deviceio_trackers) diff --git a/tests/python/core/cloudxr/test_service.py b/tests/python/core/cloudxr/test_service.py index 476a58ebb4..353287b0aa 100644 --- a/tests/python/core/cloudxr/test_service.py +++ b/tests/python/core/cloudxr/test_service.py @@ -237,6 +237,25 @@ def test_wss_log_path_set_after_construction(self, tmp_path): class TestServiceStop: """Tests for CloudXRService.stop().""" + def test_stop_failure_retains_runtime_and_restores_handlers(self, tmp_path): + """Failed drain preserves the owner and allows an ordinary cleanup retry.""" + with mock_service_deps(tmp_path, ready=True) as mocks: + service = CloudXRService() + failure = RuntimeError("owned runtime did not drain") + with ( + patch.object(service, "_terminate_runtime", side_effect=failure), + patch.object(service, "_stop_wss_proxy"), + patch.object(service, "_restore_signal_handlers") as restore, + ): + with pytest.raises(RuntimeError) as caught: + service.stop() + assert caught.value is failure + restore.assert_called_once_with() + assert not service._stopping + assert service._runtime_proc is mocks["proc"] + mocks["proc"].poll.return_value = 0 + service.stop() + @_posix_only def test_stop_terminates_runtime(self, tmp_path): """stop() sends SIGTERM to the runtime process group.""" diff --git a/tests/python/core/ovxr_session_manager/CMakeLists.txt b/tests/python/core/ovxr_session_manager/CMakeLists.txt new file mode 100644 index 0000000000..7ca5d8100f --- /dev/null +++ b/tests/python/core/ovxr_session_manager/CMakeLists.txt @@ -0,0 +1,16 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +file(GLOB TEST_FILES CONFIGURE_DEPENDS + RELATIVE "${CMAKE_CURRENT_SOURCE_DIR}" + "${CMAKE_CURRENT_SOURCE_DIR}/test_*.py") + +foreach(test_file IN LISTS TEST_FILES) + get_filename_component(test_name "${test_file}" NAME_WE) + add_test( + NAME "ovxr_session_manager_${test_name}" + COMMAND uv run --python ${ISAAC_TELEOP_PYTHON_VERSION} --extra dev pytest -v --tb=short "${test_file}" + WORKING_DIRECTORY "${CMAKE_CURRENT_SOURCE_DIR}") + set_tests_properties("ovxr_session_manager_${test_name}" PROPERTIES + ENVIRONMENT "PYTHONPATH=${CMAKE_BINARY_DIR}/python_package/$") +endforeach() diff --git a/tests/python/core/ovxr_session_manager/pyproject.toml b/tests/python/core/ovxr_session_manager/pyproject.toml new file mode 100644 index 0000000000..f65b241bd5 --- /dev/null +++ b/tests/python/core/ovxr_session_manager/pyproject.toml @@ -0,0 +1,26 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +[project] +name = "isaaccapture-ovxr-session-manager-tests" +version = "0.0.0" +description = "OVXR session manager tests for Isaac Capture" +requires-python = ">=3.11,<3.14" + +[project.optional-dependencies] +dev = [ + "pytest", + "numpy", +] +test = [ + "pytest", + "numpy", +] + +[tool.pytest.ini_options] +pythonpath = ["."] +python_files = ["test_*.py"] +python_classes = ["Test*"] +python_functions = ["test_*"] +# Prevent pytest from recursing into parent directories +norecursedirs = [".git", ".venv", "build", "dist", "*.egg", "__pycache__"] diff --git a/tests/python/core/ovxr_session_manager/test_ovxr_session.py b/tests/python/core/ovxr_session_manager/test_ovxr_session.py new file mode 100644 index 0000000000..9c79e130e3 --- /dev/null +++ b/tests/python/core/ovxr_session_manager/test_ovxr_session.py @@ -0,0 +1,1667 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Focused ownership and lifecycle tests for external OVXR presentation.""" + +from __future__ import annotations +import threading +import time +from contextlib import contextmanager +from types import ModuleType, SimpleNamespace +from typing import ClassVar +import pytest +import isaaccapture.ovxr_session_manager.ovxr_session as ovxr_session +from isaaccapture.ovxr_session_manager import frame_pump, runtime_owner + + +class _FakeOvxrError(RuntimeError): + pass + + +class _WorkerAbort(BaseException): + pass + + +class _FakeContext: + operations: ClassVar[list[tuple[str, int]]] = [] + create_error: BaseException | None = None + bind_error: BaseException | None = None + destroy_error: BaseException | None = None + request_exit_error: BaseException | None = None + request_exit_callback = None + owner_lane_threads: ClassVar[set[int]] = set() + ordinal_ledger = None + bind_ordinals: ClassVar[list[int]] = [] + destroy_ordinals: ClassVar[list[int]] = [] + seal_bind_results: ClassVar[set[int]] = {0, 1} + last_desc = None + selected_products = None + + def __init__(self, desc): + assert threading.get_native_id() in self.__class__.owner_lane_threads + self.desc = desc + self.__class__.last_desc = desc + self.last_error = "" + self.bind_results = [0] + if self.__class__.create_error is not None: + raise self.__class__.create_error + self.__class__.operations.append(("create", threading.get_native_id())) + + def get_openxr_handles(self): + return SimpleNamespace( + xr_instance=11, + xr_session=22, + xr_space=33, + xr_get_instance_proc_addr=44, + ) + + def set_frame_render_products(self, products): + assert threading.get_native_id() in self.__class__.owner_lane_threads + self.__class__.selected_products = tuple(products) + + def get_vulkan_handles(self): + return SimpleNamespace() + + def submit_openxr_blank_frame(self): + return 0, SimpleNamespace() + + def begin_frame(self): + self.__class__.operations.append(("begin", threading.get_native_id())) + return 0, SimpleNamespace(step_result=0) + + def bind_frame(self, _plan, _time, *, stage_ordinal): + self.__class__.operations.append(("bind", threading.get_native_id())) + self.__class__.bind_ordinals.append(stage_ordinal) + if self.__class__.bind_error is not None: + raise self.__class__.bind_error + if len(self.bind_results) > 1: + result = self.bind_results.pop(0) + else: + result = self.bind_results[0] + if result in self.__class__.seal_bind_results: + self.__class__.ordinal_ledger.seal_external(stage_ordinal) + return result, result in self.__class__.seal_bind_results + + def submit_frame(self, _plan, *, additional_layers=()): + self.__class__.operations.append(("submit", threading.get_native_id())) + return 0, SimpleNamespace(submitted=True) + + def cancel_frame(self, _plan): + self.__class__.operations.append(("cancel", threading.get_native_id())) + + def request_exit(self): + self.__class__.operations.append(("request-exit", threading.get_native_id())) + if self.__class__.request_exit_error is not None: + raise self.__class__.request_exit_error + if self.__class__.request_exit_callback is not None: + self.__class__.request_exit_callback() + + def destroy(self, *, stage_ordinal): + assert threading.get_native_id() in self.__class__.owner_lane_threads + self.__class__.destroy_ordinals.append(stage_ordinal) + if self.__class__.destroy_error is not None: + raise self.__class__.destroy_error + self.__class__.ordinal_ledger.seal_external(stage_ordinal) + self.__class__.operations.append(("destroy", threading.get_native_id())) + + +class _FakeContextDesc: + def __init__(self, **kwargs): + self.__dict__.update(kwargs) + + +class _FakeOrdinalLedger: + def __init__(self): + self.floor = 9 + self.active = None + self.issued = [] + self.published = [] + self.external_resolutions = [] + self.abandoned = [] + self.resolve_error = None + + def reserve(self): + assert self.active is None + ordinal = self.floor + 1 + self.active = ordinal + self.issued.append(ordinal) + return ordinal + + def publish(self, ordinal): + assert self.active == ordinal + self.published.append(ordinal) + self.floor = ordinal + self.active = None + + def seal_external(self, ordinal): + assert self.active == ordinal + self.published.append(ordinal) + self.floor = ordinal + + def resolve_external(self, ordinal, *, expected_published): + assert self.active == ordinal + if self.resolve_error is not None: + raise self.resolve_error + published = ordinal in self.published + assert expected_published is None or published == expected_published + self.external_resolutions.append((ordinal, expected_published, published)) + self.active = None + return published + + def abandon(self, ordinal): + assert self.active == ordinal + self.abandoned.append(ordinal) + self.active = None + + +def _fake_ovxr(): + return SimpleNamespace( + Context=_FakeContext, + OvxrError=_FakeOvxrError, + ContextDesc=_FakeContextDesc, + Result=SimpleNamespace(SUCCESS=0, TIMEOUT=1, NOT_READY=2, ERROR=3), + StageOrdinalMode=SimpleNamespace(PRIVATE=0, APPLICATION=1), + DisplayMode=SimpleNamespace(STEREO=3), + SessionMode=SimpleNamespace(VR=4, AR=5), + StageUpAxis=SimpleNamespace(Z=4, Y=5), + SceneInfoPolicy=SimpleNamespace(AUTO=5), + RasterPanelSpace=SimpleNamespace(VIEWER_START=0, STAGE_WORLD=1, HEAD_LOCKED=2), + result_name=str, + ) + + +class _FakeHost: + def __init__(self): + self.renderer = object() + self.stage = object() + self.ordinal_ledger = _FakeOrdinalLedger() + self.guard = threading.Lock() + self.offer_thread = None + self.attach_thread = None + self.publish_threads = [] + self.validated = [] + self.events = [] + self.publication_closed = False + self.owner_lane_entries = [] + + def offer_render_owner_transfer(self): + self.offer_thread = threading.get_native_id() + return object() + + def attach_on_render_thread(self, _capability): + self.attach_thread = threading.get_native_id() + self.events.append("attach") + _FakeContext.ordinal_ledger = self.ordinal_ledger + return ovxr_session.OvxrSessionBinding( + self.renderer, self.stage, self.ordinal_ledger + ) + + @contextmanager + def owner_lane(self): + thread_id = threading.get_native_id() + self.owner_lane_entries.append(thread_id) + _FakeContext.owner_lane_threads.add(thread_id) + try: + yield + finally: + _FakeContext.owner_lane_threads.remove(thread_id) + + @contextmanager + def publication_guard(self): + with self.guard: + yield + + def publish_complete_scene(self, publication_id): + assert not self.publication_closed + self.publish_threads.append(threading.get_native_id()) + self.events.append(("publish", publication_id)) + ordinal = self.ordinal_ledger.reserve() + self.ordinal_ledger.publish(ordinal) + return ordinal + + def validate_first_published_scene(self, ordinal): + self.validated.append(ordinal) + + def get_target_frame_T_world(self, prim_path): + return ("target", prim_path) + + def close_publication(self): + self.publication_closed = True + self.events.append("publication-close") + + def close_on_render_thread(self): + self.events.append(("host-close", threading.get_native_id())) + + def close_environment_on_render_thread(self): + self.events.append(("environment-close", threading.get_native_id())) + + def abort(self): + self.events.append("abort") + + +@pytest.fixture(autouse=True) +def _runtime_stubs(monkeypatch): + _FakeContext.operations = [] + _FakeContext.create_error = None + _FakeContext.bind_error = None + _FakeContext.destroy_error = None + _FakeContext.request_exit_error = None + _FakeContext.request_exit_callback = None + _FakeContext.owner_lane_threads = set() + _FakeContext.ordinal_ledger = None + _FakeContext.bind_ordinals = [] + _FakeContext.destroy_ordinals = [] + _FakeContext.seal_bind_results = {0, 1} + _FakeContext.last_desc = None + _FakeContext.selected_products = None + monkeypatch.setattr(frame_pump, "_import_ovxr", _fake_ovxr) + oxr = ModuleType("isaaccapture.oxr") + oxr.OpenXRSessionHandles = lambda *values: values + monkeypatch.setitem(__import__("sys").modules, "isaaccapture.oxr", oxr) + + +def _config(**kwargs): + return ovxr_session.OvxrSessionConfig( + camera_paths=("/left", "/right"), auto_launch_cloudxr=False, **kwargs + ) + + +@pytest.mark.parametrize( + "extra,required,expected", + [ + (None, (), []), + ( + ("XR_EXT_hand_tracking", "XR_NV_opaque_data_channel"), + ("XR_KHR_convert_timespec_time", "XR_NV_opaque_data_channel"), + [ + "XR_EXT_hand_tracking", + "XR_NV_opaque_data_channel", + "XR_KHR_convert_timespec_time", + ], + ), + ( + (), + ("XR_KHR_win32_convert_performance_counter_time",), + ["XR_KHR_win32_convert_performance_counter_time"], + ), + ], +) +def test_session_enables_only_device_requirements_and_application_extras( + extra, required, expected +): + config = _config() if extra is None else _config(extra_openxr_extensions=extra) + owner = ovxr_session.OvxrSession( + _FakeHost(), config, required_openxr_extensions=required + ) + with owner: + owner.start() + assert owner._pump._context.desc.extra_openxr_extensions == expected + + +def _wait_for_operation(name: str) -> None: + deadline = time.monotonic() + 1.0 + while name not in [operation for operation, _thread in _FakeContext.operations]: + if time.monotonic() >= deadline: + raise AssertionError(f"timed out waiting for {name}") + time.sleep(0.001) + + +class _ControlledLauncher: + def __init__(self, *, block_stop=False, stop_error=None): + self.block_stop = block_stop + self.stop_error = stop_error + self.stop_started = threading.Event() + self.allow_stop = threading.Event() + self.events = [] + + def interrupt_runtime(self): + self.events.append(("interrupt", threading.get_native_id())) + + def stop(self): + self.events.append(("stop-start", threading.get_native_id())) + self.stop_started.set() + if self.block_stop: + self.allow_stop.wait(timeout=1.0) + if self.stop_error is not None: + raise self.stop_error + self.events.append(("stop-end", threading.get_native_id())) + + +def _call_in_thread(callback): + errors = [] + + def call(): + try: + callback() + except BaseException as exc: + errors.append(exc) + + thread = threading.Thread(target=call) + thread.start() + return thread, errors + + +@pytest.mark.parametrize("wait_phase", ["bootstrap", "xr"]) +def test_interrupted_start_cancels_worker_before_retirement(monkeypatch, wait_phase): + entered = threading.Event() + release = threading.Event() + failure = _WorkerAbort("startup interrupted") + + def block_startup(): + entered.set() + assert release.wait(timeout=2.0) + + class Host(_FakeHost): + def attach_on_render_thread(self, capability): + if wait_phase == "bootstrap": + block_startup() + return super().attach_on_render_thread(capability) + + class InterruptedCondition(threading.Condition): + def wait_for(self, predicate, timeout=None): + if (timeout is None) == (wait_phase == "xr"): + assert entered.wait(timeout=1.0) + raise failure + return super().wait_for(predicate, timeout) + + owner = ovxr_session.OvxrSession(Host(), _config()) + owner._worker_condition = InterruptedCondition() + if wait_phase == "xr": + monkeypatch.setattr(owner._pump, "bootstrap", block_startup) + try: + with pytest.raises(_WorkerAbort) as caught: + owner.start() + assert caught.value is failure + assert owner.shutdown_requested + finally: + release.set() + owner.close() + + assert not owner._worker.is_alive() + assert owner.retirement_complete + + +def test_bootstrap_timeout_cancels_slow_attach_without_creating_context(): + """An expired attach retains its host until the same worker can retire it.""" + attach_entered = threading.Event() + allow_attach = threading.Event() + + class SlowHost(_FakeHost): + def attach_on_render_thread(self, capability): + attach_entered.set() + assert allow_attach.wait(timeout=2.0) + return super().attach_on_render_thread(capability) + + host = SlowHost() + owner = ovxr_session.OvxrSession(host, _config()) + start_thread, start_errors = _call_in_thread(lambda: owner.start(timeout_s=0.02)) + close_thread = None + close_errors = [] + try: + assert attach_entered.wait(timeout=1.0) + start_thread.join(timeout=1.0) + assert not start_thread.is_alive() + assert len(start_errors) == 1 + assert isinstance(start_errors[0], TimeoutError) + close_thread, close_errors = _call_in_thread(owner.close) + assert owner._borrowers_quiesced.wait(timeout=1.0) + assert owner.owns_environment + assert not owner.retirement_complete + assert not any(isinstance(event, tuple) for event in host.events) + finally: + allow_attach.set() + start_thread.join(timeout=1.0) + if close_thread is not None: + close_thread.join(timeout=1.0) + owner.close() + + assert close_errors == [] + assert not owner._worker.is_alive() + assert owner.retirement_complete + assert "create" not in [operation for operation, _thread in _FakeContext.operations] + assert [event for event in host.events if isinstance(event, tuple)] == [ + ("host-close", owner.render_worker_native_thread_id), + ("environment-close", owner.render_worker_native_thread_id), + ] + + +@pytest.mark.parametrize( + "timeout_s", [0, -1, float("nan"), float("inf"), threading.TIMEOUT_MAX * 2] +) +def test_invalid_bootstrap_budget_is_rejected_before_owner_transfer(timeout_s): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + with pytest.raises(ValueError, match="bootstrap timeout must be finite"): + owner.start(timeout_s=timeout_s) + assert host.offer_thread is None + assert owner._worker is None + assert not _FakeContext.operations + owner.close() + + +def test_scene_publication_is_admitted_once_per_render_demand(monkeypatch): + host = _FakeHost() + frame_entered = threading.Event() + release_frame = threading.Event() + + class BlockingSubmitContext(_FakeContext): + def submit_frame(self, plan, **kwargs): + frame_entered.set() + assert release_frame.wait(timeout=1.0) + return super().submit_frame(plan, **kwargs) + + fake_ovxr = _fake_ovxr() + fake_ovxr.Context = BlockingSubmitContext + monkeypatch.setattr(frame_pump, "_import_ovxr", lambda: fake_ovxr) + owner = ovxr_session.OvxrSession(host, _config()) + owner.start() + + try: + first_ordinal = owner.request_render("step-1") + assert frame_entered.wait(timeout=1.0) + + second_ordinal = owner.request_render("step-2") + assert second_ordinal > first_ordinal + assert owner._worker_publication_requested is False + assert owner.request_render("step-3") is None + + assert owner.request_render("reset-1", force=True) > second_ordinal + assert [ + event + for event in host.events + if isinstance(event, tuple) and event[0] == "publish" + ] == [ + ("publish", "step-1"), + ("publish", "step-2"), + ("publish", "reset-1"), + ] + finally: + release_frame.set() + owner.close() + + +@pytest.mark.parametrize("force", [False, True]) +def test_shutdown_rejects_late_publication_without_error(force): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + owner.start() + with owner: + owner.request_shutdown() + assert owner.request_render("late-step", force=force) is None + assert host.publish_threads == [] + assert owner._worker_pending_ordinal is None + assert owner.retirement_complete + + +def test_shutdown_during_publication_does_not_enqueue_a_frame(monkeypatch): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + original_publish = host.publish_complete_scene + + def publish_then_shutdown(publication_id): + ordinal = original_publish(publication_id) + owner.request_shutdown() + return ordinal + + monkeypatch.setattr(host, "publish_complete_scene", publish_then_shutdown) + owner.start() + with owner: + assert owner.request_render("last-step") is None + assert host.publish_threads == [threading.get_native_id()] + assert owner._worker_pending_ordinal is None + assert not host.publication_closed + assert owner.retirement_complete + assert "submit" not in [name for name, _thread in _FakeContext.operations] + + +@pytest.mark.parametrize("during_publication", [False, True]) +def test_shutdown_does_not_suppress_a_worker_failure(monkeypatch, during_publication): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + owner.start() + failure = RuntimeError("first worker failure") + if during_publication: + original_publish = host.publish_complete_scene + + def publish_then_fail(publication_id): + ordinal = original_publish(publication_id) + owner._record_worker_failure(failure) + return ordinal + + monkeypatch.setattr(host, "publish_complete_scene", publish_then_fail) + else: + owner._record_worker_failure(failure) + try: + with pytest.raises(RuntimeError) as caught: + owner.request_render("late-step") + assert caught.value is failure + assert len(host.publish_threads) == int(during_publication) + assert owner._worker_pending_ordinal is None + finally: + with pytest.raises(RuntimeError) as caught: + owner.close() + assert caught.value is failure + + +def test_publication_before_start_and_after_close_remains_an_error(): + owner = ovxr_session.OvxrSession(_FakeHost(), _config()) + with pytest.raises(RuntimeError, match="not accepting requests"): + owner.request_render("before-start") + owner.start() + owner.close() + with pytest.raises(RuntimeError, match="closed"): + owner.request_render("after-close") + + +@pytest.mark.parametrize("timeout_s", [0.2, 300.0]) +def test_non_isaaclab_host_lends_exact_pair_to_one_owner_thread(timeout_s): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + caller_thread = threading.get_native_id() + + owner.start(timeout_s=timeout_s) + assert host.offer_thread == caller_thread + assert host.attach_thread == owner.render_worker_native_thread_id + assert host.attach_thread != caller_thread + context = owner._pump._context + assert context.desc.ovrtx_renderer is host.renderer + assert context.desc.ovstage is host.stage + assert context.desc.ordinal_mode == 1 + assert context.desc.host_wait_render_output is False + assert owner.external_oxr_handles() == (11, 22, 33, 44) + + scene_ordinal = owner.request_render("step-1") + _wait_for_operation("submit") + owner.close() + + assert host.publish_threads == [caller_thread] + render_operations = [ + thread_id + for name, thread_id in _FakeContext.operations + if name != "request-exit" + ] + assert set(render_operations) == {host.attach_thread} + request_threads = [ + thread_id + for name, thread_id in _FakeContext.operations + if name == "request-exit" + ] + assert caller_thread in request_threads + assert host.attach_thread in request_threads + assert host.owner_lane_entries == [host.attach_thread, host.attach_thread] + assert host.validated == [scene_ordinal] + assert _FakeContext.bind_ordinals + assert _FakeContext.bind_ordinals[0] > scene_ordinal + assert _FakeContext.destroy_ordinals + assert host.events[-2:] == [ + ("host-close", host.attach_thread), + ("environment-close", host.attach_thread), + ] + + +def test_host_wait_render_output_is_forwarded_to_ovxr(): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config(host_wait_render_output=True)) + + owner.start() + with owner: + assert owner._pump._context.desc.host_wait_render_output is True + + +def test_bind_timeout_reconciles_a_sealed_ordinal_then_reserves_a_fresh_one(): + host = _FakeHost() + _FakeContext.seal_bind_results = {0, 1} + owner = ovxr_session.OvxrSession(host, _config()) + owner.start() + owner._pump._context.bind_results = [1, 0] + + scene_ordinal = owner.request_render("step-1") + _wait_for_operation("cancel") + deadline = time.monotonic() + 1.0 + while [name for name, _thread in _FakeContext.operations].count("submit") < 1: + if time.monotonic() >= deadline: + raise AssertionError("timed out waiting for retry submission") + time.sleep(0.001) + owner.close() + + names = [name for name, _thread in _FakeContext.operations] + assert names.count("bind") >= 2 + assert names.index("cancel") < names.index("submit") + assert host.validated == [scene_ordinal] + assert len(_FakeContext.bind_ordinals) >= 2 + first, second = _FakeContext.bind_ordinals[:2] + assert second > first + assert (first, True, True) in host.ordinal_ledger.external_resolutions + + +def test_missing_core_phase_method_still_rejects_before_host_transfer(monkeypatch): + runtime = _fake_ovxr() + runtime.Context = type("BrokenContext", (_FakeContext,), {"cancel_frame": None}) + monkeypatch.setattr(frame_pump, "_import_ovxr", lambda: runtime) + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + with pytest.raises(RuntimeError, match="required Context methods: cancel_frame"): + owner.start() + assert host.offer_thread is None + + +def test_bind_timeout_reconciles_an_unsealed_ordinal_then_reuses_it(): + host = _FakeHost() + _FakeContext.seal_bind_results = {0} + owner = ovxr_session.OvxrSession(host, _config()) + owner.start() + owner._pump._context.bind_results = [1, 0] + + owner.request_render("step-1") + _wait_for_operation("cancel") + deadline = time.monotonic() + 1.0 + while [name for name, _thread in _FakeContext.operations].count("submit") < 1: + assert time.monotonic() < deadline + time.sleep(0.001) + owner.close() + + assert len(_FakeContext.bind_ordinals) >= 2 + assert _FakeContext.bind_ordinals[1] == _FakeContext.bind_ordinals[0] + assert ( + _FakeContext.bind_ordinals[0], + False, + False, + ) in host.ordinal_ledger.external_resolutions + + +def test_bind_not_ready_without_writes_retries_the_unsealed_ledger_ordinal(): + host = _FakeHost() + _FakeContext.seal_bind_results = {0} + owner = ovxr_session.OvxrSession(host, _config()) + owner.start() + owner._pump._context.bind_results = [2, 0] + + owner.request_render("step-1") + _wait_for_operation("cancel") + deadline = time.monotonic() + 1.0 + while [name for name, _thread in _FakeContext.operations].count("submit") < 1: + assert time.monotonic() < deadline + time.sleep(0.001) + owner.close() + + assert len(_FakeContext.bind_ordinals) >= 2 + assert _FakeContext.bind_ordinals[1] == _FakeContext.bind_ordinals[0] + assert ( + _FakeContext.bind_ordinals[0], + False, + False, + ) in host.ordinal_ledger.external_resolutions + + +@pytest.mark.parametrize("sealed", [False, True]) +def test_fatal_bind_resolves_publication_before_reporting_failure(sealed): + host = _FakeHost() + _FakeContext.seal_bind_results = {0, 3} if sealed else {0} + owner = ovxr_session.OvxrSession(host, _config()) + owner.start() + owner._pump._context.bind_results = [3] + owner.request_render("step-1") + _wait_for_operation("cancel") + with pytest.raises(RuntimeError, match="OVXR bind_frame failed"): + owner.close() + first = _FakeContext.bind_ordinals[0] + assert (first, sealed, sealed) in host.ordinal_ledger.external_resolutions + assert "submit" not in [name for name, _thread in _FakeContext.operations] + assert host.ordinal_ledger.active is None + + +def test_create_failure_does_not_reserve_or_publish_an_ordinal(): + host = _FakeHost() + _FakeContext.create_error = RuntimeError("create failed") + owner = ovxr_session.OvxrSession(host, _config()) + + with pytest.raises(RuntimeError, match="create failed"): + owner.start() + + assert host.ordinal_ledger.issued == [] + assert host.ordinal_ledger.external_resolutions == [] + assert host.ordinal_ledger.abandoned == [] + + +@pytest.mark.parametrize("before_attach", [False, True]) +def test_shutdown_unblocks_start_while_context_is_still_waiting( + monkeypatch, before_attach +): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + entered = threading.Event() + resume = threading.Event() + original = _FakeContext.__init__ + + def create(context, desc): + entered.set() + assert resume.wait(timeout=2.0) + original(context, desc) + + if before_attach: + attach = host.attach_on_render_thread + + def blocked_attach(capability): + entered.set() + assert resume.wait(timeout=2.0) + return attach(capability) + + monkeypatch.setattr(host, "attach_on_render_thread", blocked_attach) + else: + monkeypatch.setattr(_FakeContext, "__init__", create) + results = [] + thread, errors = _call_in_thread(lambda: results.append(owner.start())) + try: + assert entered.wait(timeout=1.0) + owner.request_shutdown() + thread.join(timeout=0.2) + assert not thread.is_alive() + assert errors == [] + assert results == [False] + finally: + resume.set() + thread.join(timeout=1.0) + owner.close() + + +def test_close_wakes_context_before_releasing_ready_worker(): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + owner.start() + caller = threading.get_native_id() + wakes = [] + + def wake(): + if threading.get_native_id() == caller: + wakes.append(True) + assert not owner._borrowers_quiesced.is_set() + + _FakeContext.request_exit_callback = wake + owner.request_shutdown() + assert owner._phase is not ovxr_session._Phase.CLOSED + assert "destroy" not in [name for name, _thread in _FakeContext.operations] + owner.close() + assert wakes == [True] + assert owner.retirement_complete + + +def test_successful_shutdown_waits_for_caller_quiescence_before_retirement(): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + waiting = threading.Event() + + class ObservedQuiescence(threading.Event): + def wait(self, timeout=None): + waiting.set() + return super().wait(timeout) + + owner._borrowers_quiesced = ObservedQuiescence() + owner.start() + context = owner._pump._context + with owner: + owner.request_shutdown() + assert waiting.wait(timeout=1.0) + assert owner._worker.is_alive() + assert owner._pump._context is context + assert owner._phase is not ovxr_session._Phase.CLOSED + assert not owner.retirement_complete + assert "destroy" not in [name for name, _thread in _FakeContext.operations] + assert not any( + isinstance(event, tuple) and event[0] in {"host-close", "environment-close"} + for event in host.events + ) + + assert not owner._worker.is_alive() + assert owner.retirement_complete + assert _FakeContext.operations.count(("destroy", host.attach_thread)) == 1 + + +@pytest.mark.parametrize("abort_error_type", [None, RuntimeError, KeyboardInterrupt]) +def test_worker_thread_start_failure_rolls_back_host_and_preserves_first_error( + monkeypatch, abort_error_type, caplog +): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + start_error = RuntimeError("native thread creation failed") + original_cause = OSError("thread resources unavailable") + start_error.add_note("Original thread diagnostic") + original_notes = list(start_error.__notes__) + abort_error = abort_error_type("host abort failed") if abort_error_type else None + + def fail_start(_thread): + raise start_error from original_cause + + def abort(): + host.events.append("abort") + if abort_error is not None: + raise abort_error + + monkeypatch.setattr(threading.Thread, "start", fail_start) + monkeypatch.setattr(host, "abort", abort) + interrupted = abort_error_type is KeyboardInterrupt + with pytest.raises(KeyboardInterrupt if interrupted else RuntimeError) as error: + owner.start() + assert error.value is (abort_error if interrupted else start_error) + assert start_error.__cause__ is original_cause + assert start_error.__notes__ == original_notes + diagnostics = [ + record + for record in caplog.records + if record.getMessage() == "Cleanup after failed render-worker start also failed" + ] + assert [record.exc_info[1] for record in diagnostics] == ( + [abort_error] if abort_error is not None and not interrupted else [] + ) + assert not owner.owns_environment + assert host.events == ["abort"] + assert not _FakeContext.operations + owner.close() + assert host.events == ["abort"] + + +def test_close_before_start_aborts_host_without_creating_context(): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + + owner.close() + + assert host.events == ["abort"] + assert _FakeContext.operations == [] + + +def test_binding_rejects_an_incomplete_borrowed_tuple(): + with pytest.raises(ValueError, match="renderer, Stage, and ordinal ledger"): + ovxr_session.OvxrSessionBinding(object(), object(), None) + + +def test_cloudxr_and_ovxr_are_created_and_destroyed_on_the_owner_thread(monkeypatch): + host = _FakeHost() + launcher_events = [] + + class FakeLauncher: + def __init__(self): + launcher_events.append(("create", threading.get_native_id())) + + def interrupt_runtime(self): + launcher_events.append(("interrupt", threading.get_native_id())) + + def stop(self): + launcher_events.append(("stop", threading.get_native_id())) + + def launcher_factory(_path, *, accept_eula=False): + return FakeLauncher() + + config = _config() + config.cloudxr_env_file = "/cloudxr.env" + config.auto_launch_cloudxr = True + owner = ovxr_session.OvxrSession( + host, config, runtime_launcher_factory=launcher_factory + ) + + owner.start() + owner.close() + + assert launcher_events == [ + ("create", owner.render_worker_native_thread_id), + ("stop", owner.render_worker_native_thread_id), + ] + + +def test_destroy_failure_retains_borrowed_providers_and_forbids_retry(monkeypatch): + host = _FakeHost() + launcher = _ControlledLauncher() + + def launcher_factory(_path, *, accept_eula=False): + return launcher + + failure = RuntimeError("destroy failed") + _FakeContext.destroy_error = failure + config = _config() + config.cloudxr_env_file = "/cloudxr.env" + config.auto_launch_cloudxr = True + owner = ovxr_session.OvxrSession( + host, config, runtime_launcher_factory=launcher_factory + ) + owner.start() + context, binding = owner._pump._context, owner._pump._binding + + with pytest.raises(RuntimeError, match="destroy failed") as error: + owner.close() + + assert error.value is failure + assert owner._pump._context is context + assert owner._pump._binding is binding + assert owner._pump._runtime._launcher is launcher + assert not owner.retirement_complete + assert launcher.events == [] + assert not any( + isinstance(event, tuple) and event[0] in ("host-close", "environment-close") + for event in host.events + ) + destroy_attempts = list(_FakeContext.destroy_ordinals) + with pytest.raises(RuntimeError, match="destroy failed"): + owner.close() + assert _FakeContext.destroy_ordinals == destroy_attempts + + +@pytest.mark.parametrize("failure", ["native-result", "ffi", "ordinal", "success"]) +def test_actual_public_destroy_wrapper_distinguishes_consumed_context( + monkeypatch, failure +): + ovxr = pytest.importorskip("ovxr") + host = _FakeHost() + calls = [] + invalidations = [] + native = ovxr.Context.__new__(ovxr.Context) + native._handle = 123 + native._uses_explicit_stage_ordinals = True + native._frame_lifetime = SimpleNamespace( + invalidate_context=lambda: invalidations.append(True) + ) + + def destroy(_handle, ordinal): + calls.append(ordinal) + if failure == "ffi": + raise RuntimeError("FFI before consumption") + host.ordinal_ledger.seal_external(ordinal) + return ovxr.Result.ERROR if failure == "native-result" else ovxr.Result.SUCCESS + + native._bindings = SimpleNamespace( + destroy=destroy, get_last_error=lambda: "release failed" + ) + + class WrappedContext(_FakeContext): + def destroy(self, *, stage_ordinal): + native.destroy(stage_ordinal=0 if failure == "ordinal" else stage_ordinal) + + runtime = _fake_ovxr() + runtime.Context = WrappedContext + runtime.OvxrError = ovxr.OvxrError + monkeypatch.setattr(frame_pump, "_import_ovxr", lambda: runtime) + owner = ovxr_session.OvxrSession(host, _config()) + owner.start() + if failure == "success": + owner.close() + else: + with pytest.raises(RuntimeError): + owner.close() + consumed = failure in ("native-result", "success") + assert (owner._pump._context is None) == consumed + assert bool(invalidations) == consumed + assert (native._handle is None) == consumed + assert ( + any(isinstance(e, tuple) and e[0] == "environment-close" for e in host.events) + == consumed + ) + first_calls = calls.copy() + if failure != "success": + with pytest.raises(RuntimeError): + owner.close() + assert calls == first_calls + # Avoid destructor warnings for deliberately quarantined fixture handles. + native._handle = None + + +@pytest.mark.parametrize("accepted", [False, True]) +def test_explicit_eula_consent_reaches_generic_cloudxr_launcher(monkeypatch, accepted): + import isaaccapture.cloudxr + + captured = [] + monkeypatch.setattr( + isaaccapture.cloudxr, + "CloudXRLauncher", + lambda **kwargs: captured.append(kwargs), + ) + runtime_owner._create_cloudxr_launcher("profile.env", accept_eula=accepted) + assert captured[0]["accept_eula"] is accepted + + +def test_failed_exit_request_does_not_prevent_proven_context_retirement(): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + owner.start() + _FakeContext.request_exit_error = RuntimeError("request exit failed") + with pytest.raises(RuntimeError, match="request exit failed"): + owner.close() + assert owner._pump._context is None + assert [event[0] for event in host.events if isinstance(event, tuple)][-2:] == [ + "host-close", + "environment-close", + ] + + +def test_cloudxr_stop_failure_is_reported_without_runtime_interrupt(monkeypatch): + host = _FakeHost() + launcher = _ControlledLauncher(stop_error=RuntimeError("cloudxr stop failed")) + + def launcher_factory(_path, *, accept_eula=False): + return launcher + + config = _config() + config.cloudxr_env_file = "/cloudxr.env" + config.auto_launch_cloudxr = True + owner = ovxr_session.OvxrSession( + host, config, runtime_launcher_factory=launcher_factory + ) + owner.start() + + with pytest.raises(RuntimeError, match="cloudxr stop failed"): + owner.close() + + assert owner._pump._runtime._launcher is launcher + assert launcher.events == [("stop-start", owner.render_worker_native_thread_id)] + + launcher.stop_error = None + with pytest.raises(RuntimeError, match="cloudxr stop failed"): + owner.close() + assert owner._pump._runtime._launcher is None + assert [name for name, _thread in launcher.events] == [ + "stop-start", + "stop-start", + "stop-end", + ] + + +@pytest.mark.parametrize("failure_type", [RuntimeError, _WorkerAbort]) +def test_render_failure_waits_for_borrowers_and_preserves_first_error( + monkeypatch, failure_type +): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + owner.start() + failure = failure_type("bind failed") + _FakeContext.bind_error = failure + failed = threading.Event() + original = owner._record_worker_failure + + def record(error): + original(error) + failed.set() + + monkeypatch.setattr(owner, "_record_worker_failure", record) + + owner.request_render("step-1") + try: + assert failed.wait(timeout=1.0) + assert owner._phase is not ovxr_session._Phase.CLOSED + assert "destroy" not in [name for name, _thread in _FakeContext.operations] + assert not any( + isinstance(event, tuple) and event[0] in {"host-close", "environment-close"} + for event in host.events + ) + for read in ( + owner.external_oxr_handles, + lambda: owner.request_render("step-2"), + ): + with pytest.raises(failure_type) as caught: + read() + assert caught.value is failure + finally: + with pytest.raises(failure_type) as caught: + owner.close() + assert caught.value is failure + with pytest.raises(failure_type) as caught: + owner.close() + assert caught.value is failure + assert host.publication_closed + assert host.events[-2:] == [ + ("host-close", host.attach_thread), + ("environment-close", host.attach_thread), + ] + + +def test_bind_failure_remains_primary_when_ordinal_reconciliation_also_fails(): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + owner.start() + _FakeContext.bind_error = RuntimeError("bind failed") + host.ordinal_ledger.resolve_error = RuntimeError("reconciliation failed") + + owner.request_render("step-1") + deadline = time.monotonic() + 1.0 + while not owner.shutdown_requested: + assert time.monotonic() < deadline + time.sleep(0.001) + + with pytest.raises(RuntimeError, match="bind failed") as error: + owner.close() + assert isinstance(error.value.__cause__, RuntimeError) + assert str(error.value.__cause__) == "reconciliation failed" + + +def test_destroy_failure_remains_primary_when_ordinal_reconciliation_also_fails(): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + owner.start() + _FakeContext.destroy_error = RuntimeError("destroy failed") + host.ordinal_ledger.resolve_error = RuntimeError("reconciliation failed") + + with pytest.raises(RuntimeError, match="destroy failed") as error: + owner.close() + assert isinstance(error.value.__cause__, RuntimeError) + assert str(error.value.__cause__) == "reconciliation failed" + + +@pytest.mark.parametrize( + "bind_result", [0, 1, 2], ids=["submit", "timeout", "not-ready"] +) +def test_paused_frames_use_runtime_pacing_without_an_added_idle_wait( + monkeypatch, bind_result +): + fourth_frame = threading.Event() + release_frame = threading.Event() + idle_waits = [] + + class PacedContext(_FakeContext): + frames = 0 + + def begin_frame(self): + self.frames += 1 + if self.frames == 4: + fourth_frame.set() + assert release_frame.wait(1.0) + return super().begin_frame() + + class ObservedCondition(threading.Condition): + def wait(self, timeout=None): + if threading.get_native_id() == owner.render_worker_native_thread_id: + idle_waits.append((owner._pump._last_ordinal, timeout)) + return super().wait(timeout) + + fake = _fake_ovxr() + fake.Context = PacedContext + monkeypatch.setattr(frame_pump, "_import_ovxr", lambda: fake) + _FakeContext.request_exit_callback = release_frame.set + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + owner._worker_condition = ObservedCondition() + with owner: + owner.start() + owner._pump._context.bind_results = [bind_result] + owner.request_render("paused-scene") + assert fourth_frame.wait(1.0) + + assert not [wait for wait in idle_waits if wait[0] is not None] + assert [ + event + for event in host.events + if isinstance(event, tuple) and event[0] == "publish" + ] == [("publish", "paused-scene")] + assert owner._pump._context is None + + +@pytest.mark.parametrize( + "begin_result", [None, 1, 2], ids=["no-scene", "timeout", "not-ready"] +) +def test_idle_or_unready_frames_keep_a_wakeable_backoff(monkeypatch, begin_result): + backed_off = threading.Event() + waits = [] + + class UnreadyContext(_FakeContext): + def begin_frame(self): + assert begin_result is not None + self.__class__.operations.append(("begin", threading.get_native_id())) + return begin_result, None + + class ObservedCondition(threading.Condition): + def wait(self, timeout=None): + if threading.get_native_id() == owner.render_worker_native_thread_id: + waits.append(timeout) + if len(waits) == 3: + backed_off.set() + return super().wait(timeout) + + fake = _fake_ovxr() + fake.Context = UnreadyContext + monkeypatch.setattr(frame_pump, "_import_ovxr", lambda: fake) + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + owner._worker_condition = ObservedCondition() + with owner: + owner.start() + if begin_result is not None: + owner.request_render("unready-scene") + assert backed_off.wait(1.0) + + assert len(waits) >= 3 + assert all(timeout == 0.01 for timeout in waits) + assert "submit" not in [name for name, _thread in _FakeContext.operations] + assert not owner._worker.is_alive() + + +@pytest.mark.parametrize( + "publish_scene", [False, True], ids=["idle-owner", "rendering-owner"] +) +def test_late_peer_exposes_input_handles_after_bootstrap(monkeypatch, publish_scene): + peer_ready = threading.Event() + missing_observed = threading.Event() + handles_observed = threading.Event() + get_threads = [] + + class LatePeerContext(_FakeContext): + def get_openxr_handles(self): + get_threads.append(threading.get_native_id()) + if not peer_ready.is_set(): + missing_observed.set() + return None + handles_observed.set() + return super().get_openxr_handles() + + fake = _fake_ovxr() + fake.Context = LatePeerContext + monkeypatch.setattr(frame_pump, "_import_ovxr", lambda: fake) + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config(), max_blank_frames=1) + with owner: + owner.start() + assert missing_observed.wait(1.0) + assert owner.external_oxr_handles() is None + if publish_scene: + owner.request_render("paused-scene") + _wait_for_operation("submit") + peer_ready.set() + with owner._worker_condition: + ready = owner._worker_condition.wait_for( + lambda: owner._worker_handles is not None, timeout=1.0 + ) + assert ready, "Late XR session never reached the Teleop input borrower" + assert handles_observed.is_set() + assert owner.external_oxr_handles() == (11, 22, 33, 44) + assert set(get_threads) == {host.attach_thread} + + +@pytest.mark.parametrize("mode,expected", [("ar", 5), ("vr", 4)]) +def test_session_mode_is_forwarded_to_ovxr(mode, expected): + owner = ovxr_session.OvxrSession(_FakeHost(), _config(session_mode=mode)) + with owner: + owner.start() + assert owner._pump._context.desc.session_mode == expected + + +def test_config_rejects_unknown_session_mode(): + with pytest.raises(ValueError, match="session_mode must be 'ar' or 'vr'"): + _config(session_mode="automatic") + + +@pytest.mark.parametrize( + "scale", [float("nan"), float("inf"), float("-inf"), 0.0, -1.0] +) +def test_world_scale_rejects_nonfinite_or_nonpositive_values(scale): + with pytest.raises(ValueError): + _config(meters_per_unit=scale) + + +@pytest.mark.parametrize("error_type", [RuntimeError, KeyboardInterrupt]) +def test_cancel_failure_does_not_skip_context_and_service_cleanup( + monkeypatch, error_type +): + host = _FakeHost() + launcher = _ControlledLauncher() + + def launcher_factory(*args, **kwargs): + return launcher + + config = _config() + config.cloudxr_env_file = "/cloudxr.env" + config.auto_launch_cloudxr = True + owner = ovxr_session.OvxrSession( + host, config, runtime_launcher_factory=launcher_factory + ) + failure = error_type("cancel failed") + + def cancel(_self, _plan): + raise failure + + monkeypatch.setattr(_FakeContext, "cancel_frame", cancel) + owner.start() + owner._pump._pending_plan = SimpleNamespace(step_result=0) + with pytest.raises(error_type) as caught: + owner.close() + assert caught.value is failure + assert "destroy" in [name for name, _thread in _FakeContext.operations] + assert owner._pump._context is None + assert owner._pump._pending_plan is None + assert [name for name, _thread in launcher.events] == ["stop-start", "stop-end"] + assert [event[0] for event in host.events if isinstance(event, tuple)][-2:] == [ + "host-close", + "environment-close", + ] + assert not owner.retirement_complete # Cleanup does not erase the original error. + + +def test_shutdown_timeout_preserves_runtime_until_frame_worker_retires(monkeypatch): + host = _FakeHost() + frame_wait_entered = threading.Event() + release_frame = threading.Event() + launcher = _ControlledLauncher() + + class BlockingContext(_FakeContext): + def begin_frame(self): + frame_wait_entered.set() + assert release_frame.wait(timeout=2.0) + return super().begin_frame() + + fake_ovxr = _fake_ovxr() + fake_ovxr.Context = BlockingContext + monkeypatch.setattr(frame_pump, "_import_ovxr", lambda: fake_ovxr) + + def launcher_factory(*args, **kwargs): + return launcher + + monkeypatch.setattr(ovxr_session, "_WORKER_SHUTDOWN_TIMEOUT_SECONDS", 0.01) + config = _config() + config.cloudxr_env_file = "/cloudxr.env" + config.auto_launch_cloudxr = True + owner = ovxr_session.OvxrSession( + host, config, runtime_launcher_factory=launcher_factory + ) + owner.start() + assert owner.request_render("step-1") == 10 + assert frame_wait_entered.wait(timeout=1.0) + context = owner._pump._context + failure = RuntimeError("request exit failed") + _FakeContext.request_exit_error = failure + try: + with pytest.raises( + TimeoutError, match="OVXR session worker did not stop" + ) as error: + owner.close() + assert error.value.__cause__ is failure + assert owner._pump._context is context + assert owner._worker.is_alive() + assert not owner.retirement_complete + assert launcher.events == [] + assert "destroy" not in [name for name, _thread in _FakeContext.operations] + finally: + release_frame.set() + owner._worker.join(timeout=1.0) + assert not owner._worker.is_alive() + with pytest.raises(RuntimeError) as error: + owner.close() + assert error.value is failure + assert [name for name, _thread in launcher.events] == ["stop-start", "stop-end"] + + +@pytest.mark.parametrize("native_error", [False, True]) +def test_retired_context_is_unpublished_before_lifecycle_unlock( + monkeypatch, native_error +): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + owner.start() + context = owner._pump._context + lock = threading.Lock() + observed = [] + + class ObservedLock: + def acquire(self, blocking=True): + return lock.acquire(blocking=blocking) + + def release(self): + lock.release() + + def __enter__(self): + lock.acquire() + + def __exit__(self, *_args): + lock.release() + observed.append(owner._pump._context) + # The real close wake can enter immediately after unlock. + owner._pump.request_exit() + + owner._pump._context_lifecycle_lock = ObservedLock() + if native_error: + _FakeContext.destroy_error = _FakeOvxrError("native result error") + with pytest.raises(_FakeOvxrError): + owner.close() + else: + owner.close() + assert observed == [None] + assert owner._pump._context is None + assert context is not None + + +def test_partial_startup_failure_retires_context_before_cloudxr_stop(monkeypatch): + host = _FakeHost() + launcher = _ControlledLauncher(block_stop=True) + + def launcher_factory(_path, *, accept_eula=False): + return launcher + + _FakeContext.create_error = RuntimeError("create failed") + config = _config() + config.cloudxr_env_file = "/cloudxr.env" + config.auto_launch_cloudxr = True + owner = ovxr_session.OvxrSession( + host, config, runtime_launcher_factory=launcher_factory + ) + + start_thread, start_errors = _call_in_thread(owner.start) + assert launcher.stop_started.wait(timeout=1.0) + close_thread, close_errors = _call_in_thread(owner.close) + try: + time.sleep(0.03) + assert "interrupt" not in [name for name, _thread in launcher.events] + finally: + launcher.allow_stop.set() + start_thread.join(timeout=1.0) + close_thread.join(timeout=1.0) + + assert not start_thread.is_alive() + assert not close_thread.is_alive() + assert [str(error) for error in start_errors] == ["create failed"] + assert [str(error) for error in close_errors] == ["create failed"] + assert launcher.events == [ + ("stop-start", owner.render_worker_native_thread_id), + ("stop-end", owner.render_worker_native_thread_id), + ] + + +def test_owner_lane_exit_failure_after_destroy_still_releases_retired_owners( + monkeypatch, +): + host = _FakeHost() + launcher = _ControlledLauncher() + + def launcher_factory(*args, **kwargs): + return launcher + + config = _config() + config.cloudxr_env_file = "/cloudxr.env" + config.auto_launch_cloudxr = True + owner = ovxr_session.OvxrSession( + host, config, runtime_launcher_factory=launcher_factory + ) + owner.start() + original_lane = host.owner_lane + + @contextmanager + def failing_lane(): + with original_lane(): + yield + raise RuntimeError("owner lane exit failed") + + host.owner_lane = failing_lane + with pytest.raises(RuntimeError, match="owner lane exit failed"): + owner.close() + assert owner._pump._context is None + assert [name for name, _thread in launcher.events] == ["stop-start", "stop-end"] + assert [event[0] for event in host.events if isinstance(event, tuple)][-2:] == [ + "host-close", + "environment-close", + ] + + +def test_cloudxr_stop_can_finish_without_runtime_interruption(monkeypatch, caplog): + host = _FakeHost() + launcher = _ControlledLauncher(block_stop=True) + + def launcher_factory(_path, *, accept_eula=False): + return launcher + + config = _config() + config.cloudxr_env_file = "/cloudxr.env" + config.auto_launch_cloudxr = True + owner = ovxr_session.OvxrSession( + host, config, runtime_launcher_factory=launcher_factory + ) + owner.start() + + close_thread, close_errors = _call_in_thread(owner.close) + try: + assert launcher.stop_started.wait(timeout=1.0) + assert owner._pump._runtime._launcher is launcher + time.sleep(0.03) + assert close_thread.is_alive() + assert "interrupt" not in [name for name, _thread in launcher.events] + assert "interrupting the owned OpenXR runtime" not in caplog.text + finally: + launcher.allow_stop.set() + close_thread.join(timeout=1.0) + + assert not close_thread.is_alive() + assert close_errors == [] + assert owner._pump._runtime._launcher is None + assert launcher.events == [ + ("stop-start", owner.render_worker_native_thread_id), + ("stop-end", owner.render_worker_native_thread_id), + ] + + +def test_close_times_out_without_blocking_on_startup_teardown_lock(monkeypatch): + host = _FakeHost() + owner = ovxr_session.OvxrSession(host, _config()) + entered = threading.Event() + resume = threading.Event() + destroy = _FakeContext.destroy + failure = RuntimeError("startup failed") + + def blocked_destroy(context, *, stage_ordinal): + entered.set() + assert resume.wait(timeout=2.0) + destroy(context, stage_ordinal=stage_ordinal) + + def fail_start(): + raise failure + + monkeypatch.setattr(_FakeContext, "destroy", blocked_destroy) + monkeypatch.setattr(owner._pump, "bootstrap", fail_start) + monkeypatch.setattr(ovxr_session, "_WORKER_SHUTDOWN_TIMEOUT_SECONDS", 0.01) + try: + with pytest.raises(RuntimeError) as caught: + owner.start() + assert caught.value is failure + assert entered.wait(timeout=1.0) + closing, errors = _call_in_thread(owner.close) + closing.join(timeout=0.5) + assert not closing.is_alive() + assert len(errors) == 1 + assert isinstance(errors[0], TimeoutError) + assert owner._pump._context is not None + assert not owner.retirement_complete + finally: + resume.set() + owner._worker.join(timeout=1.0) + with pytest.raises(RuntimeError) as caught: + owner.close() + assert caught.value is failure + + +def test_slow_context_retirement_keeps_owned_runtime_alive(monkeypatch): + host = _FakeHost() + launcher = _ControlledLauncher() + entered = threading.Event() + resume = threading.Event() + destroy = _FakeContext.destroy + + def slow_destroy(context, *, stage_ordinal): + entered.set() + assert resume.wait(timeout=3.0) + destroy(context, stage_ordinal=stage_ordinal) + + monkeypatch.setattr(_FakeContext, "destroy", slow_destroy) + + def launcher_factory(*args, **kwargs): + return launcher + + config = _config() + config.cloudxr_env_file = "/cloudxr.env" + config.auto_launch_cloudxr = True + owner = ovxr_session.OvxrSession( + host, config, runtime_launcher_factory=launcher_factory + ) + owner.start() + closing, errors = _call_in_thread(owner.close) + try: + assert entered.wait(timeout=1.0) + # Slow native retirement must keep its runtime alive until completion. + time.sleep(1.1) + assert closing.is_alive() + assert launcher.events == [] + finally: + resume.set() + closing.join(timeout=2.0) + assert not closing.is_alive() + assert errors == [] + assert owner.retirement_complete + assert [name for name, _thread in launcher.events] == ["stop-start", "stop-end"] + + +@pytest.mark.parametrize("axis, expected", [("Z", 4), ("Y", 5)]) +def test_host_up_axis_reaches_openxr_context(axis, expected): + owner = ovxr_session.OvxrSession(_FakeHost(), _config(up_axis=axis)) + try: + owner.start() + assert owner._pump._context.desc.up_axis == expected + finally: + owner.close() + + +def test_stereo_camera_paths_are_required_and_axis_is_validated(): + with pytest.raises(TypeError, match="camera_paths"): + ovxr_session.OvxrSessionConfig() + with pytest.raises(ValueError, match="up_axis"): + _config(up_axis="X") + + +def test_failure_after_worker_ready_before_start_returns_retires_without_close( + monkeypatch, +): + """A failed start never transfers quiescence responsibility to its caller.""" + may_finish_bootstrap = threading.Event() + environment_closed = threading.Event() + failure = RuntimeError("first frame failed") + + class Host(_FakeHost): + def close_environment_on_render_thread(self): + super().close_environment_on_render_thread() + environment_closed.set() + + class StartupCondition(threading.Condition): + def wait_for(self, predicate, timeout=None): + if timeout is None: + may_finish_bootstrap.set() + return super().wait_for(lambda: owner.shutdown_requested, timeout=1.0) + return super().wait_for(predicate, timeout) + + def bootstrap(_pump): + assert may_finish_bootstrap.wait(timeout=1.0) + + def fail_frame(_pump): + raise failure + + monkeypatch.setattr(frame_pump.OvxrFramePump, "bootstrap", bootstrap) + monkeypatch.setattr(frame_pump.OvxrFramePump, "render_reused", fail_frame) + owner = ovxr_session.OvxrSession(Host(), _config()) + owner._worker_condition = StartupCondition() + try: + with pytest.raises(RuntimeError) as caught: + owner.start() + assert caught.value is failure + assert environment_closed.wait(timeout=1.0) + owner._worker.join(timeout=1.0) + assert not owner._worker.is_alive() + assert owner.retirement_complete + finally: + with pytest.raises(RuntimeError): + owner.close() diff --git a/tests/python/core/teleop_session_manager/test_teleop_session.py b/tests/python/core/teleop_session_manager/test_teleop_session.py index 7c37700560..1113e0866a 100644 --- a/tests/python/core/teleop_session_manager/test_teleop_session.py +++ b/tests/python/core/teleop_session_manager/test_teleop_session.py @@ -1086,8 +1086,13 @@ def test_enter_rejects_nested_active_session(self): with mock_session_dependencies(): session = TeleopSession(config) with session: + handles = session.oxr_session + device = session.deviceio_session with pytest.raises(RuntimeError, match="already active"): session.__enter__() + assert session.oxr_session is handles + assert session.deviceio_session is device + session.step() with session: pass @@ -1199,8 +1204,14 @@ def fail_setup(stack): assert "Failed to roll back TeleopSession setup" in caplog.text with mock_session_dependencies(): - with session: - pass + for operation in ( + session.__enter__, + session.step, + lambda: session.__exit__(None, None, None), + ): + with pytest.raises(ValueError, match="setup failed"): + operation() + assert session.cleanup_failed def test_enter_initializes_runtime_state(self): """__enter__ should reset frame count and record start time.""" @@ -1270,6 +1281,9 @@ def __exit__(self, *args): session.__exit__(None, None, None) assert session.oxr_session is None + with pytest.raises(RuntimeError, match="cleanup boom"): + session.__exit__(None, None, None) + assert session.cleanup_failed def test_context_manager_protocol(self): """TeleopSession works correctly as a context manager.""" @@ -1983,6 +1997,86 @@ def test_step_no_external_inputs_when_none_needed(self): class TestPipelinedRetargeting: """Tests for TeleopSession-owned pipelined async retarget execution.""" + def test_pending_reset_survives_replacement_without_overriding_control_state(self): + pipeline = CountingPipeline() + control = BlockingControlPipeline() + control.execute_pipeline = lambda _inputs: make_control_outputs( + ExecutionState.PAUSED, reset=False + ) + config = make_config(pipeline) + config.teleop_control_pipeline = control + with mock_session_dependencies(), TeleopSession(config) as session: + runner = AsyncRetargetRunner( + session._execute_step_request, + RetargetingExecutionConfig(pacing=ImmediatePacingConfig()), + ) + reset = make_step_request(1) + reset.execution_events.reset = True + ordinary = StepRequest(2, {}, None, None, time.monotonic()) + runner.submit(reset) + assert runner.submit(ordinary) == 1 + runner.start() + try: + frame = runner.wait_for_frame(2, timeout_s=1.0) + assert frame is not None + assert frame.context.execution_events.reset + assert ( + frame.context.execution_events.execution_state + == ExecutionState.PAUSED + ) + finally: + assert runner.stop(timeout_s=1.0) + + def test_completed_reset_survives_newer_output_until_acknowledged(self): + pipeline = CountingPipeline() + runner = make_async_runner(pipeline, self._pipelined_config(pipeline)) + reset = make_step_request(1) + reset.execution_events.reset = True + runner.submit(reset) + runner.start() + try: + assert runner.wait_for_frame(1, timeout_s=1.0) is not None + runner.submit(make_step_request(2)) + assert runner.wait_for_frame(2, timeout_s=1.0) is not None + frame = runner.latest() + assert frame.frame_id == 2 + assert frame.context.execution_events.reset + runner.acknowledge_reset(frame.reset_generation) + assert not runner.latest().context.execution_events.reset + + reset = make_step_request(3) + reset.execution_events.reset = True + runner.submit(reset) + assert runner.wait_for_frame(3, timeout_s=1.0) is not None + runner.acknowledge_reset(frame.reset_generation) + newer = runner.latest() + assert newer.context.execution_events.reset + runner.acknowledge_reset(newer.reset_generation) + assert not runner.latest().context.execution_events.reset + finally: + assert runner.stop(timeout_s=1.0) + + def test_failed_application_copy_does_not_acknowledge_reset(self): + config = self._pipelined_config(CountingPipeline()) + with mock_session_dependencies(), TeleopSession(config) as session: + session.step() + session.step(execution_events=ExecutionEvents(reset=True)) + runner = session._async_runner + assert runner is not None + assert runner.wait_for_frame(1, timeout_s=1.0) is not None + with ( + patch( + "isaaccapture.teleop_session_manager.teleop_session.snapshot_retargeter_io", + side_effect=TypeError("caller copy failed"), + ), + pytest.raises(TypeError, match="caller copy failed"), + ): + session.step() + assert runner.latest().context.execution_events.reset + session.step() + assert session.last_context.execution_events.reset + assert not runner.latest().context.execution_events.reset + def _pipelined_config(self, pipeline, **execution_kwargs): return make_config( pipeline, @@ -2243,8 +2337,15 @@ def test_repeated_completed_outputs_are_snapshots(self): first = session.step() first["result"][0] = 99.0 + session.last_context.execution_events.reset = True second = session.step() assert async_result_value(second) == 0.0 + assert session.last_context.execution_events.reset is False + session.last_context.execution_events.reset = True + assert ( + session._async_runner.latest().context.execution_events.reset + is False + ) def test_pipelined_uncopyable_outputs_fail_with_clear_error(self): pipeline = UncopyableOutputPipeline() @@ -2908,6 +3009,96 @@ def test_plugin_config_disabled_and_required(self, tmp_path): assert config.required is True +@pytest.mark.parametrize("boundary", ["deviceio", "plugin"]) +@pytest.mark.parametrize("cleanup_fails", [False, True]) +def test_failed_session_entry_unwinds_acquired_contexts( + tmp_path, boundary, cleanup_fails, caplog +): + """A failed __enter__ has no user-body __exit__ to release earlier contexts.""" + events = [] + startup_error = RuntimeError("startup failed") + original_cause = ValueError("invalid session input") + if hasattr(startup_error, "add_note"): + startup_error.add_note("Original startup diagnostic") + original_notes = getattr(startup_error, "__notes__", None) + # Exercise the supported Python 3.10 boundary even on a newer interpreter. + startup_error.add_note = None + cleanup_error = RuntimeError("cleanup failed") + + @contextmanager + def openxr(): + events.append("openxr-enter") + try: + yield MockOpenXRSession() + finally: + events.append("openxr-exit") + if cleanup_fails: + raise cleanup_error + + @contextmanager + def deviceio_context(): + if boundary == "deviceio": + raise startup_error from original_cause + events.append("deviceio-enter") + try: + yield MockDeviceIOSession() + finally: + events.append("deviceio-exit") + + manager = MockPluginManager(plugin_names=["test_plugin"]) + + def fail_plugin_start(*_args, **_kwargs): + raise startup_error from original_cause + + manager.start = MagicMock(side_effect=fail_plugin_start) + plugin = PluginConfig( + plugin_name="test_plugin", plugin_root_id="/root", search_paths=[tmp_path] + ) + config = make_config(MockPipeline(leaf_nodes=[]), plugins=[plugin]) + with mock_session_dependencies( + mock_oxr=openxr(), mock_dio=deviceio_context(), mock_pm=manager + ): + session = TeleopSession(config) + with pytest.raises(RuntimeError) as caught: + session.__enter__() + + assert events == ( + ["openxr-enter", "openxr-exit"] + if boundary == "deviceio" + else ["openxr-enter", "deviceio-enter", "deviceio-exit", "openxr-exit"] + ) + assert caught.value is startup_error + assert caught.value.__cause__ is original_cause + assert getattr(caught.value, "__notes__", None) == original_notes + diagnostics = [ + record + for record in caplog.records + if record.exc_info is not None and record.exc_info[1] is cleanup_error + ] + assert [record.exc_info[1] for record in diagnostics] == ( + [cleanup_error] if cleanup_fails else [] + ) + assert session.oxr_session is None + assert session.deviceio_session is None + assert not session._in_context + assert session.plugin_managers == [] + assert session.plugin_contexts == [] + with mock_session_dependencies(): + if cleanup_fails: + for operation in ( + session.__enter__, + session.step, + lambda: session.__exit__(None, None, None), + ): + with pytest.raises(RuntimeError) as repeated: + operation() + assert repeated.value is startup_error + assert session.cleanup_failed + else: + with session: + assert session.deviceio_session is not None + + class TestSessionReuse: """Test that session state is properly reset between uses.""" @@ -3678,3 +3869,26 @@ def test_bundled_plugins_dir_finds_the_root_that_holds_the_plugins(tmp_path): roots = [str(source_root), str(install_root)] assert _bundled_plugins_dir(roots) == install_root / "plugins" assert _bundled_plugins_dir(roots[:1]) == source_root / "plugins" + + +@pytest.mark.parametrize( + "stop_error", [RuntimeError("stop failed"), KeyboardInterrupt()] +) +def test_interrupted_worker_stop_can_be_retried_before_releasing_resources(stop_error): + session = TeleopSession(make_config(MockPipeline(leaf_nodes=[]))) + retired = [] + runner = MagicMock() + runner.stop.side_effect = [stop_error, True] + with mock_session_dependencies(): + session.__enter__() + session._exit_stack.callback(lambda: retired.append("resources")) + session._async_runner = runner + with pytest.raises(type(stop_error)) as caught: + session.__exit__(None, None, None) + assert caught.value is stop_error + assert session.cleanup_failed + assert retired == [] + assert session.__exit__(None, None, None) is False + assert not session.cleanup_failed + assert retired == ["resources"] + assert runner.stop.call_count == 2