Defer CloudXR headset connection until the OpenXR app is ready

docs/AGENTS.md

@@ -9,6 +9,9 @@ Rules that the docs tree cannot state for itself. Anything that describes how a
 currently put together belongs next to it instead — in the `.rst`, in the YAML header, in
 the CSS — where it moves when the code moves.
 
+Keep small API-option updates to purpose, prerequisites, and usage; avoid
+example wrappers and lifecycle walkthroughs unless needed to use the option.
+
 ## Build the way CI does
 
 `make current-docs` wipes `build/current` and runs `sphinx-build -W --keep-going`, which

docs/source/references/cloudxr.rst

@@ -276,6 +276,19 @@ container that way.
 Setting ``CI`` skips the interactive pause described under `Configuration`_,
 alongside the check for a non-interactive stdin, so automated runs never wait.
 
+.. _cloudxr-deferred-oob-connect:
+
+Connect the OOB headset after application startup
+-------------------------------------------------
+
+Set ``defer_oob_connect=True`` on ``CloudXRService`` or ``CloudXRLauncher``
+to delay OOB automation and WebSocket signaling. Call ``connect_headset()``
+after the OpenXR session and renderer are ready.
+
+Requires ``setup_oob=True`` and a fixed ``NV_DEVICE_PROFILE`` (not ``auto*``).
+For ``CloudXRLauncher``, also set ``run_embedded=True``. The default ``False``
+keeps immediate connection.
+
 Configuration
 -------------
 

src/python/isaaccapture/cloudxr/launcher.py

@@ -164,6 +164,7 @@ def __init__(
         host_client: bool = True,
         run_embedded: bool = False,
         start_wss_proxy: bool | None = None,
+        defer_oob_connect: bool = False,
     ) -> None:
         """Attach to the running runtime, or own one when *run_embedded*.
 
@@ -175,6 +176,9 @@ def __init__(
                 to one this process cannot configure.
             start_wss_proxy: Deprecated no-op; the proxy always starts with
                 the runtime.
+            defer_oob_connect: Hold OOB automation and backend WebSocket signaling
+                until :meth:`connect_headset`. Defaults to ``False`` (connect
+                immediately). Requires ``run_embedded=True`` and ``setup_oob=True``.
 
         Every other argument is forwarded to :class:`CloudXRService` and only
         applies when this process owns it.  When attaching they describe a
@@ -184,10 +188,16 @@ def __init__(
         when that is available.
 
         Raises:
+            ValueError: If *defer_oob_connect* is set without *run_embedded*
+                and *setup_oob*.
             RuntimeError: If the runtime fails to start or come up, if
                 *run_embedded* is set while a runtime is already serving, or
                 if the requested OOB/USB mode cannot be verified or matched.
         """
+        if defer_oob_connect and not (run_embedded and setup_oob):
+            raise ValueError(
+                "defer_oob_connect requires run_embedded=True and setup_oob=True"
+            )
         if start_wss_proxy is not None:
             self._warn_start_wss_proxy_deprecated()
 
@@ -206,6 +216,7 @@ def __init__(
                 setup_oob=setup_oob,
                 usb_local=usb_local,
                 host_client=host_client,
+                defer_oob_connect=defer_oob_connect,
             )
             return
 
@@ -846,6 +857,16 @@ def stop(self) -> None:
         if self._service is not None:
             self._service.stop()
 
+    def connect_headset(self) -> None:
+        """Release deferred OOB automation once the application's XR session is ready.
+
+        Requires ``defer_oob_connect=True``. Non-blocking and idempotent;
+        connection progress uses the WSS log as usual.
+        """
+        if self._service is None:
+            raise RuntimeError("connect_headset requires an embedded CloudXR service")
+        self._service.connect_headset()
+
     def health_check(self) -> None:
         """Raise :class:`RuntimeError` if the runtime is no longer available.
 

src/python/isaaccapture/cloudxr/service/_service.py

@@ -119,6 +119,7 @@ def __init__(
         usb_local: bool = False,
         host_client: bool = False,
         recovery_config=None,
+        defer_oob_connect: bool = False,
     ) -> None:
         """Start the CloudXR runtime and the WSS proxy.
 
