test(adb-off audit): name the carve-outs, drive the paths the audit names (t25)

The t24 review's four observations on the t23 ADB-off audit. No production code
changes: button.py is untouched, deliberately (see O1).

O1 — THE PAIR ADB CARVE-OUT, ACCEPTED AND NAMED. The button's press builds its own
AdbMediaSessionPoller and connects without ever reading the entry's option. I
searched for a production route that reaches the press with ADB off and found
NONE, so no gate was added:
  * `button.async_setup_entry` is the ONLY place that entity is created, and it
    creates it only while the option is on (button.py:34-35);
  * every options save runs through `NvidiaShieldOptionsFlow(OptionsFlowWithReload)`,
    which reloads the entry and re-runs that setup — so turning ADB off takes the
    button away (config_flow.py:1340-1353);
  * the press is user-initiated in any case — the same class as the already
    declared `test_connect` pairing probe.
It is now a NAMED carve-out in the audit's `CARVE_OUTS` table, with what it does,
why that is safe, and the assertions behind both. The measurement is DRIVEN rather
than quoted: the press is executed with the option OFF and ON and asserted to cost
the same 1 ctor / 1 connect / 1 disconnect, so the carve-out's substance is a test
result and not prose. A meta test checks every carve-out's named assertion really
exists, so a carve-out cannot point at a phantom. MUTATION-CHECKED: making
`async_setup_entry` add the entity unconditionally fails three of the new tests.
The intent button is explicitly NOT a carve-out (it can exist for a disabled entry
and refuses through the t23 accessor gate) — stated in the same entry.

O2 — THE AUDIT NOW EXECUTES WHAT IT NAMES. `_open_logcat_stream` and the logcat
ladder are driven both ways: two new rows in the parametrized `AUDITED_PATHS` table
(so the disabled case costs zero traffic) plus dedicated assertions — the open
RAISES "ADB is switched off" in a `pytest.raises(match=...)`, and the ladder
RETURNS at once with `_logcat_failures == []` and `_logcat_disabled is False`, i.e.
a local option is not booked as a device failure. The anti-overcorrection pin is
kept in the same class: with ADB ON, five real ladder failures DO auto-disable.
MUTATION-CHECKED: removing either t23 logcat gate fails three of these tests.

O3 — THE ROTTABLE FIGURE IS GONE. The statement_blocks bullet no longer states a
count ("currently 28" was stale within one unit of work); the band (~2%) is the
policy, the current walk lives in the header the sweep prints, and the comment now
says why no band carries a count. The relative_imports bullet lost its illustrative
figure for the same reason.

O4 — THE ONE WRITE IS NAMED AS A DEBT. The audit's module docstring now states
exactly what "nothing is cached" means: the single write a disabled path can still
make is `_app_label_failed.add(package)` in `_dump_label` — a retry DEBT, not an
answer. It is pinned by a test: with ADB off the debt is recorded while
`_app_labels` stays EMPTY and the lookup still answers None; with ADB back on the
probe runs, the label is cached and the debt is settled (the discard that the code
already does).

FAIL-CLOSED, CONFIRMED AND LOCATED. The harness change (`entry=None` now reads as
ADB off) matches production because both read the option the same way: the manager
returns "disabled" for a missing option, and `__init__.py` builds no poller for the
same missing option — asserted by two new tests (an entry with `options={}`, and a
manager with no entry at all) and by a source assertion on `__init__.py`'s own
`if not entry.options.get(CONF_ADB_ENABLED, False):`. No production path is
affected: production always constructs the manager WITH the entry, and for an entry
whose option is on nothing changed at all.

No regressions: 942 passed / 5 skipped (V1), 943 / 4 (V2) — the +19 are the new
tests. ruff clean, drift exit 0, the real-HA pattern collects and runs its 3 files.
The device side remains a scripted double: the claim proven is that nothing was
sent, not what firmware would do.

tests/ha_api_drift.py

