"""ShieldConnectionManager — entry-scoped owner of the ADB connection (Phase 1). The ADB socket, its watchdog and all ADB-derived state live here instead of on the media-player entity (plan blockers B1/B2/B5): - one persistent socket; a watchdog task reconnects with exponential backoff (5 s doubling to a 300 s cap, ±25 % jitter) when it drops, and pulls the heartbeat data (wake + media sessions + running apps) while connected; - ADB-derived values live in a :class:`StateStore` with per-field timestamps and TTL expiry — stale values read as expired (None/empty) instead of clinging on; - entities are thin callback-views: they ``register_listener()`` and write HA state when the store changes (the androidtv core ``entity.py`` pattern); - ADB is optional: when no poller factory is provided the manager is *passive* and the entry still sets up — a failed ADB connect is a background retry, never a ``ConfigEntryNotReady``. This module deliberately stays Home-Assistant-free at import time so it is unit-testable with plain pytest; every hass capability is injected (the entry builds the concrete poller factory). Task creation additionally works without hass (plain ``asyncio.create_task``) for the unit tests. """ import asyncio import contextlib import dataclasses import logging import random import time from collections.abc import Callable from typing import Any from .adb_session import AdbMediaSessionPoller from .const import ( ADB_POLL_INTERVAL, ADB_RECONNECT_BACKOFF_CAP, ADB_RECONNECT_BACKOFF_START, ADB_RECONNECT_JITTER, APP_ICON_FALLBACKS, APP_ICON_GENERIC, APP_LABEL_EXIT_MARKER, APP_LABEL_GREP_SHAPES, APP_LABEL_PROBE_LIMIT, APP_LABEL_TTL, APP_LIST_TTL, APP_SWEEP_CONCURRENCY, APP_SWEEP_SPACING, CONF_ADB_ENABLED, CONF_APP_NAME_MAP, CONF_LOGCAT_EVENTS, DEFAULT_APP_DIRECTORY, LOGCAT_SUPPRESS_WINDOW, PROTOCOL_AUTH_REJECTED, PROTOCOL_CONNECT_TIMEOUT, PROTOCOL_CONNECTED, PROTOCOL_DISCONNECTED, PROTOCOL_NEVER_CONNECTED, PROTOCOL_OFF, STATE_STORE_TTL, SYSTEM_MEDIA_PACKAGES, ) from .ha_adb_common import ( INTENT_SCAN_ADB_DISABLED, INTENT_SCAN_ADB_UNAVAILABLE, INTENT_SCAN_CACHED, INTENT_SCAN_FRESH, INTENT_SCAN_FRESH_EMPTY, INTENT_SCAN_INVALID_PACKAGE, INTENT_SCAN_UNREADABLE, IntentScanCache, IntentScanResult, MediaSessionInfo, foreground_from_logcat, has_activity_resolver_table, icon_for_package, installed_packages, intent_scan_command, intent_scan_fallback_command, is_launchable, parse_app_label, parse_audio_playback, parse_intent_report, resolve_package_name, ) _LOGGER = logging.getLogger(__name__) # StateStore field names FIELD_AWAKE = "awake" FIELD_SESSIONS = "sessions" FIELD_AUDIO_PLAYBACK = "audio_playback" FIELD_RUNNING_APPS = "running_apps" # Protocol transport (androidtvremote2) push fields FIELD_IS_ON = "is_on" FIELD_IS_AVAILABLE = "is_available" FIELD_CURRENT_APP = "current_app" FIELD_VOLUME_INFO = "volume_info" # ADB fallback foreground identity (P3-2): stamped only when the protocol # push is silent; consumers prefer FIELD_CURRENT_APP over this. FIELD_FOREGROUND = "foreground" # ADB-derived fields (cleared when the ADB connection goes down — they are # regenerated by the watchdog); protocol pushes (IS_ON/IS_AVAILABLE/ # CURRENT_APP/VOLUME_INFO) are device-reported truth and expire by TTL only, # so an ADB outage never clobbers them. ADB_FIELDS = ( FIELD_AWAKE, FIELD_SESSIONS, FIELD_AUDIO_PLAYBACK, FIELD_RUNNING_APPS, FIELD_FOREGROUND, ) PROTOCOL_FIELDS = ( FIELD_IS_ON, FIELD_IS_AVAILABLE, FIELD_CURRENT_APP, FIELD_VOLUME_INFO, ) ALL_FIELDS = ADB_FIELDS + PROTOCOL_FIELDS def volume_tuple(info) -> tuple | None: """Normalize the protocol's volume push to ``(level, max, muted)``. The library delivers a VolumeInfo TypedDict (``level``/``max``/``muted``). Readout only — there is no protocol-native VOLUME_SET (C-VOL), so nothing here ever sends one. """ if info is None: return None try: return ( float(info.get("level", 0.0)), float(info.get("max", 1.0)), bool(info.get("muted", False)), ) except (AttributeError, TypeError, ValueError): return None def _protocol_error_kind(err: BaseException) -> str: """Classify a protocol exception: "auth" or "network" (t15). Classified by NAME so the manager keeps working around a test double and stays importable where the optional androidtvremote2 is not installed. Only the library's ``InvalidAuth`` is terminal — it means the device no longer accepts our certificate, so retrying can never succeed; everything else (CannotConnect, ConnectionClosed, TimeoutError, an OSError from the TLS handshake) is a transport failure worth another attempt. """ return "auth" if type(err).__name__ == "InvalidAuth" else "network" class StateStore: """Timed per-field state cache shared by all entities of one entry. Reading an expired (or missing) field returns the caller-supplied default, so a stale "playing" session can never outlive the TTL (plan blocker B5). Values are written only via :meth:`set`, which re-stamps the field's freshness timestamp. """ def __init__(self, ttl: float = STATE_STORE_TTL, clock: Callable[[], float] = time.monotonic): self._ttl = ttl self._clock = clock self._values: dict[str, tuple[float, Any]] = {} def set(self, name: str, value: Any) -> None: """Store a value, re-stamping its freshness timestamp.""" self._values[name] = (self._clock(), value) def get(self, name: str, default: Any = None) -> Any: """Return the field's value, or ``default`` when it is missing/expired.""" stamped = self._values.get(name) if stamped is None: return default at, value = stamped if self._clock() - at > self._ttl: return default return value def clear(self, *names: str) -> None: """Drop fields immediately so they read as expired on the next get.""" for name in names: self._values.pop(name, None) @property def ttl(self) -> float: return self._ttl def values_stamped(self) -> dict[str, tuple[float, Any]]: """Public stamped-values snapshot (F6): ``{field: (stamp, value)}``.""" return dict(self._values) class ShieldConnectionManager: """Owner of one entry's persistent connections (both transports) + state. Transport A — ADB (adb_shell): the watchdog keeps a persistent socket, pulls heartbeats and reconnects with its own backoff ladder. Transport B — remote protocol (androidtvremote2, Phase 2): the library owns reconnection (``keep_reconnecting`` with internal backoff), pushes ``is_on``/``is_available``/``current_app``/``volume_info`` into the state store, and surfaces invalid certs (device hard reset) via a reauth hook. :param poller_factory: builds the ADB poller when the watchdog starts; skipped entirely for entries without ADB (passive manager) or tests that inject a ready-to-use ``poller`` directly. :param protocol_factory: builds the AndroidTVRemote client; instead pass ``protocol_settings`` (host/client_name/certfile/keyfile) and the manager builds it, or ``protocol`` directly (tests). :param invalid_auth_callback: called when the device rejects our certificates (hard reset) — the entry-owned reauth hook (t9). """ def __init__( self, hass=None, entry=None, poller_factory: Callable[[], AdbMediaSessionPoller] | None = None, poller: AdbMediaSessionPoller | None = None, store: StateStore | None = None, protocol_factory: Callable[[], Any] | None = None, protocol: Any = None, protocol_settings: dict | None = None, invalid_auth_callback: Callable[[], None] | None = None, clock: Callable[[], float] = time.monotonic, ): self.hass = hass self.entry = entry self.store = ( store if store is not None else StateStore(ttl=STATE_STORE_TTL, clock=clock) ) self._poller_factory = poller_factory self._poller = poller # Remote-protocol transport (androidtvremote2): entry passes either a # settings dict (manager itself builds the real client) or a factory / # ready-made client (tests, or the config flow when it pre-pairs). self._protocol_factory = protocol_factory self._protocol = protocol self._protocol_settings = protocol_settings self._invalid_auth_callback = invalid_auth_callback self._protocol_loop: asyncio.AbstractEventLoop | None = None self._listeners: list[Callable[[], None]] = [] # P3-2d last-known-state cache: package → (session info, stamp). # Serves the foreground app's last known playback state when it # publishes no MediaSession (TiviMate, R2). TTL-capped: 'playing' # is NEVER resurrected past STATE_STORE_TTL. self._last_known: dict[str, tuple[MediaSessionInfo, float]] = {} # P5-1: device manifest labels — package → (label, stamp). Long-TTL # (APP_LABEL_TTL) cache resolved via `pm dump` on demand, one # package per watchdog cycle so a burst never blocks the watchdog. # An entry means "a fetch RAN" (t57 UNIT 0): a tuple whose label is # "" is a definitive negative, and a fetch that could not run is not # in here at all — so the flag t55 carried to say so was dead weight # (its only writer always passed True) and is gone. self._app_labels: dict[str, tuple[str, float]] = {} self._app_label_requests: set[str] = set() # t55: packages whose label fetch is queued or in flight — a render # during the fetch must not enqueue the same package twice. self._app_label_inflight: set[str] = set() # t66: the DEVICE-level "this firmware exposes no labels" memo. Set after # APP_LABEL_PROBE_LIMIT consecutive probes that RAN and found nothing (one # label-less package proves nothing — see _dump_label), and it expires # with APP_LABEL_TTL so a changed device is re-checked within a day. self._labels_absent_at: float | None = None self._label_negative_streak: int = 0 # t66: packages whose probe FAILED (transport delivered nothing) and which # therefore still owe a retry even while the memo above is set — t57's # "a failed fetch is never cached as an answer" survives the memo. self._app_label_failed: set[str] = set() # t57: the discovered installed-apps table — (rows, stamp), or None when # no sweep has produced a fresh answer. Only a sweep that RAN writes # here (t62 F1): a failed or unavailable sweep records itself as UNKNOWN # instead, because stamping an empty answer would serve "0 apps" — and # block every retry — for the whole APP_LIST_TTL. Refreshed on demand; # the watchdog cycle never rebuilds it. self._installed_apps: tuple[list[dict], float] | None = None # t62 F1: True when the last sweep could not deliver an answer (no # poller, ADB down, a shell that raised or answered nothing) — so an # empty installed_apps() is "unknown", not "this device has no apps". self._apps_unavailable = False # ...and the clock stamp of an ATTEMPTED failure (shells really went # out). A retry then costs two listing shells, so it is spaced by # APP_SWEEP_RETRY_SECONDS instead of firing on every state poll; an # unavailable transport costs nothing and retries immediately. self._apps_failed_at: float | None = None #: t57: the in-flight sweep, so concurrent callers share one sweep #: instead of stacking a second burst on the same connection. self._apps_sweep: asyncio.Task | None = None # P5-3 (C-STREAM): the dedicated logcat line-stream consumer. # NEVER shares the poller's ADB socket — its own device object and # connection; CONF_LOGCAT_EVENTS gates it (default off) and the # heartbeat cadence remains the floor when the stream is off/disabled. self._logcat_task: asyncio.Task | None = None self._logcat_disabled = False self._logcat_failures: list[float] = [] self._logcat_last_pkg: str | None = None # G10 (t30): protocol device identity captured once per session — # (name, manufacturer, model, sw) for the registry's device card. self.protocol_device_data: tuple[str, str, str, str] | None = None self._logcat_last_stamp: float = -1e9 # Injectable ladder sleep (t26: deterministic tests / future alt-floors) self._logcat_ladder_sleep = asyncio.sleep # Injectable clock keeps the TTL expiry tests deterministic. self._clock = clock # t16: intent discovery. ONE package per explicit request — never a sweep # over the installed apps (that is ~2241 dump lines each) — cached with # INTENT_SCAN_TTL. The cache is the module's, so the "a scan that RAN and # found nothing is an answer, a scan whose transport failed is not" rule # lives in one place (IntentScanCache.put is its only writer). self._intents = IntentScanCache(clock=self._clock) #: t16: scans in flight, keyed by package, so a second request for the #: same app awaits the dump already being read instead of paying for a #: second one. self._intent_scans: dict[str, asyncio.Task] = {} self._tasks: list[asyncio.Task] = [] self._stopping = False self._failures = 0 self._down_since = False # listeners already told the connection is down # t15: the protocol transport's REAL state. keep_reconnecting() only MAINTAINS # a session (the library's loop is `while self._remote_message_protocol:`, # None until something connects first), so before this unit the transport was # never connected: no push arrived and every send raised ConnectionClosed # internally. This state is what the entities publish so the difference # between "never switched on", "connected" and "cert refused" is readable. self._protocol_state: str = ( PROTOCOL_NEVER_CONNECTED if self._protocol is not None else PROTOCOL_OFF ) self._protocol_last_error: str | None = None # t20/O5: the connect ATTEMPT's task. Held so a second start recognises a # live attempt instead of orphaning it; it is also tracked in ``_tasks`` so # ``async_stop`` cancels it (t20/O4). self._protocol_connect_task: asyncio.Task | None = None self._protocol_ever_authenticated = False self._protocol_failures = 0 # ── One-shot connection test (pairing UX) ── @staticmethod async def test_connect(host: str, port: int, key_path: str) -> bool: """Single-shot ADB connect used by the config-flow pairing step. Throws away the poller immediately — the config flow runs before any entry exists, so it must never touch an entry's persistent socket. """ poller = AdbMediaSessionPoller(host, port, key_path) try: return await poller.connect() finally: await poller.disconnect() # ── Remote-protocol transport (androidtvremote2, Phase 2) ── @staticmethod def build_protocol_client( host: str, client_name: str, certfile: str, keyfile: str, api_port: int = 6466, pair_port: int = 6467, loop=None, ): """Build a real androidtvremote2 client (cert/PIN flow hook, t9). Kept as a manager-level factory so the config flow never hand-rolls the constructor; the import stays local so this module remains importable where androidtvremote2 is not installed. """ try: import androidtvremote2 except ImportError: _LOGGER.warning( "androidtvremote2 is not installed — the remote-protocol " "transport is disabled and the entry degrades to ADB-only." ) return None return androidtvremote2.AndroidTVRemote( client_name=client_name, certfile=certfile, keyfile=keyfile, host=host, api_port=api_port, pair_port=pair_port, loop=loop, ) def _build_protocol(self): """Default factory from the entry settings (host + cert/key paths).""" settings = self._protocol_settings or {} return self.build_protocol_client( host=settings["host"], client_name=settings.get("client_name", "Home Assistant"), certfile=settings["certfile"], keyfile=settings["keyfile"], ) async def _start_protocol(self) -> None: """Spin up the protocol transport alongside the ADB watchdog. One owner per transport: the library's own ``keep_reconnecting()`` handles reconnection with its internal backoff, so no competing watchdog is built here. Push callbacks write straight into the StateStore and ping registered listeners. """ if self._stopping: return # a stopped manager must not restart itself (t20/O4/O5) if self._protocol is None: if self._protocol_factory is None: if self._protocol_settings is None: return # protocol transport disabled for this entry self._protocol_factory = self._build_protocol self._protocol = self._protocol_factory() client = self._protocol if client is None: return # t20/O5: A SECOND START IS DETERMINISTIC. A live attempt is reused and an # established session is left alone, so nothing is ever orphaned — and the # guard sits BEFORE the push callbacks below, so a second call cannot # register them twice either (which would double every device push). attempt = self._protocol_connect_task if attempt is not None and not attempt.done(): _LOGGER.debug( "Shield remote-protocol connect attempt already in flight — reusing it" ) return if self._protocol_state in (PROTOCOL_CONNECTED, PROTOCOL_DISCONNECTED): # A session exists (or existed and is being maintained): the LIBRARY's own # keep-reconnecting loop owns it from here, and calling keep_reconnecting # a second time would put a second loop over the same session — so a # second start never builds one. _LOGGER.debug( "Shield remote-protocol session already handed to the library's " "keep-reconnecting loop (state: %s) — no second attempt", self._protocol_state, ) return # Real client: make sure local certificate material exists before it # connects (fails soft — pair flows can regenerate). generate_cert = getattr(client, "async_generate_cert_if_missing", None) if generate_cert is not None: with contextlib.suppress(Exception): await generate_cert() self._protocol_loop = asyncio.get_running_loop() client.add_is_on_updated_callback(self._on_is_on) client.add_is_available_updated_callback(self._on_is_available) client.add_current_app_updated_callback(self._on_current_app) client.add_volume_info_updated_callback(self._on_volume_info) # t15: CONNECT FIRST, then hand maintenance to the library. # # Core androidtv_remote does `await api.async_connect()` (5 s timeout, # InvalidAuth -> ConfigEntryAuthFailed, CannotConnect -> ConfigEntryNotReady) # and ONLY THEN `api.keep_reconnecting(reauth_needed)`. The order is the # whole point: the library's reconnect loop is # # while self._remote_message_protocol: # exc = await self._remote_message_protocol.on_con_lost # # and `_remote_message_protocol` is None until a first async_connect() # succeeds — so calling keep_reconnecting() alone starts a task that checks # its condition once, sees None and RETURNS. That is what shipped (this # method called keep_reconnecting and nothing in the tree ever called # async_connect): no session, not one push, and every send raised # ConnectionClosed internally. The attempt below is a background task, never # setup work: an unreachable or unpaired Shield must not fail the entry — # ADB is the primary transport and stays the answer (t40 ladder). self._protocol_connect_task = self._create_task( self._protocol_connect_loop(client), "shield_protocol_connect" ) # t20/O4: TRACKED. Cancelling only the tasks the manager holds is what makes # an unload complete: an untracked connect attempt kept retrying after its # entry was gone, and since every options save reloads the entry those loops # could accumulate across saves. self._tasks.append(self._protocol_connect_task) async def _protocol_connect_loop(self, client) -> None: """Open the protocol session, then let the library keep it open (t15). A NETWORK failure is retried with the ADB watchdog's backoff (nothing to lose: the next attempt costs one TCP/TLS handshake). An AUTH rejection stops the loop — retrying a certificate the device no longer accepts can never succeed — and asks the entry to run its reauth/pair flow instead. """ backoff = ADB_RECONNECT_BACKOFF_START try: while not self._stopping: try: await asyncio.wait_for( client.async_connect(), timeout=PROTOCOL_CONNECT_TIMEOUT ) except asyncio.CancelledError: raise except Exception as err: # noqa: BLE001 - classified right here if _protocol_error_kind(err) == "auth": _LOGGER.warning( "Shield remote-protocol auth was rejected (%s) — " "pushing the entry into its reauth/pair flow", err, ) self._on_invalid_auth( reason=f"connect: {type(err).__name__}: {err}" ) return self._note_protocol_state( PROTOCOL_NEVER_CONNECTED, f"connect: {type(err).__name__}: {err}", ) _LOGGER.debug( "Shield remote-protocol connect failed (%s) — retrying in " "%.0f s", err, backoff, ) await asyncio.sleep(self._jitter(backoff)) backoff = min(backoff * 2, ADB_RECONNECT_BACKOFF_CAP) continue # Connected: remember it as an authenticated session and hand the # socket to the library's own reconnect task (one owner per # transport — the stop path closes the client, which tears it down). # t20/O2: the banner says only what THIS attempt proved. A # successful async_connect() cannot know the device stays reachable, # so it must not claim liveliness — that is the transport's own # business from here (its availability push and the library's # keep-reconnecting loop). The state announcement that follows is the # log's last word on the transport, and it is the state the entity # publishes: that is what keeps the two from contradicting. _LOGGER.info( "Shield remote-protocol session opened on %s — what this proves: " "the client accepted a session; what maintains it from here: the " "transport's own availability push (state: %s)", getattr(client, "host", self.entry.title if self.entry else "?"), PROTOCOL_CONNECTED, ) self._protocol_ever_authenticated = True self._note_protocol_state(PROTOCOL_CONNECTED, None) client.keep_reconnecting(self._on_invalid_auth) return except asyncio.CancelledError: raise except Exception: # pragma: no cover - defensive: never kill the entry _LOGGER.exception("Shield remote-protocol connect loop crashed") def _on_is_on(self, on) -> None: self._protocol_push(FIELD_IS_ON, bool(on)) def _on_is_available(self, is_available) -> None: self._protocol_push(FIELD_IS_AVAILABLE, bool(is_available)) def _on_current_app(self, app_package) -> None: self._protocol_push(FIELD_CURRENT_APP, app_package) def _on_volume_info(self, info) -> None: self._protocol_push(FIELD_VOLUME_INFO, volume_tuple(info)) def _on_invalid_auth(self, reason: str | None = None) -> None: """Device hard reset — certificates rejected. The library keeps its own backoff, but retrying can never succeed since certificate auth is gone — the entry-level callback (reauth flow) decides the user-facing response (t9). t15: the state and the error are recorded too — "the certificate was rejected" must be readable on the entity, not only in a log line. ``reason`` carries the specific failure (the connect loop's exception) so the generic wording is only used when the library reports a rejection by itself. """ self._note_protocol_state( PROTOCOL_AUTH_REJECTED, reason or "auth rejected while reconnecting (device hard reset? pair again)", ) _LOGGER.warning( "Shield remote-protocol auth is invalid (device hard reset?) — " "notifying the entry's reauth hook (transport: %s, last error: %s)", self._protocol_state, self._protocol_last_error, ) if self._invalid_auth_callback is not None: try: self._invalid_auth_callback() except Exception: _LOGGER.exception("Shield invalid-auth callback failed") def _protocol_push(self, field: str, value) -> None: """Stamp a push value into the store and ping listeners. Push callbacks may arrive off the event loop (grpc worker threads); route the listener notification back onto the captured loop. t15: a push is the ONLY proof the library offers that a session is authenticated (it exposes no is_paired/is_connected), so receiving one marks the transport authenticated — and an availability push is the transport's own connected/disconnected report. """ self._protocol_ever_authenticated = True if field == FIELD_IS_AVAILABLE: # t20/O1: the transport's own availability report is the OTHER place a # session becomes provable, so it takes the same route — a healthy push # clears a recovered failure exactly as a successful connect does. self._apply_protocol_state( PROTOCOL_CONNECTED if value else PROTOCOL_DISCONNECTED, None ) self.store.set(field, value) loop = self._protocol_loop if loop is not None and loop.is_running(): try: loop.call_soon_threadsafe(self._notify) return except RuntimeError: pass # loop closed during shutdown — skip the notify self._notify() # ── Introspection (thin views for entities) ── @property def available(self) -> bool: """True when a poller exists for this entry — a TRANSPORT fact (t23/O1). Deliberately NOT the permission to talk to the device: the option is (see :meth:`_adb_active`). Diagnostics reads this as "ADB is configured", and the t22 review's scenario — a poller injected into an entry whose own option says ADB is off — must not be able to turn it into traffic. """ return self._poller is not None @property def connected(self) -> bool: return bool(self._poller and self._poller.connected) @property def awake(self) -> bool | None: """Device wake state (exact mWakefulness parse), None = unknown.""" return self.store.get(FIELD_AWAKE) @property def active_sessions(self) -> list[MediaSessionInfo]: """Stored media sessions, sorted playing > paused > other.""" return self.store.get(FIELD_SESSIONS, []) @property def audio_playback(self) -> list[str]: """Packages that recently played audio, ordered last-played-first.""" return self.store.get(FIELD_AUDIO_PLAYBACK, []) @property def running_apps(self) -> list[dict]: """Visible recent-task apps, ordered by recency.""" return self.store.get(FIELD_RUNNING_APPS, []) @property def protocol_active(self) -> bool: """True when the remote-protocol transport is wired for this entry. Entities use this instead of reaching into ``_protocol``: a passive manager means "no transport", not "broken transport". """ return self._protocol is not None @property def protocol_state(self) -> str: """The transport's real state, in one word (t15). ``off`` (no client wired), ``never_connected`` (wired, no session yet), ``connected``, ``disconnected`` or ``auth_rejected``. The library offers no is_paired/is_connected, so these are the states we can actually prove — and the pre-t15 code was permanently ``never_connected``, which is the state the user was reading as "maybe unpaired". """ return self._protocol_state @property def protocol_connected(self) -> bool: """True only while an authenticated session is up (never assumed).""" return self._protocol_state == PROTOCOL_CONNECTED @property def protocol_ever_authenticated(self) -> bool: """True once a session connected or ANY push arrived (the only proof).""" return self._protocol_ever_authenticated @property def protocol_last_error(self) -> str | None: """The last connect/auth/send failure, verbatim, or None (t15). Readable on the entity so "the transport is fine" and "the transport could not run" are never the same picture — the masking this unit removed. """ return self._protocol_last_error @property def protocol_failures(self) -> int: """How many protocol sends could NOT run (never counted as successes).""" return self._protocol_failures def _apply_protocol_state(self, state: str, error: str | None) -> bool: """Set the published pair and announce a TRANSITION; True if it changed. One place decides what the pair MEANS (``_note_protocol_state`` documents it) and one place decides when a change is announced, so a connect attempt and a device push cannot publish two different stories. """ previous = self._protocol_state state_changed = state != previous self._protocol_state = state error_changed = False if state == PROTOCOL_CONNECTED: # t20/O1: a proven session ENDS the failure episode. The recovery stays # visible in ``protocol_failures`` (history), never as a stale message # published next to ``protocol_connected: true``. error_changed = self._forget_protocol_error() elif error is not None: error_changed = error != self._protocol_last_error self._protocol_last_error = error if state_changed: self._announce_protocol_state(previous, state) return state_changed or error_changed def _forget_protocol_error(self) -> bool: """Drop a failure the transport has since recovered from (t20/O1). Returns True when there WAS one, so the caller knows listeners must be told that a published attribute changed. """ if self._protocol_last_error is None: return False self._protocol_last_error = None return True def _announce_protocol_state(self, previous: str, state: str) -> None: """Announce a transport TRANSITION — this is what keeps banners honest. t20/O2 defines a STALE banner as one whose transport claim later state changes never followed up on: the log's last word would then contradict what the entity publishes. Every transition therefore logs one line naming both states, at INFO for the two states a user acts on (a session came up, a session went down) and at DEBUG otherwise (``never_connected`` repeats per retry, and the auth rejection is already logged with its reason by ``_on_invalid_auth``). """ _LOGGER.log( logging.INFO if state in (PROTOCOL_CONNECTED, PROTOCOL_DISCONNECTED) else logging.DEBUG, "Shield remote-protocol transport state: %s → %s", previous, state, ) def _note_protocol_state(self, state: str, error: str | None) -> None: """Record the transport state (and an error when there is one) + notify. WHAT THE PUBLISHED PAIR MEANS (t20/O1): ``protocol_state`` — and its boolean twin ``protocol_connected`` — is the transport's CURRENT condition, while ``protocol_last_error`` is the failure the current state has NOT recovered from. A session becoming proven (a successful connect, or the transport's own availability push) therefore CLEARS it, so an error can never be published beside a healthy ``protocol_connected``. A failure that happens AFTER the session came up is kept — that one is current, not stale. Before this, ``_note_protocol_state(PROTOCOL_CONNECTED, None)`` left the previous failure in place and the entity published ``protocol_last_error == "connect: Net: boom"`` beside ``protocol_connected: true`` — on the very attributes a user is told to read to answer "is the protocol working?" (t19 review, O1). ``protocol_failures`` still counts, so a recovery remains visible as history instead of as a stale message. """ if self._apply_protocol_state(state, error): self._notify() def _note_protocol_send_failure(self, what: str, err) -> None: """A protocol send that could NOT run must not look like one that did (t15). Before this unit ``send_protocol_key``/``launch_app``/``send_text_ime`` swallowed the failure into ``return False`` + a debug line, so a transport that had never connected was indistinguishable from a working one: the exact masking behind the user's "could be either" pairing question. Every failure is counted, recorded verbatim, and logged at WARNING carrying the transport state. """ self._protocol_failures += 1 detail = f"{type(err).__name__}: {err}" if isinstance(err, BaseException) else str(err) self._protocol_last_error = f"{what}: {detail}" _LOGGER.warning( "Shield protocol %s did not run — nothing was sent (transport: %s, " "last error: %s, failures: %d)", what, self._protocol_state, self._protocol_last_error, self._protocol_failures, ) self._notify() async def async_protocol_device_info(self) -> tuple[str, str] | None: """(device name, MAC) via the protocol client (P5-1 migration). None when the transport is absent or the lookup fails — callers fall back to the config-entry id. """ if self._protocol is None: return None with contextlib.suppress(Exception): name, mac = await asyncio.wait_for( self._protocol.async_get_name_and_mac(), timeout=10 ) return name, mac return None @property def adb_poller(self): """The ADB poller this entry MAY use, or None — the public accessor (F6). Read by the ``adb_command`` service, the screen-capture image and the intent shell instead of reaching into ``_poller``. It answers the same question as every other device path in this class: may this entry talk to the device? So it is None while the entry's option says ADB is off (t23/O1) — and those callers already treat None as "ADB is not available" and refuse with their own message, which is what closes the shell paths that live OUTSIDE this module without this module having to reach into them. The private ``_poller`` stays the transport object for internal bookkeeping (connect/disconnect/state), which is not traffic the user asked for. """ if self._adb_disabled(): return None return self._poller @property def is_on(self) -> bool | None: """Protocol power push (device reports on/off state).""" return self.store.get(FIELD_IS_ON) @property def is_available(self) -> bool | None: """Protocol availability push (device reachable at the protocol).""" return self.store.get(FIELD_IS_AVAILABLE) @property def current_app(self) -> str | None: """Protocol current-app push (raw package; name resolution deferred).""" return self.store.get(FIELD_CURRENT_APP) @property def foreground_app(self) -> str | None: """Best foreground identity: protocol push first, ADB parse fallback (P3-2 store priority: FIELD_CURRENT_APP > FIELD_FOREGROUND).""" return self.store.get(FIELD_CURRENT_APP) or self.store.get(FIELD_FOREGROUND) @property def foreground_source(self) -> str | None: """Which tier produced ``foreground_app``: ``"protocol"``, ``"adb"``, None (t15). The two fields are DISJOINT, so naming the tier is exact. This is what keeps the newly-reachable protocol push from silently outranking the ADB parse that the user's setup is built on: the reading still prefers the push (P3-2), and the entity says where it came from — see the media player's ``app_source`` and the remote's ``activity_source``. """ if self.store.get(FIELD_CURRENT_APP): return "protocol" if self.store.get(FIELD_FOREGROUND): return "adb" return None @property def volume_info(self) -> tuple | None: """Protocol volume push as ``(level, max, muted)`` — readout only (no protocol VOLUME_SET exists, C-VOL).""" return self.store.get(FIELD_VOLUME_INFO) def best_session( self, foreground_package: str | None = None ) -> MediaSessionInfo | None: """Best media session: foreground app first, then playing > paused. P3-2d: when there is no live session at all, the foreground app's last known state is served from the TTL-capped cache (TiviMate publishes no MediaSession — RESEARCH_SHIELD R2). """ sessions = self.active_sessions if foreground_package: for session in sessions: if session.package == foreground_package: return session if sessions: return sessions[0] if foreground_package: return self.last_known_session(foreground_package) return None def _cache_last_known_sessions(self) -> None: """Stamp every live session's state as the package's last known.""" now = self._clock() for session in self.active_sessions: if session.package: self._last_known[session.package] = (session, now) def last_known_session(self, package: str) -> MediaSessionInfo | None: """Last known session state for a package, TTL-capped. Returns None once STATE_STORE_TTL passes — 'playing' must never be resurrected from a stale cache (P3-2d expiry contract). """ cached = self._last_known.get(package) if cached is None: return None info, stamp = cached if self._clock() - stamp > STATE_STORE_TTL: self._last_known.pop(package, None) return None return info def store_snapshot(self) -> dict[str, dict]: """Public StateStore snapshot for diagnostics (F6). ``{field: {"age": seconds, "value": ...}}`` with dataclass values expanded to dicts so the payload stays JSON-serializable. """ snapshot: dict[str, dict] = {} for name, (stamp, value) in self.store.values_stamped().items(): if dataclasses.is_dataclass(value) and not isinstance(value, type): value = dataclasses.asdict(value) snapshot[name] = {"age": round(time.monotonic() - stamp, 3), "value": value} return snapshot async def send_text_ime(self, text: str) -> bool: """Text entry via the protocol client's send_text (P5-2). The Shield may swallow the text when its on-screen virtual keyboard is up (RESEARCH_IDEAS §2 availability gate) — the call reports the send outcome; callers fall back to the ADB input-text path only when the protocol transport is down or the send failed. t15: a send that could not run is logged at WARNING and recorded — the ADB fallback still runs, but silence here is what made a never-connected transport look like a working one. """ if self._protocol is None: self._note_protocol_send_failure("send_text", "no protocol transport") return False try: self._protocol.send_text(text) return True except Exception as err: self._note_protocol_send_failure("send_text", err) return False async def send_adb_input_text(self, text: str) -> bool: """ADB fallback for text entry: `input text` (P5-2). Used only when the protocol send_text path is unavailable/failed. Shell-quoted (shlex) so single quotes and spaces survive. """ import shlex poller = self._poller # t23/O1: the OPTION gates the shell, not the poller's existence. if poller is None or not self.connected or self._adb_disabled(): return False try: result = await poller.shell(f"input text {shlex.quote(text)}") return result is not None except Exception as err: _LOGGER.debug("adb input text failed: %s", err) return False def app_label(self, package: str | None) -> str | None: """Cached manifest label for a package (P5-1), or None (t55). A cache miss enqueues an on-demand `pm dump` fetch — drained one package per watchdog cycle via the manager task pool, so entity refreshes and the watchdog are never blocked by a burst. What is cached is the FETCH, not the wishful result (t55 root cause): * a fetch that RAN caches its outcome for ``APP_LABEL_TTL`` — a label when the manifest has one, and the empty string otherwise. It is definitive: the device answered, the package simply has no default label (several system/OEM packages do not). No re-enqueue, so a permanently label-less package costs ZERO further round trips (t24 F1 stays closed). * a fetch that could NOT run (no poller, ADB down, shell error) caches nothing, so the next render enqueues again — a miss is retried visibly instead of being stamped as an eternal "no label". This was the live defect: the old code stored ("", stamp) on every failure path, which silenced the label tier for the whole 24 h TTL and left app_name showing the raw package id. * while a fetch for the package is queued or in flight, the next render does not enqueue it a second time. """ if not package: return None cached = self._app_labels.get(package) if cached is not None: label, stamp = cached if self._clock() - stamp <= APP_LABEL_TTL: # empty label = negative cache (t24 F1): the pm dump RAN and # found nothing — do not re-enqueue, zero per-heartbeat tax. # A fetch that did not run is not cached at all (t57 UNIT 0), # so there is no "unresolved" shape to fall through on. return label or None self._app_labels.pop(package, None) # stale — refetch on demand if self._adb_disabled(): # t23/O1: ADB off means there is nothing to wait for — no request (so no # task and no shell later) and no cache write. The name falls back to the # user map / curated map / package id exactly as a failed fetch does, and # because nothing was cached, the next render after ADB is turned back on # asks again. _LOGGER.debug( "app label for %s not queued (ADB is switched off for this entry)", package, ) return None if package not in self._app_label_requests and ( package not in self._app_label_inflight ): self._app_label_requests.add(package) return None def _probe_still_owed(self, package: str) -> bool: """True when a package's only remaining question is a RETRY (t66). A probe that FAILED (no poller, shell error, no output at all) records the package here and caches nothing, so the t57 UNIT 0 guarantee — a failed fetch is retried, never stamped as an answer — survives the device-level memo below. A probe that RAN and found no label is not owed anything: it answered. """ return package in self._app_label_failed def _labels_absent(self) -> bool: """The device-level "no labels here" memo, expiring with APP_LABEL_TTL. A conclusion about a piece of firmware cannot be permanent — the user may swap the device, update it, or add an app that does carry a label — so after APP_LABEL_TTL the memo clears itself and one probe re-checks the device (t66). """ if self._labels_absent_at is None: return False if self._clock() - self._labels_absent_at > APP_LABEL_TTL: self._labels_absent_at = None # re-probe once per TTL self._label_negative_streak = 0 return False return True @property def app_labels_absent(self) -> bool: """Public read of the memo (tests, and the options page's own note).""" return self._labels_absent_at is not None and ( self._clock() - self._labels_absent_at <= APP_LABEL_TTL ) def _entry_app_name_map(self) -> dict: """The user's own apps map from the entry options ({} when unset).""" options = getattr(self.entry, "options", None) or {} return dict(options.get(CONF_APP_NAME_MAP) or {}) def _resolved_app_name(self, package: str, device_label=None, user_map=None) -> str: """The display name for a discovered app, or "" when nothing names it (t66). The ONE place the discovery table's name comes from, so it obeys the same chain as the media player and the Current App sensor (:func:`~.ha_adb_common.app_label.resolve_package_name`). "" is deliberate: the row's ``package`` field already carries the id, and a name that repeats it is exactly the ``pkg=pkg`` row the user reported from the apps editor. Every consumer already reads an empty name as "this app needs a name" (the editor says so out loud; the installed-apps sensor publishes it as-is). """ if user_map is None: user_map = self._entry_app_name_map() name = resolve_package_name( package, user_map=user_map, device_label=device_label, curated=DEFAULT_APP_DIRECTORY, ) return "" if not name or name == package else name def _known_label(self, package: str) -> tuple[bool, str]: """``(known, label)`` from the label cache — "known" = a fetch RAN (t57). The accessor the DISCOVERY sweep needs and ``app_label`` deliberately is not: ``app_label`` answers "what should be displayed", so a definitive negative (a package with no manifest label) comes back as ``None``, exactly like a miss. A sweep that used it would re-dump every label-less package on every run and throw away the 24 h cache it is supposed to benefit from. Here the two are distinguishable. """ cached = self._app_labels.get(package) if cached is None: return False, "" label, stamp = cached if self._clock() - stamp > APP_LABEL_TTL: self._app_labels.pop(package, None) # stale — refetch on demand return False, "" return True, label def cache_app_label(self, package: str, label: str | None) -> None: """Record a label fetch that RAN, and what it found (t55, t57). The cache's only writer. ``None``/"" is the negative result devices genuinely report for label-less packages, and it is definitive: no further ``pm dump`` for ``APP_LABEL_TTL``. A fetch that did NOT run must not call this — that was the live defect (see ``_fetch_app_label``). """ self._app_labels[package] = (label or "", self._clock()) @property def app_label_requests(self) -> set[str]: """Packages queued for a manifest-label fetch (public, for tests).""" return set(self._app_label_requests) @property def app_label_inflight(self) -> set[str]: """Packages whose label fetch is running right now (public, tests).""" return set(self._app_label_inflight) async def async_capture_protocol_device_data(self) -> None: """Capture (name, manufacturer, model, sw) once (t30 G10).""" if self._protocol is None or self.protocol_device_data is not None: return try: info = self._protocol.device_info or {} # {manufacturer, model, sw_version} name, _mac = await asyncio.wait_for( self._protocol.async_get_name_and_mac(), timeout=10.0 ) self.protocol_device_data = ( name, info.get("manufacturer", "Google"), info.get("model", "Shield TV"), info.get("sw_version", ""), ) except Exception as err: _LOGGER.debug("protocol device info unavailable: %s", err) async def _dump_label(self, package: str, *, on_demand: bool = False) -> str | None: """One label read: the label, or None for a FAILED fetch (t66). The ONE place the probe is issued, shared by the on-demand path (:meth:`_fetch_app_label`) and the discovery sweep, so the t57 UNIT 0 distinction — "the command ran and found no label" versus "the transport delivered nothing" — cannot drift between them. A successful fetch is also cached here, so both paths benefit from the 24 h cache and neither double-stamps. WHAT CHANGED IN t66, and why the old shape could not work on the user's Shield: the probe used to ship ``pm dump | head -300`` and parse the answer with ``parse_application_label`` alone. A real dump on that device is ~2241 lines of dumpsys, so ``head -300`` cut before the package sections even appeared — the "no label" answer was a truncation artefact, not a fact about the app. The probe now searches ON THE DEVICE for every label shape Android prints (``APP_LABEL_GREP_SHAPES``) and only those lines cross the wire, which is both smaller and honest. A device that has already answered "no labels" for a run of packages is NOT probed again (``APP_LABEL_PROBE_LIMIT``, expiring with ``APP_LABEL_TTL``): the sweep's per-package fan-out stops there and a package the device has never been asked about is answered "" from the memo, with no shell at all. The two exceptions are deliberate: a package whose probe FAILED still owes its retry (``_probe_still_owed``), and that retry is only ever driven by the app ON SCREEN (``on_demand``) — one shell, once, user-driven, then cached. The two meanings of "" must stay distinct in the reader's head: a probe that RAN and found nothing is a cached answer (zero further shells), and a probe that COULD NOT run is not cached at all — it is recorded as owed and retried for the app on screen. Returns the label (possibly "") when the probe RAN, and None when it did not run or delivered no output at all. The exit marker is a liveness tell, not a status code (t62 nit b): it sits after a pipeline, and the Android shell that runs this has no PIPESTATUS, so ``$?`` is the last command's status rather than the dump's. Nothing reads its VALUE — the parser ignores the marker line — and the only property tested here is that a command which RAN leaves non-blank output even when the device prints no label. That non-blankness is what separates "ran, no label" from "the transport delivered nothing", which this method turns into a cached negative versus a retry. """ if self._labels_absent(): if not self._probe_still_owed(package): # The firmware already answered; a shell here would spend a round # trip on a settled question (t66). The negative is cached so the # render path stops asking cycle after cycle. self.cache_app_label(package, "") return "" if not on_demand: # a retry is not the sweep's job: the fan-out stops here return "" _LOGGER.debug( "retrying the failed label probe for %s on the app now on " "screen (the device reports no labels, but this one never " "answered)", package, ) poller = self._poller # t23/O1: the OPTION is asked at the ONE place a label dump is issued (the # sweep's fan-out comes through here too), so this gate covers both paths. if poller is None or not self.connected or self._adb_disabled(): _LOGGER.debug( "app label for %s not fetched (ADB not connected or switched off) " "— will retry on the next render", package, ) self._app_label_failed.add(package) return None try: import shlex dump = await poller.shell( f"pm dump {shlex.quote(package)} 2>/dev/null | " f"grep -E {shlex.quote(APP_LABEL_GREP_SHAPES)} | head -40; " f"echo {APP_LABEL_EXIT_MARKER}$?" ) except Exception as err: _LOGGER.debug( "app label fetch failed for %s (%s) — will retry on the " "next render", package, err, ) self._app_label_failed.add(package) return None if dump is None or not dump.strip(): # the transport never delivered output: a FAILED fetch, not an # answer. Caching here is the t55 defect this repairs. _LOGGER.debug( "app label fetch for %s returned no output — will retry on " "the next render", package, ) self._app_label_failed.add(package) return None self._app_label_failed.discard(package) # it answered — nothing owed label = parse_app_label(dump) if label: # This firmware DOES name apps: a run of misses before it proved nothing. self._label_negative_streak = 0 else: self._label_negative_streak += 1 if ( self._label_negative_streak >= APP_LABEL_PROBE_LIMIT and self._labels_absent_at is None ): # A RUN of label-less answers, each of which really ran: this # device keeps its labels somewhere we cannot read, so every # further per-package probe is a shell spent on a settled # question (t66). The conclusion expires with APP_LABEL_TTL. self._labels_absent_at = self._clock() _LOGGER.info( "%d label probes in a row found no app label — this device " "exposes none; names now come from the curated map and the " "user's own apps map for the next %s seconds", self._label_negative_streak, APP_LABEL_TTL, ) # F1: label-less probes negatively cache too (empty label) so a later # cycle performs zero additional probes for the package. self.cache_app_label(package, label) # "" (not None) is the "the probe RAN" answer, even when the device named # nothing: None is reserved for a fetch that could not run (t57 UNIT 0). return label or "" async def _fetch_app_label(self, package: str) -> None: """One on-demand `pm dump` label fetch; stamps ONLY a fetch that RAN. The on-demand half of the label path (one package per watchdog cycle, driven by an entity render). The dump itself — including the t57 UNIT 0 distinction between a FAILED fetch and a genuinely label-less package — lives in :meth:`_dump_label`, so the sweep and this path cannot drift. """ self._app_label_inflight.add(package) try: await self._dump_label(package, on_demand=True) finally: self._app_label_inflight.discard(package) def _on_logcat_app_start(self, package: str) -> None: """Stamp FIELD_FOREGROUND instantly from a parsed START (P5-3). The next watchdog cycle confirms via poll_foreground/priority — whatever dumpsys says wins, so logcat↔dumpsys disagreement never becomes state churn. Duplicate STARTs for the same package inside LOGCAT_SUPPRESS_WINDOW are not re-stamped. """ now = self._clock() if ( package == self._logcat_last_pkg and now - self._logcat_last_stamp < LOGCAT_SUPPRESS_WINDOW ): return self._logcat_last_pkg = package self._logcat_last_stamp = now self.store.set(FIELD_FOREGROUND, package) self._notify() def _open_logcat_stream(self): """Open the dedicated streaming logcat line source (P5-3). Returns an async generator of decoded logcat lines on ITS OWN connection — a fresh AdbDeviceTcpAsync sharing only the key signer material with the poller, never the poller's socket. """ poller = self._poller if self._adb_disabled(): # t23/O1: ADB off is not a poller problem, and this stream opens its OWN # connection to the device — the option gates it like every other path. raise RuntimeError( "ADB is switched off for this entry — logcat stream cannot start" ) if poller is None: raise RuntimeError("poller gone — logcat stream cannot start") import asyncio from adb_shell.adb_device_async import AdbDeviceTcpAsync from adb_shell.auth.sign_pythonrsa import PythonRSASigner config = poller.config # public accessor (t27 F-NEW3) async def _lines(): loop = asyncio.get_running_loop() signer = await loop.run_in_executor( None, PythonRSASigner.FromRSAKeyPath, config["key_path"] ) device = AdbDeviceTcpAsync(config["host"], config["port"]) connected = await device.connect(rsa_keys=[signer], auth_timeout_s=10.0) if not connected: raise RuntimeError("dedicated logcat connection failed") # F-NEW1 (HIGH): quiet logcat is NORMAL on a bored device — the # TCP transport timeout must NOT be the idle detector or every # ≥10 s gap becomes a TcpTimeoutException streak that ends in # the session auto-disable in the field. transport_timeout_s=None # makes bulk_read block indefinitely; real loss (dead socket) is # the ladder's job (reconnect with backoff), and async_stop is # the cleanup path. async for line in device.streaming_shell( "logcat -v brief ActivityTaskManager:*", transport_timeout_s=None, ): yield line return _lines() def _note_logcat_failure(self, failure_time: float) -> None: """Record a stream failure; auto-disable after 5 within 10 minutes.""" self._logcat_failures = [ stamp for stamp in self._logcat_failures if failure_time - stamp <= 600.0 ] self._logcat_failures.append(failure_time) if len(self._logcat_failures) >= 5: self._logcat_disabled = True _LOGGER.warning( "Live events (logcat) stream failed 5 times in 10 minutes — " "disabled for this session; the watchdog heartbeat remains " "the state floor" ) self._logcat_failures.clear() async def _logcat_pass(self) -> None: """One full consumption of the logcat line-stream (P5-3). Yields lines are buffered 3-deep (t27 F-NEW2) before parsing so a START record whose cmp= continuation spans stream lines is not missed. Returns when the stream ends (the dedicated connection dropped); errors propagate to the caller's ladder. """ buffer: list[str] = [] async for line in self._open_logcat_stream(): buffer.append(str(line)) if len(buffer) < 3: # small wrap window (t27 F-NEW2) continue package = foreground_from_logcat(buffer) if package: self._on_logcat_app_start(package) buffer = buffer[-2:] # keep the tail for a pending continuation # flush the tail buffer once the stream ends if buffer: package = foreground_from_logcat(buffer) if package: self._on_logcat_app_start(package) raise RuntimeError("stream ended (device closed)") async def _run_logcat_stream(self) -> None: """Consume the logcat line-stream; backoff-reconnect on failure. The stream owns its failure ladder (separate from the ADB socket watchdog, reusing the same jitter): Stream failures reconnect with 5 s doubling to the 300 s cap ±jitter; five failures in ten minutes auto-disable the option for the session (heartbeat floor). The ladder sleep is injectable (_logcat_ladder_sleep) so tests stay deterministic. """ delay = ADB_RECONNECT_BACKOFF_START while not self._stopping and not self._logcat_disabled: if not self._adb_active(): # t23/O1: ADB off ends the stream WITHOUT counting a failure — the # device is not at fault for a local option, and five of those would # auto-disable the stream for the session. break try: await self._logcat_pass() except asyncio.CancelledError: raise except Exception as err: _LOGGER.debug("logcat stream failure: %s", err) self._note_logcat_failure(self._clock()) if self._stopping or self._logcat_disabled: break await self._logcat_ladder_sleep(self._jitter(delay)) delay = min(delay * 2, ADB_RECONNECT_BACKOFF_CAP) def _ensure_logcat_stream(self) -> None: """Start the consumer when the option is on, connected, enabled.""" if self._logcat_disabled or not self._entry_options_logcat(): return # t23/O1: option AND transport — a poller that exists while the option says # ADB is off must not start a device stream. if not self._adb_active() or not self.connected: return if self._logcat_task is None or self._logcat_task.done(): self._logcat_task = self._create_task( self._run_logcat_stream(), "shield_logcat_stream" ) def _entry_options_logcat(self) -> bool: """CONF_LOGCAT_EVENTS from the entry's options (default off).""" entry = getattr(self, "entry", None) if entry is None: return False return bool((getattr(entry, "options", {}) or {}).get(CONF_LOGCAT_EVENTS, False)) def _stop_logcat_stream(self) -> None: """Cancel the consumer at unload/stop (the stream closes in its own finally — the dedicated device never outlives the manager).""" if self._logcat_task is not None: self._logcat_task.cancel() self._logcat_task = None # ── t57: installed-app discovery ─────────────────────────────────────── # # MARK — documented, not observed (t62 nit c): the output SHAPES the two # listing commands rely on (`pm list packages -3` → `package:` lines, # `cmd package resolve-activity --brief -c …HOME` → `package/activity`) are # Android's documentation, not a captured run: no real Shield was available # when discovery was written (t57). Every fixture in # tests/test_installed_apps.py is built from those shapes and the parsers in # ``ha_adb_common.installed_apps`` are deliberately tolerant of whitespace # and ordering. Same mark as the state=6 buffering inference in # ``media_player.py``. #: t62 F1: how long after an ATTEMPTED sweep failed before another one is #: issued. A retry costs two listing shells, so it waits out one state poll #: (30 s) rather than hammering a silent socket; ``force=True`` bypasses it. #: An unavailable transport (no poller, ADB down) costs nothing and is #: retried on the very next update. APP_SWEEP_RETRY_SECONDS: float = 30.0 def installed_apps(self) -> list[dict]: """The discovered apps, ``[{package, name, icon}]``, cached (t57). Returns the cached table whenever it is fresh, and an EMPTY list when there is no fresh answer: ADB disabled, not connected, the first sweep still running, or a sweep that could not run (t62 F1). Empty is therefore "nothing to show", never by itself "this device has no apps" — ask :meth:`installed_apps_status` which of the two it is. A caller that wants the table filled calls :meth:`async_refresh_installed_apps`; this accessor never blocks and never issues a shell itself, so a state poll stays free of discovery work. """ if self._installed_apps is None: return [] rows, stamp = self._installed_apps if self._clock() - stamp > APP_LIST_TTL: return [] # stale — a refresh repopulates it, the poll does not return [dict(row) for row in rows] @property def installed_apps_stale(self) -> bool: """True when no fresh table is available (including "never swept").""" if self._installed_apps is None: return True _rows, stamp = self._installed_apps return self._clock() - stamp > APP_LIST_TTL @property def installed_apps_status(self) -> str: """Why :meth:`installed_apps` may be empty, in one word (t62 F1). * ``"ok"`` — a sweep RAN and its answer is fresh. An empty table with this status is a real answer (the device reported no launchable apps), which is the case the old stamped-empty cache could not say. * ``"unknown"`` — the last sweep could not deliver an answer (no poller, ADB down, a shell that raised or answered nothing). Nothing is known about the device, and the next update retries. * ``"pending"`` — no sweep has completed yet, or the last answer expired and a refresh is due. """ if not self.installed_apps_stale: return "ok" return "unknown" if self._apps_unavailable else "pending" async def async_refresh_installed_apps(self, force: bool = False) -> list[dict]: """(Re)build the installed-app table over ADB, bounded and shared (t57). One sweep at a time: concurrent callers await the same task rather than starting a second burst, and a fresh table is returned as-is unless ``force``. The sweep is deliberately NOT part of the watchdog: it is a handful of shells on demand, not a per-cycle cost. Failure is not an answer (t62 F1). No poller, a disconnected transport, a shell that raises, or a sweep that delivered no output records ``unknown`` and stores NO table, so the next non-forced call — an entity poll or the standard ``update_entity`` action — retries instead of serving a stamped empty answer for the whole APP_LIST_TTL. A sweep that reached the device and failed is retried at most once per ``APP_SWEEP_RETRY_SECONDS`` so a silent socket is not hammered; a sweep that RAN stores its answer, including a genuinely empty list. """ if self._adb_disabled(): # t23/O1: no sweep is STARTED, and not even the retry-spacing bookkeeping # below runs — nothing was attempted, so nothing is owed a retry. The last # known table is returned as-is (empty when none was ever built). _LOGGER.debug( "installed-app refresh skipped (ADB is switched off for this entry)" ) return self.installed_apps() if not force and not self.installed_apps_stale: return self.installed_apps() if ( not force and self._apps_failed_at is not None and self._clock() - self._apps_failed_at < self.APP_SWEEP_RETRY_SECONDS ): # The previous attempt really issued shells and failed: space the # retries by one poll interval instead of re-issuing them every # 30 s on a socket that is not answering (t62 F1). return self.installed_apps() if self._apps_sweep is not None and not self._apps_sweep.done(): await asyncio.shield(self._apps_sweep) return self.installed_apps() task = self._create_task(self._sweep_installed_apps(), "apps_sweep") self._apps_sweep = task try: rows = await task finally: if self._apps_sweep is task: self._apps_sweep = None return [dict(row) for row in rows] async def _sweep_installed_apps(self) -> list[dict]: """The sweep itself: list, label, icon — then stamp the cache (t57). Only a sweep that RAN stamps the cache (t62 F1): every failure path records ``unknown`` and leaves ``_installed_apps`` alone, so the table the entity serves is the last one that was real — never a fabricated empty answer that also blocks the retry for APP_LIST_TTL. """ poller = self._poller # t23/O1: the OPTION is asked FIRST. The review measured this path sending 4 # real shells with a poller injected into an entry whose option said ADB is # off: the poller's absence was doing the gating, and that is setup code. if self._adb_disabled(): _LOGGER.debug( "installed-app discovery skipped (ADB is switched off for this " "entry) — the options page and the app list stay on what is known" ) self._note_apps_unavailable(attempted=False) return [] if poller is None or not self.connected: _LOGGER.debug( "installed-app discovery skipped (ADB unavailable) — the " "options page and the app list stay on what is already known" ) self._note_apps_unavailable(attempted=False) return [] try: user_output = await poller.shell("pm list packages -3") launcher_output = await poller.shell( "cmd package resolve-activity --brief -c " "android.intent.category.HOME" ) except Exception as err: # pragma: no cover - shell swallows its own _LOGGER.debug("installed-app discovery failed: %s", err) self._note_apps_unavailable(attempted=True) return [] if user_output is None and launcher_output is None: # the transport delivered nothing at all: no list is NOT an answer _LOGGER.debug( "installed-app discovery got no output — will retry on the " "next update" ) self._note_apps_unavailable(attempted=True) return [] packages = [p for p in installed_packages(user_output, launcher_output) if is_launchable(p)] labels = await self._sweep_labels(packages) # t66: ONE precedence chain for the table too — the user's own map, then # the device's label (when it exposes one), then the curated map, then the # package id. An app nothing can name gets name "" rather than its own id: # the row already carries the id in ``package``, and writing it twice is # the ``pkg=pkg`` the user read as a bug in the apps editor. user_map = self._entry_app_name_map() rows = [ { "package": package, "name": self._resolved_app_name(package, labels.get(package), user_map), "icon": icon_for_package(package, APP_ICON_FALLBACKS, APP_ICON_GENERIC), } for package in packages ] self._installed_apps = (rows, self._clock()) self._apps_unavailable = False self._apps_failed_at = None _LOGGER.debug("installed-app discovery found %d apps", len(rows)) return rows def _note_apps_unavailable(self, *, attempted: bool) -> None: """Record a sweep that could not deliver an answer (t62 F1). ``attempted`` says whether shells were actually issued: only then does the retry window apply, because only then does a retry cost anything. """ self._apps_unavailable = True self._apps_failed_at = self._clock() if attempted else None async def _sweep_labels(self, packages: list[str]) -> dict[str, str]: """Labels for a whole sweep, in a bounded burst (t57). Deliberately NOT through ``_app_label_requests``: that queue is drained ONE package per watchdog cycle, which is right for the live entity path (a render must never fan out into shells) but would take a couple of minutes for a Shield with ~40 apps. Here the fetches run ``APP_SWEEP_CONCURRENCY`` at a time with a short spacing, so a typical Shield finishes in seconds while the socket still sees a trickle rather than a flood. Cached labels are reused, so a second sweep costs almost nothing; a failed fetch is NOT cached (t57 UNIT 0) and the app falls back to its package id for this table while the next sweep retries it. The check goes through :meth:`_known_label` rather than :meth:`app_label`, so a package the device has definitively answered "no label" for is not re-dumped on every sweep. """ cached: dict[str, str] = {} missing: list[str] = [] for package in packages: known, label = self._known_label(package) if not known: missing.append(package) elif label: cached[package] = label # a known NEGATIVE needs no entry: the row falls back to its # package id, and the cache already holds the definitive answer if not missing: return cached semaphore = asyncio.Semaphore(APP_SWEEP_CONCURRENCY) async def _one(package: str) -> tuple[str, str | None]: async with semaphore: # a small spacing keeps the burst polite on a single socket await asyncio.sleep(APP_SWEEP_SPACING) return package, await self._dump_label(package) results = await asyncio.gather( *(_one(package) for package in missing), return_exceptions=True ) for result in results: if isinstance(result, BaseException): continue # one bad package never fails the sweep package, label = result if label: cached[package] = label else: # None here means the fetch FAILED (nothing cached); the empty # string means the device answered with no label. Both leave # the row on its package id, and both keep the app in the list. cached[package] = "" return cached def _drain_app_label_requests(self) -> None: """Start at most one label fetch per watchdog cycle.""" # t23/O1: option AND poller. A queue that cannot be served stays queued (it is # served once ADB is turned back on) instead of starting a task that cannot run. if not self._app_label_requests or not self._adb_active(): return package = min(self._app_label_requests) self._app_label_requests.discard(package) self._tasks.append( self._create_task(self._fetch_app_label(package), f"app_label_{package}") ) # ── Intent discovery (t16) ── @property def intent_cache(self) -> IntentScanCache: """The per-package intent-scan cache (t16) — tests, and the options page.""" return self._intents def intent_scan_state(self, package: str | None) -> str | None: """Which cached ANSWER holds for a package: "candidates", "empty", None (t16). None means no answer is cached: the package was never scanned, or the last scan could not deliver one (a failed scan is never cached, so the next request retries instead of serving "this app advertises nothing"). """ return self._intents.state(package) def _adb_disabled(self) -> bool: """Is ADB switched OFF for this entry (t16)? — the gate for EVERY device path (t23). The config option is the honest signal: no poller can mean "off" OR "enabled but not connected yet", and those are different statuses with different fixes (turn ADB on / pair it, versus wait or reconnect). t23/O1: every path in this class that can reach the device ASKS this, rather than inferring it from a poller being absent. The two are not the same guarantee: ``_build_poller_factory`` returns None when ADB is off, which makes the poller's absence a property of the SETUP code, and the review measured what that costs — with a poller injected into an entry whose own option said ADB is off, the installed-apps sweep sent 4 real shells and a label probe sent 1 more. ``available`` and ``connected`` keep describing the TRANSPORT; this describes PERMISSION. An entry with no options at all is treated as OFF (fail closed, matching ``__init__.py``). """ options = getattr(self.entry, "options", None) or {} return not options.get(CONF_ADB_ENABLED, False) def _adb_active(self) -> bool: """May this entry talk to the device right now (t23/O1)? The two-part predicate the shared/background paths use: the OPTION says ADB is on, and a poller exists to talk through. ``available`` alone answers the transport half of that question, which is why it cannot be the gate. """ return self.available and not self._adb_disabled() async def async_scan_intents( self, package: str, force: bool = False ) -> IntentScanResult: """Scan ONE app's advertised intents over ADB, on demand (t16). EXPLICIT AND SINGLE (the plan's ruling): one package per call, triggered by the user scanning that app — never a sweep over the installed apps, which on the user's Shield would be ~2241 dump lines each. The caller gets an :class:`IntentScanResult` whose STATUS says what happened, so "ADB is off", "the device did not answer" and "this app advertises no intents" can never be confused for one another. Cached per package with ``INTENT_SCAN_TTL`` unless ``force``: a cached answer is returned as ``cached`` with its age and its entry state, and only a scan that RAN is ever cached (see :class:`IntentScanCache`). """ name = (package or "").strip() if not is_launchable(name): # A malformed id never reaches a shell, and never looks like an # answer: the same rule the sweep applies to the rows it keeps. return IntentScanResult(package=name, status=INTENT_SCAN_INVALID_PACKAGE) # t21/F1: THE OPTION IS CHECKED FIRST, ONCE, AND IT RETURNS BEFORE ANYTHING # ELSE — no cache read, no cache write, no shell. This check used to live # INSIDE the no-poller branch below, so an entry that HOLDS a poller while # its own option says ADB is off fell straight through to the dump: the # status was merely mislabelled, but the SHELL WAS REAL, and a scan the # entry was not allowed to run was CACHED as an answer (measured by the t18 # review: status `fresh`, one candidate cached, `pm dump com.x …` sent). # "ADB is switched off" is a fact about the ENTRY, so it is decided from the # entry — never inferred from which objects happen to exist. if self._adb_disabled(): _LOGGER.debug( "intent scan for %s not attempted (ADB is switched off for this " "entry) — no shell, nothing cached", name, ) return IntentScanResult(package=name, status=INTENT_SCAN_ADB_DISABLED) if not force: cached = self._intents.get(name) if cached is not None: return IntentScanResult( package=name, status=INTENT_SCAN_CACHED, candidates=cached.candidates, entry_state=cached.state, age_seconds=self._intents.age(name), truncated=cached.truncated, ) if self._poller is None: # t21/F1: no poller is NOT the same fact as ADB being switched off. The # option above said ADB is ON, so the honest answer is "the device # cannot be asked right now" (wait, or reconnect), never `adb_disabled` # (go turn the option on) — the two have different fixes. _LOGGER.debug( "intent scan for %s not attempted (ADB is on for this entry but " "there is no poller to ask) — no shell, nothing cached", name, ) return IntentScanResult(package=name, status=INTENT_SCAN_ADB_UNAVAILABLE) if not self.connected: _LOGGER.debug( "intent scan for %s not attempted (ADB connected: no) — the " "device cannot be asked right now", name, ) return IntentScanResult(package=name, status=INTENT_SCAN_ADB_UNAVAILABLE) running = self._intent_scans.get(name) if running is not None and not running.done(): # One dump per package at a time: a second request awaits the scan # already in flight rather than paying for a second one. return await asyncio.shield(running) task = self._create_task(self._scan_intents(name), f"intents_{name}") self._intent_scans[name] = task try: return await task finally: if self._intent_scans.get(name) is task: self._intent_scans.pop(name, None) async def _scan_intents(self, package: str) -> IntentScanResult: """The scan itself: the on-device filter, the bounded fallback, the cache (t16). The status vocabulary is the point. A capture that carries the resolver section is an ANSWER — cached even when it holds no candidates — while anything else (a shell that raised, no output, an error line, a filtered answer we cannot read) is NOT an answer, is NOT cached, and is retried. That is the discovery-cache defect's rule applied to intents, and it is why ``has_activity_resolver_table`` is asked before the parser is believed. """ if not self._adb_active(): # t23/O1: the SHELL SITE itself asks the option (the public caller asked # first), so no route reaches the device — the t21 repair, one layer down. return IntentScanResult( package=package, status=( INTENT_SCAN_ADB_DISABLED if self._adb_disabled() else INTENT_SCAN_ADB_UNAVAILABLE ), ) poller = self._poller if poller is None: # pragma: no cover - the caller checked this return IntentScanResult(package=package, status=INTENT_SCAN_ADB_UNAVAILABLE) delivered_something = False for command in ( intent_scan_command(package), intent_scan_fallback_command(package), ): try: answer = await poller.shell(command) except Exception as err: # noqa: BLE001 - the shell's failure IS the finding _LOGGER.debug("intent scan for %s failed (%s)", package, err) continue if not answer or not answer.strip(): continue delivered_something = True if not has_activity_resolver_table(answer): # The device answered, but not with the resolver section (a # "Unable to find package" line, a truncated answer): worth one # bounded fallback attempt, and never a cached answer. continue report = parse_intent_report(answer) entry = self._intents.put( package, report.candidates, truncated=report.truncated ) _LOGGER.debug( "intent scan for %s answered: %d candidate(s), state=%s, " "truncated=%s", package, len(entry.candidates), entry.state, entry.truncated, ) return IntentScanResult( package=package, status=INTENT_SCAN_FRESH if report.candidates else INTENT_SCAN_FRESH_EMPTY, candidates=entry.candidates, entry_state=entry.state, age_seconds=0.0, truncated=entry.truncated, ) return IntentScanResult( package=package, status=INTENT_SCAN_UNREADABLE if delivered_something else INTENT_SCAN_ADB_UNAVAILABLE, ) def last_known_map(self) -> dict[str, MediaSessionInfo]: """TTL-filtered snapshot of the per-package last-known cache.""" result: dict[str, MediaSessionInfo] = {} for package in tuple(self._last_known): info = self.last_known_session(package) if info is not None: result[package] = info return result def last_known_state( self, package: str ) -> tuple[MediaSessionInfo | None, float | None]: """(last known session, freshness stamp) for a package — the stamp lets callers age the served state in their UI.""" cached = self._last_known.get(package) if cached is None: return None, None info, stamp = cached if self._clock() - stamp > STATE_STORE_TTL: self._last_known.pop(package, None) return None, None return info, stamp # ── Listener API (entities are thin callback-views) ── def register_listener(self, listener: Callable[[], None]) -> Callable[[], None]: """Register a refresh callback; returns an unsubscribe callable.""" self._listeners.append(listener) def _unregister() -> None: with contextlib.suppress(ValueError): self._listeners.remove(listener) return _unregister def _notify(self) -> None: for listener in tuple(self._listeners): try: listener() except Exception: _LOGGER.exception("Shield state listener failed") # ── Delegation (commands over the managed socket) ── async def send_media_command(self, command: str) -> bool: """Send a media key command over the persistent socket.""" # t23/O1: the option AND a poller — `available` alone is the transport half # of the question, and a poller can exist while the option says ADB is off. if not self._adb_active(): return False return await self._poller.send_media_command(command) async def bring_to_foreground(self, package: str) -> bool: # t23/O1: the option AND a poller — `available` alone is the transport half # of the question, and a poller can exist while the option says ADB is off. if not self._adb_active(): return False return await self._poller.bring_to_foreground(package) # -- Protocol command delegation (Phase 2) -- async def send_protocol_key(self, key: str, direction: str = "SHORT") -> bool: """Send a remote-protocol key push (fast, non-blocking). Returns False when the transport could not run the send (no client, or the library raised ConnectionClosed because nothing is connected) — entities then fall back to the ADB keyevent path. t15: that failure is a WARNING carrying the transport state, never a debug line (see ``_note_protocol_send_failure``). """ if self._protocol is None: self._note_protocol_send_failure(f"send_key {key}", "no protocol transport") return False try: self._protocol.send_key_command(key, direction) return True except Exception as err: self._note_protocol_send_failure(f"send_key {key}", err) return False async def launch_app(self, app_link_or_app_id: str) -> bool: """Launch an app or URL over the protocol transport. ``send_launch_app_command`` auto-prefixes ``market://launch?id=`` for bare package ids (C-LAUNCH); deep links pass through unchanged. t15: False means the launch DID NOT RUN (no transport, or the library raised because nothing is connected) and says so at WARNING; callers that need to report the outcome to a user must check it, not assume it. """ if self._protocol is None: self._note_protocol_send_failure( f"launch {app_link_or_app_id}", "no protocol transport" ) return False try: self._protocol.send_launch_app_command(app_link_or_app_id) return True except Exception as err: self._note_protocol_send_failure(f"launch {app_link_or_app_id}", err) return False async def send_digit_key(self, digit: str) -> bool: """Channel digit via the ADB keyevent path (fallback for §7).""" # t23/O1: the option AND a poller — `available` alone is the transport half # of the question, and a poller can exist while the option says ADB is off. if not self._adb_active(): return False return await self._poller.send_digit_keyevent(digit) async def send_enter_key(self) -> bool: """ENTER via the ADB keyevent path (KEYCODE_ENTER = 66, §7 fallback).""" # t23/O1: the option AND a poller — `available` alone is the transport half # of the question, and a poller can exist while the option says ADB is off. if not self._adb_active(): return False return await self._poller.send_enter_keyevent() # ── Lifecycle ── async def async_start(self) -> None: """Start the persistent connection loop. Never validates the ADB connection here: setup must not fail because of ADB (it is optional), so all failure handling lives in the watchdog's background retry. """ self._stopping = False self._tasks.append(self._create_task(self._watchdog(), "shield_adb_watchdog")) await self._start_protocol() async def async_stop(self) -> None: """Cancel all manager tasks and close the socket. Called on entry unload and on EVENT_HOMEASSISTANT_STOP. """ self._stopping = True self._stop_logcat_stream() tasks, self._tasks = self._tasks, [] for task in tasks: task.cancel() # t20/O4: cancel() only SCHEDULES the cancellation, and an unload that # returns while a connect attempt is still running is exactly how a loop # outlived its entry. Await them: their exceptions (CancelledError included) # are results here, not failures — being cancelled IS how a manager task ends. if tasks: await asyncio.gather(*tasks, return_exceptions=True) self._listeners.clear() self._last_known.clear() self._app_labels.clear() self._app_label_requests.clear() self._app_label_inflight.clear() if self._poller is not None: with contextlib.suppress(Exception): await self._poller.disconnect() self._poller = None if self._protocol is not None: with contextlib.suppress(Exception): self._protocol.disconnect() self._protocol = None def _create_task(self, coro, name: str) -> asyncio.Task: if self.hass is not None: return self.hass.async_create_background_task(coro, name) return asyncio.create_task(coro, name=name) async def _watchdog(self) -> None: """Persistent loop: reconnect with backoff when down, refresh when up. A passive manager (no poller possible) exits immediately — the entry recreates it when options change (full reload). """ try: if self._adb_disabled(): # t23/O1: ADB off means there is nothing to keep alive, and this sits # BEFORE the factory — so even a factory that would build a poller # cannot put traffic on the device. The entry reloads on an options # change, which is what brings the loop back. _LOGGER.debug("ADB watchdog not started (ADB is off for this entry)") return if self._poller is None: if self._poller_factory is None: return self._poller = self._poller_factory() if self._poller is None: return backoff = ADB_RECONNECT_BACKOFF_START while not self._stopping: delay = await self._watchdog_cycle(backoff) if self._stopping: break if self.connected: backoff = ADB_RECONNECT_BACKOFF_START else: backoff = min(backoff * 2, ADB_RECONNECT_BACKOFF_CAP) await asyncio.sleep(delay) except asyncio.CancelledError: raise except Exception: _LOGGER.exception("Shield connection watchdog crashed") async def _watchdog_cycle(self, backoff: float) -> float: """Run one watchdog cycle; return seconds to sleep before the next. When the connection is up this is the heartbeat: wake check + (when awake) media sessions + running apps, all stored with timestamps. The current backoff sleeps after failures; success resets it.""" if self._adb_disabled(): # t23/O1: a DIRECT call must not connect either — a TCP connect is device # traffic. The option is read live on every cycle, so a later flip is picked # up (a check, never a latch). return ADB_POLL_INTERVAL try: if not self.connected and not await self._poller.connect(): await self._note_failure(backoff) return self._jitter(backoff) if self._failures: self._failures = 0 self._down_since = False _LOGGER.info("ADB connection recovered") await self._refresh() return ADB_POLL_INTERVAL except Exception as err: await self._note_failure(backoff) _LOGGER.debug("Watchdog cycle failed: %s", err) return self._jitter(backoff) async def _note_failure(self, backoff: float) -> None: """Bookkeep a failed cycle: log sparingly, drop stale store, notify.""" self._failures += 1 if self._failures == 1 or self._failures % 6 == 0: _LOGGER.warning( "ADB connection down (%s consecutive failures) — retrying in ~%.0fs", self._failures, backoff, ) if not self._down_since: self._down_since = True # Reconnect decision (health_check's documented role): close the # socket so the next cycle rebuilds it from scratch instead of # trusting shell()'s auto-reconnect on a broken transport. try: await self._poller.disconnect() except Exception: _LOGGER.debug("disconnect during reconnect decision failed", exc_info=True) # Only ADB-derived fields expire here: protocol pushes are # device-reported truth and must survive an ADB outage (t11 F2). self.store.clear(*ADB_FIELDS) self._notify() @staticmethod def _jitter(delay: float) -> float: """Backoff with ±jitter so many clients do not reconnect in lockstep.""" return max( 1.0, delay * (1.0 + random.uniform(-ADB_RECONNECT_JITTER, ADB_RECONNECT_JITTER)), ) async def _refresh(self) -> None: """Heartbeat: pull all ADB-derived state into the store (with stamps).""" if self._adb_disabled(): # t23/O1: the ADB half of the heartbeat — wake probe, media dump, running # apps, foreground fallback, the label drain and the logcat ensure — does # not run at all. (The protocol identity capture further down is protocol # work, and it is reached only while the watchdog is running, which needs # ADB on.) return poller = self._poller awake = await poller.is_awake() if awake is None: # Failed poll (connection dropped / device unreachable) — surface # as a cycle failure instead of clinging to the last values. raise ConnectionError("ADB wake poll failed") self.store.set(FIELD_AWAKE, awake) if awake: # Full dump, no truncation (B4) — low sessions must not be lost. await poller.poll_media_sessions(head_lines=None) self.store.set(FIELD_SESSIONS, list(poller.all_sessions)) self._cache_last_known_sessions() # Zero-cost hint from the same dump; system processes filtered # (mediashell/systemui/tv settings/diagtools report "audio"). playback = [ pkg for pkg in parse_audio_playback(poller.last_dump or "") if pkg not in SYSTEM_MEDIA_PACKAGES ] self.store.set(FIELD_AUDIO_PLAYBACK, playback) self.store.set(FIELD_RUNNING_APPS, await poller.poll_running_apps()) # Foreground fallback (P3-2): ADB parse only when the protocol # push is silent — store priority is protocol > ADB, so this # never overwrites a live push. if self.store.get(FIELD_CURRENT_APP) is None: foreground = await poller.poll_foreground() if foreground: self.store.set(FIELD_FOREGROUND, foreground) # G10: capture the protocol device identity for the registry card if self._protocol is not None: self._tasks.append( self._create_task( self.async_capture_protocol_device_data(), "protocol_device_capture", ) ) # P5-1: manifest labels resolve on demand, one package per cycle self._drain_app_label_requests() # P5-3: ensure the (optional, default-off) logcat fast-path runs self._ensure_logcat_stream() else: # Standby: playback data is not current — expire it. self.store.set(FIELD_SESSIONS, []) self.store.set(FIELD_AUDIO_PLAYBACK, []) self.store.set(FIELD_RUNNING_APPS, []) self._notify()