@@ -148,12 +149,20 @@ def __init__(
             host_client: Serve the web client at ``/client/`` on the WSS
                 proxy port.  Assets are fetched once from GitHub Pages into
                 ``TELEOP_WEB_CLIENT_STATIC_DIR`` or ``~/.cloudxr/static-client``.
+            defer_oob_connect: Hold OOB automation and backend WebSocket signaling
+                until :meth:`connect_headset`. Defaults to ``False`` (connect
+                immediately). Requires *setup_oob*; the runtime and WSS listener
+                start immediately.
 
         Raises:
+            ValueError: If *defer_oob_connect* is set without *setup_oob*.
             RuntimeError: If the EULA is not accepted, another runtime is
                 already serving *install_dir*, or the runtime or WSS proxy
                 fails to start within its timeout.
         """
+        if defer_oob_connect and not setup_oob:
+            raise ValueError("defer_oob_connect requires setup_oob=True")
+        self._connect_requested = asyncio.Event() if defer_oob_connect else None
         self._install_dir = install_dir
         self._env_config = str(env_config) if env_config is not None else None
         self._device_profile = device_profile
@@ -749,6 +758,15 @@ def _terminate_runtime_windows(_proc: subprocess.Popen) -> None:
     # WSS proxy (background thread with its own event loop)
     # ------------------------------------------------------------------
 
+    def connect_headset(self) -> None:
+        """Start deferred headset automation without blocking the caller's frame loop."""
+        if self._connect_requested is None:
+            raise RuntimeError("connect_headset requires defer_oob_connect=True")
+        self.health_check()
+        if self._wss_loop is None or self._wss_loop.is_closed():
+            raise RuntimeError("CloudXR WSS proxy is not running")
+        self._wss_loop.call_soon_threadsafe(self._connect_requested.set)
+
     def _start_wss_proxy_thread(
         self, log_path: Path, timeout_sec: float = WSS_STARTUP_TIMEOUT_SEC
     ) -> None:
@@ -801,6 +819,7 @@ def _on_listening() -> None:
                         recovery_config=self._recovery_config,
                         on_oob_status=self._publish_oob_status,
                         on_oob_fatal=self._on_oob_fatal,
+                        connect_requested=self._connect_requested,
                     )
                 )
                 if (

src/python/isaaccapture/cloudxr/wss.py

@@ -566,6 +566,22 @@ def default_cert_paths() -> CertPaths:
     return cert_paths_from_dir(Path(get_env_config().openxr_run_dir()).parent / "certs")
 
 
+async def _wait_for_connect(
+    requested: asyncio.Event | None, stop_future: asyncio.Future
+) -> bool:
+    """Gate headset automation without holding shutdown behind application startup."""
+    if requested is not None and not stop_future.done():
+        waiter = asyncio.create_task(requested.wait())
+        try:
+            await asyncio.wait(
+                (waiter, stop_future), return_when=asyncio.FIRST_COMPLETED
+            )
+        finally:
+            waiter.cancel()
+            await asyncio.gather(waiter, return_exceptions=True)
+    return not stop_future.done()
+
+
 async def run(
     log_file_path: str | Path | None,
     stop_future: asyncio.Future,
@@ -579,6 +595,7 @@ async def run(
     recovery_config=None,
     on_oob_status: Callable[[dict], object] | None = None,
     on_oob_fatal: Callable[[Exception], object] | None = None,
+    connect_requested: asyncio.Event | None = None,
 ) -> None:
     """Start the WSS proxy server and run until *stop_future* is resolved.
 
@@ -662,13 +679,27 @@ async def report_fatal(error: Exception, circumstance: str) -> None:
                 initial,
             )
 
-        def handler(ws):
-            """Route an incoming WebSocket to the OOB hub or the backend proxy."""
+        async def handler(ws) -> None:
+            """Route to the hub or proxy; WebSocket handlers do not return a value."""
             if hub is not None:
                 path = _normalize_request_path(ws.request.path or "/")
                 if path == OOB_WS_PATH:
-                    return hub.handle_connection(ws)
-            return proxy_handler(ws, backend_host, backend_port)
+                    await hub.handle_connection(ws)
+                    return
+            # An already-open headset page may reconnect before ADB automation.
+            ready = asyncio.create_task(
+                _wait_for_connect(connect_requested, stop_future)
+            )
+            closed = asyncio.create_task(ws.wait_closed())
+            try:
+                await asyncio.wait((ready, closed), return_when=asyncio.FIRST_COMPLETED)
+                # Disconnected clients must not forward queued offers when the gate opens.
+                if not closed.done() and ready.result():
+                    await proxy_handler(ws, backend_host, backend_port)
+            finally:
+                ready.cancel()
+                closed.cancel()
+                await asyncio.gather(ready, closed, return_exceptions=True)
 
         # /client/ on this WSS port for --host-client and --usb-local (same
         # files; USB-local reaches them via adb reverse of PROXY_PORT).
@@ -709,7 +740,11 @@ def handler(ws):
                 if on_listening is not None:
                     on_listening()
                 was_listening = True
-                if setup_oob and not os.getenv("TELEOP_OOB_HUB_ONLY"):
+                if (
+                    setup_oob
+                    and not os.getenv("TELEOP_OOB_HUB_ONLY")
+                    and await _wait_for_connect(connect_requested, stop_future)
+                ):
                     from .oob_teleop_env import resolve_oob_recovery_config
 
                     lifecycle = OobLifecycle(

src/python/isaaccapture/teleop_session_manager/config.py

@@ -10,7 +10,7 @@
 from __future__ import annotations
 
 import math
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from dataclasses import dataclass, field
 from enum import Enum
 from pathlib import Path
@@ -382,6 +382,11 @@ class TeleopSessionConfig:
         trackers: Optional list of manual trackers (usually not needed - auto-discovered!)
         plugins: List of plugin configurations
         verbose: Whether to print detailed progress information during setup
+        oxr_session_factory: Optional callback taking the complete required-extension
+            list and returning external OpenXR handles. Called before DeviceIO starts.
+            The caller owns the session and must stop rendering before context exit,
+            then destroy it after exit. Requires LIVE mode; mutually exclusive with
+            oxr_handles and joint_publisher.
         oxr_handles: Optional pre-existing OpenXRSessionHandles from an external runtime
             (e.g. Kit's XR system). When provided, TeleopSession will use these handles
             instead of creating its own OpenXR session via OpenXRSession.create().
@@ -472,9 +477,20 @@ class TeleopSessionConfig:
     )
     joint_publisher: Optional["RobotTwinPublisher"] = None
     twin_render: TwinRenderConfig = field(default_factory=TwinRenderConfig)
+    # Called with all tracker extensions before DeviceIO starts. The caller owns
+    # the returned session and must stop rendering before TeleopSession exits.
+    oxr_session_factory: Optional[Callable[[List[str]], OpenXRSessionHandles]] = None
 
     def __post_init__(self) -> None:
         """Validate configuration consistency."""
+        if self.oxr_session_factory is not None and (
+            self.mode != SessionMode.LIVE
+            or self.oxr_handles is not None
+            or self.joint_publisher is not None
+        ):
+            raise ValueError(
+                "oxr_session_factory requires LIVE mode without oxr_handles or joint_publisher"
+            )
         if self.mode == SessionMode.REPLAY and self.mcap_config is None:
             raise ValueError("mcap_config is required when mode is SessionMode.REPLAY")
 

src/python/isaaccapture/teleop_session_manager/teleop_session.py

@@ -1087,10 +1087,14 @@ def _enter_resources(self, stack: ExitStack) -> None:
         self._resolved_plugins = []
 
         live_mode = self.config.mode == SessionMode.LIVE
+        external_openxr = (
+            self.config.oxr_handles is not None
+            or self.config.oxr_session_factory is not None
+        )
         self._status_monitor.rebuild(
             (),
             include_openxr=live_mode,
-            external_openxr=live_mode and self.config.oxr_handles is not None,
+            external_openxr=live_mode and external_openxr,
         )
         if live_mode:
             self._resolve_monitored_plugins()
@@ -1101,7 +1105,7 @@ def _enter_resources(self, stack: ExitStack) -> None:
                     if spec is not None
                 ),
                 include_openxr=True,
-                external_openxr=self.config.oxr_handles is not None,
+                external_openxr=external_openxr,
             )
 
         # Auto-populate mcap_config from pipeline sources if recording or replaying.
@@ -1165,6 +1169,8 @@ def _add_tracker(tracker: Any) -> None:
             # Resolve OpenXR handles
             if self.config.oxr_handles is not None:
                 handles = self.config.oxr_handles
+            elif self.config.oxr_session_factory is not None:
+                handles = self.config.oxr_session_factory(required_extensions)
             elif self.config.joint_publisher is not None:
                 handles = oxr.OpenXRSessionHandles(
                     *self._start_twin(stack, required_extensions)

tests/python/core/cloudxr/conftest.py

@@ -184,6 +184,11 @@ def mock_service_deps(tmp_path, ready=True, wss=True):
         patch(
             "isaaccapture.cloudxr.service._service.check_eula",
         ) as m_eula,
+        # The worker is mocked, so lifecycle tests need no installed native SDK.
+        patch(
+            "isaaccapture.cloudxr.service._service.get_sdk_path",
+            return_value=str(tmp_path / "sdk"),
+        ),
         patch(
             "isaaccapture.cloudxr.service._service.wait_for_runtime_ready_sync",
             return_value=ready,

tests/python/core/cloudxr/test_launcher.py

@@ -141,6 +141,32 @@ def test_run_embedded_is_refused(self, tmp_path):
         mocks["from_args"].assert_not_called()
         mocks["popen"].assert_not_called()
 
+    @pytest.mark.parametrize(
+        "options", [{}, {"run_embedded": True}, {"setup_oob": True}]
+    )
+    def test_deferred_connect_requires_an_owned_oob_service(self, options):
+        with pytest.raises(ValueError, match="defer_oob_connect requires"):
+            CloudXRLauncher(defer_oob_connect=True, **options)
+
+    def test_deferred_connect_delegates_to_embedded_service(self):
+        with (
+            _live(False),
+            patch("isaaccapture.cloudxr.launcher.CloudXRService") as service,
+        ):
+            launcher = CloudXRLauncher(
+                run_embedded=True, setup_oob=True, defer_oob_connect=True
+            )
+            assert service.call_args.kwargs["defer_oob_connect"] is True
+            launcher.connect_headset()
+            service.return_value.connect_headset.assert_called_once_with()
+
+    def test_connect_cannot_control_a_borrowed_service(self, tmp_path):
+        install = _env_file(tmp_path, XR_RUNTIME_JSON="/x/openxr.json")
+        with _live():
+            launcher = CloudXRLauncher(install_dir=install)
+        with pytest.raises(RuntimeError, match="embedded"):
+            launcher.connect_headset()
+
     def test_stop_does_not_touch_a_borrowed_runtime(self, tmp_path):
         install = _env_file(tmp_path, XR_RUNTIME_JSON="/x/openxr.json")
         with _live():