@@ -158,14 +158,20 @@ HA_CALL_OWNERS: dict[str, tuple[str, str]] = {
 #:     sweep green) is what the 2 catches. This is the project's long-standing
 #:     canary (t62 wrote it as "two below the count"), now stated once instead of
 #:     being implied by each raise.
-#:   * relative_imports: ~2.5%, currently 7. The count is dominated by package-
-#:     relative imports that legitimately churn when files move; a collapse (an
-#:     unparsed module, a stale glob) is orders of magnitude, not 7.
-#:   * statement_blocks: ~2%, currently 28. This counter moves in BULK with every
-#:     edit, which is exactly why a fixed small margin would be wrong here (the
-#:     t19 complaint: at 1300 vs 1322 a 23-statement refactor tripped it). A 2%
-#:     band leaves a normal refactor alone while a parse failure still shows up as
-#:     a large drop.
+#:   * relative_imports: ~2.5%. The count is dominated by package-relative imports
+#:     that legitimately churn when files move; a collapse (an unparsed module, a
+#:     stale glob) is orders of magnitude, not a handful.
+#:   * statement_blocks: ~2%. This counter moves in BULK with every edit, which is
+#:     exactly why a fixed small margin would be wrong here (the t19 complaint: at
+#:     1300 vs 1322 a 23-statement refactor tripped it). A 2% band leaves a normal
+#:     refactor alone while a parse failure still shows up as a large drop.
+#:
+#: No band states a COUNT: the current walk is in the header above and printed by
+#: every sweep run, so a figure in this prose can only rot — a previous revision
+#: carried exactly such a figure for statement_blocks and it was stale within one
+#: unit of work (t24/O3). The BANDS are the policy and are deliberately the only
+#: numbers here; a floor, once re-derived from a re-measured walk, is named next to
+#: the counter that owns it.
 #:   * awaited_calls: 1 (measured 6, floor 5 — DELIBERATE, see below). Stated
 #:     plainly: with six awaited HA calls a percentage margin is meaningless, so
 #:     this floor is the weakest one here. It is still worth having — a walk that

tests/test_adb_off_gates_every_device_path.py

@@ -12,12 +12,27 @@ and asserts on the TRAFFIC (shells, polls, commands, connects), because the
 traffic is the harm — a correct-looking status string is not the point. The
 device side is a scripted double that would answer everything: no real Shield is
 involved, so what is proven is that nothing was sent, not what firmware would do.
+
+WHAT "NOTHING IS CACHED" MEANS EXACTLY (t24/O4): the ONE write a disabled path can
+still make is a retry DEBT — ``_app_label_failed.add(package)`` in ``_dump_label``
+— and a debt is not an answer: the label cache (``_app_labels``) stays untouched,
+``app_label`` keeps returning None, and the debt is settled the moment the probe
+finally runs with ADB back on. Asserted in
+``test_the_only_write_while_off_is_a_retry_debt_not_an_answer``.
+
+WHAT THIS FILE DOES NOT GATE (t25/O1): the paths where "ADB off must mean no
+traffic" is deliberately NOT enforced, each listed in :data:`CARVE_OUTS` with its
+mitigation and with the test that asserts that mitigation — the Pair ADB button's
+press and ``ShieldConnectionManager.test_connect``. Naming them is the point: an
+audit that silently excludes a path is not an audit, and a carve-out whose
+assertion does not exist is prose.
 """
 
 from __future__ import annotations
 
 import asyncio
 import contextlib
+from pathlib import Path
 from types import SimpleNamespace
 
 import pytest
@@ -197,10 +212,68 @@ AUDITED_PATHS = [
     ("watchdog cycle", lambda m: m._watchdog_cycle(5.0)),
     ("heartbeat refresh", lambda m: m._refresh()),
     ("logcat ensure", lambda m: _logcat(m)),
+    ("logcat open", lambda m: _open_logcat(m)),
+    ("logcat ladder", lambda m: m._run_logcat_stream()),
     ("watchdog start", lambda m: _start_watchdog(m)),
 ]
 
 
+#: NAMED CARVE-OUTS (t25/O1) — the device paths this audit does NOT gate, each with
+#: what it does, why that is safe, and the assertion that proves the mitigation. A
+#: carve-out is a claim like every other claim here, so
+#: ``test_every_carve_out_names_a_real_assertion`` checks that the named assertion
+#: actually exists in the tree: a carve-out cannot point at a phantom.
+CARVE_OUTS = (
+    {
+        "name": "Pair ADB button press (button.py::NvidiaShieldPairAdbButton)",
+        "what_it_does": (
+            "builds its OWN AdbMediaSessionPoller and connects it, and never reads "
+            "the entry's ADB option — MEASURED ctor/connect/disconnect 1/1/1 with "
+            "the option OFF, identical to option-ON (driven, not quoted)"
+        ),
+        "why_safe": (
+            "there is no production route to the press with ADB off: "
+            "button.async_setup_entry adds that entity ONLY when the option is on, "
+            "so a disabled entry has no button to press; every options save runs "
+            "through NvidiaShieldOptionsFlow(OptionsFlowWithReload), which reloads "
+            "the entry and re-runs that setup, so turning ADB off takes the button "
+            "away; and a press is user-initiated — the same class as the "
+            "test_connect pairing probe below"
+        ),
+        "asserted_by": (
+            "tests/test_intents_registration.py",
+            "test_no_pair_adb_button_while_adb_is_off",
+        ),
+        "measured_by": (
+            "tests/test_intents_registration.py",
+            "test_the_press_does_not_consult_the_option_and_that_is_pinned",
+        ),
+        "not_the_same_as": (
+            "the INTENT button, which CAN exist for an ADB-off entry and is NOT a "
+            "carve-out: it sends through run_registered_intent, which resolves the "
+            "poller through the gated adb_poller accessor (media_player.py:2317) and "
+            "refuses with `intents_need_adb` — zero traffic"
+        ),
+    },
+    {
+        "name": "ShieldConnectionManager.test_connect (the config-flow pairing probe)",
+        "what_it_does": (
+            "opens a throwaway AdbMediaSessionPoller and connects, before any entry "
+            "exists, consulting no option at all"
+        ),
+        "why_safe": (
+            "the setup menu offers that step only while the setup's own ADB option "
+            "is on (the menu row is built from _user_data[CONF_ADB_ENABLED]), and "
+            "the probe IS the user's explicit pairing action"
+        ),
+        "asserted_by": (
+            "tests/test_adb_off_gates_every_device_path.py",
+            "test_the_pairing_probe_is_offered_only_while_the_option_is_on",
+        ),
+    },
+)
+
+
 async def _maybe_await(value):
     """Drive an audited path: some are coroutines, some are plain calls (t23)."""
     if asyncio.iscoroutine(value):
@@ -231,6 +304,16 @@ def _logcat(manager):
     manager._ensure_logcat_stream()
 
 
+def _open_logcat(manager):
+    """The logcat OPEN path: it raises while disabled, so swallow it here.
+
+    ``test_the_logcat_open_refuses_with_the_disabled_message`` asserts the message;
+    this driver only has to prove the raise cost no traffic.
+    """
+    with contextlib.suppress(RuntimeError):
+        manager._open_logcat_stream()
+
+
 async def _start_watchdog(manager):
     await manager.async_start()
     await asyncio.sleep(0.01)  # would be one cycle if the loop had started
@@ -410,3 +493,154 @@ class TestTheOtherStatusesAreUntouched:
         assert (await unbuilt.async_scan_intents(PKG)).status == (
             INTENT_SCAN_ADB_UNAVAILABLE
         )
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+
+
+def _source(relative_path: str) -> str:
+    return (REPO_ROOT / relative_path).read_text(encoding="utf-8")
+
+
+class TestTheNamedCarveOuts:
+    """Every exclusion is a NAMED claim with an assertion behind it (t25/O1)."""
+
+    @pytest.mark.parametrize("carve_out", CARVE_OUTS, ids=[c["name"] for c in CARVE_OUTS])
+    def test_every_carve_out_names_a_real_assertion(self, carve_out):
+        """No phantom carve-outs: the named assertion must exist in the tree."""
+        for key in ("asserted_by", "measured_by"):
+            pointer = carve_out.get(key)
+            if pointer is None:
+                continue
+            relative, name = pointer
+            assert f"def {name}(" in _source(relative), (
+                f"{carve_out['name']} claims {key}={relative}::{name}, which does "
+                "not exist"
+            )
+        assert carve_out["what_it_does"], "a carve-out states what it does"
+        assert carve_out["why_safe"], "a carve-out states its mitigation"
+
+    @pytest.mark.parametrize("carve_out", CARVE_OUTS, ids=[c["name"] for c in CARVE_OUTS])
+    def test_every_carve_out_is_outside_the_audited_table(self, carve_out):
+        """A carve-out is an EXCLUSION — it must not also claim to be driven."""
+        driven = {name for name, _ in AUDITED_PATHS}
+        assert all(
+            carve_out["name"] not in driven_path for driven_path in driven
+        ), carve_out["name"]
+
+    def test_the_pairing_probe_is_offered_only_while_the_option_is_on(self):
+        """`test_connect` has no entry to ask, so the SETUP MENU is its gate."""
+        config_flow = _source("custom_components/nvidia_shield/config_flow.py")
+
+        menu_guard = (
+            "if self._user_data.get(CONF_ADB_ENABLED) and not getattr(\n"
+            '            self, "_paired_adb", False\n'
+            "        ):\n"
+            '            options.append("adb_pair")'
+        )
+        assert menu_guard in config_flow, (
+            "the adb_pair row (the only route to the pairing probe) must stay "
+            "behind the setup's ADB option"
+        )
+
+
+class TestTheLogcatPathsTheAuditNames:
+    """The audit text names these two paths — so they are DRIVEN here (t25/O2)."""
+
+    async def test_the_logcat_open_refuses_with_the_disabled_message(self):
+        poller = WillingPoller()
+        manager = _manager(poller, adb_enabled=False)
+
+        with pytest.raises(RuntimeError, match="ADB is switched off"):
+            manager._open_logcat_stream()
+
+        assert poller.traffic == [], "the refusal cost no traffic"
+
+    async def test_the_logcat_ladder_returns_without_counting_a_failure(self):
+        """A local option must not look like a device failure (t23, pinned t25)."""
+        poller = WillingPoller()
+        manager = _manager(poller, adb_enabled=False)
+
+        await manager._run_logcat_stream()  # returns at once: nothing to consume
+
+        assert manager._logcat_failures == [], "no failure was recorded"
+        assert manager._logcat_disabled is False, "and the stream was not disabled"
+        assert poller.traffic == []
+
+    async def test_five_real_logcat_failures_still_auto_disable(self):
+        """THE ANTI-OVERCORRECTION PIN: the ladder itself still works when ADB is on."""
+        poller = WillingPoller()
+        manager = _manager(poller, adb_enabled=True)
+
+        async def failing_pass():
+            raise RuntimeError("stream ended (device closed)")
+
+        async def no_sleep(_seconds):
+            return None
+
+        manager._logcat_pass = failing_pass
+        manager._logcat_ladder_sleep = no_sleep
+
+        await manager._run_logcat_stream()
+
+        assert manager._logcat_disabled is True, "five real failures still trip it"
+        assert manager._entry_options_logcat() is True, "the option itself is unchanged"
+
+
+class TestTheOneWriteWhileOffIsDebt:
+    """t25/O4: "nothing is cached" is exact because the one write is a DEBT."""
+
+    async def test_the_only_write_while_off_is_a_retry_debt_not_an_answer(self):
+        poller = WillingPoller()
+        manager = _manager(poller, adb_enabled=False)
+
+        # the one disabled call that reaches the probe directly (the watchdog drain
+        # and the public lookup are gated above it, so this is the route the review
+        # measured the write on)
+        await manager._fetch_app_label(PKG)
+
+        assert manager._app_labels == {}, "no label was CACHED — that is the claim"
+        assert manager._app_label_failed == {PKG}, "only the retry DEBT was recorded"
+        assert manager.app_label(PKG) is None, "and the lookup still answers nothing"
+        assert manager.app_label_inflight == set()
+        assert poller.traffic == [], "the debt cost no traffic"
+
+        manager.entry.options[CONF_ADB_ENABLED] = True  # ADB is turned back on
+        await manager._fetch_app_label(PKG)  # the owed retry finally runs
+
+        assert manager._app_labels.get(PKG) is not None, "the probe ran and cached"
+        assert manager._app_label_failed == set(), "and the debt was settled"
+
+
+class TestAMissingOptionFailsClosed:
+    """Where the `entry=None`-means-ADB-off behaviour is asserted (t25)."""
+
+    async def test_an_entry_with_no_adb_option_is_disabled(self):
+        """No option recorded = OFF, so the harness change matched production."""
+        poller = WillingPoller()
+        entry = SimpleNamespace(options={}, data={}, title="Shield", entry_id="e1")
+        manager = ShieldConnectionManager(hass=None, entry=entry, poller=poller)
+
+        assert manager._adb_disabled() is True
+        assert manager._adb_active() is False
+        assert manager.adb_poller is None
+        assert await manager.send_media_command("play") is False
+        assert await manager.async_refresh_installed_apps(force=True) == []
+        assert manager.installed_apps() == []
+        assert poller.traffic == []
+
+    async def test_a_manager_with_no_entry_at_all_is_disabled_too(self):
+        """The old harness shape (`entry=None`) is the same fact."""
+        poller = WillingPoller()
+        manager = ShieldConnectionManager(hass=None, entry=None, poller=poller)
+
+        assert manager._adb_disabled() is True
+        assert await manager.send_digit_key("5") is False
+        assert poller.traffic == []
+
+    def test_the_production_default_is_read_the_same_way(self):
+        """`__init__.py` gates the poller on the same missing-option default."""
+        entry_source = _source("custom_components/nvidia_shield/__init__.py")
+
+        assert "if not entry.options.get(CONF_ADB_ENABLED, False):" in entry_source, (
+            "production treats a missing option as OFF, exactly like the manager"
+        )

tests/test_intents_registration.py

@@ -38,6 +38,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parent))
 import entity_stublib  # noqa: E402
 from entity_stublib import load_entity_modules  # noqa: E402
 from nsh.const import (  # noqa: E402
+    CONF_ADB_ENABLED,
     CONF_REGISTERED_INTENTS,
     INTENT_FIELD_ACTION,
     INTENT_FIELD_ACTIVITY,
@@ -1370,3 +1371,164 @@ def test_a_truncated_capture_prefills_no_url(modules):
     assert "www.exam" not in values[INTENT_FIELD_DATA]
     assert values[INTENT_FIELD_ACTION] == "android.intent.action.VIEW"
     assert values[INTENT_FIELD_ACTIVITY] == "com.netflix.ninja/.MainActivity"
+
+# ── t25/O1: the Pair ADB BUTTON is a NAMED carve-out of the ADB-off audit ──
+
+
+class TestThePairAdbCarveOut:
+    """Why the ADB-off audit deliberately does NOT gate the Pair ADB press.
+
+    `NvidiaShieldPairAdbButton.async_press` builds its OWN poller and connects —
+    it never asks the entry's ADB option (the t24 review measured
+    ctor/connect/disconnect 1/1/1 with the option OFF, identical to option-ON).
+    That is accepted, not overlooked, because there is no production route to the
+    press with ADB off:
+
+    * `button.async_setup_entry` adds that entity ONLY for an entry whose option
+      is on, so a disabled entry has NO button to press (asserted below);
+    * every options save goes through `NvidiaShieldOptionsFlow(OptionsFlowWithReload)`,
+      which reloads the entry — and the reload re-runs that setup, so turning ADB
+      off takes the button away;
+    * the press itself is USER-INITIATED (somebody had to press it), which is the
+      same class as the `ShieldConnectionManager.test_connect` pairing probe.
+
+    If a future change ever makes that entity exist for a disabled entry, this
+    class fails — which is the point: the carve-out is a claim like any other.
+    """
+
+    async def _setup(self, modules, options):
+        import importlib
+
+        entry = SimpleNamespace(
+            entry_id="e1", options=options, title="Shield", runtime_data=None
+        )
+        added: list = []
+        button = importlib.import_module("nsh.button")
+        await button.async_setup_entry(None, entry, added.extend)
+        return added
+
+    @staticmethod
+    def _pair_adb(added):
+        return [entity for entity in added if getattr(entity, "_attr_name", None) == "Pair ADB"]
+
+    async def test_no_pair_adb_button_while_adb_is_off(self, modules):
+        """THE MITIGATION: a disabled entry has no button to press."""
+        added = await self._setup(modules, {})
+
+        assert self._pair_adb(added) == []
+        assert all(
+            not getattr(entity, "_attr_name", "").startswith("Pair")
+            for entity in added
+        )
+
+    async def test_no_pair_adb_button_with_an_explicit_off_option(self, modules):
+        added = await self._setup(modules, {CONF_ADB_ENABLED: False})
+
+        assert self._pair_adb(added) == []
+
+    async def test_the_button_is_back_when_adb_is_on(self, modules):
+        """The anti-overcorrection pin: the entity is not gone for good (t25)."""
+        added = await self._setup(modules, {CONF_ADB_ENABLED: True})
+
+        pair = self._pair_adb(added)
+        assert len(pair) == 1
+        assert pair[0]._attr_unique_id.endswith("_pair_adb")
+
+    async def test_the_press_does_not_consult_the_option_and_that_is_pinned(
+        self, modules
+    ):
+        """THE MEASUREMENT, executed: identical traffic whichever way the option reads.
+
+        `async_press` never asks the entry's ADB option, so a press costs one
+        constructor, one connect and one disconnect with the option OFF — the same
+        as option-ON (the t24 review measured 1/1/1 both ways). This is the
+        carve-out's substance, so it is DRIVEN rather than quoted: if a future
+        change adds a gate here, this fails and the carve-out entry (and the audit
+        text beside it) must be updated in the same change — which is the point.
+        """
+        import importlib
+
+        button = importlib.import_module("nsh.button")
+
+        class RecordingPoller:
+            constructed = 0
+            connects = 0
+            disconnects = 0
+
+            def __init__(self, host, port, key_path):
+                type(self).constructed += 1
+                self.host = host
+
+            async def connect(self):
+                type(self).connects += 1
+                return True
+
+            async def disconnect(self):
+                type(self).disconnects += 1
+
+        class NoReload:
+            async def async_reload(self, entry_id):
+                return True
+
+        original_poller = button.AdbMediaSessionPoller
+        original_key = button.ensure_adb_key
+        button.AdbMediaSessionPoller = RecordingPoller
+        button.ensure_adb_key = lambda config_dir: "/tmp/test-key.pem"
+        try:
+            measured = {}
+            for label, adb_enabled in (("off", False), ("on", True)):
+                RecordingPoller.constructed = 0
+                RecordingPoller.connects = 0
+                RecordingPoller.disconnects = 0
+                entry = SimpleNamespace(
+                    entry_id="e1",
+                    options={
+                        CONF_ADB_ENABLED: adb_enabled,
+                        "host": "shield.invalid",
+                    },
+                    title="Shield",
+                    runtime_data=None,
+                )
+                entity = button.NvidiaShieldPairAdbButton(None, entry)
+                entity.hass = SimpleNamespace(
+                    config=SimpleNamespace(config_dir="/tmp"),
+                    async_add_executor_job=FakeHass.async_add_executor_job,
+                    config_entries=NoReload(),
+                )
+                await entity.async_press()
+                measured[label] = (
+                    RecordingPoller.constructed,
+                    RecordingPoller.connects,
+                    RecordingPoller.disconnects,
+                )
+        finally:
+            button.AdbMediaSessionPoller = original_poller
+            button.ensure_adb_key = original_key
+
+        assert measured["off"] == (1, 1, 1), measured["off"]
+        assert measured["on"] == (1, 1, 1), measured["on"]
+        assert measured["off"] == measured["on"], (
+            "the option does not gate this path — that is the named carve-out"
+        )
+
+    async def test_the_reload_that_removes_it_is_the_options_flow(self):
+        """The second half of the mitigation, pinned where it lives."""
+        source = (
+            Path(__file__).resolve().parent.parent
+            / "custom_components"
+            / "nvidia_shield"
+            / "config_flow.py"
+        ).read_text(encoding="utf-8")
+
+        assert "OptionsFlowWithReload" in source, (
+            "an options save must reload the entry (which re-runs button setup)"
+        )
+        button_source = (
+            Path(__file__).resolve().parent.parent
+            / "custom_components"
+            / "nvidia_shield"
+            / "button.py"
+        ).read_text(encoding="utf-8")
+        assert "if entry.options.get(CONF_ADB_ENABLED, False):" in button_source, (
+            "the Pair ADB entity is created only while the option is on"
+        )