From 155c60edd8a2ab36a439138e564e8df639b6f21b Mon Sep 17 00:00:00 2001 From: Farbod Motlagh Date: Wed, 30 Sep 2026 16:26:35 -0700 Subject: [PATCH 1/5] Defer CloudXR headset connection until the OpenXR app is ready Add an opt-in gate for OOB automation and WebSocket signaling, with immediate connection remaining the default. Allow an external OpenXR session factory to receive the full set of tracker and plugin extensions. Cover deferred connection, cancellation, early signaling and session factory ownership; isolate lifecycle tests from the installed native SDK. Signed-off-by: Farbod Motlagh --- src/python/isaaccapture/cloudxr/launcher.py | 17 +++ .../isaaccapture/cloudxr/service/_service.py | 16 +++ src/python/isaaccapture/cloudxr/wss.py | 31 +++++- .../teleop_session_manager/config.py | 18 +++- .../teleop_session_manager/teleop_session.py | 10 +- tests/python/core/cloudxr/conftest.py | 5 + tests/python/core/cloudxr/test_launcher.py | 26 +++++ tests/python/core/cloudxr/test_service.py | 101 ++++++++++++++++++ .../core/cloudxr/test_service_oob_status.py | 1 + .../core/cloudxr/test_wss_oob_lifecycle.py | 44 +++++--- .../test_external_oxr_handles.py | 49 +++++++++ 11 files changed, 299 insertions(+), 19 deletions(-) diff --git a/src/python/isaaccapture/cloudxr/launcher.py b/src/python/isaaccapture/cloudxr/launcher.py index ed141b1a3e..a7d3a0a0c9 100644 --- a/src/python/isaaccapture/cloudxr/launcher.py +++ b/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_connect: bool = False, ) -> None: """Attach to the running runtime, or own one when *run_embedded*. @@ -175,6 +176,8 @@ def __init__( to one this process cannot configure. start_wss_proxy: Deprecated no-op; the proxy always starts with the runtime. + defer_connect: Wait for :meth:`connect_headset` before OOB headset + automation. 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 @@ -188,6 +191,10 @@ def __init__( *run_embedded* is set while a runtime is already serving, or if the requested OOB/USB mode cannot be verified or matched. """ + if defer_connect and not (run_embedded and setup_oob): + raise ValueError( + "defer_connect requires run_embedded=True and setup_oob=True" + ) if start_wss_proxy is not None: self._warn_start_wss_proxy_deprecated() @@ -206,6 +213,7 @@ def __init__( setup_oob=setup_oob, usb_local=usb_local, host_client=host_client, + defer_connect=defer_connect, ) return @@ -846,6 +854,15 @@ 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. + + 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. diff --git a/src/python/isaaccapture/cloudxr/service/_service.py b/src/python/isaaccapture/cloudxr/service/_service.py index 4cf34b58ab..68868f6e9e 100644 --- a/src/python/isaaccapture/cloudxr/service/_service.py +++ b/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_connect: bool = False, ) -> None: """Start the CloudXR runtime and the WSS proxy. @@ -148,12 +149,17 @@ 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_connect: Hold OOB headset automation until :meth:`connect_headset`. + Requires *setup_oob*; the runtime and WSS listener start immediately. Raises: 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_connect and not setup_oob: + raise ValueError("defer_connect requires setup_oob=True") + self._connect_requested = asyncio.Event() if defer_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 +755,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_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 +816,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 ( diff --git a/src/python/isaaccapture/cloudxr/wss.py b/src/python/isaaccapture/cloudxr/wss.py index 686e1b6222..47f203829c 100755 --- a/src/python/isaaccapture/cloudxr/wss.py +++ b/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,15 @@ async def report_fatal(error: Exception, circumstance: str) -> None: initial, ) - def handler(ws): + async def handler(ws): """Route an incoming WebSocket to the OOB hub or the backend proxy.""" 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) + return await hub.handle_connection(ws) + # An already-open headset page may reconnect before ADB automation. + if await _wait_for_connect(connect_requested, stop_future): + await proxy_handler(ws, backend_host, backend_port) # /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 +728,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( diff --git a/src/python/isaaccapture/teleop_session_manager/config.py b/src/python/isaaccapture/teleop_session_manager/config.py index c51d587b4f..8307a43af9 100644 --- a/src/python/isaaccapture/teleop_session_manager/config.py +++ b/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") diff --git a/src/python/isaaccapture/teleop_session_manager/teleop_session.py b/src/python/isaaccapture/teleop_session_manager/teleop_session.py index f22cc7faae..9645931141 100644 --- a/src/python/isaaccapture/teleop_session_manager/teleop_session.py +++ b/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) diff --git a/tests/python/core/cloudxr/conftest.py b/tests/python/core/cloudxr/conftest.py index 8fb3dd1dac..0a8592a9e3 100644 --- a/tests/python/core/cloudxr/conftest.py +++ b/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, diff --git a/tests/python/core/cloudxr/test_launcher.py b/tests/python/core/cloudxr/test_launcher.py index 3a58ff15bf..b9ed4b3511 100644 --- a/tests/python/core/cloudxr/test_launcher.py +++ b/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_connect requires"): + CloudXRLauncher(defer_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_connect=True + ) + assert service.call_args.kwargs["defer_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(): diff --git a/tests/python/core/cloudxr/test_service.py b/tests/python/core/cloudxr/test_service.py index 9a32069390..e645eea14d 100644 --- a/tests/python/core/cloudxr/test_service.py +++ b/tests/python/core/cloudxr/test_service.py @@ -11,6 +11,7 @@ import socket import subprocess import sys +import threading import types from pathlib import Path from unittest.mock import MagicMock, patch @@ -467,6 +468,58 @@ def test_a_proxy_that_never_listens_times_out(self, tmp_path): # The thread outlives the timeout; stop() is what reaps it. service._stop_wss_proxy() + @pytest.mark.parametrize("defer_connect", [False, True]) + def test_headset_connection_is_explicit_only_when_deferred( + self, tmp_path, defer_connect + ): + from cloudxr_py_test_ns.wss import _wait_for_connect + + connected = threading.Event() + + async def proxy(*, stop_future, on_listening, connect_requested, **kwargs): + on_listening() + if await _wait_for_connect(connect_requested, stop_future): + connected.set() + await stop_future + + with ( + mock_service_deps(tmp_path, wss=False), + _stub_wss(proxy), + patch("isaaccapture.cloudxr.oob_teleop_adb.require_adb_on_path"), + ): + service = CloudXRService(setup_oob=True, defer_connect=defer_connect) + try: + if defer_connect: + assert not connected.wait(0.05) + service.connect_headset() + service.connect_headset() # Releasing the same gate is harmless. + assert connected.wait(1) + finally: + service._stop_wss_proxy() + + def test_stopping_before_connect_does_not_launch_headset(self, tmp_path): + from cloudxr_py_test_ns.wss import _wait_for_connect + + connected = [] + + async def proxy(*, stop_future, on_listening, connect_requested, **kwargs): + on_listening() + if await _wait_for_connect(connect_requested, stop_future): + connected.append(True) + + with ( + mock_service_deps(tmp_path, wss=False), + _stub_wss(proxy), + patch("isaaccapture.cloudxr.oob_teleop_adb.require_adb_on_path"), + ): + service = CloudXRService(setup_oob=True, defer_connect=True) + thread = service._wss_thread + service._stop_wss_proxy() + assert not thread.is_alive() + assert connected == [] + with pytest.raises(RuntimeError, match="not running"): + service.connect_headset() + # ============================================================================ # TestSignalHandlers @@ -578,3 +631,51 @@ def test_sigint_handler_with_sig_ign_prev_is_noop(self, tmp_path): signal.signal(signal.SIGINT, orig_sigint) assert not stop_called, "signal handler must not call stop() directly" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("connect", [True, False]) +async def test_early_websocket_waits_for_app_or_shutdown(monkeypatch, connect): + from unittest.mock import AsyncMock + from cloudxr_py_test_ns import wss + + requested = asyncio.Event() + listening = asyncio.Event() + stop = asyncio.get_running_loop().create_future() + handlers = [] + + @contextlib.asynccontextmanager + async def serve(handler, **kwargs): + handlers.append(handler) + yield + + proxy = AsyncMock() + # Keep this signaling test independent of an installed CloudXR environment. + monkeypatch.setattr(wss, "default_cert_paths", lambda: None) + monkeypatch.setattr(wss, "ensure_certificate", lambda *args: None) + monkeypatch.setattr(wss, "build_ssl_context", lambda *args: None) + monkeypatch.setattr(wss, "ws_serve", serve) + monkeypatch.setattr(wss, "proxy_handler", proxy) + task = asyncio.create_task( + wss.run( + None, + stop, + on_listening=listening.set, + connect_requested=requested, + ) + ) + try: + await asyncio.wait_for(listening.wait(), timeout=1) + client = asyncio.create_task(handlers[0](object())) + await asyncio.sleep(0) + proxy.assert_not_called() + if connect: + requested.set() + else: + stop.set_result(None) + await asyncio.wait_for(client, timeout=1) + assert proxy.call_count == int(connect) + finally: + if not stop.done(): + stop.set_result(None) + await asyncio.wait_for(task, timeout=1) diff --git a/tests/python/core/cloudxr/test_service_oob_status.py b/tests/python/core/cloudxr/test_service_oob_status.py index 3a2cc2ef18..24aa0bbe5b 100644 --- a/tests/python/core/cloudxr/test_service_oob_status.py +++ b/tests/python/core/cloudxr/test_service_oob_status.py @@ -34,6 +34,7 @@ def _service_for_status(tmp_path): service._runtime_proc.pid = os.getpid() service._runtime_proc.poll.return_value = None service._wss_thread = None + service._connect_requested = None return service diff --git a/tests/python/core/cloudxr/test_wss_oob_lifecycle.py b/tests/python/core/cloudxr/test_wss_oob_lifecycle.py index 943081186d..294967fee5 100644 --- a/tests/python/core/cloudxr/test_wss_oob_lifecycle.py +++ b/tests/python/core/cloudxr/test_wss_oob_lifecycle.py @@ -18,13 +18,22 @@ @pytest.mark.asyncio -async def test_wss_lifecycle_waits_without_headset_and_cleans_up_in_order( - monkeypatch, tmp_path +@pytest.mark.parametrize("connection_mode", ["immediate", "deferred", "cancelled"]) +async def test_wss_lifecycle_respects_connection_gate_and_cleanup_order( + monkeypatch, tmp_path, connection_mode ): + monkeypatch.delenv("TELEOP_OOB_HUB_ONLY", raising=False) events = [] pending = asyncio.Event() + listening = asyncio.Event() + started = asyncio.Event() + requested = asyncio.Event() if connection_mode != "immediate" else None stop = asyncio.get_running_loop().create_future() + def on_listening(): + events.append("callback") + listening.set() + @asynccontextmanager async def serving(*args, **kwargs): events.append("wss listening") @@ -33,6 +42,7 @@ async def serving(*args, **kwargs): async def lifecycle_run(): events.append("lifecycle started") + started.set() # Match the package loggers that WSS forwards into its session file. logging.getLogger("isaaccapture.cloudxr.oob_teleop_lifecycle").info( "lifecycle test transition" @@ -73,18 +83,28 @@ async def lifecycle_run(): setup_oob=True, usb_local=True, recovery_config=RecoveryConfig(), - on_listening=lambda: events.append("callback"), + on_listening=on_listening, + connect_requested=requested, ) ) - for _ in range(10): - await asyncio.sleep(0) - if "lifecycle started" in events: - break - assert not task.done() - assert events[:2] == ["wss listening", "callback"] - factory.assert_called_once() - stop.set_result(None) - await task + try: + await asyncio.wait_for(listening.wait(), timeout=1) + assert not task.done() + assert events[:2] == ["wss listening", "callback"] + if requested is not None: + factory.assert_not_called() + if connection_mode == "deferred": + requested.set() + if connection_mode != "cancelled": + await asyncio.wait_for(started.wait(), timeout=1) + factory.assert_called_once() + finally: + stop.set_result(None) + await asyncio.wait_for(task, timeout=1) + if connection_mode == "cancelled": + factory.assert_not_called() + assert events == ["wss listening", "callback", "wss closed"] + return assert events.index("lifecycle stopped") < events.index("wss closed") assert "lifecycle test transition" in (tmp_path / "wss.log").read_text() assert "hub test registration" in (tmp_path / "wss.log").read_text() diff --git a/tests/python/core/teleop_session_manager/test_external_oxr_handles.py b/tests/python/core/teleop_session_manager/test_external_oxr_handles.py index 1c7d4942a5..a1537976a7 100644 --- a/tests/python/core/teleop_session_manager/test_external_oxr_handles.py +++ b/tests/python/core/teleop_session_manager/test_external_oxr_handles.py @@ -20,6 +20,7 @@ from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch +import pytest from isaaccapture.teleop_session_manager import TeleopSession, TeleopSessionConfig @@ -124,6 +125,54 @@ def test_oxr_handles_accepts_stub(self): class TestTeleopSessionExternalHandles: """Tests for TeleopSession when external OpenXR handles are provided.""" + def test_factory_receives_aggregated_extensions_before_deviceio(self): + handles = _make_stub_handles() + tracker = MagicMock() + factory = MagicMock(return_value=handles) + config = TeleopSessionConfig( + app_name="test", + pipeline=_make_empty_pipeline(), + trackers=[tracker], + oxr_session_factory=factory, + ) + with ( + _mock_deviceio_and_oxr() as mocks, + patch( + "isaaccapture.deviceio.DeviceIOSession.get_required_extensions", + return_value=["XR_EXT_hand_tracking", "XR_NVX1_push_tensor"], + ) as extensions, + ): + factory.side_effect = lambda required: ( + mocks.deviceio_run.assert_not_called() or handles + ) + with TeleopSession(config): + assert tracker in extensions.call_args.args[0] + factory.assert_called_once_with(extensions.return_value) + assert mocks.deviceio_run.call_args.args[1] is handles + mocks.oxr_cls.assert_not_called() + + def test_factory_failure_does_not_start_deviceio(self): + config = TeleopSessionConfig( + app_name="test", + pipeline=_make_empty_pipeline(), + oxr_session_factory=MagicMock(side_effect=RuntimeError("viz failed")), + ) + with _mock_deviceio_and_oxr() as mocks: + with pytest.raises(RuntimeError, match="viz failed"): + with TeleopSession(config): + pass + mocks.deviceio_run.assert_not_called() + + @pytest.mark.parametrize("field", ["oxr_handles", "joint_publisher"]) + def test_factory_rejects_competing_session_owners(self, field): + with pytest.raises(ValueError, match="oxr_session_factory requires LIVE"): + TeleopSessionConfig( + app_name="test", + pipeline=_make_empty_pipeline(), + oxr_session_factory=MagicMock(), + **{field: MagicMock()}, + ) + def test_skips_oxr_session_creation(self): """When oxr_handles is set, OpenXRSession.create() must not be called.""" external_handles = _make_stub_handles() From a683fe6f10cd0d27800e51df02f7fa8379887843 Mon Sep 17 00:00:00 2001 From: Farbod Motlagh Date: Fri, 9 Oct 2026 08:52:51 -0700 Subject: [PATCH 2/5] Use consistent void returns in the CloudXR WebSocket handler Signed-off-by: Farbod Motlagh --- src/python/isaaccapture/cloudxr/wss.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/python/isaaccapture/cloudxr/wss.py b/src/python/isaaccapture/cloudxr/wss.py index 47f203829c..4b7252deb5 100755 --- a/src/python/isaaccapture/cloudxr/wss.py +++ b/src/python/isaaccapture/cloudxr/wss.py @@ -679,12 +679,13 @@ async def report_fatal(error: Exception, circumstance: str) -> None: initial, ) - async 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 await hub.handle_connection(ws) + await hub.handle_connection(ws) + return # An already-open headset page may reconnect before ADB automation. if await _wait_for_connect(connect_requested, stop_future): await proxy_handler(ws, backend_host, backend_port) From 10eaf10c3da4c982fb53e9ac78519caa3171506a Mon Sep 17 00:00:00 2001 From: Farbod Motlagh Date: Fri, 9 Oct 2026 10:34:55 -0700 Subject: [PATCH 3/5] Clarify deferred OOB connection API and document startup ordering Signed-off-by: Farbod Motlagh --- docs/source/references/cloudxr.rst | 47 +++++++++++++++++++ src/python/isaaccapture/cloudxr/launcher.py | 18 ++++--- .../isaaccapture/cloudxr/service/_service.py | 17 ++++--- tests/python/core/cloudxr/test_launcher.py | 8 ++-- tests/python/core/cloudxr/test_service.py | 12 +++-- 5 files changed, 79 insertions(+), 23 deletions(-) diff --git a/docs/source/references/cloudxr.rst b/docs/source/references/cloudxr.rst index 658b5314b8..e2543ea116 100644 --- a/docs/source/references/cloudxr.rst +++ b/docs/source/references/cloudxr.rst @@ -276,6 +276,53 @@ 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 +------------------------------------------------- + +Applications that need their OpenXR session and renderer ready before the +headset connects can pass ``defer_oob_connect=True``. This Python option +requires ``setup_oob=True`` and, for ``CloudXRLauncher``, ``run_embedded=True``. +It is also available directly on ``CloudXRService``. The default is ``False``, +so existing applications still connect immediately. + +The runtime and WSS listener start immediately, but OOB automation and backend +WebSocket signaling wait for ``connect_headset()``. This also holds signaling +from an already-open headset page; the OOB control hub remains available. +Call ``connect_headset()`` after application initialization, then keep servicing +the application's frame loop: the call is non-blocking and does not wait for +the headset to finish connecting. Repeated calls are harmless. Stopping the +service before releasing the gate cancels the wait without launching the client. + +In this example, ``start_app()`` is an application-provided context manager +that creates the OpenXR session and waits for renderer initialization. Its +``app.run()`` method runs the frame loop, and context exit stops rendering +and destroys the session before CloudXR stops: + +.. code-block:: python + + from isaaccapture.cloudxr import CloudXRLauncher + + def run_with_cloudxr(start_app): + with CloudXRLauncher( + run_embedded=True, + setup_oob=True, + usb_local=True, + defer_oob_connect=True, + env_config="cloudxr.env", + accept_eula=True, + ) as cloudxr: + with start_app() as app: + cloudxr.connect_headset() + app.run() + +Set a fixed ``NV_DEVICE_PROFILE`` matching the headset in ``cloudxr.env`` +(for example, ``NV_DEVICE_PROFILE=Quest3``). An ``auto*`` profile requires a +connected client to discover the device and cannot support this startup order. +The gate applies to an embedded service; it cannot defer connections on an +already-running detached service. + Configuration ------------- diff --git a/src/python/isaaccapture/cloudxr/launcher.py b/src/python/isaaccapture/cloudxr/launcher.py index a7d3a0a0c9..b7f408cb5c 100644 --- a/src/python/isaaccapture/cloudxr/launcher.py +++ b/src/python/isaaccapture/cloudxr/launcher.py @@ -164,7 +164,7 @@ def __init__( host_client: bool = True, run_embedded: bool = False, start_wss_proxy: bool | None = None, - defer_connect: bool = False, + defer_oob_connect: bool = False, ) -> None: """Attach to the running runtime, or own one when *run_embedded*. @@ -176,8 +176,9 @@ def __init__( to one this process cannot configure. start_wss_proxy: Deprecated no-op; the proxy always starts with the runtime. - defer_connect: Wait for :meth:`connect_headset` before OOB headset - automation. Requires ``run_embedded=True`` and ``setup_oob=True``. + 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 @@ -187,13 +188,15 @@ 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_connect and not (run_embedded and setup_oob): + if defer_oob_connect and not (run_embedded and setup_oob): raise ValueError( - "defer_connect requires run_embedded=True and setup_oob=True" + "defer_oob_connect requires run_embedded=True and setup_oob=True" ) if start_wss_proxy is not None: self._warn_start_wss_proxy_deprecated() @@ -213,7 +216,7 @@ def __init__( setup_oob=setup_oob, usb_local=usb_local, host_client=host_client, - defer_connect=defer_connect, + defer_oob_connect=defer_oob_connect, ) return @@ -857,7 +860,8 @@ def stop(self) -> None: def connect_headset(self) -> None: """Release deferred OOB automation once the application's XR session is ready. - Non-blocking and idempotent; connection progress uses the WSS log as usual. + 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") diff --git a/src/python/isaaccapture/cloudxr/service/_service.py b/src/python/isaaccapture/cloudxr/service/_service.py index 68868f6e9e..3faa3df3a0 100644 --- a/src/python/isaaccapture/cloudxr/service/_service.py +++ b/src/python/isaaccapture/cloudxr/service/_service.py @@ -119,7 +119,7 @@ def __init__( usb_local: bool = False, host_client: bool = False, recovery_config=None, - defer_connect: bool = False, + defer_oob_connect: bool = False, ) -> None: """Start the CloudXR runtime and the WSS proxy. @@ -149,17 +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_connect: Hold OOB headset automation until :meth:`connect_headset`. - Requires *setup_oob*; the runtime and WSS listener start immediately. + 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_connect and not setup_oob: - raise ValueError("defer_connect requires setup_oob=True") - self._connect_requested = asyncio.Event() if defer_connect else None + 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 @@ -758,7 +761,7 @@ def _terminate_runtime_windows(_proc: subprocess.Popen) -> None: 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_connect=True") + 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") diff --git a/tests/python/core/cloudxr/test_launcher.py b/tests/python/core/cloudxr/test_launcher.py index b9ed4b3511..840b1ce552 100644 --- a/tests/python/core/cloudxr/test_launcher.py +++ b/tests/python/core/cloudxr/test_launcher.py @@ -145,8 +145,8 @@ def test_run_embedded_is_refused(self, tmp_path): "options", [{}, {"run_embedded": True}, {"setup_oob": True}] ) def test_deferred_connect_requires_an_owned_oob_service(self, options): - with pytest.raises(ValueError, match="defer_connect requires"): - CloudXRLauncher(defer_connect=True, **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 ( @@ -154,9 +154,9 @@ def test_deferred_connect_delegates_to_embedded_service(self): patch("isaaccapture.cloudxr.launcher.CloudXRService") as service, ): launcher = CloudXRLauncher( - run_embedded=True, setup_oob=True, defer_connect=True + run_embedded=True, setup_oob=True, defer_oob_connect=True ) - assert service.call_args.kwargs["defer_connect"] is True + assert service.call_args.kwargs["defer_oob_connect"] is True launcher.connect_headset() service.return_value.connect_headset.assert_called_once_with() diff --git a/tests/python/core/cloudxr/test_service.py b/tests/python/core/cloudxr/test_service.py index e645eea14d..98b56b70fa 100644 --- a/tests/python/core/cloudxr/test_service.py +++ b/tests/python/core/cloudxr/test_service.py @@ -468,9 +468,9 @@ def test_a_proxy_that_never_listens_times_out(self, tmp_path): # The thread outlives the timeout; stop() is what reaps it. service._stop_wss_proxy() - @pytest.mark.parametrize("defer_connect", [False, True]) + @pytest.mark.parametrize("defer_oob_connect", [False, True]) def test_headset_connection_is_explicit_only_when_deferred( - self, tmp_path, defer_connect + self, tmp_path, defer_oob_connect ): from cloudxr_py_test_ns.wss import _wait_for_connect @@ -487,9 +487,11 @@ async def proxy(*, stop_future, on_listening, connect_requested, **kwargs): _stub_wss(proxy), patch("isaaccapture.cloudxr.oob_teleop_adb.require_adb_on_path"), ): - service = CloudXRService(setup_oob=True, defer_connect=defer_connect) + service = CloudXRService( + setup_oob=True, defer_oob_connect=defer_oob_connect + ) try: - if defer_connect: + if defer_oob_connect: assert not connected.wait(0.05) service.connect_headset() service.connect_headset() # Releasing the same gate is harmless. @@ -512,7 +514,7 @@ async def proxy(*, stop_future, on_listening, connect_requested, **kwargs): _stub_wss(proxy), patch("isaaccapture.cloudxr.oob_teleop_adb.require_adb_on_path"), ): - service = CloudXRService(setup_oob=True, defer_connect=True) + service = CloudXRService(setup_oob=True, defer_oob_connect=True) thread = service._wss_thread service._stop_wss_proxy() assert not thread.is_alive() From 76017d7d6de69c35a7307d9a9f8805bc32a25e0e Mon Sep 17 00:00:00 2001 From: Farbod Motlagh Date: Mon, 12 Oct 2026 10:37:36 -0700 Subject: [PATCH 4/5] Keep deferred OOB connection docs concise Signed-off-by: Farbod Motlagh --- docs/AGENTS.md | 3 ++ docs/source/references/cloudxr.rst | 48 +++++------------------------- 2 files changed, 10 insertions(+), 41 deletions(-) diff --git a/docs/AGENTS.md b/docs/AGENTS.md index 9f7f806fa1..769b9578dd 100644 --- a/docs/AGENTS.md +++ b/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 diff --git a/docs/source/references/cloudxr.rst b/docs/source/references/cloudxr.rst index e2543ea116..e71b79d900 100644 --- a/docs/source/references/cloudxr.rst +++ b/docs/source/references/cloudxr.rst @@ -281,47 +281,13 @@ alongside the check for a non-interactive stdin, so automated runs never wait. Connect the OOB headset after application startup ------------------------------------------------- -Applications that need their OpenXR session and renderer ready before the -headset connects can pass ``defer_oob_connect=True``. This Python option -requires ``setup_oob=True`` and, for ``CloudXRLauncher``, ``run_embedded=True``. -It is also available directly on ``CloudXRService``. The default is ``False``, -so existing applications still connect immediately. - -The runtime and WSS listener start immediately, but OOB automation and backend -WebSocket signaling wait for ``connect_headset()``. This also holds signaling -from an already-open headset page; the OOB control hub remains available. -Call ``connect_headset()`` after application initialization, then keep servicing -the application's frame loop: the call is non-blocking and does not wait for -the headset to finish connecting. Repeated calls are harmless. Stopping the -service before releasing the gate cancels the wait without launching the client. - -In this example, ``start_app()`` is an application-provided context manager -that creates the OpenXR session and waits for renderer initialization. Its -``app.run()`` method runs the frame loop, and context exit stops rendering -and destroys the session before CloudXR stops: - -.. code-block:: python - - from isaaccapture.cloudxr import CloudXRLauncher - - def run_with_cloudxr(start_app): - with CloudXRLauncher( - run_embedded=True, - setup_oob=True, - usb_local=True, - defer_oob_connect=True, - env_config="cloudxr.env", - accept_eula=True, - ) as cloudxr: - with start_app() as app: - cloudxr.connect_headset() - app.run() - -Set a fixed ``NV_DEVICE_PROFILE`` matching the headset in ``cloudxr.env`` -(for example, ``NV_DEVICE_PROFILE=Quest3``). An ``auto*`` profile requires a -connected client to discover the device and cannot support this startup order. -The gate applies to an embedded service; it cannot defer connections on an -already-running detached service. +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 ------------- From a15f2fdb153e0ee629ef425b778927963cee9b11 Mon Sep 17 00:00:00 2001 From: Farbod Motlagh Date: Mon, 12 Oct 2026 13:50:39 -0700 Subject: [PATCH 5/5] Stop waiting for deferred signaling when clients disconnect Signed-off-by: Farbod Motlagh --- src/python/isaaccapture/cloudxr/wss.py | 15 ++- tests/python/core/cloudxr/test_service.py | 124 ++++++++++++++++++++-- 2 files changed, 129 insertions(+), 10 deletions(-) diff --git a/src/python/isaaccapture/cloudxr/wss.py b/src/python/isaaccapture/cloudxr/wss.py index 4b7252deb5..c76b271c6e 100755 --- a/src/python/isaaccapture/cloudxr/wss.py +++ b/src/python/isaaccapture/cloudxr/wss.py @@ -687,8 +687,19 @@ async def handler(ws) -> None: await hub.handle_connection(ws) return # An already-open headset page may reconnect before ADB automation. - if await _wait_for_connect(connect_requested, stop_future): - await proxy_handler(ws, backend_host, backend_port) + 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). diff --git a/tests/python/core/cloudxr/test_service.py b/tests/python/core/cloudxr/test_service.py index 98b56b70fa..d32eea5385 100644 --- a/tests/python/core/cloudxr/test_service.py +++ b/tests/python/core/cloudxr/test_service.py @@ -636,16 +636,28 @@ def test_sigint_handler_with_sig_ign_prev_is_noop(self, tmp_path): @pytest.mark.asyncio -@pytest.mark.parametrize("connect", [True, False]) -async def test_early_websocket_waits_for_app_or_shutdown(monkeypatch, connect): +@pytest.mark.parametrize( + "outcome", ["connect", "shutdown", "disconnect", "disconnect_and_connect", "cancel"] +) +async def test_early_websocket_waits_for_app_or_disconnect(monkeypatch, outcome): from unittest.mock import AsyncMock from cloudxr_py_test_ns import wss requested = asyncio.Event() listening = asyncio.Event() + closed = asyncio.Event() + watching_close = asyncio.Event() + close_wait_finished = asyncio.Event() stop = asyncio.get_running_loop().create_future() handlers = [] + async def wait_closed(): + watching_close.set() + try: + await closed.wait() + finally: + close_wait_finished.set() + @contextlib.asynccontextmanager async def serve(handler, **kwargs): handlers.append(handler) @@ -666,18 +678,114 @@ async def serve(handler, **kwargs): connect_requested=requested, ) ) + client = None try: await asyncio.wait_for(listening.wait(), timeout=1) - client = asyncio.create_task(handlers[0](object())) - await asyncio.sleep(0) + client = asyncio.create_task( + handlers[0](types.SimpleNamespace(wait_closed=wait_closed)) + ) + await asyncio.wait_for(watching_close.wait(), timeout=1) proxy.assert_not_called() - if connect: + if outcome in ("disconnect", "disconnect_and_connect"): + closed.set() + if outcome in ("connect", "disconnect_and_connect"): requested.set() - else: + if outcome == "shutdown": stop.set_result(None) - await asyncio.wait_for(client, timeout=1) - assert proxy.call_count == int(connect) + if outcome == "cancel": + client.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(client, timeout=1) + else: + await asyncio.wait_for(client, timeout=1) + assert proxy.call_count == int(outcome == "connect") + assert close_wait_finished.is_set() + assert stop.done() == (outcome == "shutdown") finally: + if client is not None: + client.cancel() + await asyncio.gather(client, return_exceptions=True) if not stop.done(): stop.set_result(None) await asyncio.wait_for(task, timeout=1) + + +@pytest.mark.asyncio +async def test_disconnected_websockets_do_not_forward_stale_offers(monkeypatch): + from websockets.asyncio.client import connect + from websockets.asyncio.server import serve + from cloudxr_py_test_ns import wss + + requested = asyncio.Event() + listening = asyncio.Event() + stop = asyncio.get_running_loop().create_future() + finished = asyncio.Queue() + connections = [] + offers = [] + proxy_port = None + + async def backend_handler(ws): + connections.append(ws.request.path) + async for offer in ws: + offers.append(offer) + await ws.send(offer) + + @contextlib.asynccontextmanager + async def serve_proxy(handler, **kwargs): + nonlocal proxy_port + + async def tracked_handler(ws): + try: + await handler(ws) + finally: + finished.put_nowait(ws.request.path) + + kwargs.update(host="127.0.0.1", port=0) + async with serve(tracked_handler, **kwargs) as server: + proxy_port = server.sockets[0].getsockname()[1] + yield + + monkeypatch.setattr(wss, "default_cert_paths", lambda: None) + monkeypatch.setattr(wss, "ensure_certificate", lambda *args: None) + monkeypatch.setattr(wss, "build_ssl_context", lambda *args: None) + monkeypatch.setattr(wss, "ws_serve", serve_proxy) + + async with serve(backend_handler, "127.0.0.1", 0) as backend: + task = asyncio.create_task( + wss.run( + None, + stop, + backend_host="127.0.0.1", + backend_port=backend.sockets[0].getsockname()[1], + proxy_port=0, + on_listening=listening.set, + connect_requested=requested, + ) + ) + try: + await asyncio.wait_for(listening.wait(), timeout=2) + uri = f"ws://127.0.0.1:{proxy_port}" + for index in range(3): + async with connect(f"{uri}/stale-{index}") as client: + await client.send(f"stale-offer-{index}") + + # Disconnected handlers must exit while the startup gate is still held. + exited = { + await asyncio.wait_for(finished.get(), timeout=2) for _ in range(3) + } + assert exited == {f"/stale-{index}" for index in range(3)} + assert not connections + assert not offers + + async with connect(f"{uri}/live") as client: + await client.send("fresh-offer") + requested.set() + assert await asyncio.wait_for(client.recv(), timeout=2) == "fresh-offer" + assert await asyncio.wait_for(finished.get(), timeout=2) == "/live" + finally: + if not stop.done(): + stop.set_result(None) + await asyncio.wait_for(task, timeout=2) + + assert connections == ["/live"] + assert offers == ["fresh-offer"]