diff --git a/docs/source/references/cloudxr.rst b/docs/source/references/cloudxr.rst index 707d1eb987..01f5e3954e 100644 --- a/docs/source/references/cloudxr.rst +++ b/docs/source/references/cloudxr.rst @@ -135,6 +135,36 @@ 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. ``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_adb.py b/src/python/isaaccapture/cloudxr/oob_teleop_adb.py index abfc30e7d1..2fcf89c248 100644 --- a/src/python/isaaccapture/cloudxr/oob_teleop_adb.py +++ b/src/python/isaaccapture/cloudxr/oob_teleop_adb.py @@ -340,18 +340,15 @@ def headset_wakefulness() -> str: return m.group(1) if m else "" -def assert_headset_awake(*, timeout: float = 15.0) -> 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. +def assert_headset_awake(*, timeout: float = 15.0, require_awake: bool = False) -> None: + """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": @@ -384,10 +381,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 d8d3f18f98..b7ceeabafe 100644 --- a/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py +++ b/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py @@ -9,13 +9,17 @@ import logging import os import re +import secrets import socket +import subprocess 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, @@ -37,6 +41,8 @@ def __init__(self, reason: str, *, adb_ready: bool, network_present: bool) -> No @dataclass(frozen=True) class RecoveryConfig: + """Timeout and retry budget for headset transport and browser recovery.""" + timeout_sec: float = 60.0 interval_sec: float = 5.0 client_reconnect_enabled: bool = True @@ -91,6 +97,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 +113,23 @@ 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 + # 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 # Headset selection remains pinned after the first successful choice. self.selected = os.environ.get("ANDROID_SERIAL", "").strip() or None @@ -209,6 +235,36 @@ 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 + ), + "usbTransferResponseBodyMbps": ( + self._usb_transfer.get("hostToHeadsetResponseBodyMbps") + 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, } transport_ready = bool( @@ -288,6 +344,27 @@ 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: + 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 @@ -490,6 +567,340 @@ 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" + 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: + """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: + 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: + """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") + 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, + subprocess.TimeoutExpired, + RuntimeError, + 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 " + "wall-clock; response-body-only: %s Mb/s (%s)", + self._usb_transfer.get("hostToHeadsetMbps"), + self._usb_transfer.get("hostToHeadsetResponseBodyMbps"), + 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 + self._usb_waiting_for_awake = 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._usb_waiting_for_awake = True + 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: + """Best-effort persist report sections without interrupting recovery.""" + 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: + """Best-effort persist one USB startup or recovery transition.""" + 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: + """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 + 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 @@ -1052,6 +1463,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" @@ -1063,6 +1475,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( @@ -1125,6 +1538,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", @@ -1145,6 +1570,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", @@ -1158,6 +1584,8 @@ async def run(self) -> None: if network.state is adb.HeadsetNetworkState.NO_NETWORK: # An absent non-loopback interface cannot satisfy transport # readiness or support browser automation in either mode. + if self.usb_local: + self._record_usb_wait("WAITING_FOR_NETWORK", "network_absent") await self._enter_transport_recovery( "Headset has no non-loopback network", "PREPARING_DEVICE", @@ -1234,6 +1662,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 = ( @@ -1259,10 +1691,41 @@ async def run(self) -> None: and not self.browser_ready and not self.connect_dispatched ): + 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, ) @@ -1290,9 +1753,49 @@ 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 + self._usb_waiting_for_awake = 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._record_usb_wait( + "REBUILDING_USB", type(exc).__name__ + ) + self._persist_usb_health( + 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 @@ -1336,6 +1839,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..32c9a8c736 --- /dev/null +++ b/src/python/isaaccapture/cloudxr/usb_health.py @@ -0,0 +1,576 @@ +# 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 contextlib import suppress +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]$") +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" + + +def _read_serial_salt(path: Path) -> bytes: + """Reject symlinks and incomplete salts before hashing device identifiers.""" + + with os.fdopen(os.open(path, os.O_RDONLY | os.O_NOFOLLOW), "rb") as stream: + salt = stream.read(33) + if len(salt) != 32: + raise OSError("USB health serial salt is not a complete 32-byte value") + return salt + + +def _load_or_create_serial_salt(directory: Path) -> bytes: + """Publish a complete per-install salt before concurrent readers can see it.""" + salt_path = directory / ".serial-salt" + try: + return _read_serial_salt(salt_path) + except FileNotFoundError: + # No salt is published yet; race to publish a complete candidate below. + pass + temporary = directory / f".serial-salt.{secrets.token_hex(8)}.tmp" + descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + try: + with os.fdopen(descriptor, "wb") as stream: + stream.write(secrets.token_bytes(32)) + stream.flush() + os.fsync(stream.fileno()) + try: + os.link(temporary, salt_path) + except FileExistsError: + # Another creator published its complete salt first; use that one. + pass + finally: + temporary.unlink(missing_ok=True) + return _read_serial_salt(salt_path) + + +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): + return 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): + return 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"), + "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]: + """Run pinned ADB shell, preserving timeout/cancellation after bounded cleanup.""" + 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[:65536].decode("ascii", "replace") + 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.kill() + try: + await asyncio.wait_for(child.communicate(), 0.5) + return True + except TimeoutError: + # 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: + """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, + "hostToHeadsetResponseBodyMbps": None, + "headsetToHostMbps": None, + "completedBytes": 0, + "completedRequests": 0, + "method": None, + "pathCoverage": None, + "directionCoverage": "host_to_headset", + "browserRequired": False, + "newListenerCreated": False, + "newReverseRuleCreated": False, + "childCleanup": {"reaped": True}, + "measurementRevision": 2, + "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" + # 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}:%{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) + samples = [ + (int(match.group(1)), float(match.group(2)), float(match.group(3))) + for line in output.splitlines() + 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 + 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 + 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" + 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 _terminate_fallback_child(child) -> bool: + """Bound termination and reaping of the ADB sink child.""" + if child.returncode is None: + # The child may exit between the returncode check and the signal. + with suppress(ProcessLookupError): + child.terminate() + try: + await asyncio.wait_for(child.wait(), 0.5) + return True + except TimeoutError: + if child.returncode is None: + with suppress(ProcessLookupError): + child.kill() + 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( + "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 + failure: BaseException | None = None + output = b"" + 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 + # 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: + 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) + # 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: + 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()) + 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", + ) -> 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") + 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" + ) + self._serial_salt = _load_or_create_serial_salt(self.directory) + 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: + """Store an installation-salted serial hash, never the raw serial.""" + 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: + """Append one timestamped report state transition.""" + self.payload["events"].append( + {"at": datetime.now(timezone.utc).isoformat(), "state": state} + ) + self.update() + + def update(self, **sections) -> None: + """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( + 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: + """Keep at most 20 reports, dropping retained files older than 14 days.""" + 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..8816c768a8 --- /dev/null +++ b/tests/python/core/cloudxr/test_usb_health.py @@ -0,0 +1,1348 @@ +# 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 concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from threading import Barrier +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock, 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: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) + 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["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): + 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"] + assert result["hostToHeadsetResponseBodyMbps"] is None + 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.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 + + 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 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 + return b"", b"" + + def terminate(self): + if self.signal_race == "terminate": + 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): + 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 + + +@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 + + +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 + if self.drain_behavior == "abort": + raise _ProbeAbort + + 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 + if self.read_behavior == "abort": + raise _ProbeAbort + 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_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) + + 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.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, +): + 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"]) + 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:" + ) + + +def test_concurrent_reports_share_only_a_complete_published_salt(tmp_path): + barrier = Barrier(2) + original_link = usb_health.os.link + + def race_publish(source, destination): + barrier.wait(timeout=5) + return original_link(source, destination) + + with ( + patch.object(usb_health.os, "link", side_effect=race_publish), + ThreadPoolExecutor(max_workers=2) as pool, + ): + reports = list( + pool.map( + lambda _: usb_health.UsbHealthReport( + tmp_path, "session", 0, "same-serial" + ), + range(2), + ) + ) + directory = tmp_path / "usb-health" + assert len((directory / ".serial-salt").read_bytes()) == 32 + assert not list(directory.glob(".serial-salt.*.tmp")) + assert ( + reports[0].payload["device"]["serialHash"] + == reports[1].payload["device"]["serialHash"] + ) + + +@pytest.mark.parametrize("invalid_salt", [b"", b"too-short"]) +def test_incomplete_existing_serial_salt_is_rejected(tmp_path, invalid_salt): + directory = tmp_path / "usb-health" + directory.mkdir() + (directory / ".serial-salt").write_bytes(invalid_salt) + with pytest.raises(OSError, match="complete 32-byte"): + usb_health.UsbHealthReport(tmp_path, "session", 0, "selected") + + +@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_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( + 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.parametrize("failure", ["timeout", "symlink_loop"]) +@pytest.mark.asyncio +async def test_link_probe_failure_preserves_report_and_warn_launch( + tmp_path, monkeypatch, failure +): + 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 + browser_reached = [] + + async def no_op(): + pass + + async def awake(): + lifecycle._usb_wake_confirmed = True + + async def transfer(): + lifecycle._usb_transfer = {"completed": True, "outcome": "pass"} + + async def browser(): + browser_reached.append(True) + raise asyncio.CancelledError + + async def stop(_seconds): + raise asyncio.CancelledError + + def listing(args, **_kwargs): + if failure == "timeout": + raise subprocess.TimeoutExpired(args, 4) + return subprocess.CompletedProcess( + args, 0, "selected device usb:1-2.1 model:A9210\n", "" + ) + + def inspect(_serial, _listing): + raise RuntimeError("sysfs symlink loop") + + lifecycle._prepare_device = no_op + lifecycle._rebuild_usb = no_op + lifecycle._confirm_usb_awake = awake + lifecycle._check_usb_transfer = transfer + lifecycle._automate = browser + lifecycle.sleep = stop + monkeypatch.setattr(adb, "_adb_run", listing) + monkeypatch.setattr(usb_health, "inspect_link", inspect) + 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 = json.loads( + next((tmp_path / "usb-health").glob("usb-health-*.json")).read_text() + ) + assert browser_reached == [True] + assert payload["linkProbe"]["probeSucceeded"] is False + assert payload["topology"]["negotiatedSpeedMbps"] is None + assert payload["topology"]["errors"] + + +@pytest.mark.asyncio +async def test_reverse_failure_retains_bounded_structured_error_history( + 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 + lifecycle._ensure_usb_report() + for index in range(20): + lifecycle._record_usb_wait("WAITING_FOR_ADB", f"prior_{index}") + + async def no_op(): + pass + + async def fail_reverse(): + raise adb.OobAdbError("private reverse detail") + + async def stop(_seconds): + raise asyncio.CancelledError + + lifecycle._prepare_device = no_op + lifecycle._check_usb_link = no_op + lifecycle._rebuild_usb = fail_reverse + 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 len(payload["errors"]) == 20 + assert all(isinstance(item, dict) for item in payload["errors"]) + assert payload["errors"][0]["code"] == "prior_1" + assert payload["errors"][-1]["state"] == "REBUILDING_USB" + assert payload["errors"][-1]["code"] == "OobAdbError" + assert payload["decision"]["overallOutcome"] == "incomplete" + assert payload["decision"]["browserLaunchAllowed"] is False + assert "private reverse detail" 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"), (None, "degraded")], +) +@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, + "hostToHeadsetResponseBodyMbps": 900, + } + devices = adb.AdbDevices((("selected", "device"),)) + lifecycle._last_observation = (devices.devices, devices.diagnostic) + lifecycle._ready_count = 1 + speeds = iter((5000, reconnected_speed)) + observed = [] + listing_calls = 0 + + 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 + + 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 + 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["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"] == ( + "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 + assert lifecycle.snapshot["usbTransferMethod"] is None + assert lifecycle.snapshot["usbTransferOutcome"] == "stale"