feat(connection,const,strings): complete the P5-3 logcat consumer wiring lost in an earlier amend — dedicated stream, instant stamp, ladder + auto-disable, default-off toggle strings

custom_components/nvidia_shield/connection.py

@@ -37,11 +37,14 @@ from .const import (
     ADB_RECONNECT_BACKOFF_START,
     ADB_RECONNECT_JITTER,
     APP_LABEL_TTL,
+    CONF_LOGCAT_EVENTS,
+    LOGCAT_SUPPRESS_WINDOW,
     STATE_STORE_TTL,
     SYSTEM_MEDIA_PACKAGES,
 )
 from .ha_adb_common import (
     MediaSessionInfo,
+    foreground_from_logcat,
     parse_application_label,
     parse_audio_playback,
 )
@@ -204,6 +207,15 @@ class ShieldConnectionManager:
         # (APP_LABEL_TTL) cache resolved via `pm dump` on demand, one
         # package per watchdog cycle so a burst never blocks the watchdog.
         self._app_labels: dict[str, tuple[str, float]] = {}
+        # 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
+        self._logcat_last_stamp: float = -1e9
         self._app_label_requests: set[str] = set()
         # Injectable clock keeps the TTL expiry tests deterministic.
         self._clock = clock
@@ -565,6 +577,130 @@ class ShieldConnectionManager:
         if label:
             self._app_labels[package] = (label, self._clock())
 
+    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 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
+
+        async def _lines():
+            key_path = getattr(poller, "_key_path", None) or getattr(
+                poller, "key_path", None
+            )
+            host = getattr(poller, "_host", None) or getattr(poller, "host", None)
+            port = getattr(poller, "_port", None) or getattr(poller, "port", None)
+            loop = asyncio.get_running_loop()
+            signer = await loop.run_in_executor(
+                None, PythonRSASigner.FromRSAKeyPath, key_path
+            )
+            device = AdbDeviceTcpAsync(host, port, default_transport_timeout_s=10.0)
+            connected = await device.connect(rsa_keys=[signer], auth_timeout_s=10.0)
+            if not connected:
+                raise RuntimeError("dedicated logcat connection failed")
+            async for line in device.streaming_shell(
+                "logcat -v brief ActivityTaskManager:*"
+            ):
+                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 _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): Store failures and 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).
+        """
+        delay = ADB_RECONNECT_BACKOFF_START
+        while not self._stopping and not self._logcat_disabled:
+            try:
+                async for line in self._open_logcat_stream():
+                    package = foreground_from_logcat([str(line)])
+                    if package:
+                        self._on_logcat_app_start(package)
+                raise RuntimeError("stream ended (device closed)")
+            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 asyncio.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
+        if self._poller is None 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
+
     def _drain_app_label_requests(self) -> None:
         """Start at most one label fetch per watchdog cycle."""
         if not self._app_label_requests or self._poller is None:
@@ -693,6 +829,7 @@ class ShieldConnectionManager:
         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()
@@ -828,6 +965,8 @@ class ShieldConnectionManager:
                     self.store.set(FIELD_FOREGROUND, foreground)
             # 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, [])