diff --git a/src/python/isaaccapture/cloudxr/oob_teleop_env.py b/src/python/isaaccapture/cloudxr/oob_teleop_env.py index 467ebc30d6..dfc84681ba 100644 --- a/src/python/isaaccapture/cloudxr/oob_teleop_env.py +++ b/src/python/isaaccapture/cloudxr/oob_teleop_env.py @@ -18,6 +18,7 @@ from urllib.parse import urlencode, urljoin from urllib.request import Request, urlopen +from ..logging_config._core import logging_enabled from .oob_teleop_hub import OOB_WS_PATH log = logging.getLogger("isaaccapture.cloudxr.oob_teleop_env") @@ -567,14 +568,13 @@ def resolve_lan_host_for_oob() -> str: def oob_progress(stage: str, msg: str) -> None: """One-line progress update for ``--setup-oob`` / ``--usb-local`` steps. - Goes to stderr in dim cyan so the operator can see *where* the launcher - is in its sequence of steps without these lines competing with the - success banner (stdout) or error prints (red). - - This stays a print so progress remains visible when the console log level - is set to warning. + Logged at INFO as ``[] ``; printed to stderr instead when + ``ISAACCAPTURE_LOGGING=off``, where no handler of ours would show it. """ - print(f"\033[36m[{stage}]\033[0m {msg}", file=sys.stderr, flush=True) + if logging_enabled(): + log.info("[%s] %s", stage, msg) + else: + print(f"\033[36m[{stage}]\033[0m {msg}", file=sys.stderr, flush=True) def print_oob_hub_startup_banner( diff --git a/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py b/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py index f5267b1f88..9701b642b5 100644 --- a/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py +++ b/src/python/isaaccapture/cloudxr/oob_teleop_lifecycle.py @@ -23,6 +23,7 @@ from .oob_teleop_env import ( USB_TURN_CREDENTIAL, USB_TURN_USER, + oob_progress, redact_control_token, usb_backend_port, ) @@ -187,6 +188,9 @@ def __init__( self._last_health: str | None = None self._readiness_stage: str | None = None self._readiness_since: float | None = None + # Whether the operator was told the headset streams now, and ever did. + self._streaming = False + self._streamed = False self._prerequisite_signature: tuple | None = None self.snapshot: dict = {} @@ -260,6 +264,23 @@ async def _publish(self, health: str, state: str, reason: str, **flags) -> None: self._restart_episode() self._last_health = health reason = re.sub(r"https?://\S+", "", redact_control_token(reason))[:300] + # The operator sees the headset connect once ACTIVE confirms the stream, and + # disconnect only on an explicit streaming=False: stale metrics drop health + # below ACTIVE while the browser still streams, and a publication that omits + # the flag does not know. The explicit False matches the snapshot default but + # is what announces the loss. Other transitions stay on this module's logger. + stage = "usb-local" if self.usb_local else "setup-oob" + if health == "active" and not self._streaming: + oob_progress( + stage, + "headset reconnected; streaming resumed" + if self._streamed + else "headset connected; streaming", + ) + self._streaming = self._streamed = True + elif self._streaming and flags.get("streaming") is False: + oob_progress(stage, f"headset disconnected: {reason}") + self._streaming = False snapshot = { "schemaVersion": 1, "generation": self.generation, @@ -458,8 +479,21 @@ async def _enter_transport_recovery( adb_ready: bool, network_present: bool, adb_diagnostic: str = "", + transport_verified: bool = False, ) -> None: - """Preserve browser intent and publish one USB-loss transition.""" + """Preserve browser intent and publish one USB-loss transition. + + *transport_verified* means this iteration's USB transport checks passed, so + the disruption came from observing the browser, not from the stream's path. + """ + # USB-local streams over this transport; once it is verified, only the + # browser's own report shows the stream lost. Over Wi-Fi only a headset that + # ADB reports without a network has certainly lost its stream. Read the + # report before the browser client is forgotten below. + stream_lost = (adb_ready and not network_present) or ( + self.usb_local + and not (transport_verified and await self._browser_reports_streaming()) + ) await self._remember_transport_loss() if self._transport_lost: self._invalidate_completed_repair( @@ -472,6 +506,17 @@ async def _enter_transport_recovery( adbReady=adb_ready, networkPresent=network_present, adbDiagnostic=redact_control_token(adb_diagnostic), + **({"streaming": False} if stream_lost else {}), + ) + + async def _browser_reports_streaming(self) -> bool: + """Whether this lifecycle's browser still reports a running stream to the hub.""" + state = await self.hub.get_snapshot() + return any( + headset["clientId"] == self.browser_client + and headset.get("streaming") + and headset.get("streamPhase") == "streaming" + for headset in state["headsets"] ) def _transport_retry_interval(self) -> float: @@ -989,6 +1034,7 @@ async def _automate(self) -> None: networkPresent=True, reverseRulesVerified=self.usb_local, turnPrerequisitesReady=self.usb_local, + streaming=False, ) async def on_client_loaded() -> None: @@ -1078,6 +1124,7 @@ async def on_client_loaded() -> None: networkPresent=True, reverseRulesVerified=self.usb_local, turnPrerequisitesReady=self.usb_local, + streaming=False, ) def on_dispatched() -> Awaitable[None]: @@ -1096,6 +1143,7 @@ async def publish() -> None: reverseRulesVerified=self.usb_local, turnPrerequisitesReady=self.usb_local, connectDispatched=True, + streaming=False, ) return publish() @@ -1121,6 +1169,7 @@ async def publish() -> None: "Selected headset must be Awake before CONNECT; wear or unlock it", adbReady=True, networkPresent=True, + streaming=False, ) return None except adb.OobAdbError: @@ -1290,7 +1339,8 @@ async def _recover_existing_browser(self) -> None: turnPrerequisitesReady=True, browserRegistered=True, healthProbeAcknowledged=True, - streaming=False, + # Unconfirmed is not lost: announce only if the browser stopped streaming. + **({} if report.get("streaming") else {"streaming": False}), clientMetricsFresh=False, streamPhase=phase, terminalEventId=terminal_event, @@ -1859,10 +1909,12 @@ async def run(self) -> None: ) await self.sleep(self.config.interval_sec) continue + transport_verified = False try: # In USB-local mode, ACTIVE/VERIFYING_BROWSER revalidate USB first. if self.browser_ready or self.connect_dispatched: await self._check_usb_prerequisites() + transport_verified = True # Checked after prerequisites (a genuine USB/ADB/network failure keeps # priority) but before per-state dispatch below, so a CDP-observed # terminal error is handled the same way regardless of which state @@ -1943,6 +1995,7 @@ async def run(self) -> None: "REBUILDING_USB", adb_ready=exc.adb_ready, network_present=exc.network_present, + transport_verified=transport_verified, ) continue except asyncio.CancelledError: diff --git a/src/python/isaaccapture/logging_config/_console.py b/src/python/isaaccapture/logging_config/_console.py index 84353c29e7..ba32048922 100644 --- a/src/python/isaaccapture/logging_config/_console.py +++ b/src/python/isaaccapture/logging_config/_console.py @@ -51,7 +51,10 @@ def filter(self, record: logging.LogRecord) -> bool: _SGR_ESCAPE = re.compile(r"(?:\x1b\[[0-9;]*m)+") # Exact logger name to ANSI emphasis. -_logger_colors: dict[str, str] = {} +_logger_colors: dict[str, str] = { + # cloudxr.oob_teleop_env contains [setup-oob] / [usb-local] progress. + "isaaccapture.cloudxr.oob_teleop_env": "\033[36m", +} #: Color only WARNING and ERROR+ so severity remains distinctive. diff --git a/tests/python/core/cloudxr/test_oob_teleop_env.py b/tests/python/core/cloudxr/test_oob_teleop_env.py index 7832703840..729f91f233 100644 --- a/tests/python/core/cloudxr/test_oob_teleop_env.py +++ b/tests/python/core/cloudxr/test_oob_teleop_env.py @@ -5,6 +5,7 @@ from __future__ import annotations +import logging from urllib.parse import parse_qs, urlparse import cloudxr_py_test_ns.oob_teleop_env as oob_teleop_env_under_test @@ -611,6 +612,31 @@ def test_startup_banner_hides_control_and_turn_credentials( assert "turnCredential=" in out +def test_oob_progress_logs_at_info( + monkeypatch: pytest.MonkeyPatch, capsys, caplog +) -> None: + """With logging on, progress is an INFO record on the module logger, not a print.""" + monkeypatch.delenv("ISAACCAPTURE_LOGGING", raising=False) + with caplog.at_level(logging.INFO, logger="isaaccapture.cloudxr.oob_teleop_env"): + oob_teleop_env_under_test.oob_progress("setup-oob", "teleop tab found") + [record] = caplog.records + assert record.name == "isaaccapture.cloudxr.oob_teleop_env" + assert record.levelno == logging.INFO + assert record.getMessage() == "[setup-oob] teleop tab found" + assert capsys.readouterr().err == "" + + +def test_oob_progress_prints_when_logging_is_off( + monkeypatch: pytest.MonkeyPatch, capsys, caplog +) -> None: + """``ISAACCAPTURE_LOGGING=off`` keeps the cyan stderr line and logs nothing.""" + monkeypatch.setenv("ISAACCAPTURE_LOGGING", "off") + with caplog.at_level(logging.INFO, logger="isaaccapture.cloudxr.oob_teleop_env"): + oob_teleop_env_under_test.oob_progress("setup-oob", "teleop tab found") + assert capsys.readouterr().err == "\033[36m[setup-oob]\033[0m teleop tab found\n" + assert not caplog.records + + # Host preflight (H5) ------------------------------------------------------ diff --git a/tests/python/core/cloudxr/test_oob_teleop_lifecycle.py b/tests/python/core/cloudxr/test_oob_teleop_lifecycle.py index 1b9becd282..4e50ac6ba3 100644 --- a/tests/python/core/cloudxr/test_oob_teleop_lifecycle.py +++ b/tests/python/core/cloudxr/test_oob_teleop_lifecycle.py @@ -773,6 +773,316 @@ async def test_ignored_serial_display_is_bounded_and_safe(monkeypatch): assert "\n" not in status["selectedSerial"] +async def test_stale_metrics_do_not_announce_a_headset_disconnect(monkeypatch): + announced = [] + monkeypatch.setattr( + "isaaccapture.cloudxr.oob_teleop_lifecycle.oob_progress", + lambda _stage, message: announced.append(message), + ) + lifecycle = OobLifecycle( + hub=FakeHub(), + resolved_port=48322, + usb_local=False, + host_client=False, + config=RecoveryConfig(), + ) + confirmed = "Stream and fresh metrics confirmed" + await lifecycle._publish("active", "ACTIVE", confirmed, streaming=True) + # Metrics went stale while the browser still reports streaming. + stale = "Browser ready; waiting for stream or fresh metrics" + await lifecycle._publish("browser_ready", "ACTIVE", stale, streaming=True) + await lifecycle._publish("active", "ACTIVE", confirmed, streaming=True) + # The stream itself stopped. + retrying = "Browser is retrying the stream" + await lifecycle._publish("browser_ready", "ACTIVE", retrying, streaming=False) + await lifecycle._publish("active", "ACTIVE", confirmed, streaming=True) + + assert announced == [ + "headset connected; streaming", + f"headset disconnected: {retrying}", + "headset reconnected; streaming resumed", + ] + + +@pytest.mark.parametrize( + ("usb_local", "adb_ready", "network_present", "lost"), + [ + (False, False, False, False), # Wi-Fi carries the stream; ADB loss is moot. + (False, True, False, True), # A headset with no network has no stream. + (True, False, False, True), # USB-local streams over the lost transport. + ], +) +async def test_transport_loss_announces_a_disconnect_only_when_it_carries_the_stream( + monkeypatch, usb_local, adb_ready, network_present, lost +): + announced = [] + monkeypatch.setattr( + "isaaccapture.cloudxr.oob_teleop_lifecycle.oob_progress", + lambda _stage, message: announced.append(message), + ) + lifecycle = OobLifecycle( + hub=FakeHub(), + resolved_port=48322, + usb_local=usb_local, + host_client=False, + config=RecoveryConfig(), + ) + await lifecycle._publish("active", "ACTIVE", "confirmed", streaming=True) + await lifecycle._enter_transport_recovery( + "transport lost", + "WAITING_FOR_ADB", + adb_ready=adb_ready, + network_present=network_present, + ) + + assert announced[1:] == (["headset disconnected: transport lost"] if lost else []) + + +async def test_reopening_the_browser_announces_the_wifi_stream_it_ends(monkeypatch): + announced = [] + monkeypatch.setattr( + "isaaccapture.cloudxr.oob_teleop_lifecycle.oob_progress", + lambda _stage, message: announced.append(message), + ) + lifecycle = OobLifecycle( + hub=FakeHub(), + resolved_port=48322, + usb_local=False, + host_client=False, + config=RecoveryConfig(), + ) + await lifecycle._publish("active", "ACTIVE", "confirmed", streaming=True) + await lifecycle._enter_transport_recovery( + "Selected headset offline; reconnect the USB cable", + "WAITING_FOR_ADB", + adb_ready=False, + network_present=False, + ) + + async def connect(**_kwargs): + return asyncio.create_task(asyncio.Event().wait()) + + async def noop(): + return None + + with ( + patch( + "isaaccapture.cloudxr.oob_teleop_lifecycle.adb.run_oob_connect", + side_effect=connect, + ), + patch.object(lifecycle, "_verify_browser", new=noop), + ): + await lifecycle._automate() + await lifecycle._stop_monitor() + + assert announced == [ + "headset connected; streaming", + "headset disconnected: Opening browser and dispatching CONNECT", + ] + + +async def test_same_tab_connect_announces_the_stream_it_replaces(monkeypatch): + announced = [] + monkeypatch.setattr( + "isaaccapture.cloudxr.oob_teleop_lifecycle.oob_progress", + lambda _stage, message: announced.append(message), + ) + lifecycle = OobLifecycle( + hub=FakeHub(), + resolved_port=48322, + usb_local=False, + host_client=False, + config=RecoveryConfig(), + ) + await lifecycle._publish("active", "ACTIVE", "confirmed", streaming=True) + + async def attach(*, click_connect, on_dispatched=None, **_kwargs): + dispatched = on_dispatched() + if asyncio.iscoroutine(dispatched): + await dispatched + return asyncio.create_task(asyncio.Event().wait()) + + lifecycle.client_loaded = True + with patch( + "isaaccapture.cloudxr.oob_teleop_lifecycle.adb.attach_existing_oob_tab", + side_effect=attach, + ): + assert await lifecycle._same_tab_connect("terminal:1") + await lifecycle._stop_monitor() + + assert announced == [ + "headset connected; streaming", + "headset disconnected: Existing browser CONNECT dispatched; " + "waiting for fresh stream evidence", + ] + + +async def test_same_tab_connect_blocked_by_sleep_announces_the_stream_it_replaces( + monkeypatch, +): + announced = [] + monkeypatch.setattr( + "isaaccapture.cloudxr.oob_teleop_lifecycle.oob_progress", + lambda _stage, message: announced.append(message), + ) + lifecycle = OobLifecycle( + hub=FakeHub(), + resolved_port=48322, + usb_local=False, + host_client=False, + config=RecoveryConfig(), + ) + await lifecycle._publish("active", "ACTIVE", "confirmed", streaming=True) + + with patch( + "isaaccapture.cloudxr.oob_teleop_lifecycle.adb.attach_existing_oob_tab", + side_effect=adb.HeadsetNotAwakeError(adb.HEADSET_NOT_AWAKE_DIAGNOSTIC), + ): + assert await lifecycle._same_tab_connect("terminal:1") is None + + assert announced == [ + "headset connected; streaming", + "headset disconnected: Selected headset must be Awake before CONNECT; " + "wear or unlock it", + ] + + +async def test_cdp_monitor_loss_does_not_announce_a_usb_stream_that_kept_running( + monkeypatch, +): + announced = [] + monkeypatch.setattr( + "isaaccapture.cloudxr.oob_teleop_lifecycle.oob_progress", + lambda _stage, message: announced.append(message), + ) + client = { + "clientId": "surviving-page", + "streaming": True, + "lastMetricsAt": time.time() * 1000, + "streamPhase": "streaming", + "terminalEventId": None, + } + hub = FakeHub() + hub.get_snapshot = lambda: asyncio.sleep(0, result={"headsets": [client]}) + # The first probe after the repair predates the browser's next metrics report. + reports = iter([client]) + hub.probe_browser = lambda *_args, **_kwargs: asyncio.sleep( + 0, result=next(reports, {**client, "lastMetricsAt": time.time() * 1000}) + ) + sleeps = [] + + async def sleep(seconds): + sleeps.append(seconds) + if len(sleeps) == 2: + raise asyncio.CancelledError + + lifecycle = OobLifecycle( + hub=hub, + resolved_port=48322, + usb_local=True, + host_client=True, + turn_port=3478, + config=RecoveryConfig(), + sleep=sleep, + host_listener_probe=lambda _port: True, + ) + ready = AdbDevices((("original", "device"),)) + lifecycle.selected = "original" + lifecycle._ready_count = 2 + lifecycle._last_observation = (ready.devices, ready.diagnostic) + lifecycle.last_network_state = HeadsetNetworkState.NETWORK_PRESENT + lifecycle.browser_ready = True + lifecycle.browser_client = "surviving-page" + lifecycle.last_stream_at = time.time() + await lifecycle._publish("active", "ACTIVE", "confirmed", streaming=True) + # The CDP monitor cannot reattach once, while the hub still reports streaming. + attaches = iter([False]) + + async def attach_monitor(): + return next(attaches, True) + + async def no_coturn_restart(): + return False + + async def command(_command, *, timeout): + return subprocess.CompletedProcess([], 0, "", "") + + rules = "\n".join( + f"original tcp:{port} tcp:{port}" for port in (48322, 49100, 3478) + ) + with ( + patch( + "isaaccapture.cloudxr.oob_teleop_lifecycle.adb.enumerate_adb_devices", + return_value=ready, + ), + patch( + "isaaccapture.cloudxr.oob_teleop_lifecycle.adb.probe_headset_network", + return_value=HeadsetNetworkProbe(HeadsetNetworkState.NETWORK_PRESENT), + ), + patch( + "isaaccapture.cloudxr.oob_teleop_lifecycle.adb.probe_adb_reverse_rules", + return_value=AdbReverseProbe(True, ()), + ), + patch( + "isaaccapture.cloudxr.oob_teleop_lifecycle.adb._run_adb", + return_value=rules, + ), + patch("isaaccapture.cloudxr.oob_teleop_lifecycle.adb.assert_headset_awake"), + patch.object(lifecycle, "_ensure_coturn", new=no_coturn_restart), + patch.object(lifecycle, "_run_adb_command", new=command), + patch.object(lifecycle, "_attach_existing_monitor", new=attach_monitor), + patch.object(lifecycle, "_automate") as automate, + ): + with pytest.raises(asyncio.CancelledError): + await lifecycle.run() + + automate.assert_not_called() + assert lifecycle.snapshot["health"] == "active" + assert announced == ["headset connected; streaming"] + + +@pytest.mark.parametrize( + ("reported", "lost"), + [ + ({"streaming": True, "streamPhase": "streaming"}, False), + ({"streaming": False, "streamPhase": "retrying"}, True), + ({"streaming": False, "streamPhase": "terminal"}, True), + (None, True), # The browser's control client is gone. + ], +) +async def test_verified_usb_transport_defers_to_the_browser_stream_report( + monkeypatch, reported, lost +): + announced = [] + monkeypatch.setattr( + "isaaccapture.cloudxr.oob_teleop_lifecycle.oob_progress", + lambda _stage, message: announced.append(message), + ) + headsets = [] if reported is None else [{"clientId": "page", **reported}] + hub = FakeHub() + hub.get_snapshot = lambda: asyncio.sleep(0, result={"headsets": headsets}) + lifecycle = OobLifecycle( + hub=hub, + resolved_port=48322, + usb_local=True, + host_client=False, + config=RecoveryConfig(), + ) + lifecycle.browser_client = "page" + await lifecycle._publish("active", "ACTIVE", "confirmed", streaming=True) + await lifecycle._enter_transport_recovery( + "observation failed", + "REBUILDING_USB", + adb_ready=True, + network_present=True, + transport_verified=True, + ) + + assert announced[1:] == ( + ["headset disconnected: observation failed"] if lost else [] + ) + + async def test_cleanup_commands_stay_on_selected_serial_when_it_is_absent(monkeypatch): monkeypatch.setenv("ANDROID_SERIAL", "original") hub = FakeHub() diff --git a/tests/python/core/logging_config/test_console.py b/tests/python/core/logging_config/test_console.py index 9f85fc3969..dbc5d94e4f 100644 --- a/tests/python/core/logging_config/test_console.py +++ b/tests/python/core/logging_config/test_console.py @@ -25,11 +25,13 @@ def restore_console(): """This process shares the one handler under test with every other test.""" handler = _console.ensure_handler() level = handler.level + colors = dict(_console._logger_colors) published = os.environ.get("ISAACCAPTURE_LOG_LEVEL") yield logging_config.set_console_filter(None) handler.setLevel(level) _console._logger_colors.clear() + _console._logger_colors.update(colors) if published is None: os.environ.pop("ISAACCAPTURE_LOG_LEVEL", None) else: @@ -155,6 +157,10 @@ def test_a_logger_colour_on_an_uncoloured_level_resets_after_the_name(self): logging_config.set_logger_colors({"isaaccapture.core.Probe": "\033[36m"}) assert f"[\033[36misaaccapture.core.Probe{_RESET}]" in render(_Tty()) + def test_oob_progress_logger_is_cyan_by_default(self): + name = "isaaccapture.cloudxr.oob_teleop_env" + assert f"[\033[36m{name}{_RESET}]" in render(_Tty(), name=name) + def test_the_record_is_handed_back_unmodified(self): # The file handler shares the record and must stay escape-free. logging_config.set_logger_colors({"isaaccapture.core.Probe": "\033[36m"})