From c98a342fe596dd2829ff8d45b30734f532a23297 Mon Sep 17 00:00:00 2001 From: Andrew Wilson Date: Wed, 7 Oct 2026 16:07:38 -0500 Subject: [PATCH 1/9] feat(cloudxr): check USB health before local OOB browser Signed-off-by: Andrew Wilson --- .pre-commit-config.yaml | 4 +- AGENTS.md | 3 + docs/source/references/cloudxr.rst | 24 + .../isaaccapture/cloudxr/oob_teleop_adb.py | 10 +- .../cloudxr/oob_teleop_lifecycle.py | 454 +++++++++++++- src/python/isaaccapture/cloudxr/usb_health.py | 429 +++++++++++++ src/python/isaaccapture/cloudxr/wss.py | 6 + tests/python/core/cloudxr/test_usb_health.py | 574 ++++++++++++++++++ 8 files changed, 1497 insertions(+), 7 deletions(-) create mode 100644 src/python/isaaccapture/cloudxr/usb_health.py create mode 100644 tests/python/core/cloudxr/test_usb_health.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 41849867c0..95942cfb0e 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -24,8 +24,8 @@ repos: - id: check-executables-have-shebangs - id: check-toml - id: end-of-file-fixer - # This LFS SVG is an asset; rewriting its final byte creates a new LFS object. - exclude: ^docs/source/_static/camera-viz-controls\.svg$ + # These LFS SVGs are assets; rewriting their final byte creates new LFS objects. + exclude: ^(docs/source/_static/camera-viz-controls|examples/vehicle_teleop/assets/vehicle_teleop_flow)\.svg$ - id: check-shebang-scripts-are-executable - id: detect-private-key - id: debug-statements diff --git a/AGENTS.md b/AGENTS.md index b75eea35c6..4bf6027521 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -226,6 +226,9 @@ pre-commit install --hook-type commit-msg `git ls-files`, so a file you have created but not yet `git add`ed is skipped and the run passes for the wrong reason — REUSE and `ruff format` alike. Stage first (`git add -A`), then run the hooks. +- LFS SVG assets without a trailing newline need an `end-of-file-fixer` exclusion; + rewriting them creates an unrelated LFS object. Verify the LFS attribute and + restore any hook-only asset change before rerunning. - **REUSE:** files covered by the REUSE hook need **`SPDX-FileCopyrightText`** and **`SPDX-License-Identifier`** in the form the repo already uses (for example the HTML comment block at the top of `README.md` also applies to **`AGENTS.md`** and similar docs). - **C++ formatting is enforced by CI, not pre-commit.** The hook set runs `ruff` for Python but does **not** run `clang-format`; CI (`build-ubuntu.yml`) installs **`clang-format-14`** and rejects unformatted C++ as `-Wclang-format-violations`. Before pushing, format touched C++ with the system `clang-format` (match CI's version 14) and verify: diff --git a/docs/source/references/cloudxr.rst b/docs/source/references/cloudxr.rst index 707d1eb987..76d9f41292 100644 --- a/docs/source/references/cloudxr.rst +++ b/docs/source/references/cloudxr.rst @@ -135,6 +135,30 @@ control how the headset connects and how the web client is delivered. prints a banner as it comes up, so ``run`` is often the more useful command while setting those modes up for the first time. +USB-local startup checks +------------------------ + +On a clean ``--usb-local`` start, the service records the selected headset's +negotiated USB speed, hub and controller path, and evidence of faster USB +root hubs on the same controller. After its existing adb reverse rules are +ready, it measures three seconds of host-to-headset transfer using the +already served ``/client/bundle.js`` over HTTPS. This finishes before the +browser is opened. The selected headset must first report +``mWakefulness=Awake``; if that cannot be confirmed, no transfer or browser +launch occurs and the JSON report records the incomplete check. If the headset +lacks a usable curl or timer, the service uses a three-second ADB-only transfer +and labels the narrower coverage in the result. + +One JSON report is saved under ``/logs/usb-health/`` for +each USB-local start. It begins before ADB selection, records partial results +when device preparation or a check fails, and refreshes link topology after +a preserved browser reconnect without repeating the active transfer. It +uses a hashed device serial, and keeps at most 20 reports or 14 days of +history. A link below 5000 Mb/s or an incomplete transfer is logged as a +warning and startup continues. The measured Mbps is one-way application +goodput, not USB line rate or CloudXR stream throughput. A faster root hub +is a topology clue, not proof that an accessible physical port is free. + Re-open the client on the headset --------------------------------- diff --git a/src/python/isaaccapture/cloudxr/oob_teleop_adb.py b/src/python/isaaccapture/cloudxr/oob_teleop_adb.py index c2eb0a0ff5..19a1d9cd28 100644 --- a/src/python/isaaccapture/cloudxr/oob_teleop_adb.py +++ b/src/python/isaaccapture/cloudxr/oob_teleop_adb.py @@ -339,7 +339,7 @@ def headset_wakefulness() -> str: return m.group(1) if m else "" -def assert_headset_awake(*, timeout: float = 15.0) -> None: +def assert_headset_awake(*, timeout: float = 15.0, require_awake: bool = False) -> None: """Warn-and-wait when the headset is asleep before launching OOB automation. Quest / PICO devices sleep when the proximity sensor is uncovered @@ -350,7 +350,8 @@ def assert_headset_awake(*, timeout: float = 15.0) -> None: Sends ``KEYCODE_WAKEUP`` once and then polls ``mWakefulness`` for up to ``timeout`` seconds. Returns silently once the device is ``Awake``. Otherwise logs a warning and returns — downstream automation may still - succeed if ``am start`` wakes the device. + succeed if ``am start`` wakes the device. When ``require_awake`` is set, + raise instead; pre-browser USB transfer must have an observed Awake state. """ wake = headset_wakefulness() if wake == "Awake": @@ -383,10 +384,13 @@ def assert_headset_awake(*, timeout: float = 15.0) -> None: return log.warning( - "Headset still appears asleep after %.0fs (wakefulness=%s); continuing anyway.", + "Headset still appears asleep after %.0fs (wakefulness=%s); %s.", timeout, wake or "?", + "aborting USB transfer" if require_awake else "continuing anyway", ) + if require_awake: + raise OobAdbError("Headset wakefulness was not confirmed as Awake") def adb_device_state() -> str: diff --git a/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py b/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py index 8f640b516e..65321b3893 100644 --- a/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py +++ b/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py @@ -9,13 +9,16 @@ import logging import os import re +import secrets import socket import time +from pathlib import Path from collections.abc import Awaitable, Callable from inspect import isawaitable from dataclasses import dataclass from . import oob_teleop_adb as adb +from . import usb_health from .oob_teleop_env import ( USB_TURN_CREDENTIAL, USB_TURN_USER, @@ -91,6 +94,9 @@ def __init__( sleep: Callable[[float], object] = asyncio.sleep, metrics_stale_sec: float = 5.0, host_listener_probe: Callable[[int], bool] = _host_listener_ready, + usb_health_logs_dir: Path | None = None, + web_client_static_dir: Path | None = None, + usb_health_policy: str = "warn", ) -> None: # Immutable topology and injected collaborators. self.hub = hub @@ -104,6 +110,21 @@ def __init__( self.sleep = sleep self.metrics_stale_sec = metrics_stale_sec self.host_listener_probe = host_listener_probe + self.usb_health_logs_dir = usb_health_logs_dir + self.web_client_static_dir = web_client_static_dir + if usb_health_policy not in {"off", "warn", "block"}: + raise ValueError("usb_health_policy must be off, warn, or block") + self.usb_health_policy = usb_health_policy + self._usb_health_report: usb_health.UsbHealthReport | None = None + self._usb_health_session = secrets.token_hex(16) + self._usb_report_serial: str | None = None + self._usb_wait_signature: tuple[str, str] | None = None + self._usb_link_history: list[dict] = [] + self._usb_transfer_history: list[dict] = [] + self._usb_transfer_stale = False + self._usb_wake_confirmed: bool | None = None + self._usb_link: dict | None = None + self._usb_transfer: dict | None = None # Headset selection remains pinned after the first successful choice. self.selected = os.environ.get("ANDROID_SERIAL", "").strip() or None @@ -205,6 +226,31 @@ async def _publish(self, health: str, state: str, reason: str, **flags) -> None: "connectDispatched": self.connect_dispatched, "streaming": False, "clientMetricsFresh": False, + "usbHealthReport": ( + self._usb_health_report.path.name if self._usb_health_report else None + ), + "usbNegotiatedSpeedMbps": ( + self._usb_link.get("negotiatedSpeedMbps") if self._usb_link else None + ), + "usbTransferMbps": ( + self._usb_transfer.get("hostToHeadsetMbps") + if self._usb_transfer and not self._usb_transfer_stale + else None + ), + "usbLinkOutcome": self._link_outcome(), + "usbTransferOutcome": ( + "stale" + if self._usb_transfer_stale + else self._usb_transfer.get("outcome") + if self._usb_transfer + else None + ), + "usbTransferMethod": ( + self._usb_transfer.get("method") + if self._usb_transfer and not self._usb_transfer_stale + else None + ), + "usbWakeConfirmed": self._usb_wake_confirmed, **flags, } self.snapshot = snapshot @@ -242,6 +288,26 @@ async def _stop_monitor(self) -> None: async def _remember_transport_loss(self) -> None: """Drop transport-owned state while preserving the prior browser intent.""" + if self._usb_wake_confirmed is not None: + self._usb_wake_confirmed = None + if self._usb_health_report is not None: + previous = self._usb_health_report.payload.get("wakeCheck") + history = list(self._usb_health_report.payload.get("wakeHistory", [])) + if previous is not None: + history.append(dict(previous)) + self._persist_usb_health( + wakeHistory=history[-20:], + wakeCheck={ + "confirmedAwake": None, + "outcome": "unknown_after_transport_loss", + }, + decision={ + "overallOutcome": "incomplete", + "browserLaunchAllowed": None, + "shortMessage": "USB transport lost; current wakefulness is unknown.", + }, + ) + self._record_usb_event("USB_WAKE_INVALIDATED") if not self._transport_lost: self._restore_existing_browser = bool( self.browser_ready @@ -443,6 +509,324 @@ async def _prepare_device(self) -> None: ) self._cache_cleared = True + def _ensure_usb_report(self) -> None: + if ( + not self.usb_local + or self.usb_health_policy == "off" + or self.usb_health_logs_dir is None + ): + return + try: + if self._usb_health_report is None: + self._usb_health_report = usb_health.UsbHealthReport( + self.usb_health_logs_dir, + self._usb_health_session, + self.generation, + None, + self.usb_health_policy, + ) + self._usb_health_report.event("WAITING_FOR_ADB") + if self.selected and self.selected != self._usb_report_serial: + self._usb_health_report.set_device(self.selected) + self._usb_report_serial = self.selected + except OSError: + log.warning( + "USB health report could not be created or updated", exc_info=True + ) + + def _record_usb_wait(self, state: str, code: str) -> None: + self._ensure_usb_report() + signature = (state, code) + if self._usb_health_report is None or signature == self._usb_wait_signature: + return + self._usb_wait_signature = signature + self._record_usb_event(state) + errors = self._usb_health_report.payload["errors"] + errors.append({"state": state, "code": code, "at": time.time()}) + self._persist_usb_health( + errors=errors[-20:], + decision={ + "overallOutcome": "incomplete", + "browserLaunchAllowed": bool( + self.browser_ready or self.connect_dispatched + ), + "shortMessage": f"USB-local startup waiting at {state}.", + }, + ) + + def _link_outcome(self) -> str | None: + if self._usb_link is None: + return None + speed = self._usb_link.get("negotiatedSpeedMbps") + return "unknown" if speed is None else "pass" if speed >= 5000 else "slow_link" + + async def _check_usb_link(self, *, passive: bool = False) -> None: + """Record Linux USB topology for the selected ADB transport.""" + assert self.selected is not None + started = self.clock() + await self._publish( + "degraded", + "CHECKING_USB_LINK", + "Refreshing negotiated USB link after reconnect" + if passive + else "Inspecting negotiated USB link", + adbReady=True, + networkPresent=True, + ) + self._ensure_usb_report() + self._record_usb_event( + "REFRESHING_USB_LINK" if passive else "CHECKING_USB_LINK" + ) + try: + listing = await asyncio.to_thread( + adb._adb_run, + ["adb", "devices", "-l"], + capture_output=True, + text=True, + timeout=4, + check=False, + ) + if listing.returncode: + raise adb.OobAdbError("ADB device listing failed") + self._usb_link = await asyncio.to_thread( + usb_health.inspect_link, self.selected, listing.stdout or "" + ) + except (OSError, TimeoutError, adb.OobAdbError) as exc: + self._usb_link = { + "probeSucceeded": False, + "negotiatedSpeedMbps": None, + "isBetterPortAvailable": None, + "errors": [str(exc)[:160]], + } + self._usb_link["observedAt"] = time.time() + self._usb_link["probeDurationMs"] = round((self.clock() - started) * 1000, 2) + self._usb_link["passiveRefresh"] = passive + self._usb_link_history.append(dict(self._usb_link)) + self._usb_link_history = self._usb_link_history[-32:] + self._usb_wait_signature = None + speed = self._usb_link.get("negotiatedSpeedMbps") + outcome = self._link_outcome() + if passive and self._usb_transfer is not None and not self._usb_transfer_stale: + self._usb_transfer_history.append(dict(self._usb_transfer)) + self._usb_transfer_history = self._usb_transfer_history[-20:] + self._usb_transfer_stale = True + self._persist_usb_health( + topology=self._usb_link, + linkHistory=self._usb_link_history, + transferHistory=self._usb_transfer_history, + transferTest=( + { + "attempted": False, + "completed": False, + "outcome": "pending_retest", + "method": None, + "hostToHeadsetMbps": None, + "headsetToHostMbps": None, + "staleAfterReconnect": True, + } + if self._usb_transfer_stale and self._usb_transfer is not None + else self._usb_transfer + ), + linkProbe={ + "probeSucceeded": self._usb_link.get("probeSucceeded", False), + "negotiatedSpeedMbps": speed, + "isBetterPortAvailable": self._usb_link.get("isBetterPortAvailable"), + "outcome": outcome, + "shortMessage": ( + f"USB link negotiated at {speed:g} Mb/s." + if speed is not None + else "Negotiated USB link could not be determined." + ), + }, + **( + { + "decision": { + "overallOutcome": ( + "incomplete" if outcome == "pass" else "degraded" + ), + "browserLaunchAllowed": None, + "existingBrowserPreserved": True, + "transferRetestDeferred": True, + "shortMessage": "USB link refreshed; active transfer pending retest.", + } + } + if passive + else {} + ), + ) + self._record_usb_event("USB_LINK_COMPLETE") + # The state published before inspection still contains the old speed + # during reconnect; publish again with the refreshed link evidence. + await self._publish( + "degraded", + "USB_LINK_CHECKED", + ( + f"USB link negotiated at {speed:g} Mb/s" + if speed is not None + else "USB link speed unavailable" + ), + adbReady=True, + networkPresent=True, + ) + log.log( + logging.WARNING if outcome != "pass" else logging.INFO, + "USB-local negotiated link: %s Mb/s; faster root available: %s; report: %s", + self._usb_link.get("negotiatedSpeedMbps"), + self._usb_link.get("isBetterPortAvailable"), + self._usb_health_report.path if self._usb_health_report else "unavailable", + ) + + async def _check_usb_transfer(self) -> None: + """Use the existing static listener and verified reverse before browser launch.""" + assert self.selected is not None + if self._usb_wake_confirmed is not True: + self._persist_usb_health( + transferTest={ + "attempted": False, + "completed": False, + "outcome": "not_started_headset_not_awake", + "requestedMeasurementDurationMs": 3000, + "hostToHeadsetMbps": None, + } + ) + self._record_usb_wait("WAITING_FOR_AWAKE", "awake_not_confirmed") + raise adb.OobAdbError("USB transfer requires confirmed Awake wakefulness") + await self._publish( + "degraded", + "CHECKING_USB_TRANSFER", + "Measuring three-second pre-browser host-to-headset transfer", + adbReady=True, + networkPresent=True, + reverseRulesVerified=True, + ) + self._record_usb_event("CHECKING_USB_TRANSFER") + try: + if self.web_client_static_dir is None: + raise OSError("Static web client directory is unavailable") + size = (self.web_client_static_dir / "bundle.js").stat().st_size + self._usb_transfer = await usb_health.measure_transfer( + self.selected, self.resolved_port, size + ) + except asyncio.CancelledError: + self._usb_transfer = { + "attempted": True, + "completed": False, + "outcome": "cancelled", + "requestedMeasurementDurationMs": 3000, + "errors": ["USB health measurement cancelled"], + } + self._persist_usb_health(transferTest=self._usb_transfer) + raise + except (OSError, TimeoutError) as exc: + self._usb_transfer = { + "attempted": True, + "completed": False, + "outcome": "unsupported", + "requestedMeasurementDurationMs": 3000, + "errors": [str(exc)[:160]], + } + self._usb_transfer_stale = False + self._persist_usb_health(transferTest=self._usb_transfer) + self._record_usb_event("USB_TRANSFER_COMPLETE") + log.log( + logging.WARNING + if not self._usb_transfer.get("completed") + else logging.INFO, + "USB-local pre-browser host-to-headset application goodput: %s Mb/s (%s)", + self._usb_transfer.get("hostToHeadsetMbps"), + self._usb_transfer.get("pathCoverage"), + ) + + async def _confirm_usb_awake(self) -> None: + """Require a fresh Awake observation immediately before active transfer.""" + self._usb_wake_confirmed = False + started = self.clock() + await self._publish( + "degraded", + "CHECKING_USB_WAKE", + "Confirming selected headset is awake before USB transfer", + adbReady=True, + networkPresent=True, + reverseRulesVerified=True, + ) + self._record_usb_event("CHECKING_USB_WAKE") + try: + # Device preparation's soft wake helper may return while still asleep. + # Require mWakefulness=Awake after reverse setup and before transfer. + await asyncio.to_thread( + adb.assert_headset_awake, timeout=10.0, require_awake=True + ) + except (Exception, asyncio.CancelledError): + self._persist_usb_health( + wakeCheck={ + "confirmedAwake": False, + "source": "adb-shell-dumpsys-power", + "durationMs": round((self.clock() - started) * 1000, 2), + }, + transferTest={ + "attempted": False, + "completed": False, + "outcome": "not_started_headset_not_awake", + "requestedMeasurementDurationMs": 3000, + "hostToHeadsetMbps": None, + }, + ) + self._record_usb_wait("WAITING_FOR_AWAKE", "awake_not_confirmed") + raise + self._usb_wake_confirmed = True + self._persist_usb_health( + wakeCheck={ + "confirmedAwake": True, + "source": "adb-shell-dumpsys-power", + "durationMs": round((self.clock() - started) * 1000, 2), + } + ) + self._record_usb_event("USB_WAKE_CONFIRMED") + + def _persist_usb_health(self, **sections) -> None: + if self._usb_health_report is None: + return + try: + self._usb_health_report.update(**sections) + except OSError: + log.warning("USB health report update failed", exc_info=True) + + def _record_usb_event(self, state: str) -> None: + if self._usb_health_report is None: + return + try: + self._usb_health_report.event(state) + except OSError: + log.warning("USB health report event update failed", exc_info=True) + + def _usb_launch_allowed(self) -> bool: + speed = self._usb_link.get("negotiatedSpeedMbps") if self._usb_link else None + transfer_ok = bool(self._usb_transfer and self._usb_transfer.get("completed")) + degraded = speed is None or speed < 5000 or not transfer_ok + allowed = self.usb_health_policy != "block" or not degraded + if degraded: + log.warning( + "USB-local health is degraded: link=%s, transfer=%s; continuing=%s", + self._link_outcome(), + self._usb_transfer.get("outcome") if self._usb_transfer else None, + allowed, + ) + self._persist_usb_health( + decision={ + "overallOutcome": "degraded" if degraded else "pass", + "browserLaunchAllowed": allowed, + "shortMessage": ( + "USB health checks are degraded" + if degraded + else "USB health checks passed" + ), + } + ) + self._record_usb_event( + "USB_HEALTH_COMPLETE" if allowed else "USB_HEALTH_BLOCKED" + ) + return allowed + async def _automate(self) -> None: await self._stop_monitor() self._transport_lost = False @@ -945,6 +1329,7 @@ async def run(self) -> None: """ token = adb.SELECTED_ADB_SERIAL.set(self.selected) if self.selected else None try: + self._ensure_usb_report() if not self.snapshot: await self._publish( "starting", "WAITING_FOR_ADB", "Waiting for headset" @@ -956,6 +1341,7 @@ async def run(self) -> None: if self.selected is None and len(ready) == 1: self.selected = ready[0] token = adb.SELECTED_ADB_SERIAL.set(self.selected) + self._ensure_usb_report() self.ignored_serials = ( tuple( sorted( @@ -1018,6 +1404,18 @@ async def run(self) -> None: reason = "Headset offline; reconnect the USB cable" else: reason = devices.diagnostic or "Waiting for selected headset" + wait_code = ( + "ambiguous_devices" + if self.selected is None and len(ready) > 1 + else "unauthorized" + if selected_state == "unauthorized" + else "offline" + if selected_state == "offline" + else "selected_missing" + if self.selected + else "no_ready_device" + ) + self._record_usb_wait("WAITING_FOR_ADB", wait_code) await self._enter_transport_recovery( reason, "WAITING_FOR_ADB", @@ -1038,6 +1436,7 @@ async def run(self) -> None: self._restart_episode() self.last_network_state = network.state if network.state is adb.HeadsetNetworkState.ADB_UNAVAILABLE: + self._record_usb_wait("WAITING_FOR_ADB", "adb_unavailable") await self._enter_transport_recovery( "ADB unavailable: " + network.diagnostic, "WAITING_FOR_ADB", @@ -1054,6 +1453,7 @@ async def run(self) -> None: ): # Preserve the page while the headset network returns; # transport repair runs before any browser fallback. + self._record_usb_wait("WAITING_FOR_NETWORK", "network_absent") await self._enter_transport_recovery( "Headset has no non-loopback network", "PREPARING_DEVICE", @@ -1130,6 +1530,10 @@ async def run(self) -> None: await asyncio.to_thread( adb.assert_headset_awake, timeout=10.0 ) + if self.usb_health_policy != "off": + # A surviving page gets passive link evidence; a new + # three-second transfer would disturb its recovery. + await self._check_usb_link(passive=True) await self._rebuild_usb() self._repair_started_at = time.time() self._client_grace_deadline = ( @@ -1155,6 +1559,7 @@ async def run(self) -> None: and not self.browser_ready and not self.connect_dispatched ): + self._record_usb_wait("WAITING_FOR_ADB", "recovery_deadline") await self._publish( "degraded", "WAITING_FOR_ADB", @@ -1186,9 +1591,48 @@ async def run(self) -> None: self.episode_start + self.config.timeout_sec - self.clock() ) async with asyncio.timeout(max(0.001, remaining)): - await self._prepare_device() + self._record_usb_event("PREPARING_DEVICE") + if self.usb_local: + self._usb_wake_confirmed = False + try: + await self._prepare_device() + except (Exception, asyncio.CancelledError) as exc: + self._record_usb_wait( + "PREPARING_DEVICE", type(exc).__name__ + ) + raise if self.usb_local: - await self._rebuild_usb() + if self.usb_health_policy != "off": + await self._check_usb_link() + try: + await self._rebuild_usb() + except (Exception, asyncio.CancelledError) as exc: + self._persist_usb_health( + errors=[ + f"USB reverse setup failed: {type(exc).__name__}" + ], + decision={ + "overallOutcome": "incomplete", + "browserLaunchAllowed": False, + }, + ) + raise + if self.usb_health_policy != "off": + # _automate opens the browser; qualify the existing + # HTTPS reverse path before changing headset UI state. + await self._confirm_usb_awake() + await self._check_usb_transfer() + if not self._usb_launch_allowed(): + await self._publish( + "degraded", + "USB_HEALTH_BLOCKED", + "USB-local health policy blocked browser launch", + adbReady=True, + networkPresent=True, + reverseRulesVerified=True, + ) + await self.sleep(self.config.interval_sec) + continue await self._automate() except _TransportDisrupted as exc: # Enumeration can miss a quick cable flap. Reverse/TURN @@ -1232,6 +1676,12 @@ async def run(self) -> None: ) await self.sleep(self.config.interval_sec) finally: + if ( + self._usb_health_report is not None + and self._usb_health_report.payload["decision"]["overallOutcome"] + == "pending" + ): + self._record_usb_wait("STOPPED", "stopped_before_checks") await self._stop_monitor() await self._shielded_cleanup() if self.coturn is not None: diff --git a/src/python/isaaccapture/cloudxr/usb_health.py b/src/python/isaaccapture/cloudxr/usb_health.py new file mode 100644 index 0000000000..0389047a1e --- /dev/null +++ b/src/python/isaaccapture/cloudxr/usb_health.py @@ -0,0 +1,429 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Pre-browser USB-local topology, transfer, and private report helpers.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import os +import re +import secrets +import shlex +import stat +import time +from datetime import datetime, timezone +from pathlib import Path + +USB_NODE = re.compile(r"^\d+-\d+(?:\.\d+)*$") +PCI_NODE = re.compile(r"^\d{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-7]$") +BYTE_LINE = re.compile(r"^B:(\d+)$") +TIME_LINE = re.compile(r"^T:(\d+):(\d+)$") +SCHEMA = "isaac-capture.usb-health/v1" + + +def _read(path: Path) -> str | None: + try: + return path.read_text(encoding="utf-8").strip()[:160] + except (OSError, UnicodeError): + return None + + +def _number(path: Path) -> float | None: + try: + return float(_read(path)) + except (TypeError, ValueError): + return None + + +def _pci(path: Path) -> str | None: + return next( + (p.name for p in path.resolve().parents if PCI_NODE.fullmatch(p.name)), None + ) + + +def _node(path: Path) -> dict: + return { + "node": path.name, + "speedMbps": _number(path / "speed"), + "usbVersion": _read(path / "version"), + "vendorId": _read(path / "idVendor"), + "productId": _read(path / "idProduct"), + "manufacturer": _read(path / "manufacturer"), + "product": _read(path / "product"), + "authorized": _read(path / "authorized"), + "maxPower": _read(path / "bMaxPower"), + } + + +def inspect_link( + serial: str, listing: str, root: Path = Path("/sys/bus/usb/devices") +) -> dict: + """Correlate the already-selected ADB transport with Linux's negotiated link.""" + result = { + "probeSucceeded": False, + "negotiatedSpeedMbps": None, + "isBetterPortAvailable": None, + "betterPortCandidates": [], + "parentChain": [], + "errors": [], + } + fields = next( + (line.split() for line in listing.splitlines() if line.split()[:1] == [serial]), + None, + ) + if not fields or len(fields) < 2 or fields[1] != "device": + result["errors"].append("Selected ADB device is absent or not ready") + return result + props = dict(part.split(":", 1) for part in fields[2:] if ":" in part) + result["adbState"] = fields[1] + result["model"] = props.get("model", "")[:80] + result["product"] = props.get("product", "")[:80] + usb_path = props.get("usb") + device = root / usb_path if usb_path and USB_NODE.fullmatch(usb_path) else None + if device is None or not device.exists(): + # The descriptor fallback covers ADB versions that omit usb:. + matches = [ + entry.parent + for entry in root.glob("*/serial") + if USB_NODE.fullmatch(entry.parent.name) and _read(entry) == serial + ] + if len(matches) != 1: + result["errors"].append( + "ADB transport could not be uniquely mapped to sysfs" + ) + return result + device = matches[0] + speed = _number(device / "speed") + if speed is None: + result["errors"].append("Negotiated speed is unavailable in sysfs") + return result + chain = [ + _node(node) + for node in (device.resolve(), *device.resolve().parents) + if USB_NODE.fullmatch(node.name) or re.fullmatch(r"usb\d+", node.name) + ] + controller = _pci(device) + root_hub = next( + (node["node"] for node in chain if node["node"].startswith("usb")), None + ) + candidates = [ + _node(other) + for other in sorted(root.glob("usb*")) + if re.fullmatch(r"usb\d+", other.name) + and other.name != root_hub + and controller is not None + and _pci(other) == controller + and (_number(other / "speed") or 0) > speed + ] + result.update( + probeSucceeded=True, + negotiatedSpeedMbps=speed, + speedClass="usb3" if speed >= 5000 else "usb2_or_lower", + isBetterPortAvailable=bool(candidates) if controller else None, + betterPortCandidates=candidates, + betterPortInference="same-controller faster root hub; physical connector unverified", + deviceNode=device.name, + busNumber=_read(device / "busnum"), + deviceNumber=_read(device / "devnum"), + controller=controller, + parentChain=chain, + power={ + "advertisedMaxPower": _read(device / "bMaxPower"), + "runtimeStatus": _read(device / "power/runtime_status"), + "control": _read(device / "power/control"), + }, + ) + return result + + +async def _adb_shell(serial: str, script: str, timeout: float) -> tuple[int, str]: + child = await asyncio.create_subprocess_exec( + "adb", + "-s", + serial, + "shell", + "-T", + "sh -c " + shlex.quote(script), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.DEVNULL, + ) + try: + output, _ = await asyncio.wait_for(child.communicate(), timeout) + return child.returncode, output[:16384].decode("ascii", "replace") + except (asyncio.CancelledError, TimeoutError): + child.terminate() + try: + await asyncio.wait_for(child.communicate(), 0.5) + except TimeoutError: + child.kill() + await child.communicate() + raise + + +async def measure_transfer(serial: str, port: int, asset_size: int) -> dict: + """Measure three seconds over the existing HTTPS listener and reverse rule.""" + started = time.monotonic() + result = { + "attempted": True, + "completed": False, + "requestedMeasurementDurationMs": 3000, + "effectiveMeasurementDurationMs": None, + "hostToHeadsetMbps": None, + "headsetToHostMbps": None, + "completedBytes": 0, + "completedRequests": 0, + "method": None, + "pathCoverage": None, + "directionCoverage": "host_to_headset", + "browserRequired": False, + "newListenerCreated": False, + "newReverseRuleCreated": False, + "childCleanup": {"reaped": True}, + "errors": [], + } + try: + code, capabilities = await _adb_shell( + serial, + "command -v curl; command -v timeout; date +%s%N; command -v cat", + 2, + ) + lines = capabilities.splitlines() + curl_ready = code == 0 and any("curl" in line for line in lines) + timer_ready = any(re.fullmatch(r"\d{16,20}", line) for line in lines) + timeout_ready = any("timeout" in line for line in lines) + result["prerequisites"] = { + "deviceCurl": curl_ready, + "deviceTimeout": timeout_ready, + "nanosecondDate": timer_ready, + "bundleSizeBytes": asset_size, + "existingProxyReverseVerified": True, + } + if curl_ready and timeout_ready and timer_ready and asset_size >= 262144: + result["method"] = "device_curl_static_asset_loop" + result["pathCoverage"] = "https_static_over_adb_reverse" + # Each B line is printed only after curl exits successfully; a truncated + # final response contributes zero bytes. + script = ( + "start=$(date +%s%N); " + "timeout 3 sh -c 'while :; do " + 'n=$(curl -fksS --max-time 2 -o /dev/null -w "%{size_download}" ' + f"https://127.0.0.1:{port}/client/bundle.js); " + 'rc=$?; if test "$rc" = 0; then printf "B:%s\\n" "$n"; fi; ' + 'done\'; end=$(date +%s%N); printf \'T:%s:%s\\n\' "$start" "$end"' + ) + code, output = await _adb_shell(serial, script, 5) + byte_counts = [ + int(match.group(1)) + for line in output.splitlines() + if (match := BYTE_LINE.fullmatch(line)) + ] + timing = next( + ( + match + for line in output.splitlines() + if (match := TIME_LINE.fullmatch(line)) + ), + None, + ) + if timing is None: + raise ValueError("Device did not return measurement timing") + duration = (int(timing.group(2)) - int(timing.group(1))) / 1e6 + result["effectiveMeasurementDurationMs"] = round(duration, 2) + result["timingValid"] = 2950 <= duration <= 3250 + result["completedBytes"] = sum(byte_counts) + result["completedRequests"] = len(byte_counts) + result["partialTailExcluded"] = True + if ( + code != 0 + or not result["timingValid"] + or not byte_counts + or any(count != asset_size for count in byte_counts) + ): + raise ValueError("Device transfer window was invalid") + result["hostToHeadsetMbps"] = round( + sum(byte_counts) * 8000 / duration / 1e6, 2 + ) + result["completed"] = True + else: + result["method"] = "adb_shell_sink" + result["pathCoverage"] = "adb_transport_only" + result["fallbackReason"] = ( + "Device curl/timing unavailable or static asset too small" + ) + await _measure_fallback(serial, result) + except asyncio.CancelledError: + result["errors"].append("Measurement cancelled") + raise + except (OSError, TimeoutError, ValueError) as exc: + result["errors"].append(str(exc)[:160]) + result["wholeStageDurationMs"] = round((time.monotonic() - started) * 1000, 2) + result["outcome"] = "pass" if result["completed"] else "unsupported" + return result + + +async def _measure_fallback(serial: str, result: dict) -> None: + """Bound a host-to-device ADB-only sink when device curl is unavailable.""" + child = await asyncio.create_subprocess_exec( + "adb", + "-s", + serial, + "shell", + "-T", + "cat | wc -c", + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.DEVNULL, + ) + start = time.monotonic() + written = 0 + try: + payload = b"\0" * 65536 + while time.monotonic() - start < 3: + child.stdin.write(payload) + await asyncio.wait_for(child.stdin.drain(), 0.5) + written += len(payload) + duration = (time.monotonic() - start) * 1000 + result["effectiveMeasurementDurationMs"] = round(duration, 2) + result["timingValid"] = 2950 <= duration <= 3250 + result["hostBytesSubmitted"] = written + finally: + child.stdin.close() + try: + output = await asyncio.wait_for(child.stdout.read(64), 0.5) + await asyncio.wait_for(child.wait(), 0.5) + except TimeoutError: + child.terminate() + try: + await asyncio.wait_for(child.wait(), 0.5) + except TimeoutError: + child.kill() + await child.wait() + raise + if child.returncode != 0 or not output.strip().isdigit(): + raise ValueError("ADB sink did not confirm received byte count") + received = int(output.strip()) + result["completedBytes"] = received + result["hostToHeadsetMbps"] = round(received * 8000 / duration / 1e6, 2) + result["completed"] = result["timingValid"] + + +class UsbHealthReport: + """Own one private, atomically replaced JSON report for a clean start.""" + + def __init__( + self, + logs_dir: Path, + session_id: str, + generation: int, + serial: str | None, + policy: str = "warn", + ): + self.directory = logs_dir / "usb-health" + if self.directory.is_symlink(): + raise OSError("USB health report directory cannot be a symlink") + self.directory.mkdir(mode=0o700, parents=True, exist_ok=True) + os.chmod(self.directory, 0o700) + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + self.path = ( + self.directory + / f"usb-health-{stamp}-{secrets.token_hex(4)}-g{generation}.json" + ) + salt_path = self.directory / ".serial-salt" + try: + salt = salt_path.read_bytes() + except FileNotFoundError: + try: + descriptor = os.open( + salt_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600 + ) + with os.fdopen(descriptor, "wb") as stream: + stream.write(secrets.token_bytes(32)) + except FileExistsError: + pass + salt = salt_path.read_bytes() + self._serial_salt = salt + now = datetime.now(timezone.utc).isoformat() + self.payload = { + "schema": SCHEMA, + "createdAt": now, + "updatedAt": now, + "context": { + "mode": "usb-local", + "policy": policy, + "generation": generation, + "sessionIdPrefix": session_id[:8], + "requestedMeasurementDurationMs": 3000, + "reportPath": str(Path("logs/usb-health") / self.path.name), + }, + "device": { + "selectedBy": "oobLifecycle", + "serialHash": None, + "serialPersistedRaw": False, + }, + "topology": None, + "transferTest": None, + "decision": {"overallOutcome": "pending", "browserLaunchAllowed": None}, + "events": [], + "errors": [], + "redaction": { + "rawSerial": False, + "rawUrlOrQuery": False, + "subprocessArgv": False, + "controlOrTurnCredential": False, + }, + } + if serial is not None: + self.set_device(serial) + + def set_device(self, serial: str) -> None: + digest = hashlib.sha256(self._serial_salt + serial.encode()).hexdigest() + self.payload["device"]["serialHash"] = f"sha256-install-salted:{digest}" + self.update() + + def event(self, state: str) -> None: + self.payload["events"].append( + {"at": datetime.now(timezone.utc).isoformat(), "state": state} + ) + self.update() + + def update(self, **sections) -> None: + self.payload.update(sections) + self.payload["updatedAt"] = datetime.now(timezone.utc).isoformat() + temporary = self.path.with_name( + self.path.name + "." + secrets.token_hex(4) + ".tmp" + ) + descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as stream: + json.dump(self.payload, stream, indent=2, sort_keys=True) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, self.path) + directory_fd = os.open(self.directory, os.O_RDONLY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + self._prune() + finally: + temporary.unlink(missing_ok=True) + + def _prune(self) -> None: + files = sorted( + ( + p + for p in self.directory.glob("usb-health-*.json") + if p.is_file() and not p.is_symlink() and stat.S_ISREG(p.stat().st_mode) + ), + key=lambda p: p.stat().st_mtime, + reverse=True, + ) + for old in files[20:]: + old.unlink() + cutoff = time.time() - 14 * 86400 + for old in files[:20]: + if old != self.path and old.stat().st_mtime < cutoff: + old.unlink() diff --git a/src/python/isaaccapture/cloudxr/wss.py b/src/python/isaaccapture/cloudxr/wss.py index bce7f63685..686e1b6222 100755 --- a/src/python/isaaccapture/cloudxr/wss.py +++ b/src/python/isaaccapture/cloudxr/wss.py @@ -720,6 +720,12 @@ def handler(ws): turn_port=usb_turn_port() if usb_local else None, config=recovery_config or resolve_oob_recovery_config(), on_status=on_oob_status, + usb_health_logs_dir=( + Path(log_file_path).parent + if log_file_path is not None + else None + ), + web_client_static_dir=_host_client_static_dir, ) lifecycle_task = asyncio.create_task( lifecycle.run(), name="cloudxr-oob-lifecycle" diff --git a/tests/python/core/cloudxr/test_usb_health.py b/tests/python/core/cloudxr/test_usb_health.py new file mode 100644 index 0000000000..8f34b0abbd --- /dev/null +++ b/tests/python/core/cloudxr/test_usb_health.py @@ -0,0 +1,574 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Focused topology, timing, report and startup ordering checks.""" + +from __future__ import annotations + +import asyncio +import json +import subprocess +from pathlib import Path +from unittest.mock import patch + +import pytest + +from isaaccapture.cloudxr import oob_teleop_adb as adb +from isaaccapture.cloudxr import usb_health +from isaaccapture.cloudxr.oob_teleop_lifecycle import OobLifecycle, RecoveryConfig + + +def _field(directory: Path, name: str, value: str) -> None: + (directory / name).write_text(value) + + +def test_pinned_adb_sysfs_topology_and_faster_root(tmp_path): + controller = tmp_path / "0000:00:14.0" + device = controller / "usb1" / "1-2" / "1-2.1" + faster = controller / "usb2" + device.mkdir(parents=True) + faster.mkdir() + root = tmp_path / "devices" + root.mkdir() + (root / "1-2.1").symlink_to(device, target_is_directory=True) + (root / "usb1").symlink_to(controller / "usb1", target_is_directory=True) + (root / "usb2").symlink_to(faster, target_is_directory=True) + _field(device, "speed", "480") + _field(device, "product", "PICO 4 Ultra") + _field(faster, "speed", "20000") + listing = "List of devices attached\nheadset device usb:1-2.1 model:A9210\nother device usb:3-1\n" + status = usb_health.inspect_link("headset", listing, root) + assert status["probeSucceeded"] is True + assert status["negotiatedSpeedMbps"] == 480 + assert status["isBetterPortAvailable"] is True + assert status["betterPortCandidates"][0]["speedMbps"] == 20000 + assert status["controller"] == "0000:00:14.0" + assert "headset" not in json.dumps(status) + + +def test_ambiguous_or_missing_transport_has_unknown_result(tmp_path): + status = usb_health.inspect_link("selected", "other device usb:1-2", tmp_path) + assert status["probeSucceeded"] is False + assert status["isBetterPortAvailable"] is None + assert status["negotiatedSpeedMbps"] is None + + +@pytest.mark.asyncio +async def test_curl_measurement_counts_only_completed_responses(monkeypatch): + calls = [] + + async def shell(serial, script, timeout): + calls.append((serial, script, timeout)) + if len(calls) == 1: + return ( + 0, + "/system/bin/curl\n/system/bin/timeout\n1791403150633088373\n/system/bin/cat\n", + ) + return 0, "B:500000\nB:500000\nT:1000000000000000000:1000000003000000000\n" + + monkeypatch.setattr(usb_health, "_adb_shell", shell) + status = await usb_health.measure_transfer("selected", 48322, 500000) + assert status["completed"] is True + assert status["effectiveMeasurementDurationMs"] == 3000 + assert status["completedBytes"] == 1000000 + assert status["hostToHeadsetMbps"] == pytest.approx(2.67, abs=0.01) + assert status["method"] == "device_curl_static_asset_loop" + assert calls[1][0] == "selected" + assert calls[1][2] == 5 + + +@pytest.mark.asyncio +async def test_missing_curl_uses_labeled_adb_fallback(monkeypatch): + async def shell(_serial, _script, _timeout): + return 0, "/system/bin/timeout\n1791403150633088373\n/system/bin/cat\n" + + async def fallback(_serial, result): + result.update( + completed=True, + completedBytes=1024, + effectiveMeasurementDurationMs=3000, + hostToHeadsetMbps=0.003, + ) + + monkeypatch.setattr(usb_health, "_adb_shell", shell) + monkeypatch.setattr(usb_health, "_measure_fallback", fallback) + result = await usb_health.measure_transfer("selected", 48322, 500000) + assert result["completed"] is True + assert result["method"] == "adb_shell_sink" + assert result["pathCoverage"] == "adb_transport_only" + assert result["fallbackReason"] + + +def test_atomic_report_is_private_and_redacts_serial(tmp_path): + report = usb_health.UsbHealthReport(tmp_path, "session-id", 0, "SECRET-SERIAL") + report.update(topology={"probeSucceeded": False}, errors=["sysfs unavailable"]) + content = report.path.read_text() + assert "SECRET-SERIAL" not in content + assert json.loads(content)["schema"] == usb_health.SCHEMA + assert report.path.stat().st_mode & 0o777 == 0o600 + assert report.directory.stat().st_mode & 0o777 == 0o700 + assert not list(report.directory.glob("*.tmp")) + + +def test_report_can_begin_before_adb_selection(tmp_path): + report = usb_health.UsbHealthReport(tmp_path, "session-id", 0, None) + report.event("WAITING_FOR_ADB") + assert json.loads(report.path.read_text())["device"]["serialHash"] is None + report.set_device("SECRET-SERIAL") + payload = report.path.read_text() + assert "SECRET-SERIAL" not in payload + assert json.loads(payload)["device"]["serialHash"].startswith( + "sha256-install-salted:" + ) + + +@pytest.mark.parametrize("wakefulness", ["Asleep", ""]) +def test_strict_wake_confirmation_fails_closed(wakefulness): + with ( + patch.object(adb, "headset_wakefulness", return_value=wakefulness), + patch.object(adb, "_adb_run"), + ): + with pytest.raises(adb.OobAdbError, match="not confirmed"): + adb.assert_headset_awake(timeout=0, require_awake=True) + adb.assert_headset_awake(timeout=0) + + +def test_strict_wake_confirmation_accepts_observed_awake(): + with ( + patch.object(adb, "headset_wakefulness", return_value="Awake"), + patch.object(adb, "_adb_run") as command, + ): + adb.assert_headset_awake(timeout=0, require_awake=True) + command.assert_not_called() + + +@pytest.mark.asyncio +async def test_strict_wake_check_uses_lifecycle_pinned_adb_device(): + lifecycle = OobLifecycle( + hub=_Hub(), + resolved_port=48322, + usb_local=True, + host_client=True, + turn_port=3478, + config=RecoveryConfig(), + ) + lifecycle.selected = "selected-headset" + token = adb.SELECTED_ADB_SERIAL.set(lifecycle.selected) + try: + with patch.object( + adb.subprocess, + "run", + return_value=subprocess.CompletedProcess([], 0, "mWakefulness=Awake\n", ""), + ) as command: + await lifecycle._confirm_usb_awake() + finally: + adb.SELECTED_ADB_SERIAL.reset(token) + assert lifecycle._usb_wake_confirmed is True + assert command.call_args.args[0] == [ + "adb", + "-s", + "selected-headset", + "shell", + "dumpsys", + "power", + ] + + +class _Hub: + async def set_lifecycle_snapshot(self, snapshot): + pass + + async def get_snapshot(self): + return {"headsets": []} + + +@pytest.mark.parametrize( + ("usb_local", "policy", "speed", "expected"), + [ + ( + True, + "warn", + 480, + ["prepare", "link", "rebuild", "awake", "transfer", "browser"], + ), + (True, "block", 480, ["prepare", "link", "rebuild", "awake", "transfer"]), + (False, "warn", None, ["prepare", "browser"]), + ], +) +@pytest.mark.asyncio +async def test_startup_checks_finish_before_browser( + monkeypatch, usb_local, policy, speed, expected +): + events = [] + selected = "selected" + lifecycle = OobLifecycle( + hub=_Hub(), + resolved_port=48322, + usb_local=usb_local, + host_client=True, + turn_port=3478 if usb_local else None, + config=RecoveryConfig(), + usb_health_policy=policy, + ) + lifecycle.selected = selected + devices = adb.AdbDevices(((selected, "device"),)) + lifecycle._last_observation = (devices.devices, devices.diagnostic) + lifecycle._ready_count = 1 + + async def step(name): + events.append(name) + + async def link(): + events.append("link") + lifecycle._usb_link = {"negotiatedSpeedMbps": speed} + + async def transfer(): + assert lifecycle._usb_wake_confirmed is True + events.append("transfer") + lifecycle._usb_transfer = {"completed": True} + + async def confirm_awake(): + events.append("awake") + lifecycle._usb_wake_confirmed = True + + async def browser(): + events.append("browser") + raise asyncio.CancelledError + + async def sleep(_seconds): + raise asyncio.CancelledError + + monkeypatch.setattr(lifecycle, "_prepare_device", lambda: step("prepare")) + monkeypatch.setattr(lifecycle, "_check_usb_link", link) + monkeypatch.setattr(lifecycle, "_rebuild_usb", lambda: step("rebuild")) + monkeypatch.setattr(lifecycle, "_check_usb_transfer", transfer) + monkeypatch.setattr(lifecycle, "_confirm_usb_awake", confirm_awake) + monkeypatch.setattr(lifecycle, "_automate", browser) + lifecycle.sleep = sleep + with ( + patch.object(adb, "enumerate_adb_devices", return_value=devices), + patch.object( + adb, + "probe_headset_network", + return_value=adb.HeadsetNetworkProbe( + adb.HeadsetNetworkState.NETWORK_PRESENT + ), + ), + patch.object(adb, "_run_adb", return_value=""), + ): + with pytest.raises(asyncio.CancelledError): + await lifecycle.run() + assert events == expected + + +@pytest.mark.asyncio +async def test_missing_adb_writes_partial_usb_report(tmp_path, monkeypatch): + lifecycle = OobLifecycle( + hub=_Hub(), + resolved_port=48322, + usb_local=True, + host_client=True, + turn_port=3478, + config=RecoveryConfig(), + usb_health_logs_dir=tmp_path, + ) + + async def stop(_seconds): + raise asyncio.CancelledError + + lifecycle.sleep = stop + monkeypatch.setattr(adb, "enumerate_adb_devices", lambda: adb.AdbDevices(())) + with pytest.raises(asyncio.CancelledError): + await lifecycle.run() + payload = json.loads( + next((tmp_path / "usb-health").glob("usb-health-*.json")).read_text() + ) + assert payload["decision"]["overallOutcome"] == "incomplete" + assert payload["errors"][0]["code"] == "no_ready_device" + assert payload["device"]["serialHash"] is None + + +@pytest.mark.asyncio +async def test_prepare_failure_writes_partial_usb_report(tmp_path, monkeypatch): + lifecycle = OobLifecycle( + hub=_Hub(), + resolved_port=48322, + usb_local=True, + host_client=True, + turn_port=3478, + config=RecoveryConfig(), + usb_health_logs_dir=tmp_path, + ) + lifecycle.selected = "selected" + devices = adb.AdbDevices((("selected", "device"),)) + lifecycle._last_observation = (devices.devices, devices.diagnostic) + lifecycle._ready_count = 1 + + async def fail_prepare(): + raise RuntimeError("private headset detail") + + async def stop(_seconds): + raise asyncio.CancelledError + + lifecycle._prepare_device = fail_prepare + lifecycle.sleep = stop + with ( + patch.object(adb, "enumerate_adb_devices", return_value=devices), + patch.object( + adb, + "probe_headset_network", + return_value=adb.HeadsetNetworkProbe( + adb.HeadsetNetworkState.NETWORK_PRESENT + ), + ), + patch.object(adb, "_run_adb", return_value=""), + ): + with pytest.raises(asyncio.CancelledError): + await lifecycle.run() + payload_text = next((tmp_path / "usb-health").glob("usb-health-*.json")).read_text() + payload = json.loads(payload_text) + assert payload["decision"]["overallOutcome"] == "incomplete" + assert any( + item["state"] == "PREPARING_DEVICE" and item["code"] == "RuntimeError" + for item in payload["errors"] + ) + assert "private headset detail" not in payload_text + assert '"selected"' not in payload_text + + +@pytest.mark.asyncio +async def test_cancelled_transfer_persists_partial_json(tmp_path, monkeypatch): + static = tmp_path / "static" + static.mkdir() + (static / "bundle.js").write_bytes(b"x" * 300000) + lifecycle = OobLifecycle( + hub=_Hub(), + resolved_port=48322, + usb_local=True, + host_client=True, + turn_port=3478, + config=RecoveryConfig(), + usb_health_logs_dir=tmp_path, + web_client_static_dir=static, + ) + lifecycle.selected = "selected" + lifecycle._ensure_usb_report() + lifecycle._usb_wake_confirmed = True + + async def cancel(_serial, _port, _size): + raise asyncio.CancelledError + + monkeypatch.setattr(usb_health, "measure_transfer", cancel) + with pytest.raises(asyncio.CancelledError): + await lifecycle._check_usb_transfer() + payload = json.loads( + next((tmp_path / "usb-health").glob("usb-health-*.json")).read_text() + ) + assert payload["transferTest"]["outcome"] == "cancelled" + assert payload["transferTest"]["completed"] is False + + +@pytest.mark.asyncio +async def test_unconfirmed_wake_never_starts_transfer_or_browser(tmp_path, monkeypatch): + lifecycle = OobLifecycle( + hub=_Hub(), + resolved_port=48322, + usb_local=True, + host_client=True, + turn_port=3478, + config=RecoveryConfig(), + usb_health_logs_dir=tmp_path, + ) + lifecycle.selected = "selected" + devices = adb.AdbDevices((("selected", "device"),)) + lifecycle._last_observation = (devices.devices, devices.diagnostic) + lifecycle._ready_count = 1 + events = [] + + async def step(name): + events.append(name) + + async def link(): + events.append("link") + lifecycle._usb_link = {"negotiatedSpeedMbps": 5000} + + async def forbidden(): + pytest.fail("transfer/browser must not start before confirmed Awake") + + async def stop(_seconds): + raise asyncio.CancelledError + + def wake(*, timeout, require_awake=False): + assert require_awake is True + events.append("wake_unconfirmed") + raise adb.OobAdbError("Headset wakefulness was not confirmed as Awake") + + monkeypatch.setattr(lifecycle, "_prepare_device", lambda: step("prepare")) + monkeypatch.setattr(lifecycle, "_check_usb_link", link) + monkeypatch.setattr(lifecycle, "_rebuild_usb", lambda: step("rebuild")) + monkeypatch.setattr(lifecycle, "_check_usb_transfer", forbidden) + monkeypatch.setattr(lifecycle, "_automate", forbidden) + lifecycle.sleep = stop + with ( + patch.object(adb, "enumerate_adb_devices", return_value=devices), + patch.object( + adb, + "probe_headset_network", + return_value=adb.HeadsetNetworkProbe( + adb.HeadsetNetworkState.NETWORK_PRESENT + ), + ), + patch.object(adb, "_run_adb", return_value=""), + patch.object(adb, "assert_headset_awake", side_effect=wake), + ): + with pytest.raises(asyncio.CancelledError): + await lifecycle.run() + assert events == ["prepare", "link", "rebuild", "wake_unconfirmed"] + payload = json.loads( + next((tmp_path / "usb-health").glob("usb-health-*.json")).read_text() + ) + assert payload["wakeCheck"]["confirmedAwake"] is False + assert payload["transferTest"]["attempted"] is False + assert payload["decision"]["overallOutcome"] == "incomplete" + assert lifecycle.snapshot["usbWakeConfirmed"] is False + + +@pytest.mark.asyncio +async def test_transport_loss_invalidates_current_wake_but_preserves_history(tmp_path): + lifecycle = OobLifecycle( + hub=_Hub(), + resolved_port=48322, + usb_local=True, + host_client=True, + turn_port=3478, + config=RecoveryConfig(), + usb_health_logs_dir=tmp_path, + ) + lifecycle.selected = "selected" + lifecycle.browser_ready = True + lifecycle._usb_wake_confirmed = True + lifecycle._ensure_usb_report() + lifecycle._persist_usb_health( + wakeCheck={"confirmedAwake": True, "source": "adb-shell-dumpsys-power"}, + decision={"overallOutcome": "pass", "browserLaunchAllowed": True}, + ) + + await lifecycle._enter_transport_recovery( + "ADB disconnected", "WAITING_FOR_ADB", adb_ready=False, network_present=False + ) + await lifecycle._remember_transport_loss() + + payload = json.loads(lifecycle._usb_health_report.path.read_text()) + assert lifecycle.snapshot["usbWakeConfirmed"] is None + assert lifecycle._usb_wake_confirmed is None + assert payload["wakeCheck"] == { + "confirmedAwake": None, + "outcome": "unknown_after_transport_loss", + } + assert payload["wakeHistory"] == [ + {"confirmedAwake": True, "source": "adb-shell-dumpsys-power"} + ] + assert payload["decision"]["overallOutcome"] == "incomplete" + + +@pytest.mark.parametrize( + ("reconnected_speed", "decision"), + [(480, "degraded"), (5000, "incomplete")], +) +@pytest.mark.asyncio +async def test_preserved_browser_reconnect_refreshes_link_without_active_transfer( + tmp_path, monkeypatch, reconnected_speed, decision +): + lifecycle = OobLifecycle( + hub=_Hub(), + resolved_port=48322, + usb_local=True, + host_client=True, + turn_port=3478, + config=RecoveryConfig(), + usb_health_logs_dir=tmp_path, + ) + lifecycle.selected = "selected" + lifecycle._transport_lost = True + lifecycle._restore_existing_browser = True + lifecycle._usb_transfer = { + "completed": True, + "outcome": "pass", + "method": "device_curl_static_asset_loop", + "hostToHeadsetMbps": 300, + } + devices = adb.AdbDevices((("selected", "device"),)) + lifecycle._last_observation = (devices.devices, devices.diagnostic) + lifecycle._ready_count = 1 + speeds = iter((5000, reconnected_speed)) + observed = [] + + def inspect(_serial, _listing): + speed = next(speeds) + return { + "probeSucceeded": True, + "negotiatedSpeedMbps": speed, + "isBetterPortAvailable": speed < 5000, + "errors": [], + } + + async def rebuild(): + observed.append(("rebuild", lifecycle._usb_link["negotiatedSpeedMbps"])) + + async def recover(): + raise asyncio.CancelledError + + async def forbidden_transfer(): + pytest.fail("preserved browser must not run active transfer") + + async def stop(_seconds): + raise asyncio.CancelledError + + monkeypatch.setattr(usb_health, "inspect_link", inspect) + monkeypatch.setattr( + adb, + "_adb_run", + lambda args, **kwargs: subprocess.CompletedProcess( + args, 0, "selected device usb:1-2.1 model:A9210\n", "" + ), + ) + lifecycle._rebuild_usb = rebuild + lifecycle._recover_existing_browser = recover + lifecycle._check_usb_transfer = forbidden_transfer + lifecycle.sleep = stop + await lifecycle._check_usb_link() + with ( + patch.object(adb, "enumerate_adb_devices", return_value=devices), + patch.object( + adb, + "probe_headset_network", + return_value=adb.HeadsetNetworkProbe( + adb.HeadsetNetworkState.NETWORK_PRESENT + ), + ), + patch.object(adb, "_run_adb", return_value=""), + patch.object(adb, "assert_headset_awake") as awake, + ): + with pytest.raises(asyncio.CancelledError): + await lifecycle.run() + payload = json.loads( + next((tmp_path / "usb-health").glob("usb-health-*.json")).read_text() + ) + assert observed == [("rebuild", reconnected_speed)] + assert [item["negotiatedSpeedMbps"] for item in payload["linkHistory"]] == [ + 5000, + reconnected_speed, + ] + assert payload["transferTest"]["staleAfterReconnect"] is True + assert payload["transferTest"]["hostToHeadsetMbps"] is None + assert payload["transferTest"]["outcome"] == "pending_retest" + assert payload["transferHistory"][0]["hostToHeadsetMbps"] == 300 + assert payload["decision"]["transferRetestDeferred"] is True + assert payload["decision"]["overallOutcome"] == decision + awake.assert_called_once_with(timeout=10.0) + assert lifecycle.snapshot["usbLinkOutcome"] == ( + "pass" if reconnected_speed >= 5000 else "slow_link" + ) + assert lifecycle.snapshot["usbTransferMbps"] is None + assert lifecycle.snapshot["usbTransferMethod"] is None + assert lifecycle.snapshot["usbTransferOutcome"] == "stale" From f5acb9f8d04e3ef38bf8fcc51cafe16e1a581018 Mon Sep 17 00:00:00 2001 From: Andrew Wilson Date: Thu, 8 Oct 2026 10:33:25 -0500 Subject: [PATCH 2/9] fix(cloudxr): separate USB response-body rate from request overhead Signed-off-by: Andrew Wilson --- docs/source/references/cloudxr.rst | 12 +++- .../cloudxr/oob_teleop_lifecycle.py | 9 ++- src/python/isaaccapture/cloudxr/usb_health.py | 47 ++++++++++++--- tests/python/core/cloudxr/test_usb_health.py | 58 ++++++++++++++++++- 4 files changed, 113 insertions(+), 13 deletions(-) diff --git a/docs/source/references/cloudxr.rst b/docs/source/references/cloudxr.rst index 76d9f41292..01f5e3954e 100644 --- a/docs/source/references/cloudxr.rst +++ b/docs/source/references/cloudxr.rst @@ -155,9 +155,15 @@ when device preparation or a check fails, and refreshes link topology after a preserved browser reconnect without repeating the active transfer. It uses a hashed device serial, and keeps at most 20 reports or 14 days of history. A link below 5000 Mb/s or an incomplete transfer is logged as a -warning and startup continues. The measured Mbps is one-way application -goodput, not USB line rate or CloudXR stream throughput. A faster root hub -is a topology clue, not proof that an accessible physical port is free. +warning and startup continues. ``hostToHeadsetMbps`` is the completed bytes +divided by the full three-second wall-clock interval. Each curl request opens +a new connection, so this includes request and TLS setup. The optional +``hostToHeadsetResponseBodyMbps`` instead divides those same completed bytes +by the sum of curl's ``time_total - time_starttransfer`` intervals. It excludes +setup and gaps between requests; it is a response-body estimate, not sustained +goodput or USB line rate. Neither metric measures headset-to-host traffic or +CloudXR stream throughput. A faster root hub is a topology clue, not proof +that an accessible physical port is free. Re-open the client on the headset --------------------------------- diff --git a/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py b/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py index 65321b3893..4154510f93 100644 --- a/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py +++ b/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py @@ -237,6 +237,11 @@ async def _publish(self, health: str, state: str, reason: str, **flags) -> None: if self._usb_transfer and not self._usb_transfer_stale else None ), + "usbTransferResponseBodyMbps": ( + self._usb_transfer.get("hostToHeadsetResponseBodyMbps") + if self._usb_transfer and not self._usb_transfer_stale + else None + ), "usbLinkOutcome": self._link_outcome(), "usbTransferOutcome": ( "stale" @@ -732,8 +737,10 @@ async def _check_usb_transfer(self) -> None: logging.WARNING if not self._usb_transfer.get("completed") else logging.INFO, - "USB-local pre-browser host-to-headset application goodput: %s Mb/s (%s)", + "USB-local pre-browser host-to-headset application goodput: %s Mb/s " + "wall-clock; response-body-only: %s Mb/s (%s)", self._usb_transfer.get("hostToHeadsetMbps"), + self._usb_transfer.get("hostToHeadsetResponseBodyMbps"), self._usb_transfer.get("pathCoverage"), ) diff --git a/src/python/isaaccapture/cloudxr/usb_health.py b/src/python/isaaccapture/cloudxr/usb_health.py index 0389047a1e..945f59b465 100644 --- a/src/python/isaaccapture/cloudxr/usb_health.py +++ b/src/python/isaaccapture/cloudxr/usb_health.py @@ -19,7 +19,9 @@ USB_NODE = re.compile(r"^\d+-\d+(?:\.\d+)*$") PCI_NODE = re.compile(r"^\d{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-7]$") -BYTE_LINE = re.compile(r"^B:(\d+)$") +CURL_LINE = re.compile( + r"^B:(\d{1,12}):(\d{1,4}(?:\.\d{1,9})?):(\d{1,4}(?:\.\d{1,9})?)$" +) TIME_LINE = re.compile(r"^T:(\d+):(\d+)$") SCHEMA = "isaac-capture.usb-health/v1" @@ -152,7 +154,7 @@ async def _adb_shell(serial: str, script: str, timeout: float) -> tuple[int, str ) try: output, _ = await asyncio.wait_for(child.communicate(), timeout) - return child.returncode, output[:16384].decode("ascii", "replace") + return child.returncode, output[:65536].decode("ascii", "replace") except (asyncio.CancelledError, TimeoutError): child.terminate() try: @@ -172,6 +174,7 @@ async def measure_transfer(serial: str, port: int, asset_size: int) -> dict: "requestedMeasurementDurationMs": 3000, "effectiveMeasurementDurationMs": None, "hostToHeadsetMbps": None, + "hostToHeadsetResponseBodyMbps": None, "headsetToHostMbps": None, "completedBytes": 0, "completedRequests": 0, @@ -182,6 +185,7 @@ async def measure_transfer(serial: str, port: int, asset_size: int) -> dict: "newListenerCreated": False, "newReverseRuleCreated": False, "childCleanup": {"reaped": True}, + "measurementRevision": 2, "errors": [], } try: @@ -204,22 +208,29 @@ async def measure_transfer(serial: str, port: int, asset_size: int) -> dict: if curl_ready and timeout_ready and timer_ready and asset_size >= 262144: result["method"] = "device_curl_static_asset_loop" result["pathCoverage"] = "https_static_over_adb_reverse" - # Each B line is printed only after curl exits successfully; a truncated - # final response contributes zero bytes. + # A line is emitted only after a complete curl response; the cut-off + # request contributes no bytes or response-body timing. script = ( "start=$(date +%s%N); " "timeout 3 sh -c 'while :; do " - 'n=$(curl -fksS --max-time 2 -o /dev/null -w "%{size_download}" ' + "n=$(curl -fksS --max-time 2 -o /dev/null " + '-w "%{size_download}:%{time_starttransfer}:%{time_total}" ' f"https://127.0.0.1:{port}/client/bundle.js); " 'rc=$?; if test "$rc" = 0; then printf "B:%s\\n" "$n"; fi; ' 'done\'; end=$(date +%s%N); printf \'T:%s:%s\\n\' "$start" "$end"' ) code, output = await _adb_shell(serial, script, 5) - byte_counts = [ - int(match.group(1)) + samples = [ + (int(match.group(1)), float(match.group(2)), float(match.group(3))) for line in output.splitlines() - if (match := BYTE_LINE.fullmatch(line)) + if (match := CURL_LINE.fullmatch(line)) ] + malformed_samples = [ + line + for line in output.splitlines() + if line.startswith("B:") and not CURL_LINE.fullmatch(line) + ] + byte_counts = [size for size, _, _ in samples] timing = next( ( match @@ -236,16 +247,36 @@ async def measure_transfer(serial: str, port: int, asset_size: int) -> dict: result["completedBytes"] = sum(byte_counts) result["completedRequests"] = len(byte_counts) result["partialTailExcluded"] = True + result["responseBodyTimingMethod"] = "curl_total_minus_starttransfer" if ( code != 0 or not result["timingValid"] or not byte_counts + or malformed_samples or any(count != asset_size for count in byte_counts) ): raise ValueError("Device transfer window was invalid") result["hostToHeadsetMbps"] = round( sum(byte_counts) * 8000 / duration / 1e6, 2 ) + # Each curl starts a new TLS connection; do not substitute the + # response-body-only rate for three-second wall-clock goodput. + phase_valid = all(0 <= first < total <= 2.5 for _, first, total in samples) + result["responseBodyTimingValid"] = phase_valid + if phase_valid: + first_byte_wait_ms = sum(first for _, first, _ in samples) * 1000 + body_ms = sum(total - first for _, first, total in samples) * 1000 + result["completedFirstByteWaitDurationMs"] = round( + first_byte_wait_ms, 2 + ) + result["completedResponseBodyDurationMs"] = round(body_ms, 2) + result["hostToHeadsetResponseBodyMbps"] = round( + sum(byte_counts) * 8000 / body_ms / 1e6, 2 + ) + else: + result["responseBodyTimingUnavailableReason"] = ( + "curl returned invalid per-request timing" + ) result["completed"] = True else: result["method"] = "adb_shell_sink" diff --git a/tests/python/core/cloudxr/test_usb_health.py b/tests/python/core/cloudxr/test_usb_health.py index 8f34b0abbd..7fbd112d99 100644 --- a/tests/python/core/cloudxr/test_usb_health.py +++ b/tests/python/core/cloudxr/test_usb_health.py @@ -64,7 +64,13 @@ async def shell(serial, script, timeout): 0, "/system/bin/curl\n/system/bin/timeout\n1791403150633088373\n/system/bin/cat\n", ) - return 0, "B:500000\nB:500000\nT:1000000000000000000:1000000003000000000\n" + return ( + 0, + "B:500000:0.010000:0.040000\n" + "B:500000:0.020000:0.060000\n" + "curl: (28) partial final response was cut off\n" + "T:1000000000000000000:1000000003000000000\n", + ) monkeypatch.setattr(usb_health, "_adb_shell", shell) status = await usb_health.measure_transfer("selected", 48322, 500000) @@ -72,11 +78,56 @@ async def shell(serial, script, timeout): assert status["effectiveMeasurementDurationMs"] == 3000 assert status["completedBytes"] == 1000000 assert status["hostToHeadsetMbps"] == pytest.approx(2.67, abs=0.01) + assert status["hostToHeadsetResponseBodyMbps"] == pytest.approx(114.29) + assert status["completedResponseBodyDurationMs"] == pytest.approx(70) + assert status["completedFirstByteWaitDurationMs"] == pytest.approx(30) + assert status["responseBodyTimingValid"] is True + assert status["measurementRevision"] == 2 assert status["method"] == "device_curl_static_asset_loop" assert calls[1][0] == "selected" + assert "%{time_starttransfer}:%{time_total}" in calls[1][1] assert calls[1][2] == 5 +@pytest.mark.asyncio +async def test_curl_phase_metric_unavailable_when_timings_are_invalid(monkeypatch): + async def shell(_serial, _script, _timeout): + if "command -v curl" in _script: + return 0, "/system/bin/curl\n/system/bin/timeout\n1791403150633088373\n" + return ( + 0, + "B:500000:0.060000:0.040000\nT:1000000000000000000:1000000003000000000\n", + ) + + monkeypatch.setattr(usb_health, "_adb_shell", shell) + status = await usb_health.measure_transfer("selected", 48322, 500000) + assert status["completed"] is True + assert status["hostToHeadsetMbps"] == pytest.approx(1.33, abs=0.01) + assert status["hostToHeadsetResponseBodyMbps"] is None + assert status["responseBodyTimingValid"] is False + + +@pytest.mark.parametrize( + "second_sample", ["B:123:bad:0.030000", "B:123:0.010000:0.030000"] +) +@pytest.mark.asyncio +async def test_curl_rejects_malformed_or_wrong_size_samples(monkeypatch, second_sample): + async def shell(_serial, script, _timeout): + if "command -v curl" in script: + return 0, "/system/bin/curl\n/system/bin/timeout\n1791403150633088373\n" + return ( + 0, + "B:500000:0.010000:0.040000\n" + f"{second_sample}\nT:1000000000000000000:1000000003000000000\n", + ) + + monkeypatch.setattr(usb_health, "_adb_shell", shell) + status = await usb_health.measure_transfer("selected", 48322, 500000) + assert status["completed"] is False + assert status["hostToHeadsetResponseBodyMbps"] is None + assert status["outcome"] == "unsupported" + + @pytest.mark.asyncio async def test_missing_curl_uses_labeled_adb_fallback(monkeypatch): async def shell(_serial, _script, _timeout): @@ -97,6 +148,8 @@ async def fallback(_serial, result): assert result["method"] == "adb_shell_sink" assert result["pathCoverage"] == "adb_transport_only" assert result["fallbackReason"] + assert result["hostToHeadsetResponseBodyMbps"] is None + assert result["headsetToHostMbps"] is None def test_atomic_report_is_private_and_redacts_serial(tmp_path): @@ -496,6 +549,7 @@ async def test_preserved_browser_reconnect_refreshes_link_without_active_transfe "outcome": "pass", "method": "device_curl_static_asset_loop", "hostToHeadsetMbps": 300, + "hostToHeadsetResponseBodyMbps": 900, } devices = adb.AdbDevices((("selected", "device"),)) lifecycle._last_observation = (devices.devices, devices.diagnostic) @@ -563,6 +617,7 @@ async def stop(_seconds): assert payload["transferTest"]["hostToHeadsetMbps"] is None assert payload["transferTest"]["outcome"] == "pending_retest" assert payload["transferHistory"][0]["hostToHeadsetMbps"] == 300 + assert payload["transferHistory"][0]["hostToHeadsetResponseBodyMbps"] == 900 assert payload["decision"]["transferRetestDeferred"] is True assert payload["decision"]["overallOutcome"] == decision awake.assert_called_once_with(timeout=10.0) @@ -570,5 +625,6 @@ async def stop(_seconds): "pass" if reconnected_speed >= 5000 else "slow_link" ) assert lifecycle.snapshot["usbTransferMbps"] is None + assert lifecycle.snapshot["usbTransferResponseBodyMbps"] is None assert lifecycle.snapshot["usbTransferMethod"] is None assert lifecycle.snapshot["usbTransferOutcome"] == "stale" From 446028116751fc01362639145b10d73db7011297 Mon Sep 17 00:00:00 2001 From: Andrew Wilson Date: Thu, 8 Oct 2026 12:10:06 -0500 Subject: [PATCH 3/9] fix(cloudxr): preserve USB fallback cancellation during cleanup Signed-off-by: Andrew Wilson --- src/python/isaaccapture/cloudxr/usb_health.py | 72 ++++++-- tests/python/core/cloudxr/test_usb_health.py | 165 ++++++++++++++++++ 2 files changed, 227 insertions(+), 10 deletions(-) diff --git a/src/python/isaaccapture/cloudxr/usb_health.py b/src/python/isaaccapture/cloudxr/usb_health.py index 9d8307f919..c7c28dfa0c 100644 --- a/src/python/isaaccapture/cloudxr/usb_health.py +++ b/src/python/isaaccapture/cloudxr/usb_health.py @@ -330,6 +330,43 @@ async def measure_transfer(serial: str, port: int, asset_size: int) -> dict: return result +async def _terminate_fallback_child(child) -> bool: + """Bound termination and reaping of the ADB sink child.""" + if child.returncode is None: + try: + child.terminate() + except ProcessLookupError: + pass + try: + await asyncio.wait_for(child.wait(), 0.5) + return True + except TimeoutError: + if child.returncode is None: + try: + child.kill() + except ProcessLookupError: + pass + try: + await asyncio.wait_for(child.wait(), 0.5) + return True + except TimeoutError: + return False + + +async def _reap_fallback_child(child) -> bool: + """Finish bounded reaping even if cancellation recurs during cleanup.""" + cleanup = asyncio.create_task(_terminate_fallback_child(child)) + cancellation = None + while not cleanup.done(): + try: + await asyncio.shield(cleanup) + except asyncio.CancelledError as exc: + cancellation = exc + if cancellation is not None: + raise cancellation + return cleanup.result() + + async def _measure_fallback(serial: str, result: dict) -> None: """Bound a host-to-device ADB-only sink when device curl is unavailable.""" child = await asyncio.create_subprocess_exec( @@ -345,6 +382,8 @@ async def _measure_fallback(serial: str, result: dict) -> None: ) start = time.monotonic() written = 0 + failure: BaseException | None = None + output = b"" try: payload = b"\0" * 65536 while time.monotonic() - start < 3: @@ -355,19 +394,29 @@ async def _measure_fallback(serial: str, result: dict) -> None: result["effectiveMeasurementDurationMs"] = round(duration, 2) result["timingValid"] = 2950 <= duration <= 3250 result["hostBytesSubmitted"] = written - finally: + except BaseException as exc: + failure = exc + try: child.stdin.close() + except Exception as exc: + if failure is None: + failure = exc + if failure is None: try: output = await asyncio.wait_for(child.stdout.read(64), 0.5) await asyncio.wait_for(child.wait(), 0.5) - except TimeoutError: - child.terminate() - try: - await asyncio.wait_for(child.wait(), 0.5) - except TimeoutError: - child.kill() - await child.wait() - raise + except BaseException as exc: + failure = exc + if failure is not None: + try: + result["childCleanup"] = {"reaped": await _reap_fallback_child(child)} + except asyncio.CancelledError as exc: + failure = exc + except Exception: + result["childCleanup"] = {"reaped": False} + # Cleanup failure must not turn shutdown cancellation into warn-mode success. + raise failure + result["childCleanup"] = {"reaped": True} if child.returncode != 0 or not output.strip().isdigit(): raise ValueError("ADB sink did not confirm received byte count") received = int(output.strip()) @@ -445,7 +494,10 @@ def event(self, state: str) -> None: self.update() def update(self, **sections) -> None: - """Atomically persist new report sections with private permissions.""" + """Replace whole top-level sections, then synchronously persist and prune. + + I/O failures propagate to the caller. + """ self.payload.update(sections) self.payload["updatedAt"] = datetime.now(timezone.utc).isoformat() temporary = self.path.with_name( diff --git a/tests/python/core/cloudxr/test_usb_health.py b/tests/python/core/cloudxr/test_usb_health.py index a7ccde5aba..6ad971b58d 100644 --- a/tests/python/core/cloudxr/test_usb_health.py +++ b/tests/python/core/cloudxr/test_usb_health.py @@ -11,6 +11,7 @@ from concurrent.futures import ThreadPoolExecutor from pathlib import Path from threading import Barrier +from types import SimpleNamespace from unittest.mock import patch import pytest @@ -154,6 +155,170 @@ async def fallback(_serial, result): assert result["headsetToHostMbps"] is None +class _FallbackProcess: + def __init__(self, *, drain: str, read: str, wait_timeout: bool = False): + self.drain_behavior = drain + self.read_behavior = read + self.wait_timeout = wait_timeout + self.drain_started = asyncio.Event() + self.read_started = asyncio.Event() + self.wait_started = asyncio.Event() + self.wait_released = None + self.stdin = self + self.stdout = self + self.returncode = None + self.closed = False + self.terminated = False + self.killed = False + self.reaped = False + + def write(self, _payload): + pass + + async def drain(self): + self.drain_started.set() + if self.drain_behavior == "stall": + await asyncio.Future() + if self.drain_behavior == "timeout": + raise TimeoutError + + def close(self): + self.closed = True + + async def read(self, _size): + self.read_started.set() + if self.read_behavior == "stall": + await asyncio.Future() + if self.read_behavior == "timeout": + raise TimeoutError + return b"0\n" + + def terminate(self): + self.terminated = True + + def kill(self): + self.killed = True + + async def wait(self): + self.wait_started.set() + if self.wait_released is not None: + await self.wait_released.wait() + if self.wait_timeout and not self.killed: + raise TimeoutError + self.reaped = True + self.returncode = -9 if self.killed else -15 if self.terminated else 0 + return self.returncode + + +async def _missing_curl(_serial, _script, _timeout): + return 0, "/system/bin/timeout\n1791403150633088373\n" + + +@pytest.mark.asyncio +async def test_fallback_cancel_during_drain_survives_cleanup_timeout(monkeypatch): + child = _FallbackProcess(drain="stall", read="timeout", wait_timeout=True) + + async def spawn(*_args, **_kwargs): + return child + + monkeypatch.setattr(usb_health, "_adb_shell", _missing_curl) + monkeypatch.setattr(usb_health.asyncio, "create_subprocess_exec", spawn) + task = asyncio.create_task(usb_health.measure_transfer("selected", 48322, 500000)) + await asyncio.wait_for(child.drain_started.wait(), 1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert child.closed + assert child.terminated + assert child.killed + assert child.reaped + + +@pytest.mark.asyncio +async def test_fallback_repeated_cancel_waits_for_bounded_reaping(monkeypatch): + child = _FallbackProcess(drain="stall", read="timeout") + child.wait_released = asyncio.Event() + + async def spawn(*_args, **_kwargs): + return child + + monkeypatch.setattr(usb_health, "_adb_shell", _missing_curl) + monkeypatch.setattr(usb_health.asyncio, "create_subprocess_exec", spawn) + task = asyncio.create_task(usb_health.measure_transfer("selected", 48322, 500000)) + await asyncio.wait_for(child.drain_started.wait(), 1) + task.cancel() + await asyncio.wait_for(child.wait_started.wait(), 1) + task.cancel() + child.wait_released.set() + with pytest.raises(asyncio.CancelledError): + await task + assert child.terminated + assert child.reaped + + +@pytest.mark.asyncio +async def test_fallback_cancel_during_cleanup_read_reaps_child(monkeypatch): + child = _FallbackProcess(drain="complete", read="stall") + ticks = iter((0.0, 0.0, 0.0, 3.0, 3.0)) + + async def spawn(*_args, **_kwargs): + return child + + monkeypatch.setattr(usb_health, "_adb_shell", _missing_curl) + monkeypatch.setattr(usb_health.asyncio, "create_subprocess_exec", spawn) + monkeypatch.setattr( + usb_health, "time", SimpleNamespace(monotonic=lambda: next(ticks, 3.0)) + ) + task = asyncio.create_task(usb_health.measure_transfer("selected", 48322, 500000)) + await asyncio.wait_for(child.read_started.wait(), 1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert child.closed + assert child.terminated + assert child.reaped + + +@pytest.mark.asyncio +async def test_fallback_normal_drain_timeout_returns_unsupported_after_reap( + monkeypatch, +): + child = _FallbackProcess(drain="timeout", read="complete") + + async def spawn(*_args, **_kwargs): + return child + + monkeypatch.setattr(usb_health, "_adb_shell", _missing_curl) + monkeypatch.setattr(usb_health.asyncio, "create_subprocess_exec", spawn) + result = await usb_health.measure_transfer("selected", 48322, 500000) + assert result["outcome"] == "unsupported" + assert result["completed"] is False + assert result["childCleanup"]["reaped"] is True + assert child.closed + assert child.reaped + + +@pytest.mark.asyncio +async def test_fallback_cleanup_timeout_returns_unsupported_after_reap(monkeypatch): + child = _FallbackProcess(drain="complete", read="timeout") + ticks = iter((0.0, 0.0, 0.0, 3.0, 3.0)) + + async def spawn(*_args, **_kwargs): + return child + + monkeypatch.setattr(usb_health, "_adb_shell", _missing_curl) + monkeypatch.setattr(usb_health.asyncio, "create_subprocess_exec", spawn) + monkeypatch.setattr( + usb_health, "time", SimpleNamespace(monotonic=lambda: next(ticks, 3.0)) + ) + result = await usb_health.measure_transfer("selected", 48322, 500000) + assert result["outcome"] == "unsupported" + assert result["childCleanup"]["reaped"] is True + assert child.closed + assert child.terminated + assert child.reaped + + def test_atomic_report_is_private_and_redacts_serial(tmp_path): report = usb_health.UsbHealthReport(tmp_path, "session-id", 0, "SECRET-SERIAL") report.update(topology={"probeSucceeded": False}, errors=["sysfs unavailable"]) From 89cd1c62859bb8c6567d9e615532d1d8e5988c44 Mon Sep 17 00:00:00 2001 From: Andrew Wilson Date: Thu, 8 Oct 2026 12:17:19 -0500 Subject: [PATCH 4/9] chore(cloudxr): clarify USB fallback cleanup races Signed-off-by: Andrew Wilson --- src/python/isaaccapture/cloudxr/usb_health.py | 11 +++++----- tests/python/core/cloudxr/test_usb_health.py | 20 ++++++++++++++++++- 2 files changed, 24 insertions(+), 7 deletions(-) diff --git a/src/python/isaaccapture/cloudxr/usb_health.py b/src/python/isaaccapture/cloudxr/usb_health.py index c7c28dfa0c..c4ecae4f46 100644 --- a/src/python/isaaccapture/cloudxr/usb_health.py +++ b/src/python/isaaccapture/cloudxr/usb_health.py @@ -14,6 +14,7 @@ import shlex import stat import time +from contextlib import suppress from datetime import datetime, timezone from pathlib import Path @@ -333,19 +334,16 @@ async def measure_transfer(serial: str, port: int, asset_size: int) -> dict: async def _terminate_fallback_child(child) -> bool: """Bound termination and reaping of the ADB sink child.""" if child.returncode is None: - try: + # The child may exit between the returncode check and the signal. + with suppress(ProcessLookupError): child.terminate() - except ProcessLookupError: - pass try: await asyncio.wait_for(child.wait(), 0.5) return True except TimeoutError: if child.returncode is None: - try: + with suppress(ProcessLookupError): child.kill() - except ProcessLookupError: - pass try: await asyncio.wait_for(child.wait(), 0.5) return True @@ -394,6 +392,7 @@ async def _measure_fallback(serial: str, result: dict) -> None: result["effectiveMeasurementDurationMs"] = round(duration, 2) result["timingValid"] = 2950 <= duration <= 3250 result["hostBytesSubmitted"] = written + # Preserve every interruption through cleanup, including process-exit signals. except BaseException as exc: failure = exc try: diff --git a/tests/python/core/cloudxr/test_usb_health.py b/tests/python/core/cloudxr/test_usb_health.py index 6ad971b58d..260e9761fa 100644 --- a/tests/python/core/cloudxr/test_usb_health.py +++ b/tests/python/core/cloudxr/test_usb_health.py @@ -12,7 +12,7 @@ from pathlib import Path from threading import Barrier from types import SimpleNamespace -from unittest.mock import patch +from unittest.mock import AsyncMock, Mock, patch import pytest @@ -214,6 +214,24 @@ async def _missing_curl(_serial, _script, _timeout): return 0, "/system/bin/timeout\n1791403150633088373\n" +@pytest.mark.asyncio +async def test_fallback_signal_races_still_reap_exited_child(): + exited_before_terminate = SimpleNamespace( + returncode=None, + terminate=Mock(side_effect=ProcessLookupError), + wait=AsyncMock(return_value=0), + ) + assert await usb_health._terminate_fallback_child(exited_before_terminate) + + exited_before_kill = SimpleNamespace( + returncode=None, + terminate=Mock(), + kill=Mock(side_effect=ProcessLookupError), + wait=AsyncMock(side_effect=[TimeoutError, 0]), + ) + assert await usb_health._terminate_fallback_child(exited_before_kill) + + @pytest.mark.asyncio async def test_fallback_cancel_during_drain_survives_cleanup_timeout(monkeypatch): child = _FallbackProcess(drain="stall", read="timeout", wait_timeout=True) From 64cef2b60646a715f5822e69556340f8c3b96267 Mon Sep 17 00:00:00 2001 From: Andrew Wilson Date: Thu, 8 Oct 2026 13:43:00 -0500 Subject: [PATCH 5/9] fix(cloudxr): preserve ADB shell interruption on process exit Signed-off-by: Andrew Wilson --- src/python/isaaccapture/cloudxr/usb_health.py | 7 +- tests/python/core/cloudxr/test_usb_health.py | 90 +++++++++++++++++-- 2 files changed, 86 insertions(+), 11 deletions(-) diff --git a/src/python/isaaccapture/cloudxr/usb_health.py b/src/python/isaaccapture/cloudxr/usb_health.py index c4ecae4f46..f27f616603 100644 --- a/src/python/isaaccapture/cloudxr/usb_health.py +++ b/src/python/isaaccapture/cloudxr/usb_health.py @@ -192,11 +192,14 @@ async def _adb_shell(serial: str, script: str, timeout: float) -> tuple[int, str output, _ = await asyncio.wait_for(child.communicate(), timeout) return child.returncode, output[:65536].decode("ascii", "replace") except (asyncio.CancelledError, TimeoutError): - child.terminate() + # ADB may exit before the signal; still drain and reap its pipe. + with suppress(ProcessLookupError): + child.terminate() try: await asyncio.wait_for(child.communicate(), 0.5) except TimeoutError: - child.kill() + with suppress(ProcessLookupError): + child.kill() await child.communicate() raise diff --git a/tests/python/core/cloudxr/test_usb_health.py b/tests/python/core/cloudxr/test_usb_health.py index 260e9761fa..5e3745e5ae 100644 --- a/tests/python/core/cloudxr/test_usb_health.py +++ b/tests/python/core/cloudxr/test_usb_health.py @@ -155,6 +155,69 @@ async def fallback(_serial, result): assert result["headsetToHostMbps"] is None +class _AdbShellProcess: + def __init__(self, *, first: str, cleanup: str, signal_race: str): + self.first = first + self.cleanup = cleanup + self.signal_race = signal_race + self.started = asyncio.Event() + self.calls = 0 + self.reaped = False + self.returncode = None + + async def communicate(self): + self.calls += 1 + if self.calls == 1: + self.started.set() + if self.first == "stall": + await asyncio.Future() + raise TimeoutError + if self.calls == 2 and self.cleanup == "timeout": + raise TimeoutError + self.reaped = True + self.returncode = 0 + return b"", b"" + + def terminate(self): + if self.signal_race == "terminate": + raise ProcessLookupError + + def kill(self): + if self.signal_race == "kill": + raise ProcessLookupError + + +@pytest.mark.asyncio +async def test_adb_shell_terminate_race_preserves_measurement_cancellation(monkeypatch): + child = _AdbShellProcess(first="stall", cleanup="complete", signal_race="terminate") + + async def spawn(*_args, **_kwargs): + return child + + monkeypatch.setattr(usb_health.asyncio, "create_subprocess_exec", spawn) + task = asyncio.create_task(usb_health.measure_transfer("selected", 48322, 500000)) + await asyncio.wait_for(child.started.wait(), 1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert child.calls == 2 + assert child.reaped + + +@pytest.mark.asyncio +async def test_adb_shell_kill_race_preserves_timeout_and_reaps(monkeypatch): + child = _AdbShellProcess(first="timeout", cleanup="timeout", signal_race="kill") + + async def spawn(*_args, **_kwargs): + return child + + monkeypatch.setattr(usb_health.asyncio, "create_subprocess_exec", spawn) + with pytest.raises(TimeoutError): + await usb_health._adb_shell("selected", "date +%s%N", 1) + assert child.calls == 3 + assert child.reaped + + class _FallbackProcess: def __init__(self, *, drain: str, read: str, wait_timeout: bool = False): self.drain_behavior = drain @@ -884,7 +947,7 @@ async def test_transport_loss_invalidates_current_wake_but_preserves_history(tmp @pytest.mark.parametrize( ("reconnected_speed", "decision"), - [(480, "degraded"), (5000, "incomplete")], + [(480, "degraded"), (5000, "incomplete"), (None, "degraded")], ) @pytest.mark.asyncio async def test_preserved_browser_reconnect_refreshes_link_without_active_transfer( @@ -914,6 +977,7 @@ async def test_preserved_browser_reconnect_refreshes_link_without_active_transfe lifecycle._ready_count = 1 speeds = iter((5000, reconnected_speed)) observed = [] + listing_calls = 0 def inspect(_serial, _listing): speed = next(speeds) @@ -936,14 +1000,17 @@ async def forbidden_transfer(): async def stop(_seconds): raise asyncio.CancelledError - monkeypatch.setattr(usb_health, "inspect_link", inspect) - monkeypatch.setattr( - adb, - "_adb_run", - lambda args, **kwargs: subprocess.CompletedProcess( + def listing(args, **_kwargs): + nonlocal listing_calls + listing_calls += 1 + if reconnected_speed is None and listing_calls == 2: + raise subprocess.TimeoutExpired(args, 4) + return subprocess.CompletedProcess( args, 0, "selected device usb:1-2.1 model:A9210\n", "" - ), - ) + ) + + monkeypatch.setattr(usb_health, "inspect_link", inspect) + monkeypatch.setattr(adb, "_adb_run", listing) lifecycle._rebuild_usb = rebuild lifecycle._recover_existing_browser = recover lifecycle._check_usb_transfer = forbidden_transfer @@ -978,9 +1045,14 @@ async def stop(_seconds): assert payload["transferHistory"][0]["hostToHeadsetResponseBodyMbps"] == 900 assert payload["decision"]["transferRetestDeferred"] is True assert payload["decision"]["overallOutcome"] == decision + assert payload["linkProbe"]["probeSucceeded"] is (reconnected_speed is not None) awake.assert_called_once_with(timeout=10.0) assert lifecycle.snapshot["usbLinkOutcome"] == ( - "pass" if reconnected_speed >= 5000 else "slow_link" + "unknown" + if reconnected_speed is None + else "pass" + if reconnected_speed >= 5000 + else "slow_link" ) assert lifecycle.snapshot["usbTransferMbps"] is None assert lifecycle.snapshot["usbTransferResponseBodyMbps"] is None From 143a86631d1379f9f65af047229fa0e8df4b9ce2 Mon Sep 17 00:00:00 2001 From: Andrew Wilson Date: Thu, 8 Oct 2026 13:50:02 -0500 Subject: [PATCH 6/9] test(cloudxr): document fallback interruption cleanup Signed-off-by: Andrew Wilson --- src/python/isaaccapture/cloudxr/usb_health.py | 5 ++- tests/python/core/cloudxr/test_usb_health.py | 33 +++++++++++++++++++ 2 files changed, 37 insertions(+), 1 deletion(-) diff --git a/src/python/isaaccapture/cloudxr/usb_health.py b/src/python/isaaccapture/cloudxr/usb_health.py index f27f616603..6fe2044342 100644 --- a/src/python/isaaccapture/cloudxr/usb_health.py +++ b/src/python/isaaccapture/cloudxr/usb_health.py @@ -395,7 +395,8 @@ async def _measure_fallback(serial: str, result: dict) -> None: result["effectiveMeasurementDurationMs"] = round(duration, 2) result["timingValid"] = 2950 <= duration <= 3250 result["hostBytesSubmitted"] = written - # Preserve every interruption through cleanup, including process-exit signals. + # drain() can raise cancellation or a non-Exception signal; retain it until + # the owned ADB child is reaped, then re-raise that same failure below. except BaseException as exc: failure = exc try: @@ -407,6 +408,8 @@ async def _measure_fallback(serial: str, result: dict) -> None: try: output = await asyncio.wait_for(child.stdout.read(64), 0.5) await asyncio.wait_for(child.wait(), 0.5) + # The final read/wait can also be interrupted; Exception would miss + # CancelledError and skip reaping before the original signal propagates. except BaseException as exc: failure = exc if failure is not None: diff --git a/tests/python/core/cloudxr/test_usb_health.py b/tests/python/core/cloudxr/test_usb_health.py index 5e3745e5ae..4caba5fd22 100644 --- a/tests/python/core/cloudxr/test_usb_health.py +++ b/tests/python/core/cloudxr/test_usb_health.py @@ -218,6 +218,10 @@ async def spawn(*_args, **_kwargs): assert child.reaped +class _ProbeAbort(BaseException): + pass + + class _FallbackProcess: def __init__(self, *, drain: str, read: str, wait_timeout: bool = False): self.drain_behavior = drain @@ -244,6 +248,8 @@ async def drain(self): await asyncio.Future() if self.drain_behavior == "timeout": raise TimeoutError + if self.drain_behavior == "abort": + raise _ProbeAbort def close(self): self.closed = True @@ -254,6 +260,8 @@ async def read(self, _size): await asyncio.Future() if self.read_behavior == "timeout": raise TimeoutError + if self.read_behavior == "abort": + raise _ProbeAbort return b"0\n" def terminate(self): @@ -360,6 +368,31 @@ async def spawn(*_args, **_kwargs): assert child.reaped +@pytest.mark.parametrize("phase", ["drain", "final_read"]) +@pytest.mark.asyncio +async def test_fallback_non_exception_abort_is_rethrown_after_reap(monkeypatch, phase): + child = _FallbackProcess( + drain="abort" if phase == "drain" else "complete", + read="abort" if phase == "final_read" else "complete", + ) + + async def spawn(*_args, **_kwargs): + return child + + monkeypatch.setattr(usb_health, "_adb_shell", _missing_curl) + monkeypatch.setattr(usb_health.asyncio, "create_subprocess_exec", spawn) + if phase == "final_read": + ticks = iter((0.0, 0.0, 0.0, 3.0, 3.0)) + monkeypatch.setattr( + usb_health, "time", SimpleNamespace(monotonic=lambda: next(ticks, 3.0)) + ) + with pytest.raises(_ProbeAbort): + await usb_health.measure_transfer("selected", 48322, 500000) + assert child.closed + assert child.terminated + assert child.reaped + + @pytest.mark.asyncio async def test_fallback_normal_drain_timeout_returns_unsupported_after_reap( monkeypatch, From 682c3925337f073f47d5ce31fabb96106e567a0f Mon Sep 17 00:00:00 2001 From: Andrew Wilson Date: Thu, 8 Oct 2026 14:41:49 -0500 Subject: [PATCH 7/9] fix(cloudxr): bound ADB shell cleanup under cancellation Signed-off-by: Andrew Wilson --- .../cloudxr/oob_teleop_lifecycle.py | 6 ++ src/python/isaaccapture/cloudxr/usb_health.py | 48 +++++++++++++--- tests/python/core/cloudxr/test_usb_health.py | 55 ++++++++++++++++++- 3 files changed, 101 insertions(+), 8 deletions(-) diff --git a/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py b/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py index 2a40345a7a..9ef3024922 100644 --- a/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py +++ b/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py @@ -565,6 +565,7 @@ async def _prepare_device(self) -> None: self._cache_cleared = True def _ensure_usb_report(self) -> None: + """Open the USB-local report and bind it to the selected device when known.""" if ( not self.usb_local or self.usb_health_policy == "off" @@ -590,6 +591,7 @@ def _ensure_usb_report(self) -> None: ) def _record_usb_wait(self, state: str, code: str) -> None: + """Persist one deduplicated startup wait with a bounded error history.""" self._ensure_usb_report() signature = (state, code) if self._usb_health_report is None or signature == self._usb_wait_signature: @@ -610,6 +612,7 @@ def _record_usb_wait(self, state: str, code: str) -> None: ) def _link_outcome(self) -> str | None: + """Classify the current negotiated link, or leave it unknown if absent.""" if self._usb_link is None: return None speed = self._usb_link.get("negotiatedSpeedMbps") @@ -847,6 +850,7 @@ async def _confirm_usb_awake(self) -> None: self._record_usb_event("USB_WAKE_CONFIRMED") def _persist_usb_health(self, **sections) -> None: + """Best-effort persist report sections without interrupting recovery.""" if self._usb_health_report is None: return try: @@ -855,6 +859,7 @@ def _persist_usb_health(self, **sections) -> None: log.warning("USB health report update failed", exc_info=True) def _record_usb_event(self, state: str) -> None: + """Best-effort persist one USB startup or recovery transition.""" if self._usb_health_report is None: return try: @@ -863,6 +868,7 @@ def _record_usb_event(self, state: str) -> None: log.warning("USB health report event update failed", exc_info=True) def _usb_launch_allowed(self) -> bool: + """Apply the configured warn/block policy to current USB evidence.""" speed = self._usb_link.get("negotiatedSpeedMbps") if self._usb_link else None transfer_ok = bool(self._usb_transfer and self._usb_transfer.get("completed")) degraded = speed is None or speed < 5000 or not transfer_ok diff --git a/src/python/isaaccapture/cloudxr/usb_health.py b/src/python/isaaccapture/cloudxr/usb_health.py index 6fe2044342..32c9a8c736 100644 --- a/src/python/isaaccapture/cloudxr/usb_health.py +++ b/src/python/isaaccapture/cloudxr/usb_health.py @@ -63,6 +63,7 @@ def _load_or_create_serial_salt(directory: Path) -> bytes: def _read(path: Path) -> str | None: + """Read bounded sysfs text, returning unknown for inaccessible fields.""" try: return path.read_text(encoding="utf-8").strip()[:160] except (OSError, UnicodeError): @@ -70,6 +71,7 @@ def _read(path: Path) -> str | None: def _number(path: Path) -> float | None: + """Parse an optional numeric sysfs field without assuming it exists.""" try: return float(_read(path)) except (TypeError, ValueError): @@ -77,12 +79,14 @@ def _number(path: Path) -> float | None: def _pci(path: Path) -> str | None: + """Find the PCI controller that owns a resolved USB sysfs node.""" return next( (p.name for p in path.resolve().parents if PCI_NODE.fullmatch(p.name)), None ) def _node(path: Path) -> dict: + """Capture the non-secret sysfs attributes of one USB topology node.""" return { "node": path.name, "speedMbps": _number(path / "speed"), @@ -178,6 +182,7 @@ def inspect_link( async def _adb_shell(serial: str, script: str, timeout: float) -> tuple[int, str]: + """Run pinned ADB shell, preserving timeout/cancellation after bounded cleanup.""" child = await asyncio.create_subprocess_exec( "adb", "-s", @@ -191,17 +196,44 @@ async def _adb_shell(serial: str, script: str, timeout: float) -> tuple[int, str try: output, _ = await asyncio.wait_for(child.communicate(), timeout) return child.returncode, output[:65536].decode("ascii", "replace") - except (asyncio.CancelledError, TimeoutError): - # ADB may exit before the signal; still drain and reap its pipe. + except (asyncio.CancelledError, TimeoutError) as original_failure: + cleanup = asyncio.create_task(_terminate_adb_shell_child(child)) + later_cancellation = None + while not cleanup.done(): + try: + await asyncio.shield(cleanup) + except asyncio.CancelledError as exc: + # A later cancellation must not abandon the owned child. + later_cancellation = exc + # Cleanup failure cannot replace the original timeout or cancellation. + with suppress(Exception): + cleanup.result() + if later_cancellation is not None: + raise later_cancellation from None + raise original_failure from None + + +async def _terminate_adb_shell_child(child) -> bool: + """Attempt bounded drain and reap, even when a descendant holds stdout.""" + # ADB may exit before either signal; communicate still drains its pipe. + with suppress(ProcessLookupError): + child.terminate() + try: + await asyncio.wait_for(child.communicate(), 0.5) + return True + except TimeoutError: with suppress(ProcessLookupError): - child.terminate() + child.kill() try: await asyncio.wait_for(child.communicate(), 0.5) + return True except TimeoutError: - with suppress(ProcessLookupError): - child.kill() - await child.communicate() - raise + # A descendant may hold stdout open after ADB itself exits. + try: + await asyncio.wait_for(child.wait(), 0.5) + return True + except TimeoutError: + return False async def measure_transfer(serial: str, port: int, asset_size: int) -> dict: @@ -441,6 +473,7 @@ def __init__( serial: str | None, policy: str = "warn", ) -> None: + """Prepare private report state and a salted identity for this session.""" self.directory = logs_dir / "usb-health" if self.directory.is_symlink(): raise OSError("USB health report directory cannot be a symlink") @@ -525,6 +558,7 @@ def update(self, **sections) -> None: temporary.unlink(missing_ok=True) def _prune(self) -> None: + """Keep at most 20 reports, dropping retained files older than 14 days.""" files = sorted( ( p diff --git a/tests/python/core/cloudxr/test_usb_health.py b/tests/python/core/cloudxr/test_usb_health.py index 4caba5fd22..ba6cccf1a5 100644 --- a/tests/python/core/cloudxr/test_usb_health.py +++ b/tests/python/core/cloudxr/test_usb_health.py @@ -161,7 +161,11 @@ def __init__(self, *, first: str, cleanup: str, signal_race: str): self.cleanup = cleanup self.signal_race = signal_race self.started = asyncio.Event() + self.cleanup_started = asyncio.Event() + self.cleanup_released = asyncio.Event() self.calls = 0 + self.wait_calls = 0 + self.killed = False self.reaped = False self.returncode = None @@ -172,7 +176,12 @@ async def communicate(self): if self.first == "stall": await asyncio.Future() raise TimeoutError - if self.calls == 2 and self.cleanup == "timeout": + if self.calls == 2 and self.cleanup in {"timeout", "pipe_held_after_kill"}: + raise TimeoutError + if self.calls == 2 and self.cleanup == "stall": + self.cleanup_started.set() + await self.cleanup_released.wait() + if self.calls == 3 and self.cleanup == "pipe_held_after_kill": raise TimeoutError self.reaped = True self.returncode = 0 @@ -183,9 +192,16 @@ def terminate(self): raise ProcessLookupError def kill(self): + self.killed = True if self.signal_race == "kill": raise ProcessLookupError + async def wait(self): + self.wait_calls += 1 + self.reaped = True + self.returncode = -9 + return self.returncode + @pytest.mark.asyncio async def test_adb_shell_terminate_race_preserves_measurement_cancellation(monkeypatch): @@ -218,6 +234,43 @@ async def spawn(*_args, **_kwargs): assert child.reaped +@pytest.mark.asyncio +async def test_adb_shell_reaps_killed_child_when_descendant_holds_stdout(monkeypatch): + child = _AdbShellProcess( + first="timeout", cleanup="pipe_held_after_kill", signal_race="none" + ) + + async def spawn(*_args, **_kwargs): + return child + + monkeypatch.setattr(usb_health.asyncio, "create_subprocess_exec", spawn) + with pytest.raises(TimeoutError): + await usb_health._adb_shell("selected", "date +%s%N", 1) + assert child.killed + assert child.calls == 3 + assert child.wait_calls == 1 + assert child.reaped + + +@pytest.mark.asyncio +async def test_adb_shell_repeated_cancellation_reaps_before_return(monkeypatch): + child = _AdbShellProcess(first="stall", cleanup="stall", signal_race="none") + + async def spawn(*_args, **_kwargs): + return child + + monkeypatch.setattr(usb_health.asyncio, "create_subprocess_exec", spawn) + task = asyncio.create_task(usb_health._adb_shell("selected", "date +%s%N", 1)) + await asyncio.wait_for(child.started.wait(), 1) + task.cancel() + await asyncio.wait_for(child.cleanup_started.wait(), 1) + task.cancel() + child.cleanup_released.set() + with pytest.raises(asyncio.CancelledError): + await task + assert child.reaped + + class _ProbeAbort(BaseException): pass From 104c9beabd7ea709c21aa36962e1e0e91c5790d0 Mon Sep 17 00:00:00 2001 From: Andrew Wilson Date: Thu, 8 Oct 2026 14:57:39 -0500 Subject: [PATCH 8/9] fix(cloudxr): resume USB startup after headset wakes Signed-off-by: Andrew Wilson --- .../cloudxr/oob_teleop_lifecycle.py | 42 +++- tests/python/core/cloudxr/test_usb_health.py | 202 ++++++++++++++++++ 2 files changed, 241 insertions(+), 3 deletions(-) diff --git a/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py b/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py index 9ef3024922..b7ceeabafe 100644 --- a/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py +++ b/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py @@ -126,6 +126,8 @@ def __init__( self._usb_transfer_history: list[dict] = [] self._usb_transfer_stale = False self._usb_wake_confirmed: bool | None = None + # Only a failed strict wake check permits polling after episode expiry. + self._usb_waiting_for_awake = False self._usb_link: dict | None = None self._usb_transfer: dict | None = None @@ -342,6 +344,7 @@ async def _stop_monitor(self) -> None: async def _remember_transport_loss(self) -> None: """Drop transport-owned state while preserving the prior browser intent.""" + self._usb_waiting_for_awake = False if self._usb_wake_confirmed is not None: self._usb_wake_confirmed = None if self._usb_health_report is not None: @@ -806,6 +809,7 @@ async def _check_usb_transfer(self) -> None: async def _confirm_usb_awake(self) -> None: """Require a fresh Awake observation immediately before active transfer.""" self._usb_wake_confirmed = False + self._usb_waiting_for_awake = False started = self.clock() await self._publish( "degraded", @@ -823,6 +827,7 @@ async def _confirm_usb_awake(self) -> None: adb.assert_headset_awake, timeout=10.0, require_awake=True ) except (Exception, asyncio.CancelledError): + self._usb_waiting_for_awake = True self._persist_usb_health( wakeCheck={ "confirmedAwake": False, @@ -1686,11 +1691,41 @@ async def run(self) -> None: and not self.browser_ready and not self.connect_dispatched ): - self._record_usb_wait("WAITING_FOR_ADB", "recovery_deadline") + waiting_for_awake = ( + self.usb_local + and self.usb_health_policy != "off" + and self._usb_waiting_for_awake + ) + if waiting_for_awake: + try: + wake = await asyncio.to_thread(adb.headset_wakefulness) + except asyncio.CancelledError: + raise + except Exception: + log.debug( + "Passive USB wake observation failed", exc_info=True + ) + wake = "" + if wake == "Awake": + self._usb_waiting_for_awake = False + self._record_usb_event("USB_WAKE_OBSERVED") + self._restart_episode() + continue + wait_state = ( + "WAITING_FOR_AWAKE" if waiting_for_awake else "WAITING_FOR_ADB" + ) + self._record_usb_wait( + wait_state, + "awake_recovery_deadline" + if waiting_for_awake + else "recovery_deadline", + ) await self._publish( "degraded", - "WAITING_FOR_ADB", - "Recovery episode expired; observing for a change", + wait_state, + "Recovery episode expired; waiting for selected headset to wake" + if waiting_for_awake + else "Recovery episode expired; observing for a change", adbReady=True, networkPresent=True, ) @@ -1721,6 +1756,7 @@ async def run(self) -> None: self._record_usb_event("PREPARING_DEVICE") if self.usb_local: self._usb_wake_confirmed = False + self._usb_waiting_for_awake = False try: await self._prepare_device() except (Exception, asyncio.CancelledError) as exc: diff --git a/tests/python/core/cloudxr/test_usb_health.py b/tests/python/core/cloudxr/test_usb_health.py index ba6cccf1a5..8816c768a8 100644 --- a/tests/python/core/cloudxr/test_usb_health.py +++ b/tests/python/core/cloudxr/test_usb_health.py @@ -686,6 +686,208 @@ async def sleep(_seconds): assert events == expected +@pytest.mark.asyncio +async def test_expired_usb_wake_wait_resumes_when_selected_headset_wakes( + tmp_path, monkeypatch +): + now = [0.0] + wake_polls = [] + strict_checks = [] + transfers = [] + launches = [] + statuses = [] + selected = "selected-headset" + devices = adb.AdbDevices(((selected, "device"),)) + lifecycle = OobLifecycle( + hub=_Hub(), + resolved_port=48322, + usb_local=True, + host_client=True, + turn_port=3478, + config=RecoveryConfig(timeout_sec=60, interval_sec=5), + clock=lambda: now[0], + on_status=statuses.append, + usb_health_logs_dir=tmp_path, + ) + lifecycle.selected = selected + lifecycle._last_observation = (devices.devices, devices.diagnostic) + lifecycle._ready_count = 1 + + async def sleep(seconds): + now[0] += seconds + if now[0] >= 90: + raise asyncio.CancelledError + + def wakefulness(): + assert adb.SELECTED_ADB_SERIAL.get() == selected + wake_polls.append(now[0]) + return "Awake" if now[0] >= 65 else "Asleep" + + def strict_wake(*, timeout, require_awake): + assert adb.SELECTED_ADB_SERIAL.get() == selected + assert timeout == 10.0 and require_awake + strict_checks.append(now[0]) + if now[0] < 65: + raise adb.OobAdbError("still asleep") + + async def link(): + lifecycle._usb_link = {"negotiatedSpeedMbps": 5000} + lifecycle._persist_usb_health(topology=lifecycle._usb_link) + + async def transfer(): + assert lifecycle._usb_wake_confirmed is True + transfers.append(now[0]) + lifecycle._usb_transfer = {"completed": True, "outcome": "pass"} + lifecycle._persist_usb_health(transferTest=lifecycle._usb_transfer) + + async def browser(): + launches.append(now[0]) + raise asyncio.CancelledError + + lifecycle.sleep = sleep + monkeypatch.setattr(lifecycle, "_prepare_device", AsyncMock()) + monkeypatch.setattr(lifecycle, "_check_usb_link", link) + monkeypatch.setattr(lifecycle, "_rebuild_usb", AsyncMock()) + monkeypatch.setattr(lifecycle, "_check_usb_transfer", transfer) + monkeypatch.setattr(lifecycle, "_automate", browser) + with ( + patch.object(adb, "enumerate_adb_devices", return_value=devices), + patch.object( + adb, + "probe_headset_network", + return_value=adb.HeadsetNetworkProbe( + adb.HeadsetNetworkState.NETWORK_PRESENT + ), + ), + patch.object(adb, "_run_adb", return_value="stable-reverse-rules"), + patch.object(adb, "headset_wakefulness", side_effect=wakefulness), + patch.object(adb, "assert_headset_awake", side_effect=strict_wake), + ): + with pytest.raises(asyncio.CancelledError): + await lifecycle.run() + + assert wake_polls == [60.0, 65.0] + assert strict_checks[-1] == 65.0 + assert transfers == launches == [65.0] + assert any( + status["state"] == "WAITING_FOR_AWAKE" + and status["adbReady"] + and status["reason"] + == "Recovery episode expired; waiting for selected headset to wake" + for status in statuses + ) + report = json.loads(lifecycle._usb_health_report.path.read_text()) + assert report["wakeCheck"]["confirmedAwake"] is True + assert report["decision"]["overallOutcome"] == "pass" + assert any(error["code"] == "awake_not_confirmed" for error in report["errors"]) + assert any( + error["state"] == "WAITING_FOR_AWAKE" + and error["code"] == "awake_recovery_deadline" + for error in report["errors"] + ) + assert any(event["state"] == "USB_WAKE_OBSERVED" for event in report["events"]) + + +@pytest.mark.parametrize( + ("usb_local", "policy", "waiting_for_awake"), + [(True, "warn", False), (True, "off", True), (False, "warn", True)], +) +@pytest.mark.asyncio +async def test_unrelated_expired_episode_does_not_poll_usb_wake( + usb_local, policy, waiting_for_awake +): + now = [0.0] + selected = "selected-headset" + devices = adb.AdbDevices(((selected, "device"),)) + lifecycle = OobLifecycle( + hub=_Hub(), + resolved_port=48322, + usb_local=usb_local, + host_client=True, + turn_port=3478 if usb_local else None, + config=RecoveryConfig(timeout_sec=60, interval_sec=5), + clock=lambda: now[0], + usb_health_policy=policy, + ) + lifecycle.selected = selected + lifecycle._last_observation = (devices.devices, devices.diagnostic) + lifecycle._ready_count = 1 + lifecycle.last_network_state = adb.HeadsetNetworkState.NETWORK_PRESENT + lifecycle._usb_waiting_for_awake = waiting_for_awake + now[0] = 65.0 + + async def stop(_seconds): + raise asyncio.CancelledError + + lifecycle.sleep = stop + with ( + patch.object(adb, "enumerate_adb_devices", return_value=devices), + patch.object( + adb, + "probe_headset_network", + return_value=adb.HeadsetNetworkProbe( + adb.HeadsetNetworkState.NETWORK_PRESENT + ), + ), + patch.object(adb, "_run_adb", return_value="stable-reverse-rules"), + patch.object(adb, "headset_wakefulness") as wake, + ): + with pytest.raises(asyncio.CancelledError): + await lifecycle.run() + wake.assert_not_called() + assert lifecycle.episode_start == 0.0 + assert lifecycle.snapshot["state"] == "WAITING_FOR_ADB" + + +@pytest.mark.asyncio +async def test_prepare_failure_clears_prior_wake_wait_before_deadline(): + now = [0.0] + selected = "selected-headset" + devices = adb.AdbDevices(((selected, "device"),)) + lifecycle = OobLifecycle( + hub=_Hub(), + resolved_port=48322, + usb_local=True, + host_client=True, + turn_port=3478, + config=RecoveryConfig(timeout_sec=5, interval_sec=5), + clock=lambda: now[0], + ) + lifecycle.selected = selected + lifecycle._last_observation = (devices.devices, devices.diagnostic) + lifecycle._ready_count = 1 + lifecycle.last_network_state = adb.HeadsetNetworkState.NETWORK_PRESENT + lifecycle._usb_waiting_for_awake = True + + async def sleep(seconds): + now[0] += seconds + if now[0] >= 10: + raise asyncio.CancelledError + + lifecycle.sleep = sleep + with ( + patch.object(adb, "enumerate_adb_devices", return_value=devices), + patch.object( + adb, + "probe_headset_network", + return_value=adb.HeadsetNetworkProbe( + adb.HeadsetNetworkState.NETWORK_PRESENT + ), + ), + patch.object(adb, "_run_adb", return_value="stable-reverse-rules"), + patch.object(adb, "headset_wakefulness") as wake, + patch.object( + lifecycle, + "_prepare_device", + side_effect=adb.OobAdbError("prepare failed"), + ), + ): + with pytest.raises(asyncio.CancelledError): + await lifecycle.run() + wake.assert_not_called() + assert lifecycle._usb_waiting_for_awake is False + + @pytest.mark.asyncio async def test_missing_adb_writes_partial_usb_report(tmp_path, monkeypatch): lifecycle = OobLifecycle( From 9fe8baec9c7f9fc584168d87ecc70a591fd2c52f Mon Sep 17 00:00:00 2001 From: Andrew Wilson Date: Thu, 8 Oct 2026 18:09:38 -0500 Subject: [PATCH 9/9] chore(cloudxr): narrow PR scope and document wake helper Signed-off-by: Andrew Wilson --- .pre-commit-config.yaml | 4 ++-- AGENTS.md | 3 --- .../isaaccapture/cloudxr/oob_teleop_adb.py | 20 ++++++++----------- 3 files changed, 10 insertions(+), 17 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 95942cfb0e..41849867c0 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -24,8 +24,8 @@ repos: - id: check-executables-have-shebangs - id: check-toml - id: end-of-file-fixer - # These LFS SVGs are assets; rewriting their final byte creates new LFS objects. - exclude: ^(docs/source/_static/camera-viz-controls|examples/vehicle_teleop/assets/vehicle_teleop_flow)\.svg$ + # This LFS SVG is an asset; rewriting its final byte creates a new LFS object. + exclude: ^docs/source/_static/camera-viz-controls\.svg$ - id: check-shebang-scripts-are-executable - id: detect-private-key - id: debug-statements diff --git a/AGENTS.md b/AGENTS.md index 4bf6027521..b75eea35c6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -226,9 +226,6 @@ pre-commit install --hook-type commit-msg `git ls-files`, so a file you have created but not yet `git add`ed is skipped and the run passes for the wrong reason — REUSE and `ruff format` alike. Stage first (`git add -A`), then run the hooks. -- LFS SVG assets without a trailing newline need an `end-of-file-fixer` exclusion; - rewriting them creates an unrelated LFS object. Verify the LFS attribute and - restore any hook-only asset change before rerunning. - **REUSE:** files covered by the REUSE hook need **`SPDX-FileCopyrightText`** and **`SPDX-License-Identifier`** in the form the repo already uses (for example the HTML comment block at the top of `README.md` also applies to **`AGENTS.md`** and similar docs). - **C++ formatting is enforced by CI, not pre-commit.** The hook set runs `ruff` for Python but does **not** run `clang-format`; CI (`build-ubuntu.yml`) installs **`clang-format-14`** and rejects unformatted C++ as `-Wclang-format-violations`. Before pushing, format touched C++ with the system `clang-format` (match CI's version 14) and verify: diff --git a/src/python/isaaccapture/cloudxr/oob_teleop_adb.py b/src/python/isaaccapture/cloudxr/oob_teleop_adb.py index a87fcd9c25..2fcf89c248 100644 --- a/src/python/isaaccapture/cloudxr/oob_teleop_adb.py +++ b/src/python/isaaccapture/cloudxr/oob_teleop_adb.py @@ -341,18 +341,14 @@ def headset_wakefulness() -> str: def assert_headset_awake(*, timeout: float = 15.0, require_awake: bool = False) -> None: - """Warn-and-wait when the headset is asleep before launching OOB automation. - - Quest / PICO devices sleep when the proximity sensor is uncovered - (e.g. the headset is sitting on a desk). In that state, ``am start`` - may still register but the screen can return to sleep before the - CONNECT click lands, and WebXR session entry will fail. - - Sends ``KEYCODE_WAKEUP`` once and then polls ``mWakefulness`` for up to - ``timeout`` seconds. Returns silently once the device is ``Awake``. - Otherwise logs a warning and returns — downstream automation may still - succeed if ``am start`` wakes the device. When ``require_awake`` is set, - raise instead; pre-browser USB transfer must have an observed Awake state. + """Check wakefulness before OOB automation, sending one wake key if asleep. + + A sleeping Quest/PICO may accept ``am start`` yet sleep before CONNECT, + preventing WebXR session entry. + + Args: + timeout: Seconds to poll after ``KEYCODE_WAKEUP``; zero skips retry polling. + require_awake: Raise ``OobAdbError`` if still not Awake; otherwise warn and return. """ wake = headset_wakefulness() if wake == "Awake":