diff --git a/README.md b/README.md
index 4cb6e55..27fa24c 100644
--- a/README.md
+++ b/README.md
@@ -419,6 +419,14 @@ already submitted. The two pairing rows appear only while that transport still n
 pairing (a skip leaves its row there to retry; a successful pairing removes it), and
 the menu's description names what is still to pair.
 
+Turning **Add the Nvidia Remote Protocol** on in the **Main setup** row of the
+*options* page does not silently enable a client that cannot connect: an entry that has
+never completed a pairing is taken to the same **Pair the remote protocol** step, and
+its **Continue without pairing** escape turns the option back off. Pairing is remembered
+per entry, so switching the transport off and on again on a paired entry asks for no new
+PIN (the certificate from the first pairing is reused — see
+[is my transport paired?](#is-my-protocol-transport-paired-connected-or-not-running-at-all)).
+
 ### The Apps table
 
 The **Apps** row opens a small editor rather than one long form. Home Assistant has no
@@ -767,6 +775,88 @@ All four variables are always present, including their defaults, so a script can
 on the shape. Unset — the default — nothing changes: keys go out over the protocol
 exactly as before.
 
+### The remote entity: the app / activity picker
+
+The `remote` entity speaks Home Assistant's own remote contract, matching the
+`androidtv_remote` entity it replaces — so a replaced setup keeps its automations
+**and** gets the app picker the frontend only renders for a remote that declares it:
+
+| Piece | Value |
+| --- | --- |
+| `supported_features` | includes `RemoteEntityFeature.ACTIVITY` — without that flag Home Assistant offers **no activity picker at all**, which is why the remote card can look empty |
+| `activity_list` | the friendly app names, in your order: your **Apps table** (or the built-in table when you have none), plus the apps **ADB discovery** actually found on the device |
+| `current_activity` | the app on screen as a friendly name: the protocol's current-app push when it arrives, the ADB foreground reading otherwise. `None` while nothing is known — no activity is ever invented |
+| `turn_on` with `activity:` | wakes the Shield when it reports itself off, then launches that app over the protocol |
+
+```yaml
+# launch Netflix by the name from the activity list / Apps table
+action: remote.turn_on
+target:
+  entity_id: remote.living_room
+data:
+  activity: Netflix
+```
+
+An `activity` resolves in this order: a **name** from the activity list
+(case-insensitive), then a **package id** or a **detail link** (`https://…`,
+`market://…`) passed through unchanged — that is how an app your table does not carry
+is still launchable, and what the other integration does with a name it does not know.
+Anything else raises a clear error naming the value instead of silently doing nothing.
+Leaving `activity` out (or empty) keeps the old behaviour exactly: wake through the
+configured script, else `POWER` when the device reports it is off.
+
+**Where the list comes from, and why.** `androidtvremote2` exposes **no app-list API
+at all** (measured against the installed library) — which is exactly why
+`androidtv_remote` keeps a user-maintained app map. This integration has the same kind
+of source *plus* something the other one lacks: ADB discovery, so the picker can offer
+what is actually installed and name it through the one chain (your Apps table → the
+device's own label → the built-in table → the package id). Apps nothing can name are
+not listed as bare package ids; pass their id to `activity` if you need them.
+
+#### Is my protocol transport paired, connected, or not running at all?
+
+The library offers no `is_paired` / `is_connected`, so the remote entity publishes what
+can be **proven** (Developer Tools > States, or
+`state_attr('remote.living_room', 'protocol_state')`):
+
+| Attribute | Meaning |
+| --- | --- |
+| `protocol_state` | `off` (no transport configured), `never_connected`, `connected`, `disconnected`, or `auth_rejected` |
+| `protocol_connected` | `true` only while the device reports the protocol session up |
+| `protocol_ever_authenticated` | `true` once a session connected **or any push arrived** — the only proof of a paired client the library gives |
+| `protocol_current_app_seen` / `protocol_is_on_seen` / `protocol_volume_seen` | whether each device push has **ever** arrived (sticky: the state cache expires, these do not) |
+| `protocol_last_error` / `protocol_failures` | the last protocol connect/send failure verbatim, and how many sends could **not** run |
+| `protocol_certificate` | `present` / `missing` — the certificate from your setup pairing, which is why enabling the transport on a paired entry asks for **no new PIN** |
+| `power_on_route` / `power_off_route` | `protocol`, or `script:<script>` when a command script owns that command |
+| `activity_source` (and the media player's `app_source`) | which tier produced the app identity: `protocol` (the device's own push) or `adb` (the dumpsys parse) — the push is preferred, and it is never a silent override |
+
+Reading it:
+
+* **`auth_rejected`** — the device refused our certificate (a factory reset, or a
+  deleted `.storage/nvidia_shield/client.pem`). Home Assistant also flags the entry
+  for reauthentication; that flow re-pairs.
+* **`never_connected`** while the transport is enabled — switched on but with no
+  session. Pairing again is the answer (the options flow routes an unpaired entry
+  straight into the pairing step rather than enabling a client that cannot connect).
+* **`off`** — the transport is not configured. Everything else here still works: ADB
+  is the state source, so an ADB-only setup degrades **only** the protocol surface.
+* A send that could not run is **never silent**: it is counted, recorded verbatim in
+  `protocol_last_error`, and logged at WARNING together with `protocol_state`.
+
+**The other Android TV integration.** `androidtv_remote` pairs **its own client** and
+shares nothing with us: its certificate lives in its own storage and its entities are
+its own. Ours (`client.pem` under `.storage/nvidia_shield/`) is reused between pairing
+and runtime by design, so *not* being asked for a PIN on an already-paired entry is
+expected — and it says nothing about whether the session is up. `protocol_state`
+answers that.
+
+**A configured script bypasses the protocol.** When a command script owns `turn_on`
+(or `turn_off`), that script runs and the protocol is **never touched** — so a power
+press that worked through it is *not* evidence that the transport is paired, or even
+connected (`power_on_route` reads `script:…`). A configured **Remote command script**
+is the same story for `remote.send_command`. Only a command that went out over the
+protocol, with `protocol_state` = `connected`, exercises the transport.
+
 ## Entities
 
 | Entity | Type | Description |
@@ -774,6 +864,7 @@ exactly as before.
 | Media Player | `media_player` | Main entity with playback controls, metadata and the flat `now_playing_*` attributes (below) |
 | Current app (`sensor.current_app`) | `sensor` | Currently running app |
 | Now playing (`sensor.now_playing`) | `sensor` | Formatted now-playing string, with its own flat attributes (below) |
+| Remote | `remote` | Key commands, the activity/app picker, and the transport-state attributes (see [the activity picker](#the-remote-entity-the-app--activity-picker)) |
 | Pair ADB | `button` | Initiate or re-pair ADB connection |
 | Screen capture (`image.screen_capture`) | `image` | The Shield's current screen — **opt-in, off by default**, manual/diagnostic only (see [Screen capture](#screen-capture-opt-in-manual)) |
 
diff --git a/custom_components/nvidia_shield/config_flow.py b/custom_components/nvidia_shield/config_flow.py
index e3d879f..5964288 100644
--- a/custom_components/nvidia_shield/config_flow.py
+++ b/custom_components/nvidia_shield/config_flow.py
@@ -28,6 +28,7 @@ from .const import (
     CONF_PAIRING_CODE,
     CONF_POWER_ENTITY,
     CONF_PROTOCOL_ENABLED,
+    CONF_PROTOCOL_PAIRED,
     CONF_RECEIVER_ENTITY,
     CONF_RECEIVER_INPUT,
     CONF_REMOTE_COMMAND_SCRIPT,
@@ -338,6 +339,67 @@ def _build_selectors():
     }
 
 
+async def _async_start_protocol_pairing(
+    hass, host: str, client_name: str
+) -> tuple[object | None, str | None]:
+    """Build a throwaway pairing client and make the Shield show its PIN (t15).
+
+    Certificates live under ``.storage/<domain>/`` (core androidtv_remote helpers.py
+    pattern) and are SHARED between pairing and runtime — an entry paired later reuses
+    them, which is why enabling the transport on a paired entry asks for no new PIN.
+    Returns ``(client, None)`` or ``(None, error_key)``: no host, a missing
+    ``androidtvremote2`` and an unreachable Shield are all keys the form can show,
+    never an exception out of a step.
+
+    t44 F1: androidtvremote2 raises ``CannotConnect`` from both calls below, and the
+    protocol transport is ON by default, so an asleep, rebooting or mis-addressed
+    Shield is the ordinary case rather than an edge one. Left uncaught it escaped the
+    config flow entirely, and the protocol_pair form — the address field AND the
+    "continue without pairing" escape — never rendered.
+
+    Module-level (t15) because BOTH flows drive a pairing session: the setup flow on
+    its way in, and the OPTIONS flow when the transport is switched on for an entry
+    that never paired.
+    """
+    if not host:
+        return None, "protocol_no_host"
+    certfile, keyfile = await hass.async_add_executor_job(protocol_cert_paths, hass)
+    client = ShieldConnectionManager.build_protocol_client(
+        host,
+        client_name=client_name,
+        certfile=certfile,
+        keyfile=keyfile,
+    )
+    if client is None:
+        # build_protocol_client already logged the missing library; pairing cannot
+        # start, and the form still offers the skip path
+        return None, "protocol_unavailable"
+    try:
+        await client.async_generate_cert_if_missing()
+        await client.async_start_pairing()
+    except Exception as err:  # noqa: BLE001 - the library's error IS the finding
+        _LOGGER.warning(
+            "protocol pairing could not start at %s (%s: %s) — showing the "
+            "form so the address can be corrected or pairing skipped",
+            host,
+            type(err).__name__,
+            err,
+        )
+        _disconnect_pairing_client(client)
+        return None, "protocol_unreachable"
+    return client, None
+
+
+def _disconnect_pairing_client(client) -> None:
+    """Drop a THROWAWAY pairing client (t15) — both flows hold exactly one.
+
+    A pairing client never becomes the entry's runtime transport, so a session whose
+    address was corrected, or that was skipped, must be closed here rather than left
+    dangling with a live TLS connection.
+    """
+    if client is not None:
+        with contextlib.suppress(Exception):
+            client.disconnect()
 class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
     """Handle a config flow for Nvidia Shield."""
 
@@ -542,6 +604,10 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
                 CONF_HOST: d.get(CONF_HOST, ""),
                 CONF_ADB_PORT: d.get(CONF_ADB_PORT, DEFAULT_ADB_PORT),
                 CONF_PROTOCOL_ENABLED: d.get(CONF_PROTOCOL_ENABLED, True),
+                # t15: the pairing marker travels with the entry: the pairing step set
+                # it (True) or the skip path cleared it (False), and the options flow
+                # reads it before it lets the transport run unpaired.
+                CONF_PROTOCOL_PAIRED: bool(d.get(CONF_PROTOCOL_PAIRED, False)),
             },
         )
 
@@ -563,57 +629,20 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
         """Drop a pairing client whose session became void (host corrected)."""
         client = self._protocol_client
         self._protocol_client = None
-        if client is not None:
-            with contextlib.suppress(Exception):
-                client.disconnect()
+        _disconnect_pairing_client(client)
 
     async def _start_protocol_pairing(self):
-        """Build a throwaway protocol client and make the Shield show its PIN.
+        """Build a throwaway protocol client and make the Shield show its PIN (t15).
 
-        Certificates live under ``.storage/<domain>/`` (core androidtv_remote
-        helpers.py pattern) and are shared between pairing and runtime. Returns
-        ``(client, None)`` or ``(None, error_key)`` — no host, a missing
-        ``androidtvremote2`` and an unreachable Shield are all keys the form can
-        show, never an exception out of the step.
-
-        t44 F1: androidtvremote2 raises ``CannotConnect`` from both calls below,
-        and the protocol transport is ON by default, so an asleep, rebooting or
-        mis-addressed Shield is the ordinary case rather than an edge one. Left
-        uncaught it escaped the config flow entirely, and the protocol_pair form
-        — the address field AND the "continue without pairing" escape — never
-        rendered.
+        Thin wrapper: the construction is module-level because the OPTIONS flow needs
+        exactly the same thing when the transport is switched on for an entry that
+        never paired (``_async_start_protocol_pairing``).
         """
-        host = self._host()
-        if not host:
-            return None, "protocol_no_host"
-        certfile, keyfile = await self.hass.async_add_executor_job(
-            protocol_cert_paths, self.hass
+        return await _async_start_protocol_pairing(
+            self.hass,
+            self._host(),
+            self._user_data.get("name", DEFAULT_NAME) or DEFAULT_NAME,
         )
-        client = ShieldConnectionManager.build_protocol_client(
-            host,
-            client_name=self._user_data.get("name", DEFAULT_NAME) or DEFAULT_NAME,
-            certfile=certfile,
-            keyfile=keyfile,
-        )
-        if client is None:
-            # build_protocol_client already logged the missing library;
-            # pairing cannot start, and the form still offers the skip path
-            return None, "protocol_unavailable"
-        try:
-            await client.async_generate_cert_if_missing()
-            await client.async_start_pairing()
-        except Exception as err:  # noqa: BLE001 - the library's error IS the finding
-            _LOGGER.warning(
-                "protocol pairing could not start at %s (%s: %s) — showing the "
-                "form so the address can be corrected or pairing skipped",
-                host,
-                type(err).__name__,
-                err,
-            )
-            with contextlib.suppress(Exception):
-                client.disconnect()
-            return None, "protocol_unreachable"
-        return client, None
 
     async def _async_begin_protocol_pairing(self):
         """Start a PIN pairing session: ``(client, None)`` or ``(None, error)``."""
@@ -647,6 +676,9 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
         """
         self._discard_protocol_client()
         self._user_data[CONF_PROTOCOL_ENABLED] = False
+        # t15: skipped means NOT paired — the marker is what keeps an unpaired client
+        # off instead of letting it look like a dead remote later.
+        self._user_data[CONF_PROTOCOL_PAIRED] = False
         if self._reauth_entry is not None:
             # an existing entry cannot be re-created: leave it untouched and
             # say so instead of pretending the re-pair worked
@@ -735,6 +767,12 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
         # t56: the protocol is paired now, so the setup menu must not offer to
         # pair it again — it returns to the menu (Receiver / Connection /
         # Finish) instead of creating the entry here.
+        #
+        # t15: and the pairing is PERSISTED. Nothing else can tell a paired entry
+        # from one that merely switched the transport on (the library has no
+        # is_paired, and certificate material is generated either way), so the
+        # options flow reads this marker before it lets the transport run unpaired.
+        self._user_data[CONF_PROTOCOL_PAIRED] = True
         self._paired_protocol = True
         return self._show_setup_menu()
 
@@ -851,6 +889,9 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
             "options": {
                 **entry.options,
                 CONF_PROTOCOL_ENABLED: True,
+                # t15: this reconfigure JUST paired — persist the marker the options
+                # flow reads before it lets an unpaired transport run.
+                CONF_PROTOCOL_PAIRED: True,
                 CONF_HOST: self._host(),
             },
         }
@@ -874,6 +915,8 @@ class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
             options={
                 **entry.options,
                 CONF_PROTOCOL_ENABLED: True,
+                # t15: this reauth JUST paired — persist the marker (see above).
+                CONF_PROTOCOL_PAIRED: True,
                 CONF_HOST: self._host(),
             },
         )
@@ -1194,6 +1237,9 @@ class NvidiaShieldOptionsFlow(config_entries.OptionsFlowWithReload):
         self._discover_note: str | None = None
         self._editing_package: str | None = None
         self._removing_package: str | None = None
+        # t15: the throwaway pairing client of a protocol_pair visit (see
+        # async_step_protocol_pair). Flow state, never an option.
+        self._protocol_client = None
 
     @property
     def _options_dict(self) -> dict:
@@ -1416,9 +1462,118 @@ class NvidiaShieldOptionsFlow(config_entries.OptionsFlowWithReload):
         """Main setup: the link, both transports and the linked entities."""
         errors: dict = {}
         if user_input is not None and self._apply_submission(dict(user_input), errors):
+            if self._protocol_enable_needs_pairing(user_input):
+                # t15: switching the transport ON for an entry that was never paired
+                # must PAIR now, not enable an unauthenticated client. The setup flow
+                # always paired (or turned the transport off); this is the way back in
+                # for an entry created in passive mode, and the only honest reading of
+                # "pair or stay off".
+                return await self.async_step_protocol_pair()
             return await self.async_step_menu_again()
         return self._show_connection_form(errors)
 
+    def _protocol_paired(self) -> bool:
+        """True when this entry's protocol client was really paired (t15).
+
+        The persisted marker is the ONLY honest source: the library exposes no
+        ``is_paired``, and certificate material proves only that something generated a
+        key — which an entry that merely enabled the transport has too (the runtime
+        calls ``async_generate_cert_if_missing``).
+        """
+        data = getattr(self.config_entry, "data", None) or {}
+        return bool(self._options_dict.get(CONF_PROTOCOL_PAIRED) or data.get(CONF_PROTOCOL_PAIRED))
+
+    def _protocol_enable_needs_pairing(self, submission: dict) -> bool:
+        """Does this submission turn the transport ON without a pairing to back it?"""
+        if CONF_PROTOCOL_ENABLED not in submission:
+            return False
+        if not self._options_dict.get(CONF_PROTOCOL_ENABLED):
+            return False  # turned off (or left off): nothing runs, nothing to pair
+        return not self._protocol_paired()
+
+    async def async_step_protocol_pair(self, user_input=None):
+        """Pair the remote protocol from the OPTIONS flow (t15).
+
+        Reached only by switching the transport ON for an entry that never paired. The
+        alternative is the state the user was stuck in: a client that connects nowhere,
+        no pushes, every send swallowed — a remote that looks unpaired whether it is or
+        not. ``skip_pairing`` leaves the transport OFF (persisted), so the entry cannot
+        end up enabled-but-unpaired either way.
+        """
+        errors: dict = {}
+
+        if user_input is None:
+            return await self._async_restart_options_pairing(errors)
+
+        if user_input.get(CONF_SKIP_PAIRING):
+            return await self._async_skip_options_pairing()
+
+        typed_host = str(user_input.get(CONF_HOST) or "").strip()
+        if typed_host and typed_host != entry_host(self.config_entry):
+            # A host typed HERE is authoritative (same rule as the setup flow): the
+            # session started against the old address is dropped and the Shield is
+            # asked for a NEW code at the corrected address.
+            self._options_dict[CONF_HOST] = typed_host
+            self._discard_options_pairing_client()
+
+        code = str(user_input.get(CONF_PAIRING_CODE, "") or "").strip()
+        if self._protocol_client is None or not code:
+            return await self._async_restart_options_pairing(errors)
+
+        try:
+            await self._protocol_client.async_finish_pairing(code)
+        except Exception:
+            errors["base"] = "protocol_pair_failed"
+            return self._show_options_pair_form(errors)
+
+        self._discard_options_pairing_client()
+        # Paired: the transport may run, and the marker keeps every later visit from
+        # asking again (the certificate is reused by design — no new PIN).
+        self._options_dict[CONF_PROTOCOL_PAIRED] = True
+        return await self.async_step_menu_again()
+
+    def _show_options_pair_form(self, errors=None):
+        """protocol_pair form for the OPTIONS flow: PIN, address, skip escape."""
+        host = entry_host(self.config_entry)
+        return self.async_show_form(
+            step_id="protocol_pair",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_PAIRING_CODE): str,
+                    _host_field_key(CONF_HOST, host): str,
+                    vol.Optional(CONF_SKIP_PAIRING, default=False): bool,
+                }
+            ),
+            errors=errors or {},
+        )
+
+    async def _async_restart_options_pairing(self, errors: dict):
+        """Start a fresh pairing session and render the form for its code (t15)."""
+        self._discard_options_pairing_client()
+        client, error = await _async_start_protocol_pairing(
+            self.hass,
+            entry_host(self.config_entry),
+            self.config_entry.title or DEFAULT_NAME,
+        )
+        if error:
+            errors["base"] = error
+        else:
+            self._protocol_client = client
+        return self._show_options_pair_form(errors)
+
+    async def _async_skip_options_pairing(self):
+        """Continue with the transport OFF (t15) — never enabled-but-unpaired."""
+        self._discard_options_pairing_client()
+        self._options_dict[CONF_PROTOCOL_ENABLED] = False
+        self._options_dict[CONF_PROTOCOL_PAIRED] = False
+        return await self.async_step_menu_again()
+
+    def _discard_options_pairing_client(self) -> None:
+        """Drop the options flow's throwaway pairing client (t15)."""
+        client = self._protocol_client
+        self._protocol_client = None
+        _disconnect_pairing_client(client)
+
     def _show_connection_form(self, errors=None):
         """Main setup: the same settings the first setup form carries.
 
diff --git a/custom_components/nvidia_shield/connection.py b/custom_components/nvidia_shield/connection.py
index cc95097..3d8bdc0 100644
--- a/custom_components/nvidia_shield/connection.py
+++ b/custom_components/nvidia_shield/connection.py
@@ -49,6 +49,12 @@ from .const import (
     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,
 )
@@ -121,6 +127,18 @@ def volume_tuple(info) -> tuple | None:
         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.
 
@@ -278,6 +296,18 @@ class ShieldConnectionManager:
         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
+        self._protocol_ever_authenticated = False
+        self._protocol_failures = 0
 
     # ── One-shot connection test (pairing UX) ──
 
@@ -369,10 +399,84 @@ class ShieldConnectionManager:
         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)
-        # One owner per transport: the library's keep_reconnecting() creates
-        # and owns its reconnect task internally — we never wrap or duplicate
-        # it (the stop path closes the client, which tears the task down).
-        client.keep_reconnecting(self._on_invalid_auth)
+        # 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"
+        )
+
+    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).
+                _LOGGER.info(
+                    "Shield remote-protocol transport connected to %s",
+                    getattr(client, "host", self.entry.title if self.entry else "?"),
+                )
+                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))
@@ -386,16 +490,27 @@ class ShieldConnectionManager:
     def _on_volume_info(self, info) -> None:
         self._protocol_push(FIELD_VOLUME_INFO, volume_tuple(info))
 
-    def _on_invalid_auth(self) -> None:
+    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 certificated auth is gone — the entry-level callback (reauth
+        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"
+            "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:
@@ -408,7 +523,17 @@ class ShieldConnectionManager:
 
         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:
+            self._protocol_state = (
+                PROTOCOL_CONNECTED if value else PROTOCOL_DISCONNECTED
+            )
         self.store.set(field, value)
         loop = self._protocol_loop
         if loop is not None and loop.is_running():
@@ -459,6 +584,75 @@ class ShieldConnectionManager:
         """
         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 _note_protocol_state(self, state: str, error: str | None) -> None:
+        """Record the transport state (and an error when there is one) + notify."""
+        changed = state != self._protocol_state
+        self._protocol_state = state
+        if error is not None:
+            self._protocol_last_error = error
+        if changed:
+            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).
 
@@ -501,6 +695,22 @@ class ShieldConnectionManager:
         (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
@@ -569,14 +779,19 @@ class ShieldConnectionManager:
         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:
-            _LOGGER.debug("protocol send_text failed: %s", err)
+            self._note_protocol_send_failure("send_text", err)
             return False
 
     async def send_adb_input_text(self, text: str) -> bool:
@@ -1348,16 +1563,19 @@ class ShieldConnectionManager:
     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 is down or the send raises — entities
-        then fall back to the ADB keyevent path.
+        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:
-            _LOGGER.debug("Protocol key %s failed: %s", key, 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:
@@ -1365,14 +1583,21 @@ class ShieldConnectionManager:
 
         ``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:
-            _LOGGER.debug("Protocol launch %s failed: %s", app_link_or_app_id, 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:
diff --git a/custom_components/nvidia_shield/const.py b/custom_components/nvidia_shield/const.py
index 6b284a1..c5b362f 100644
--- a/custom_components/nvidia_shield/const.py
+++ b/custom_components/nvidia_shield/const.py
@@ -37,6 +37,15 @@ CONF_PAIRING_CODE = "pairing_code"
 # steps — the entry still finishes (passive mode) instead of stranding the
 # user on a form with no way forward.
 CONF_SKIP_PAIRING = "skip_pairing"
+#: t15: a PERSISTED marker that this entry's protocol client really paired (a
+#: successful ``async_finish_pairing``). Pairing cannot be inferred from anything
+#: else: the library exposes no ``is_paired``, and certificate material proves only
+#: that SOMETHING generated a key — which an entry that merely enabled the transport
+#: without pairing has too. Written by the setup flow's pairing step, the options
+#: flow's, and the reauth/reconfigure finishers, so enabling the transport later can
+#: tell "already paired" (reuse the certificate, no new PIN) from "never paired"
+#: (must pair now, or stay off).
+CONF_PROTOCOL_PAIRED = "protocol_paired"
 DEFAULT_PROTOCOL_API_PORT = 6466
 DEFAULT_PROTOCOL_PAIR_PORT = 6467
 
@@ -52,6 +61,22 @@ ADB_RECONNECT_BACKOFF_CAP = 300   # seconds; backoff never sleeps longer
 ADB_RECONNECT_JITTER = 0.25       # ±25 % randomization against reconnect storms
 STATE_STORE_TTL = 60              # ADB-sourced fields older than this are stale
 
+# t15: the remote-protocol transport's own state machine. The library offers NO
+# ``is_paired``/``is_connected`` (measured on androidtvremote2), so these are the
+# states we can actually prove — and the one the user was reading as "maybe
+# unpaired" was ``never_connected``: keep_reconnecting() alone never connects
+# (its loop is ``while self._remote_message_protocol:``, None until a first
+# ``async_connect()``), so the transport had never been switched on at all.
+PROTOCOL_OFF = "off"                          # no client wired (disabled/no library)
+PROTOCOL_NEVER_CONNECTED = "never_connected"  # wired, no session yet, no rejection
+PROTOCOL_CONNECTED = "connected"              # device reachable at the protocol
+PROTOCOL_DISCONNECTED = "disconnected"        # session was up, now reported down
+PROTOCOL_AUTH_REJECTED = "auth_rejected"      # cert refused: pair again (reauth)
+
+#: Seconds allowed for the FIRST protocol connect before it counts as unreachable.
+#: Core androidtv_remote uses the same 5 s before deciding ready/not-ready.
+PROTOCOL_CONNECT_TIMEOUT = 5.0
+
 DEFAULT_NAME = "Nvidia Shield"
 
 # Config-entry schema version (t41): v1 → v2 added the remote-protocol fields,
diff --git a/custom_components/nvidia_shield/media_player.py b/custom_components/nvidia_shield/media_player.py
index 45558ef..2854e89 100644
--- a/custom_components/nvidia_shield/media_player.py
+++ b/custom_components/nvidia_shield/media_player.py
@@ -418,6 +418,91 @@ def resolve_app_name(
     )
 
 
+def normalize_app_directory(directory) -> dict[str, str]:
+    """The t51 rule for an apps table: strip the edges, drop blanks, keep order.
+
+    ONE implementation for every consumer (the media player's source list, the
+    remote's activity list, the reverse lookup), so display and match key stay one
+    string: an entry like ``com.x=  TiviMate  `` shows as "TiviMate" AND resolves as
+    "TiviMate". An entry whose name — or whose package — is blank is dropped, so no
+    surface offers a child with no package behind it or a name no launch can reach.
+    Two packages normalizing to the same name are BOTH KEPT (each stays reachable by
+    its own id) and the FIRST in map order wins the name (t53).
+    """
+    normalized: dict[str, str] = {}
+    for package, name in (directory or {}).items():
+        key = str(package).strip()
+        label = str(name).strip()
+        if not key or not label:
+            continue
+        normalized[key] = label
+    return normalized
+
+
+def app_directory(options) -> dict[str, str]:
+    """Launchable apps for an entry: the user's map, else the curated default (t61).
+
+    ONE source for the source list and the reverse lookup, so the dropdown can
+    never offer a name the launch path cannot resolve (``_app_directory``, and now
+    the remote's activity list, both read this).
+    """
+    return normalize_app_directory(
+        (options or {}).get(CONF_APP_NAME_MAP) or DEFAULT_APP_DIRECTORY
+    )
+
+
+def distinct_app_names(names) -> list[str]:
+    """Each name ONCE, compared case-insensitively, keeping the first spelling (t53).
+
+    ``Dup``/``dup`` would otherwise show two rows while the case-insensitive
+    reverse lookup resolves both to the first entry — one row can only launch the
+    app it shows. The shadowed package is not lost: it stays reachable by id.
+    """
+    seen: set[str] = set()
+    offered: list[str] = []
+    for name in names:
+        text = str(name)
+        folded = text.casefold()
+        if not text or folded in seen:
+            continue
+        seen.add(folded)
+        offered.append(text)
+    return offered
+
+
+def activity_names(options, discovered=()) -> list[str]:
+    """The remote entity's ACTIVITY list: friendly names, stable order (t15).
+
+    Core ``androidtv_remote`` builds its list from a user-maintained app map
+    because ``AndroidTVRemote`` exposes NO app-list API (measured on the installed
+    library, not assumed). Ours is the same shape of source plus what ADB discovery
+    actually found on THIS device:
+
+    1. the entry's app directory (the user's map, else the curated table) in map
+       order — the same names the media player's source list offers;
+    2. apps discovery found that no directory row covers, named by the one chain
+       (user map → device label → curated → id).
+
+    A discovered app NOTHING can name is skipped rather than offered as its own
+    package id (the ``pkg=pkg`` noise this contract forbids); it stays launchable by
+    id through ``remote.turn_on(activity=...)``. Nothing is invented here: an empty
+    directory with no discovery yields an empty list, and the protocol's pairing
+    state cannot add an entry.
+    """
+    directory = app_directory(options)
+    names = list(directory.values())
+    covered = set(directory)
+    for row in discovered or ():
+        package = str(row.get("package") or "").strip()
+        if not package or package in covered:
+            continue
+        covered.add(package)
+        label = str(row.get("name") or "").strip()
+        if not label or label == package:
+            continue
+        names.append(label)
+    return distinct_app_names(names)
+
 class NvidiaShieldMediaPlayer(MediaPlayerEntity):
 
     _attr_device_class = MediaPlayerDeviceClass.TV
@@ -948,44 +1033,14 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
 
     @property
     def _app_directory(self) -> dict:
-        """Launchable apps: the user's map, else the curated default (P4-1).
+        """Launchable apps: the user's map, else the curated default (P4-1; t15).
 
-        ONE source for the source list and the reverse lookup (t61 removed the
-        browse tree that used to be the third consumer), so the dropdown can
-        never offer a name the launch path cannot resolve.
-
-        The map is hand-editable, so it is NORMALIZED here — once, so that every
-        consumer sees the same thing (t51):
-
-        * a name is stripped, so display and match key are one string: an entry
-          like ``com.x=  TiviMate  `` shows as "TiviMate" AND resolves as
-          "TiviMate", instead of being shown padded and missing every lookup;
-        * an entry whose name — or whose package — is blank/whitespace-only is
-          dropped, so no surface can offer a child with no package behind it or a
-          name no launch can reach (the t48 invariant, extended to the table);
-        * two packages normalizing to the same name are BOTH KEPT — each stays
-          reachable by its own package id, which ``select_source`` accepts (t61:
-          the browse tree that used to publish those ids is gone) — while
-          the FIRST in map order wins the name: that is the single entry
-          source_list offers and the package select_source resolves. Map order is
-          the user's order, so the winner does not move between rebuilds. "The same
-          name" is judged case-insensitively (t53), because the reverse lookup is:
-          names differing only by case would otherwise give the dropdown a second
-          row that launches the FIRST entry's app.
-
-        A well-formed map (what the editor and the curated default produce) passes
-        through byte-identical: stripping only touches the edges, so a name like
-        "My App" keeps its inner space.
+        The rule lives in the module-level :func:`app_directory` (t51: strip the
+        edges, drop blanks, keep map order, first name wins case-insensitively, the
+        shadowed package still reachable by id) so the source list, the reverse
+        lookup AND the remote's activity list cannot disagree about it.
         """
-        directory = self._entry.options.get(CONF_APP_NAME_MAP) or DEFAULT_APP_DIRECTORY
-        normalized: dict[str, str] = {}
-        for package, name in directory.items():
-            key = str(package).strip()
-            label = str(name).strip()
-            if not key or not label:
-                continue
-            normalized[key] = label
-        return normalized
+        return app_directory(self._entry.options)
 
     @property
     def source_list(self) -> list[str]:
@@ -1004,15 +1059,7 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
         row per name means a row can only launch the app it shows. The shadowed
         package is not lost: ``select_source`` accepts its own package id.
         """
-        seen: set[str] = set()
-        offered: list[str] = []
-        for name in self._app_directory.values():
-            folded = name.casefold()
-            if folded in seen:
-                continue
-            seen.add(folded)
-            offered.append(name)
-        return offered
+        return distinct_app_names(self._app_directory.values())
 
     @property
     def source(self) -> str | None:
@@ -1300,6 +1347,13 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
         attrs = {
             "app_id": self.app_id,
             "app_name": self.app_name,
+            # t15: which tier produced the app identity ("protocol" push / ADB parse),
+            # so the newly-reachable push cannot silently outrank the ADB reading the
+            # user's setup is built on — the same disclosure now_playing_via makes.
+            "app_source": (
+                self._manager.foreground_source if self._manager is not None else None
+            ),
+
             ATTR_MEDIA_DAG_ROLE: "source",
         }
         # t65: the reading is FLATTENED into separate top-level attributes.
diff --git a/custom_components/nvidia_shield/remote.py b/custom_components/nvidia_shield/remote.py
index 7e90941..7a0f068 100644
--- a/custom_components/nvidia_shield/remote.py
+++ b/custom_components/nvidia_shield/remote.py
@@ -9,31 +9,58 @@ remote-protocol transport (androidtvremote2) owned by ShieldConnectionManager:
   long presses are expressed as the START_LONG/END_LONG pair instead;
 - ``num_repeats`` / ``delay_secs`` are honored; long-press duration uses the
   ``hold_secs`` parameter;
-- ``activity`` = the protocol current-app push (package; name resolution is
-  deferred app-map work);
-- turn on/off send the POWER key and honor the is_on push (no superfluous
-  toggling when the state already matches).
+- ``RemoteEntityFeature.ACTIVITY`` with an ``activity_list`` built from OUR app
+  directory — the library exposes NO app-list API (measured), which is exactly why
+  core's list is user-maintained too — so the app/activity picker matches core;
+- ``current_activity`` resolves the foreground push (protocol first, ADB parse as
+  the fallback) through the one ``resolve_app_name`` chain, and stays None while
+  nothing is known: no fabricated activity, ever;
+- ``turn_on(activity=...)`` wakes when the device says it is off and then launches
+  through the same path ``select_source`` uses; an activity name the directory does
+  not carry is a ``ServiceValidationError``, never a silent no-op;
+- the transport's REAL state is published in ``extra_state_attributes`` so a
+  never-connected transport cannot look like a paired one (no ``is_paired`` exists:
+  a session or a push is the only proof);
+- a configured command script still owns turn_on/turn_off and never touches the
+  protocol, so a press that worked through it says NOTHING about pairing — the
+  attributes name which route each power command takes.
 """
 
 import asyncio
+import os
 
-from homeassistant.components.remote import RemoteEntity
+from homeassistant.components.remote import (
+    ATTR_ACTIVITY,
+    RemoteEntity,
+    RemoteEntityFeature,
+)
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant, callback
+from homeassistant.exceptions import ServiceValidationError
 from homeassistant.helpers.entity import DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
 from .connection import ShieldConnectionManager
 from .const import (
+    CONF_APP_NAME_MAP,
     DOMAIN,
     PROTOCOL_REMOTE_KEYS,
     SCRIPT_VAR_COMMAND,
     SCRIPT_VAR_DELAY_SECS,
     SCRIPT_VAR_HOLD_SECS,
     SCRIPT_VAR_NUM_REPEATS,
+    command_scripts,
     remote_command_script,
 )
-from .media_player import async_call_script, async_run_command_script, entity_available
+from .helpers import protocol_cert_paths
+from .media_player import (
+    activity_names,
+    app_directory,
+    async_call_script,
+    async_run_command_script,
+    entity_available,
+    resolve_app_name,
+)
 
 _START_PREFIX = "START_LONG:"
 _END_PREFIX = "END_LONG:"
@@ -64,6 +91,19 @@ def _key_code_name(command: str) -> str:
     return upper  # unknown names: the library reports/validates them
 
 
+def _looks_launchable(value: str) -> bool:
+    """True for a package id or a detail link — forms the launch path accepts (t15).
+
+    The rule is the repo's own definition of a package (``_valid_app_row``: dotted,
+    no spaces) plus URL-shaped deep links, which is exactly what
+    ``send_launch_app_command`` handles. It is deliberately NOT a guess that an id
+    exists: the device answers that. It exists so "Netflx" (a typo of a listed name)
+    is refused while an app the directory does not carry can still be launched.
+    """
+    if not value or " " in value:
+        return False
+    return "." in value or "://" in value
+
 async def async_setup_entry(
     hass: HomeAssistant,
     entry: ConfigEntry,
@@ -74,11 +114,21 @@ async def async_setup_entry(
 
 
 class NvidiaShieldRemoteEntity(RemoteEntity):
-    """Remote entity over the manager's androidtvremote2 transport."""
+    """Remote entity over the manager's androidtvremote2 transport.
+
+    t15: this is the entity the app/activity picker hangs off. HA renders the picker
+    only when ``supported_features`` carries ``RemoteEntityFeature.ACTIVITY`` — the
+    user's "the UI doesn't actually show any buttons like the android tv remote
+    does" — and the list itself has to come from OUR directory, because
+    ``AndroidTVRemote`` exposes no app-list API at all (measured on the installed
+    library; core androidtv_remote keeps a user-maintained map for the same reason).
+    """
 
     _attr_has_entity_name = True
     _attr_name = None
     _attr_should_poll = False
+    #: The feature flag that makes HA offer the activity/app picker at all.
+    _attr_supported_features = RemoteEntityFeature.ACTIVITY
 
     def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
         self.hass = hass
@@ -87,10 +137,58 @@ class NvidiaShieldRemoteEntity(RemoteEntity):
             entry, "runtime_data", None
         )
         self._attr_unique_id = f"{DOMAIN}_{entry.entry_id}_remote"
+        # t15: the activity list is available from the FIRST render (a real-HA state
+        # write needs no entity platform), and is refreshed with discovered apps once
+        # ADB can answer (``async_added_to_hass``).
+        self._attr_activity_list = activity_names(self._entry.options)
+        self._attr_current_activity = self._resolve_activity_name()
+        #: Which protocol pushes have EVER arrived — the only pairing evidence the
+        #: library can offer (it has no is_paired). Sticky on purpose: the store
+        #: expires every field after STATE_STORE_TTL, so "has it ever arrived?"
+        #: cannot be read from the store.
+        self._protocol_pushes_seen: set[str] = set()
+        self._note_protocol_pushes()
+        #: "present"/"missing"/"unknown" — the shared setup certificate, which is
+        #: why enabling the transport on a paired entry asks for no new PIN.
+        self._protocol_certificate = "unknown"
+        self._activity_refresh_task = None
 
     def _manager_of(self) -> ShieldConnectionManager | None:
         return getattr(self._entry, "runtime_data", None)
 
+    def _resolve_activity_name(self) -> str | None:
+        """The foreground app as a friendly name (t15), or None when nothing is known.
+
+        The same chain as the media player's ``app_name`` and the Current App sensor
+        (``resolve_app_name``): the user's map, then the cached device label, then the
+        curated table, then the raw id — so an activity and a source can never
+        disagree about what is on screen. The foreground identity prefers the protocol
+        push and falls back to the ADB parse, which keeps this entity useful on an
+        ADB-only setup (its protocol surface alone degrades).
+        """
+        manager = self._manager_of()
+        if not isinstance(manager, ShieldConnectionManager):
+            return None
+        return resolve_app_name(
+            manager.foreground_app,
+            self._entry.options.get(CONF_APP_NAME_MAP) or {},
+            manager.app_label,
+        )
+
+    def _note_protocol_pushes(self) -> None:
+        """Remember which device pushes have arrived (t15) — sticky, see __init__."""
+        manager = self._manager_of()
+        if not isinstance(manager, ShieldConnectionManager):
+            return
+        if manager.current_app is not None:
+            self._protocol_pushes_seen.add("current_app")
+        if manager.is_on is not None:
+            self._protocol_pushes_seen.add("is_on")
+        if manager.is_available is not None:
+            self._protocol_pushes_seen.add("is_available")
+        if manager.volume_info is not None:
+            self._protocol_pushes_seen.add("volume")
+
     @property
     def device_info(self) -> DeviceInfo:
         return DeviceInfo(
@@ -113,24 +211,158 @@ class NvidiaShieldRemoteEntity(RemoteEntity):
 
     @property
     def activity(self) -> str | None:
-        """Current foreground app reported by the protocol push."""
+        """The raw foreground package (protocol push; ADB parse as the fallback).
+
+        Kept alongside HA's ``current_activity``: this is the id (what
+        ``turn_on(activity=...)`` accepts and what the Current App sensor stamps),
+        while ``current_activity`` is the friendly name HA displays.
+        """
         manager = self._manager_of()
         if isinstance(manager, ShieldConnectionManager):
-            return manager.current_app
+            return manager.foreground_app
         return None
 
+    @property
+    def extra_state_attributes(self) -> dict:
+        """The transport's real state — the pairing answer, readable from HA (t15).
+
+        There is no "is it paired?" API to call (the library exposes none either,
+        measured), so everything here is a fact that can be PROVEN:
+
+        * ``protocol_state`` — ``off`` / ``never_connected`` / ``connected`` /
+          ``disconnected`` / ``auth_rejected``. ``never_connected`` is the state
+          that used to be permanent (``keep_reconnecting()`` alone never connects),
+          and it must never look like ``connected``.
+        * ``protocol_connected`` / ``protocol_ever_authenticated`` — a session or a
+          device push is the only proof of an authenticated (paired) client.
+        * ``protocol_current_app_seen`` / ``protocol_is_on_seen`` /
+          ``protocol_volume_seen`` — whether each device push has EVER arrived.
+        * ``protocol_last_error`` / ``protocol_failures`` — a send that could not
+          run is recorded here and logged, never swallowed into a debug line.
+        * ``protocol_certificate`` — the setup pairing's shared ``client.pem`` is
+          still on disk, which is exactly why enabling the transport asks for no
+          new PIN (the other Android TV integration pairs its own client).
+        * ``power_on_route`` / ``power_off_route`` — ``script:<id>`` when a command
+          script owns that command. That path never touches the protocol, so a
+          success through it says NOTHING about pairing.
+        """
+        return {
+            "activity": self.activity,
+            # t15: which tier produced it — a protocol push must never win SILENTLY
+            # over the ADB parse the user's setup is built on (see the manager's
+            # ``foreground_source``).
+            "activity_source": self._protocol_read("foreground_source"),
+            "power_on_route": self._power_route("turn_on"),
+            "power_off_route": self._power_route("turn_off"),
+            "protocol_active": self._protocol_active(),
+            "protocol_state": self._protocol_state(),
+            "protocol_connected": self._protocol_flag("protocol_connected"),
+            "protocol_ever_authenticated": self._protocol_flag(
+                "protocol_ever_authenticated"
+            ),
+            "protocol_last_error": self._protocol_read("protocol_last_error"),
+            "protocol_failures": self._protocol_read("protocol_failures"),
+            "protocol_certificate": self._protocol_certificate,
+            "protocol_current_app_seen": "current_app" in self._protocol_pushes_seen,
+            "protocol_is_on_seen": "is_on" in self._protocol_pushes_seen,
+            "protocol_volume_seen": "volume" in self._protocol_pushes_seen,
+        }
+
+    def _protocol_read(self, name: str, default=None):
+        """Read one transport fact off the manager, or the default (stub managers)."""
+        manager = self._manager_of()
+        if not isinstance(manager, ShieldConnectionManager):
+            return default
+        return getattr(manager, name, default)
+
+    def _protocol_active(self) -> bool:
+        return bool(self._protocol_read("protocol_active", False))
+
+    def _protocol_state(self) -> str:
+        """The transport state, with ``off`` for a manager that carries no client."""
+        if not self._protocol_active():
+            return "off"
+        return self._protocol_read("protocol_state", "never_connected")
+
+    def _protocol_flag(self, name: str) -> bool:
+        return bool(self._protocol_read(name, False))
+
+    def _power_route(self, command: str) -> str:
+        """Which path ``turn_on``/``turn_off`` takes: a script, or the protocol (t15).
+
+        A configured command script owns the sequence and never touches the
+        protocol — so "the power press worked" through it is NOT evidence that the
+        transport is paired, or even connected. That is the conclusion the user drew
+        and then doubted, made readable instead of inferred.
+        """
+        script = (command_scripts(self._entry.options) or {}).get(command)
+        return f"script:{script}" if script else "protocol"
+
+
     @callback
     def _refresh_state(self) -> None:
-        """Manager store changed — re-evaluate state (activity/available)."""
+        """Manager store changed — re-resolve activity, availability, transport."""
         self._manager = self._manager_of()
+        self._note_protocol_pushes()
+        self._attr_current_activity = self._resolve_activity_name()
         self.async_write_ha_state()
 
     async def async_added_to_hass(self) -> None:
         manager = self._manager_of()
         if isinstance(manager, ShieldConnectionManager):
-            self._manager_unregistered = manager.register_listener(
-                self._refresh_state
-            )
+            self._manager_unregistered = manager.register_listener(self._refresh_state)
+        self._note_protocol_pushes()
+        await self._async_note_certificate()
+        self._schedule_activity_refresh()
+
+    async def _async_note_certificate(self) -> None:
+        """Record whether the shared pairing certificate exists (t15).
+
+        This is the "no new PIN is expected" evidence, and it is a design fact, not a
+        guess: ``protocol_cert_paths`` is shared between the config flow's pairing step
+        and this runtime, so an entry paired at setup reuses that ``client.pem`` when
+        the transport is (re-)enabled. The OTHER Android TV integration pairs its own
+        client and shares nothing with us.
+        """
+        add_executor_job = getattr(self.hass, "async_add_executor_job", None)
+        if add_executor_job is None:
+            return  # a hass stand-in: leave it "unknown" rather than guess
+        try:
+            certfile, keyfile = await add_executor_job(protocol_cert_paths, self.hass)
+        except Exception:  # pragma: no cover - a filesystem we cannot read
+            return
+        self._protocol_certificate = (
+            "present" if os.path.exists(certfile) and os.path.exists(keyfile) else "missing"
+        )
+
+    def _schedule_activity_refresh(self) -> None:
+        """Add the apps ADB discovery actually found to the activity list (t15).
+
+        One bounded sweep, on demand — the t57 path the options editor uses — never a
+        per-state-write shell. The list is already complete for a configured
+        directory, and a sweep that cannot run changes nothing (no fabrication).
+        """
+        manager = self._manager_of()
+        create_task = getattr(self.hass, "async_create_task", None)
+        if create_task is None or not isinstance(manager, ShieldConnectionManager):
+            return
+        if not callable(getattr(manager, "async_refresh_installed_apps", None)):
+            return  # a manager stand-in: the directory alone is the honest list
+        self._activity_refresh_task = create_task(
+            self._async_refresh_activity_list(manager),
+            "nvidia_shield_remote_activities",
+        )
+
+    async def _async_refresh_activity_list(self, manager) -> None:
+        """Fold discovered apps into the activity list (t15) — one write if it grew."""
+        try:
+            rows = await manager.async_refresh_installed_apps()
+        except Exception:  # pragma: no cover - discovery never breaks the entity
+            return
+        names = activity_names(self._entry.options, rows)
+        if names != self._attr_activity_list:
+            self._attr_activity_list = names
+            self.async_write_ha_state()
 
     async def async_will_remove_from_hass(self) -> None:
         unregistered = getattr(self, "_manager_unregistered", None)
@@ -139,9 +371,31 @@ class NvidiaShieldRemoteEntity(RemoteEntity):
             self._manager_unregistered = None
 
     async def async_turn_on(self, **kwargs) -> None:
-        """Wake the device — via the configured script, else the POWER key."""
-        # t44: `turn_on` is one configuration, and the remote entity is the other
-        # route to it; a script that owns the power sequence owns it here too.
+        """Wake the device, and launch a requested activity (t15).
+
+        With NO activity — or an empty one — this behaves exactly as before: the
+        configured script owns the wake sequence when there is one, otherwise POWER
+        goes out only when the device says it is off (unknown counts as "maybe on",
+        so nothing is toggled blindly).
+
+        With ``activity``, the same wake happens first (a script that owns turn_on
+        still owns it — it never touches the protocol), and then the app is launched
+        over the protocol through the same path ``select_source`` uses. A name the
+        activity list does not carry, and that is not a package/detail link either,
+        raises ``ServiceValidationError``: an activity request that cannot launch
+        must never look like it worked.
+        """
+        activity = str(kwargs.get(ATTR_ACTIVITY) or "").strip()
+        await self._async_wake()
+        if activity:
+            await self._async_launch_activity(activity)
+
+    async def _async_wake(self) -> None:
+        """Turn the device on: the configured script first, else the POWER key (t44).
+
+        t44: `turn_on` is one configuration, and the remote entity is the other route
+        to it; a script that owns the power sequence owns it here too.
+        """
         if await async_run_command_script(self.hass, self._entry.options, "turn_on"):
             return
         manager = self._manager_of()
@@ -152,6 +406,51 @@ class NvidiaShieldRemoteEntity(RemoteEntity):
             return
         await manager.send_protocol_key(_POWER_KEY, "SHORT")
 
+    async def _async_launch_activity(self, activity: str) -> None:
+        """Launch an activity through the shared launch path (t15).
+
+        The target comes from the same directory the activity list is built from, so
+        anything the picker offers resolves. A package id or a deep link passes
+        through: that is how an app the directory does not carry can still be
+        launched, and what core does with an activity it does not know. Anything else
+        is a request nothing can launch and raises instead of doing nothing.
+        """
+        target = self._activity_target(activity)
+        manager = self._manager_of()
+        if not isinstance(manager, ShieldConnectionManager):
+            raise ServiceValidationError(
+                translation_domain=DOMAIN,
+                translation_key="protocol_unavailable",
+                translation_placeholders={"activity": activity},
+            )
+        if not await manager.launch_app(target):
+            raise ServiceValidationError(
+                translation_domain=DOMAIN,
+                translation_key="activity_launch_failed",
+                translation_placeholders={"activity": activity},
+            )
+
+    def _activity_target(self, activity: str) -> str:
+        """Activity name (or id) → what the launch path takes (t15).
+
+        A listed NAME wins, compared case-insensitively — the same rule the
+        ``source_list``/``select_source`` pair uses, so the picker can never offer a
+        name this does not resolve. A ``CONF_APP_NAME_MAP``-less entry still resolves
+        the curated names, because both read the one ``app_directory``.
+        """
+        directory = app_directory(self._entry.options)
+        wanted = activity.casefold()
+        for package, name in directory.items():
+            if name.casefold() == wanted:
+                return package
+        if activity in directory or _looks_launchable(activity):
+            return activity
+        raise ServiceValidationError(
+            translation_domain=DOMAIN,
+            translation_key="unknown_activity",
+            translation_placeholders={"activity": activity},
+        )
+
     async def async_turn_off(self, **kwargs) -> None:
         """Sleep the device — via the configured script, else the POWER key."""
         if await async_run_command_script(self.hass, self._entry.options, "turn_off"):
diff --git a/custom_components/nvidia_shield/strings.json b/custom_components/nvidia_shield/strings.json
index eeb913a..304ead0 100644
--- a/custom_components/nvidia_shield/strings.json
+++ b/custom_components/nvidia_shield/strings.json
@@ -377,6 +377,18 @@
     "ambiguous_target": {
       "title": "More than one Shield selected",
       "message": "{entity_id} covers more than one adb_command target — select exactly one Shield (one media_player entity) — command not sent."
+    },
+    "unknown_activity": {
+      "title": "Unknown activity",
+      "message": "'{activity}' is not one of this Shield's activities. Use a name from the remote's activity list (the Apps map names, or the built-in list when no map is set), or pass an app's package id / a detail link."
+    },
+    "activity_launch_failed": {
+      "title": "Activity could not launch",
+      "message": "'{activity}' was not launched: the remote protocol could not run the command. Check the remote entity's protocol_* attributes (protocol_state says whether the transport is connected) — nothing was sent."
+    },
+    "protocol_unavailable": {
+      "title": "Remote-protocol transport unavailable",
+      "message": "'{activity}' was not launched: this entry has no remote-protocol transport (the option is off, or the androidtvremote2 library is missing). Turn the transport on and pair it, then try again."
     }
   },
   "entity": {
diff --git a/custom_components/nvidia_shield/translations/en.json b/custom_components/nvidia_shield/translations/en.json
index eeb913a..304ead0 100644
--- a/custom_components/nvidia_shield/translations/en.json
+++ b/custom_components/nvidia_shield/translations/en.json
@@ -377,6 +377,18 @@
     "ambiguous_target": {
       "title": "More than one Shield selected",
       "message": "{entity_id} covers more than one adb_command target — select exactly one Shield (one media_player entity) — command not sent."
+    },
+    "unknown_activity": {
+      "title": "Unknown activity",
+      "message": "'{activity}' is not one of this Shield's activities. Use a name from the remote's activity list (the Apps map names, or the built-in list when no map is set), or pass an app's package id / a detail link."
+    },
+    "activity_launch_failed": {
+      "title": "Activity could not launch",
+      "message": "'{activity}' was not launched: the remote protocol could not run the command. Check the remote entity's protocol_* attributes (protocol_state says whether the transport is connected) — nothing was sent."
+    },
+    "protocol_unavailable": {
+      "title": "Remote-protocol transport unavailable",
+      "message": "'{activity}' was not launched: this entry has no remote-protocol transport (the option is off, or the androidtvremote2 library is missing). Turn the transport on and pair it, then try again."
     }
   },
   "entity": {
diff --git a/tests/entity_stublib.py b/tests/entity_stublib.py
index 8aa20ce..c60fa8c 100644
--- a/tests/entity_stublib.py
+++ b/tests/entity_stublib.py
@@ -243,7 +243,40 @@ class MediaPlayerEntity:
         return None
 
 
+class RemoteEntityFeature(enum.IntFlag):
+    """homeassistant.components.remote.RemoteEntityFeature (the real values)."""
+
+    LEARN_COMMAND = 1
+    DELETE_COMMAND = 2
+    ACTIVITY = 4
+
+
 class RemoteEntity:
+    """The RemoteEntity surface HA reads, backed by the ``_attr_*`` convention.
+
+    t15: HA renders the app/activity picker off ``supported_features`` and reads
+    ``activity_list``/``current_activity``; both are cached properties on the real
+    base, reading ``_attr_*``. The stub carries that contract (the same shape as the
+    ImageEntity face above) so a test can read what the entity publishes instead of
+    asserting against our own attribute names.
+    """
+
+    _attr_activity_list: list | None = None
+    _attr_current_activity: str | None = None
+    _attr_supported_features = RemoteEntityFeature(0)
+
+    @property
+    def supported_features(self):
+        return self._attr_supported_features
+
+    @property
+    def activity_list(self):
+        return self._attr_activity_list
+
+    @property
+    def current_activity(self):
+        return self._attr_current_activity
+
     def async_write_ha_state(self) -> None:
         return None
 
@@ -474,6 +507,14 @@ def install() -> None:
 
     remote = types.ModuleType("homeassistant.components.remote")
     remote.RemoteEntity = RemoteEntity
+    # t15: the feature flag that makes HA offer the activity/app picker, the service
+    # field the picker submits, and the two attributes HA reads off the base class.
+    remote.RemoteEntityFeature = RemoteEntityFeature
+    remote.ATTR_ACTIVITY = "activity"
+    remote.RemoteEntityStateAttribute = types.SimpleNamespace(
+        ACTIVITY_LIST="activity_list",
+        CURRENT_ACTIVITY="current_activity",
+    )
     sys.modules["homeassistant.components.remote"] = remote
 
     button = types.ModuleType("homeassistant.components.button")
diff --git a/tests/ha_api_drift.py b/tests/ha_api_drift.py
index 5e0617a..fb3ffbe 100644
--- a/tests/ha_api_drift.py
+++ b/tests/ha_api_drift.py
@@ -142,15 +142,19 @@ HA_CALL_OWNERS: dict[str, tuple[str, str]] = {
 
 #: Non-vacuity floors — the ONE source of truth for the standalone entry point
 #: and the pytest wrapper (t39 F1). They sit below the current walk (25 modules,
-#: 73 imports, 193 relative imports, 54 checked arguments, 6 awaited HA calls,
-#: 1235 statement blocks) so a broken or mis-targeted walk fails loudly instead of
+#: 76 imports, 206 relative imports, 57 checked arguments, 6 awaited HA calls,
+#: 1322 statement blocks) so a broken or mis-targeted walk fails loudly instead of
 #: printing zeros and exiting 0.
 FLOORS: dict[str, int] = {
     # t41: the sweep is recursive, so the floor sits above the top-level-only
     # count (13) — a regression to the shallow glob trips it.
     "modules": 20,
-    "imports": 40,
-    "relative_imports": 100,
+    # t15: raised from 40/100 to just under the current 76/206, because the remote
+    # parity work is an IMPORTS-and-coupling feature: `RemoteEntityFeature`,
+    # `ATTR_ACTIVITY`, the transport-state accessors and the shared app-directory
+    # helpers all arrive as imports, and none of them may vanish unnoticed.
+    "imports": 70,
+    "relative_imports": 195,
     # t62 F2: raised from 20 to just under the current 46. A floor's job is to
     # notice an API OWNER going away, not only a broken walk: at 20, deleting
     # `async_show_menu` from HA_CALL_OWNERS left 40 checked arguments and still
@@ -163,11 +167,17 @@ FLOORS: dict[str, int] = {
     # registration added two owners (`async_register`'s `schema`/`supports_response`
     # arguments and the module-level description writer). At 44 those owners could go
     # away without the gate noticing; at 52 they are load-bearing.
-    "call_args": 52,
+    #
+    # t15: raised from 52 to just under the current 57 — the options-flow pairing
+    # step is a real flow-display surface of its own (its own async_show_form with
+    # step_id/data_schema/errors), so those call sites must not be able to vanish.
+    "call_args": 55,
     "awaited_calls": 5,
-    # t62 F2: the unreachable-statement pass walks ~1100 statement blocks. A
+    # t62 F2: the unreachable-statement pass walks ~1300 statement blocks. A
     # floor here means a pass that silently stopped parsing cannot look green.
-    "statement_blocks": 500,
+    # t15: raised 500 -> 1200 -> 1300, just under the walk, for the same reason the
+    # imports floor moved: the exhaustion pass covers the new surfaces too.
+    "statement_blocks": 1300,
 }
 
 #: The interpreter floor Home Assistant itself declares: PyPI metadata for
diff --git a/tests/test_command_routing.py b/tests/test_command_routing.py
index 4920fd4..ae49008 100644
--- a/tests/test_command_routing.py
+++ b/tests/test_command_routing.py
@@ -16,7 +16,22 @@ import pytest
 from conftest import _SHIELD_DIR
 from entity_stublib import install, load_entity_modules
 from nsh.connection import ShieldConnectionManager
-from nsh.const import APP_LABEL_EXIT_MARKER, TIVIMATE_PACKAGE
+from nsh.const import (
+    APP_LABEL_EXIT_MARKER,
+    CONF_APP_NAME_MAP,
+    CONF_COMMAND_SCRIPTS,
+    DEFAULT_APP_DIRECTORY,
+    TIVIMATE_PACKAGE,
+)
+
+
+def _service_validation_error():
+    """HA's ServiceValidationError, as the stub harness installs it (t15).
+
+    Read at TEST time: the autouse fixture installs the faces per test, so the class
+    cannot be bound at import time.
+    """
+    return sys.modules["homeassistant.exceptions"].ServiceValidationError
 
 
 @pytest.fixture(autouse=True)
@@ -356,6 +371,241 @@ class TestRemoteEntity:
         assert protocol.key_calls == []
 
 
+class TestRemoteActivityContract:
+    """t15: the app/activity picker HA renders off ``RemoteEntityFeature.ACTIVITY``.
+
+    The user compared our remote to the default Android TV app: "it doesn't display
+    the full set of options" and "the UI ... doesn't actually show any buttons". HA
+    only offers the picker when the feature flag is set, and the list has to come
+    from OUR directory — ``AndroidTVRemote`` exposes no app-list API (measured), the
+    same reason core keeps a user-maintained map.
+    """
+
+    def test_the_activity_feature_is_declared(self, modules):
+        entity = _remote_entity(modules)
+        feature = modules["remote"].RemoteEntityFeature
+
+        assert entity.supported_features & feature.ACTIVITY
+
+    def test_activity_list_follows_the_directory_order(self, modules):
+        entity = _remote_entity(modules)
+        entity._entry.options[CONF_APP_NAME_MAP] = {
+            "com.a.one": "One",
+            "com.b.two": "Two",
+            "com.c.three": "Three",
+        }
+
+        assert modules["remote"].activity_names(
+            {CONF_APP_NAME_MAP: entity._entry.options[CONF_APP_NAME_MAP]}
+        ) == ["One", "Two", "Three"]
+
+    def test_the_curated_table_is_the_floor_and_names_nothing_else(self, modules):
+        """Nothing is fabricated: with no user map the curated table IS the list."""
+        names = modules["remote"].activity_names({})
+
+        assert names == list(DEFAULT_APP_DIRECTORY.values())
+        assert modules["remote"].activity_names({CONF_APP_NAME_MAP: {}}) == names
+
+    def test_nameless_discovered_apps_are_skipped_not_doubled(self, modules):
+        """Never "pkg=pkg": a row nothing can name is not offered as a name."""
+        names = modules["remote"].activity_names(
+            {CONF_APP_NAME_MAP: {"com.a.one": "One"}},
+            [
+                {"package": "com.a.one", "name": "One"},  # already listed
+                {"package": "com.d.four", "name": "Four"},  # a real label
+                {"package": "com.e.five", "name": ""},  # nameless
+                {"package": "com.f.six", "name": "com.f.six"},  # id as its name
+            ],
+        )
+
+        assert names == ["One", "Four"]
+
+    def test_a_duplicate_name_is_offered_once(self, modules):
+        names = modules["remote"].activity_names(
+            {CONF_APP_NAME_MAP: {"com.a.one": "One", "com.b.one": "one", "com.c.two": "Two"}}
+        )
+
+        assert names == ["One", "Two"]
+
+    def test_a_blank_row_cannot_be_offered_as_a_name(self, modules):
+        names = modules["remote"].activity_names(
+            {CONF_APP_NAME_MAP: {"com.a.one": "One"}},
+            [{"package": "com.g.seven", "name": "   "}],
+        )
+
+        assert names == ["One"]
+
+    def test_current_activity_resolves_the_foreground_app(self, modules):
+        entity = _remote_entity(modules)
+        entity._entry.options[CONF_APP_NAME_MAP] = {"com.netflix.ninja": "Netflix"}
+
+        assert entity._resolve_activity_name() is None, "nothing known yet"
+
+        entity._manager.store.set("foreground", "com.netflix.ninja")
+        assert entity._resolve_activity_name() == "Netflix"
+
+    def test_current_activity_uses_the_protocol_push_first(self, modules):
+        entity = _remote_entity(modules)
+        entity._entry.options[CONF_APP_NAME_MAP] = {
+            "com.netflix.ninja": "Netflix",
+            "com.google.android.youtube.tv": "YouTube",
+        }
+        entity._manager.store.set("foreground", "com.google.android.youtube.tv")
+        entity._manager._on_current_app("com.netflix.ninja")
+
+        assert entity._resolve_activity_name() == "Netflix"
+        assert entity.activity == "com.netflix.ninja"
+
+    async def test_turn_on_with_an_activity_launches_it(self, modules):
+        entity = _remote_entity(modules)
+        entity._entry.options[CONF_APP_NAME_MAP] = {"com.netflix.ninja": "Netflix"}
+        entity._manager._on_is_on(False)
+        protocol = entity._manager._protocol
+
+        await entity.async_turn_on(activity="Netflix")
+
+        assert protocol.key_calls == [("KEYCODE_POWER", "SHORT")]
+        assert protocol.launch_calls == ["com.netflix.ninja"]
+
+    async def test_turn_on_launches_an_unknown_package_verbatim(self, modules):
+        """A package id (or deep link) the directory lacks is still launchable."""
+        entity = _remote_entity(modules)
+        entity._entry.options[CONF_APP_NAME_MAP] = {"com.netflix.ninja": "Netflix"}
+        entity._manager._on_is_on(True)  # already on: no POWER
+        protocol = entity._manager._protocol
+
+        await entity.async_turn_on(activity="com.example.other")
+
+        assert protocol.key_calls == []
+        assert protocol.launch_calls == ["com.example.other"]
+
+    async def test_an_unknown_activity_name_is_a_clear_error(self, modules):
+        entity = _remote_entity(modules)
+        entity._entry.options[CONF_APP_NAME_MAP] = {"com.netflix.ninja": "Netflix"}
+        protocol = entity._manager._protocol
+
+        with pytest.raises(_service_validation_error()) as raised:
+            await entity.async_turn_on(activity="Netflx")
+
+        assert raised.value.translation_key == "unknown_activity"
+        assert protocol.launch_calls == [], "nothing was silently launched"
+
+    def test_an_empty_activity_keeps_the_old_behaviour(self, modules):
+        entity = _remote_entity(modules)
+        entity._entry.options[CONF_APP_NAME_MAP] = {"com.netflix.ninja": "Netflix"}
+        entity._manager._on_is_on(True)
+        protocol = entity._manager._protocol
+
+        asyncio.run(entity.async_turn_on(activity=""))
+
+        assert protocol.key_calls == []
+        assert protocol.launch_calls == []
+
+    async def test_a_command_script_still_owns_the_wake_sequence(self, modules):
+        """The script masks the protocol: a press through it proves nothing (t15)."""
+        entity = _remote_entity(modules)
+        served = []
+        entity.hass = SimpleNamespace(services=_script_services(served))
+        entity._entry.options[CONF_COMMAND_SCRIPTS] = {"turn_on": "script.wake"}
+        entity._manager._on_is_on(False)  # the native path would fire POWER
+
+        await entity.async_turn_on(activity="com.netflix.ninja")
+
+        assert [call[2]["entity_id"] for call in served] == ["script.wake"]
+        assert entity._manager._protocol.key_calls == [], "the script owns POWER"
+        assert entity.extra_state_attributes["power_on_route"] == "script:script.wake"
+
+    def test_the_transport_attributes_distinguish_the_states(self, modules):
+        entity = _remote_entity(modules)
+        attrs = entity.extra_state_attributes
+
+        assert attrs["protocol_active"] is True
+        assert attrs["protocol_state"] == "never_connected", (
+            "the pre-t15 transport state: wired, never connected. It must never "
+            "look like a healthy one"
+        )
+        assert attrs["protocol_connected"] is False
+        assert attrs["protocol_ever_authenticated"] is False
+        assert attrs["protocol_current_app_seen"] is False
+        assert attrs["protocol_last_error"] is None
+        assert attrs["protocol_failures"] == 0
+        assert attrs["power_on_route"] == "protocol"
+
+    def test_no_protocol_reads_as_off_not_as_unpaired(self, modules):
+        entity = _remote_entity(modules)
+        entity._manager._protocol = None
+        attrs = entity.extra_state_attributes
+
+        assert attrs["protocol_active"] is False
+        assert attrs["protocol_state"] == "off"
+        assert attrs["protocol_connected"] is False
+
+    def test_a_push_is_remembered_even_after_the_store_expires(self, modules):
+        entity = _remote_entity(modules)
+        entity._manager._on_current_app("com.netflix.ninja")
+
+        entity._refresh_state()
+        assert entity.extra_state_attributes["protocol_current_app_seen"] is True
+        assert entity.extra_state_attributes["protocol_ever_authenticated"] is True
+
+        entity._manager.store.clear()  # TTL expiry / a cleared store
+        entity._refresh_state()
+
+        assert entity.extra_state_attributes["protocol_current_app_seen"] is True, (
+            "\"has the current app push EVER arrived?\" cannot be answered from a "
+            "TTL-capped store"
+        )
+
+    def test_a_failed_launch_is_never_silent(self, modules):
+        entity = _remote_entity(modules)
+        protocol = entity._manager._protocol
+        protocol.fail_all = True
+
+        with pytest.raises(_service_validation_error()) as raised:
+            asyncio.run(entity._async_launch_activity("com.netflix.ninja"))
+
+        assert raised.value.translation_key == "activity_launch_failed"
+        assert entity._manager.protocol_failures == 1
+        assert "com.netflix.ninja" in (entity._manager.protocol_last_error or "")
+
+    async def test_discovery_grows_the_activity_list(self, modules):
+        entity = _remote_entity(modules)
+        entity._entry.options[CONF_APP_NAME_MAP] = {"com.a.one": "One"}
+        entity._attr_activity_list = modules["remote"].activity_names(
+            entity._entry.options
+        )
+        written = []
+        entity.async_write_ha_state = lambda: written.append(True)
+
+        async def _rows():
+            return [{"package": "com.d.four", "name": "Four"}]
+
+        entity._manager.async_refresh_installed_apps = _rows
+
+        await entity._async_refresh_activity_list(entity._manager)
+
+        assert entity.activity_list == ["One", "Four"]
+        assert written == [True]
+
+    async def test_a_failed_sweep_changes_nothing(self, modules):
+        """A failed sweep must not be cached as an answer (t62) — here, not a list."""
+        entity = _remote_entity(modules)
+        entity._entry.options[CONF_APP_NAME_MAP] = {"com.a.one": "One"}
+        entity._attr_activity_list = ["One"]
+        written = []
+        entity.async_write_ha_state = lambda: written.append(True)
+
+        async def _boom():
+            raise RuntimeError("sweep failed")
+
+        entity._manager.async_refresh_installed_apps = _boom
+
+        await entity._async_refresh_activity_list(entity._manager)
+
+        assert entity.activity_list == ["One"]
+        assert written == []
+
+
 class TestIntentRetryWithoutATransport:
     """t41: an intent no transport served surrenders instead of delegating."""
 
@@ -590,6 +840,7 @@ class TestStateModelT55:
     no source reported anything at all.
     """
 
+
     # ── the five states ──
 
     async def test_no_data_at_all_is_the_only_on(self, modules):
@@ -833,6 +1084,117 @@ class TestStateModelT55:
         assert np == {"state": "buffering", "title": "Sky News", "via": "app"}
 
 
+
+class TestProtocolArmedKeepsTheAdbLadder:
+    """t15: the pushes this unit turns ON must not change what ADB already answers.
+
+    The protocol had NEVER been connected here, so every media-session reading the user
+    relies on arrived while is_on / current_app / volume stayed permanently empty.
+    Arming the transport adds that traffic; these tests pin the boundary the user's
+    constraint names — ADB media-session tracking stays the state source, no ADB store
+    field is touched, and an entry with no protocol at all behaves as before with only
+    the protocol surface degraded.
+    """
+
+    @staticmethod
+    def _adb_player(modules):
+        """A playing Netflix session, an ADB foreground, and no media entity."""
+        from nsh.ha_adb_common import MediaSessionInfo
+
+        player, _served, _intents = make_player(modules)
+        manager = player._manager
+        manager.store.set("awake", True)
+        manager.store.set(
+            "sessions",
+            [MediaSessionInfo(package="com.netflix.ninja", playing=True, title="Movie")],
+        )
+        manager.store.set("foreground", "com.netflix.ninja")
+        player._read = lambda eid: None
+        return player, manager
+
+    def test_a_full_protocol_push_set_leaves_state_and_now_playing_alone(self, modules):
+        player, manager = self._adb_player(modules)
+        before_state = player.state.value
+        before_now = dict(player._get_adb_now_playing())
+
+        # the transport is armed: every push the library offers arrives at once
+        manager._on_is_available(True)
+        manager._on_is_on(True)
+        manager._on_current_app("com.google.android.youtube.tv")
+        manager._on_volume_info({"level": 5, "max": 15, "muted": False})
+
+        assert player.state.value == before_state == "playing"
+        assert dict(player._get_adb_now_playing()) == before_now
+        # the ONE field whose documented priority IS the push (P3-2) — and the ADB
+        # reading it shadows is still in the store, untouched (see the manager suite)
+        assert manager.store.get("foreground") == "com.netflix.ninja"
+        assert player.app_id == "com.google.android.youtube.tv"
+        # ...and the entity NAMES the tier that won, so the push is never a silent
+        # override of the ADB reading the user's setup is built on (t15)
+        assert player.extra_state_attributes["app_source"] == "protocol"
+
+    def test_a_protocol_reported_off_is_off_while_adb_keeps_its_truth(self, modules):
+        """A device that says it is off outranks a stale session — newly reachable.
+
+        Before t15 this signal could not arrive at all, so an off Shield kept showing
+        its last session as playing. The ADB fields are still there (the push writes
+        only its own), which is why a later poll answers immediately again.
+        """
+        player, manager = self._adb_player(modules)
+        manager._on_is_on(False)
+
+        assert player.state.value == "off"
+        assert manager.store.get("sessions"), "the ADB session reading was not cleared"
+
+    def test_an_adb_only_entry_still_answers_everything_it_did(self, modules):
+        """Protocol disabled: only the protocol surface degrades (the t40 ladder)."""
+        player, manager = self._adb_player(modules)
+        manager._protocol = None  # the entry never enabled the transport
+        remote = _remote_entity(modules)
+        remote._manager = manager
+        remote._entry.runtime_data = manager
+        remote._entry.options = dict(player._entry.options)
+        remote._refresh_state()  # the manager ping that re-resolves the activity
+
+        assert player.state.value == "playing"
+        assert player.app_name == "Netflix"  # curated table, no protocol involved
+        assert player.source_list  # the app dropdown is independent of the transport
+        assert remote.available is True  # a passive transport is not a broken one
+        assert remote.activity == "com.netflix.ninja"  # the ADB parse names it
+        assert remote.current_activity == "Netflix"
+        assert remote.activity_list == player.source_list
+        assert remote.extra_state_attributes["protocol_state"] == "off"
+        assert remote.extra_state_attributes["protocol_connected"] is False
+        # an ADB-named activity says so: the tier is disclosed, not implied
+        assert remote.extra_state_attributes["activity_source"] == "adb"
+
+
+def test_the_new_exception_keys_have_copy_in_both_files():
+    """A raised translation_key with no copy shows the user a bare key (t15).
+
+    The three keys the remote raises: an activity name nothing can launch, a launch the
+    protocol could not run, and an entry with no protocol transport at all.
+    """
+    import json
+    from pathlib import Path
+
+    root = (
+        Path(__file__).resolve().parent.parent
+        / "custom_components"
+        / "nvidia_shield"
+    )
+    for name in ("strings.json", "translations/en.json"):
+        exceptions = json.loads((root / name).read_text(encoding="utf-8"))["exceptions"]
+        for key in ("unknown_activity", "activity_launch_failed", "protocol_unavailable"):
+            assert exceptions[key]["title"], (name, key)
+            assert exceptions[key]["message"], (name, key)
+        # the placeholder the entity fills in is the one the copy names
+        assert "{activity}" in exceptions["unknown_activity"]["message"]
+        assert "{activity}" in exceptions["activity_launch_failed"]["message"]
+        # ...and the message tells the user where the answer is readable
+        assert "protocol_state" in exceptions["activity_launch_failed"]["message"]
+
+
 class TestChannelLeakGuard:
     """F1: the channel number never leaks to a non-channel foreground app."""
 
@@ -959,6 +1321,14 @@ def _remote_entity(modules):
     entity._manager = manager
     entity._attr_unique_id = None
     entity._manager_unregistered = None
+    # t15: mirrors __init__ — the activity contract and the transport evidence the
+    # attribute payload publishes (a __new__-built entity skips __init__).
+    entity._attr_supported_features = remote_module.RemoteEntityFeature.ACTIVITY
+    entity._attr_activity_list = remote_module.activity_names(entry.options)
+    entity._attr_current_activity = None
+    entity._protocol_pushes_seen = set()
+    entity._protocol_certificate = "unknown"
+    entity._activity_refresh_task = None
     return entity
 
 
@@ -966,6 +1336,16 @@ async def _noop_async(*args, **kwargs):
     return None
 
 
+def _script_services(served):
+    """A hass.services stand-in that records (domain, service, data) (t15)."""
+
+    class _RecordingServices:
+        async def async_call(self, domain, service, data=None, **kwargs):
+            served.append((domain, service, data))
+
+    return _RecordingServices()
+
+
 class FakeServices:
     """hass.services stand-in recording registrations."""
 
diff --git a/tests/test_config_flow_real_ha.py b/tests/test_config_flow_real_ha.py
index 01148a4..4993332 100644
--- a/tests/test_config_flow_real_ha.py
+++ b/tests/test_config_flow_real_ha.py
@@ -862,6 +862,229 @@ class TestRealSetupMenu:
 # (``selector.text.multiline``) — the value the frontend actually reads.
 
 
+class _FakePairingClient:
+    """The async pairing surface the flows drive (t15) — a double, no socket.
+
+    Only what the flows call: generate/start/finish plus ``disconnect`` for the
+    throwaway session. ``finish_error`` reproduces a wrong PIN.
+    """
+
+    def __init__(self, finish_error: Exception | None = None):
+        self.finish_calls: list[str] = []
+        self.disconnect_calls = 0
+        self._finish_error = finish_error
+
+    async def async_generate_cert_if_missing(self) -> None:
+        return None
+
+    async def async_start_pairing(self) -> None:
+        return None
+
+    async def async_finish_pairing(self, code: str) -> None:
+        self.finish_calls.append(code)
+        if self._finish_error is not None:
+            raise self._finish_error
+
+    def disconnect(self) -> None:
+        self.disconnect_calls += 1
+
+    async def async_get_name_and_mac(self) -> tuple[str, str]:
+        """The setup flow asks for the device identity right after pairing (t15)."""
+        return "Shield", "AA:BB:CC:DD:EE:FF"
+
+
+class TestRealOptionsProtocolPairing:
+    """t15: switching the transport ON on a never-paired entry must PAIR.
+
+    The live report: the transport was switched on, NO PIN prompt appeared, and the
+    remote then looked unpaired whether it was or not. Measured cause: the certificate
+    from setup is reused by design (so a PAIRED entry correctly asks for nothing), and
+    the library has no ``is_paired`` — which is why the marker is persisted and why an
+    unpaired entry is routed into the pairing step instead of being let loose.
+    """
+
+    @staticmethod
+    def _connection_page(entry, **overrides):
+        """The options Connection page as the real frontend submits it (t60)."""
+        page = {
+            "host": entry.options.get("host", "10.0.0.9"),
+            "adb_port": 5555,
+            "adb_enabled": True,
+            "adb_store_app_info": True,
+            "protocol_enabled": True,
+            "logcat_events": False,
+            "adb_command_enabled": False,
+            "screen_capture_enabled": False,
+            "jellyfin_entity": "",
+            "power_entity": "",
+        }
+        page.update(overrides)
+        return page
+
+    @staticmethod
+    def _fake_pairing(monkeypatch, client):
+        """Answer the module-level pairing builder with a client double (t15)."""
+
+        async def _start(hass, host, client_name):
+            return client, None
+
+        monkeypatch.setattr(cf, "_async_start_protocol_pairing", _start)
+
+    async def _open_connection(self, ha, entry):
+        manager, menu = await open_options(ha, entry)
+        page = await manager.async_configure(
+            menu["flow_id"], {"next_step_id": "connection"}
+        )
+        assert page["type"] == "form" and page["step_id"] == "connection"
+        return manager, page
+
+    async def test_enabling_the_transport_on_an_unpaired_entry_pairs(self, ha, monkeypatch):
+        entry = make_entry(ha, options={"host": "10.0.0.9", "protocol_enabled": False})
+        client = _FakePairingClient()
+        self._fake_pairing(monkeypatch, client)
+        manager, page = await self._open_connection(ha, entry)
+
+        result = await manager.async_configure(
+            page["flow_id"], self._connection_page(entry, protocol_enabled=True)
+        )
+
+        assert result["type"] == "form", result
+        assert result["step_id"] == "protocol_pair", (
+            "an unpaired entry must not end up running an unauthenticated client"
+        )
+        assert result["data_schema"] is not None
+
+    async def test_the_pin_then_marks_the_entry_paired_and_keeps_it_enabled(
+        self, ha, monkeypatch
+    ):
+        entry = make_entry(ha, options={"host": "10.0.0.9", "protocol_enabled": False})
+        client = _FakePairingClient()
+        self._fake_pairing(monkeypatch, client)
+        manager, page = await self._open_connection(ha, entry)
+        pair_form = await manager.async_configure(
+            page["flow_id"], self._connection_page(entry, protocol_enabled=True)
+        )
+
+        after_code = await manager.async_configure(
+            pair_form["flow_id"], {"pairing_code": "A1B2C3", "host": "10.0.0.9"}
+        )
+
+        assert after_code["type"] == "menu", after_code
+        assert after_code["step_id"] == "init"
+        assert client.finish_calls == ["A1B2C3"]
+        assert client.disconnect_calls == 1, "the throwaway session is closed"
+
+        saved = await manager.async_configure(
+            after_code["flow_id"], {"next_step_id": "finish"}
+        )
+        assert saved["type"] == "create_entry"
+        assert saved["data"]["protocol_paired"] is True
+        assert saved["data"]["protocol_enabled"] is True
+
+    async def test_skipping_the_pairing_leaves_the_transport_off(self, ha, monkeypatch):
+        entry = make_entry(ha, options={"host": "10.0.0.9", "protocol_enabled": False})
+        self._fake_pairing(monkeypatch, _FakePairingClient())
+        manager, page = await self._open_connection(ha, entry)
+        pair_form = await manager.async_configure(
+            page["flow_id"], self._connection_page(entry, protocol_enabled=True)
+        )
+
+        after_skip = await manager.async_configure(
+            pair_form["flow_id"],
+            {"pairing_code": "", "host": "10.0.0.9", "skip_pairing": True},
+        )
+
+        assert after_skip["type"] == "menu", after_skip
+        saved = await manager.async_configure(
+            after_skip["flow_id"], {"next_step_id": "finish"}
+        )
+        assert saved["data"]["protocol_enabled"] is False, "stay off, never half-paired"
+        assert saved["data"]["protocol_paired"] is False
+
+    async def test_a_paired_entry_is_not_asked_for_a_pin_again(self, ha, monkeypatch):
+        """(a) documented: the setup certificate is REUSED, so no new PIN."""
+        entry = make_entry(
+            ha,
+            options={
+                "host": "10.0.0.9",
+                "protocol_enabled": False,
+                "protocol_paired": True,
+            },
+        )
+
+        async def _must_not_run(hass, host, client_name):  # pragma: no cover
+            raise AssertionError("a paired entry must not start a pairing session")
+
+        monkeypatch.setattr(cf, "_async_start_protocol_pairing", _must_not_run)
+        manager, page = await self._open_connection(ha, entry)
+
+        result = await manager.async_configure(
+            page["flow_id"], self._connection_page(entry, protocol_enabled=True)
+        )
+
+        assert result["type"] == "menu", result
+        assert result["step_id"] == "init"
+
+
+class TestRealSetupProtocolPairing:
+    """t15: the SETUP pairing writes the marker the options flow reads.
+
+    Both halves of the pairing contract in one place: the flow that pairs records it,
+    and the skip path records the opposite, so no entry can be enabled-but-unpaired
+    after a reload — the state the old code could not tell apart from a working one.
+    """
+
+    async def _pair_through_setup(self, ha, monkeypatch, client, **form_overrides):
+        """Drive setup → protocol_pair → the PIN → the setup menu → create_entry."""
+
+        async def _begin(self):
+            return client, None
+
+        monkeypatch.setattr(
+            cf.NvidiaShieldConfigFlow, "_async_begin_protocol_pairing", _begin
+        )
+        manager, form = await open_setup(ha)
+        pairing = await manager.async_configure(
+            form["flow_id"],
+            first_form(adb_enabled=False, protocol_enabled=True, **form_overrides),
+        )
+        assert pairing["type"] == "form" and pairing["step_id"] == "protocol_pair"
+        return manager, pairing
+
+    async def test_a_successful_setup_pairing_sets_the_marker(self, ha, monkeypatch):
+        client = _FakePairingClient()
+        manager, pairing = await self._pair_through_setup(ha, monkeypatch, client)
+
+        after_code = await manager.async_configure(
+            pairing["flow_id"], {"pairing_code": "1234", "host": "10.0.0.9"}
+        )
+
+        assert after_code["type"] == "menu", after_code
+        assert client.finish_calls == ["1234"]
+        created = await manager.async_configure(
+            after_code["flow_id"], {"next_step_id": "finish"}
+        )
+        assert created["type"] == "create_entry", created
+        # t15: the marker travels with the new entry — the options flow reads it
+        # before it lets the transport run.
+        assert created["options"]["protocol_paired"] is True
+        assert created["options"]["protocol_enabled"] is True
+
+    async def test_skipping_the_setup_pairing_records_not_paired(self, ha, monkeypatch):
+        client = _FakePairingClient()
+        manager, pairing = await self._pair_through_setup(ha, monkeypatch, client)
+
+        after_skip = await manager.async_configure(
+            pairing["flow_id"],
+            {"pairing_code": "", "host": "10.0.0.9", "skip_pairing": True},
+        )
+
+        assert after_skip["type"] == "create_entry", after_skip
+        assert after_skip["options"]["protocol_enabled"] is False
+        assert after_skip["options"]["protocol_paired"] is False
+        assert client.finish_calls == []
+
+
 class TestRealAppsMultilineField:
     """t64: the apps field is a MULTILINE text field — the live bug's cause."""
 
diff --git a/tests/test_ha_api_drift.py b/tests/test_ha_api_drift.py
index f8380fc..ba1520f 100644
--- a/tests/test_ha_api_drift.py
+++ b/tests/test_ha_api_drift.py
@@ -369,9 +369,69 @@ def test_the_media_player_extra_attributes_reach_the_state_machine():
 
     _entity_id, _state, attrs = _write_ha_state(entity, states)
 
-    assert {"app_id", "app_name"} <= set(attrs), sorted(attrs)
+    # t15: ``app_source`` names the tier that produced the app identity, so the
+    # protocol push can never look like the ADB reading it outranked.
+    assert {"app_id", "app_name", "app_source"} <= set(attrs), sorted(attrs)
 
 
+def test_the_remote_entity_publishes_the_activity_contract_under_real_ha():
+    """t15: the app/activity picker contract, measured against real HA.
+
+    Core androidtv_remote's remote is the parity target. HA renders the picker from
+    ``RemoteEntityFeature.ACTIVITY`` plus ``activity_list``/``current_activity``,
+    published through the base class's FINAL ``state_attributes`` — so this asserts
+    what the STATE MACHINE receives, not what our own attributes happen to say. The
+    transport evidence rides the same entity, which is what makes "paired?" and
+    "switched on?" answerable without a device test.
+    """
+    _require_installed_homeassistant()
+
+    from homeassistant.components.remote import RemoteEntityFeature
+
+    entity, states = _real_ha_entity(
+        "remote",
+        "NvidiaShieldRemoteEntity",
+        options={
+            "app_name_map": {
+                "com.netflix.ninja": "Netflix",
+                "ar.tvplayer.tv": "TiviMate",
+            }
+        },
+        entity_id="remote.shield",
+    )
+
+    assert entity.supported_features == RemoteEntityFeature.ACTIVITY  # exactly core's
+    assert entity.activity_list == ["Netflix", "TiviMate"]
+    assert entity.current_activity is None, "nothing is known yet — never fabricated"
+
+    _entity_id, _state, attrs = _write_ha_state(entity, states)
+
+    assert attrs["activity_list"] == ["Netflix", "TiviMate"]
+    assert attrs["current_activity"] is None
+    # ...and the transport's real state, on the same entity (no runtime_data here,
+    # so there is no transport at all — which must read as "off", not as unpaired)
+    assert attrs["protocol_state"] == "off"
+    assert attrs["protocol_connected"] is False
+    assert attrs["protocol_ever_authenticated"] is False
+    assert attrs["power_on_route"] == "protocol"
+    assert attrs["protocol_current_app_seen"] is False
+
+    # The LIVE update paths our entity uses: HA's own metaclass turns the class-level
+    # ``_attr_*`` declarations into properties backed by ``__attr_*`` plus cache
+    # invalidation, so both assignments must take effect on the published state. A
+    # private attribute name would leave ``activity_list``/``current_activity`` frozen
+    # at their first value — silent in a unit test, wrong in a dashboard.
+    entity._attr_current_activity = "Netflix"
+    entity._attr_activity_list = ["Netflix", "TiviMate", "YouTube"]
+
+    assert entity.current_activity == "Netflix"
+    assert entity.activity_list == ["Netflix", "TiviMate", "YouTube"]
+
+    _entity_id, _state, attrs = _write_ha_state(entity, states)
+
+    assert attrs["current_activity"] == "Netflix"
+    assert attrs["activity_list"] == ["Netflix", "TiviMate", "YouTube"]
+
 #: The same real 2x1 PNG fixture tests/test_screen_capture_image.py uses (the
 #: wire form ``screencap -p | base64`` produces) — a real PNG, not a stand-in.
 _SCREEN_PNG_B64 = (
diff --git a/tests/test_protocol_transport.py b/tests/test_protocol_transport.py
index 5ed32e3..bbf3a5b 100644
--- a/tests/test_protocol_transport.py
+++ b/tests/test_protocol_transport.py
@@ -1,10 +1,12 @@
 """Unit tests for the protocol transport (androidtvremote2, Phase 2, t8).
 
-Fake client mirrors AndroidTVRemote's surface (callback registration,
-``keep_reconnecting`` that owns its own reconnect task, sync ``disconnect``);
+Fake client mirrors AndroidTVRemote's surface (callback registration, an
+``async_connect`` that must run BEFORE ``keep_reconnecting``, a self-owned
+reconnect task that is a no-op until a session exists, sync ``disconnect``);
 no real gRPC / no Home Assistant. Also covers `volume_tuple` normalization
 against the real library's VolumeInfo TypedDict when the package is present
 (`uvx --with androidtvremote2 pytest`).
+
 """
 
 import asyncio
@@ -13,9 +15,12 @@ import time
 
 import pytest
 from nsh.connection import (
+    FIELD_AUDIO_PLAYBACK,
     FIELD_AWAKE,
+    FIELD_FOREGROUND,
     FIELD_IS_AVAILABLE,
     FIELD_IS_ON,
+    FIELD_RUNNING_APPS,
     FIELD_SESSIONS,
     FIELD_VOLUME_INFO,
     ShieldConnectionManager,
@@ -24,10 +29,38 @@ from nsh.connection import (
 )
 
 
-class FakeProtocolClient:
-    """AndroidTVRemote-shaped fake: callbacks + self-owned reconnect task."""
+class InvalidAuth(Exception):  # noqa: N818 - the library's own class name
+    """The library's InvalidAuth, matched by NAME (the manager never imports it)."""
 
-    def __init__(self):
+
+class CannotConnect(Exception):  # noqa: N818 - the library's own class name
+    """The library's CannotConnect — a retryable transport failure."""
+
+
+class ConnectionClosed(Exception):  # noqa: N818 - the library's own class name
+    """The library's ConnectionClosed — raised when nothing is connected."""
+
+
+class FakeProtocolClient:
+    """AndroidTVRemote-shaped fake: the library's ORDER, reproduced.
+
+    t15: the real ``AndroidTVRemote`` keeps ``_remote_message_protocol = None`` until
+    ``async_connect()`` succeeds, and ``keep_reconnecting()`` starts a loop whose
+    condition is ``while self._remote_message_protocol:`` — so calling it alone
+    starts a task that returns immediately and NOTHING ever connects (no push, and
+    every send raises ConnectionClosed). The fake reproduces that, and records which
+    of the two calls the manager made, so a regression back to keep_reconnecting-only
+    fails a test instead of silently disabling the transport.
+    """
+
+    def __init__(self, connect_error: Exception | None = None):
+        self._remote_message_protocol = None  # library attribute: None until connected
+        self._connect_error = connect_error
+        self.connect_calls = 0
+        self.keep_reconnecting_calls = 0
+        self.keys_sent: list = []
+        self.apps_launched: list = []
+        self.texts_sent: list = []
         self._is_on_callbacks: list = []
         self._is_available_callbacks: list = []
         self._current_app_callbacks: list = []
@@ -66,19 +99,51 @@ class FakeProtocolClient:
         for cb in list(self._volume_callbacks):
             cb(info)
 
+
     # -- lifecycle --
     def keep_reconnecting(self, invalid_auth_callback=None):
-        """Own task like the real library (0.1 s → 30 s backoff internally)."""
+        """Own task like the real library — and, like it, a NO-OP until connected.
+
+        The condition is the library's own: with no session the task runs once,
+        finds nothing to maintain and returns. That is exactly why
+        ``async_connect()`` has to come first.
+        """
+        self.keep_reconnecting_calls += 1
         self._invalid_auth_callback = invalid_auth_callback
         loop = asyncio.get_event_loop()
         self._reconnect_task = loop.create_task(self._reconnect_loop())
 
     async def _reconnect_loop(self):
-        while True:  # the real task sleeps through its internal backoff
+        while self._remote_message_protocol:  # the library's own guard
             await asyncio.sleep(0.05)
 
+    # -- connect (t15: the call the manager used to be missing entirely) --
+    async def async_connect(self):
+        """Connect; raise ``self._connect_error`` when the test wants a failure."""
+        self.connect_calls += 1
+        if self._connect_error is not None:
+            raise self._connect_error
+        self._remote_message_protocol = object()  # library sets its RemoteProtocol
+
+    # -- sends (t15: the library raises when nothing is connected) --
+    def send_key_command(self, key, direction="SHORT"):
+        if not self._remote_message_protocol:
+            raise ConnectionClosed("Called send_key_command after disconnect")
+        self.keys_sent.append((key, direction))
+
+    def send_launch_app_command(self, app_link_or_app_id):
+        if not self._remote_message_protocol:
+            raise ConnectionClosed("Called send_launch_app_command after disconnect")
+        self.apps_launched.append(app_link_or_app_id)
+
+    def send_text(self, text):
+        if not self._remote_message_protocol:
+            raise ConnectionClosed("Called send_text after disconnect")
+        self.texts_sent.append(text)
+
     def fire_invalid_auth(self):
-        self._invalid_auth_callback()
+        if self._invalid_auth_callback is not None:
+            self._invalid_auth_callback()
 
     def disconnect(self):
         self.disconnect_calls += 1
@@ -95,6 +160,16 @@ async def _drain_loop():
     await asyncio.sleep(0.01)
 
 
+async def _connect(manager, client, tries: int = 40):
+    """Wait for the manager's background connect task to make its attempt (t15)."""
+    for _ in range(tries):
+        if client.connect_calls:
+            await _drain_loop()
+            return
+        await asyncio.sleep(0.005)
+    raise AssertionError("the manager never attempted to connect the protocol")
+
+
 class TestPushCallbackWiring:
     async def test_is_on_and_availability_push_into_store(self):
         client = FakeProtocolClient()
@@ -142,6 +217,7 @@ class TestInvalidAuthHook:
             client, invalid_auth_callback=lambda: fired.append(True)
         )
         await manager._start_protocol()
+        await _connect(manager, client)  # t15: a session first, then pushes/auth
 
         client.fire_invalid_auth()
 
@@ -152,10 +228,202 @@ class TestInvalidAuthHook:
         client = FakeProtocolClient()
         manager = _protocol_manager(client)
         await manager._start_protocol()
+        await _connect(manager, client)
         client.fire_invalid_auth()  # no callback registered — must not raise
         await manager.async_stop()
 
 
+def _fast_backoff(manager, delay: float = 0.01) -> None:
+    """Make the connect-retry backoff fast for tests (t15), per manager."""
+    manager._jitter = lambda _delay: delay
+
+
+class TestInitialConnect:
+    """t15: the transport must CONNECT, not merely be told to keep reconnecting.
+
+    The shipped bug: `_start_protocol` called `keep_reconnecting()` and nothing in
+    the tree ever called `async_connect()`. The library's reconnect task is
+    `while self._remote_message_protocol:` — None until a first connect — so the
+    task returned immediately: no session, no push, every send raising
+    ConnectionClosed inside a swallowed handler, for every install. These tests pin
+    the order (connect first) and the state it reports.
+    """
+
+    async def test_connects_before_handing_over_to_keep_reconnecting(self):
+        client = FakeProtocolClient()
+        manager = _protocol_manager(client)
+        await manager._start_protocol()
+        await _connect(manager, client)
+
+        assert client.connect_calls == 1
+        assert client.keep_reconnecting_calls == 1
+        assert client._remote_message_protocol is not None, "a session exists now"
+        assert client._reconnect_task is not None, (
+            "keep_reconnecting only maintains a session; a task with no session "
+            "would be the pre-t15 no-op"
+        )
+        assert manager.protocol_state == "connected"
+        assert manager.protocol_connected is True
+        assert manager.protocol_ever_authenticated is True
+        assert manager.protocol_last_error is None
+        await manager.async_stop()
+
+    async def test_keep_reconnecting_alone_would_stay_unconnected(self):
+        """The fake reproduces the library: without async_connect nothing happens."""
+        client = FakeProtocolClient()
+        client.keep_reconnecting(lambda: None)
+        await _drain_loop()
+
+        assert client._reconnect_task.done(), (
+            "the library's loop returns at once with no session — which is why "
+            "keep_reconnecting() alone never connected anything"
+        )
+        assert client.connect_calls == 0
+
+    async def test_network_failure_keeps_retrying_and_stays_never_connected(self):
+        client = FakeProtocolClient(connect_error=CannotConnect("unreachable"))
+        manager = _protocol_manager(client)
+        _fast_backoff(manager)
+        await manager._start_protocol()
+        await _connect(manager, client)
+
+        assert manager.protocol_state == "never_connected"
+        assert manager.protocol_connected is False
+        assert "CannotConnect" in (manager.protocol_last_error or "")
+        assert client.keep_reconnecting_calls == 0, (
+            "nothing to maintain until a session exists"
+        )
+
+        for _ in range(200):  # the retry loop must keep trying
+            if client.connect_calls >= 2:
+                break
+            await asyncio.sleep(0.005)
+        assert client.connect_calls >= 2, "a network failure is retried with backoff"
+        assert manager.protocol_ever_authenticated is False
+        await manager.async_stop()
+
+    async def test_auth_rejection_stops_and_runs_the_reauth_hook(self):
+        client = FakeProtocolClient(connect_error=InvalidAuth("Need to pair"))
+        fired = []
+        manager = _protocol_manager(
+            client, invalid_auth_callback=lambda: fired.append(True)
+        )
+        await manager._start_protocol()
+        await _connect(manager, client)
+
+        assert manager.protocol_state == "auth_rejected"
+        assert fired == [True], "the entry's reauth/pair flow is the answer"
+        assert "InvalidAuth" in (manager.protocol_last_error or "")
+        assert client.keep_reconnecting_calls == 0
+
+        connect_calls = client.connect_calls
+        await asyncio.sleep(0.05)
+        assert client.connect_calls == connect_calls, (
+            "retrying a rejected certificate can never succeed"
+        )
+        await manager.async_stop()
+
+    async def test_auth_rejection_after_a_session_marks_the_state(self):
+        client = FakeProtocolClient()
+        manager = _protocol_manager(client)
+        await manager._start_protocol()
+        await _connect(manager, client)
+        assert manager.protocol_state == "connected"
+
+        client.fire_invalid_auth()
+
+        assert manager.protocol_state == "auth_rejected"
+        assert manager.protocol_last_error  # readable, not only logged
+        await manager.async_stop()
+
+
+class TestHonestSendFailures:
+    """t15: a send that could not run must never look like one that did."""
+
+    async def test_send_without_a_session_warns_and_records(self, caplog):
+        client = FakeProtocolClient()
+        manager = _protocol_manager(client)
+        await manager._start_protocol()
+        await _connect(manager, client)
+        client._remote_message_protocol = None  # session lost (library no-ops sends)
+
+        with caplog.at_level("WARNING"):
+            sent = await manager.send_protocol_key("POWER")
+
+        assert sent is False
+        assert manager.protocol_failures == 1
+        assert "POWER" in (manager.protocol_last_error or "")
+        assert "ConnectionClosed" in (manager.protocol_last_error or "")
+        assert any(
+            "did not run" in rec.message for rec in caplog.records
+        ), "the failure is a WARNING carrying what happened"
+        assert client.keys_sent == []
+        await manager.async_stop()
+
+    async def test_launch_without_a_transport_warns_and_records(self, caplog):
+        manager = _protocol_manager(None)  # protocol disabled for this entry
+        with caplog.at_level("WARNING"):
+            sent = await manager.launch_app("com.netflix.ninja")
+
+        assert sent is False
+        assert manager.protocol_failures == 1
+        assert "no protocol transport" in (manager.protocol_last_error or "")
+        assert any("did not run" in rec.message for rec in caplog.records)
+        await manager.async_stop()
+
+    async def test_a_working_send_is_not_counted_as_a_failure(self):
+        client = FakeProtocolClient()
+        manager = _protocol_manager(client)
+        await manager._start_protocol()
+        await _connect(manager, client)
+
+        assert await manager.send_protocol_key("POWER") is True
+        assert await manager.launch_app("com.netflix.ninja") is True
+
+        assert client.keys_sent == [("POWER", "SHORT")]
+        assert client.apps_launched == ["com.netflix.ninja"]
+        assert manager.protocol_failures == 0
+        assert manager.protocol_last_error is None
+        await manager.async_stop()
+
+    async def test_availability_push_drives_the_connected_state(self):
+        client = FakeProtocolClient()
+        manager = _protocol_manager(client)
+        await manager._start_protocol()
+        await _connect(manager, client)
+        client.push_is_available(False)
+        await _drain_loop()
+        assert manager.protocol_state == "disconnected"
+        assert manager.protocol_connected is False
+
+        client.push_is_available(True)
+        await _drain_loop()
+        assert manager.protocol_state == "connected"
+        await manager.async_stop()
+
+    async def test_a_push_proves_an_authenticated_session(self):
+        """The library exposes no is_paired: a push is the only proof there is."""
+        client = FakeProtocolClient(connect_error=CannotConnect("down"))
+        manager = _protocol_manager(client)
+        await manager._start_protocol()
+        await _connect(manager, client)
+        assert manager.protocol_ever_authenticated is False
+
+        client.push_current_app("com.netflix.ninja")
+        await _drain_loop()
+
+        assert manager.protocol_ever_authenticated is True
+        await manager.async_stop()
+
+    async def test_transport_state_is_off_without_a_client(self):
+        manager = _protocol_manager(None)
+        await manager._start_protocol()
+        assert manager.protocol_state == "off"
+        assert manager.protocol_connected is False
+        assert manager.protocol_ever_authenticated is False
+        await manager.async_stop()
+
+
 class TestLifecycle:
     async def test_stop_disconnects_lib_owned_reconnect_task(self):
         """The library owns keep_reconnecting's task; closing the client
@@ -163,6 +431,7 @@ class TestLifecycle:
         client = FakeProtocolClient()
         manager = _protocol_manager(client)
         await manager._start_protocol()
+        await _connect(manager, client)
         task = client._reconnect_task
         assert task is not None
 
@@ -315,6 +584,115 @@ class TestAdbOutageKeepsProtocolTruth:
         await manager.async_stop()
 
 
+class TestAdbStaysThePrimarySource:
+    """t15: arming the protocol must not promote its pushes over ADB's readings.
+
+    The pushes this unit turns ON (is_on / is_available / current_app / volume) had
+    NEVER been live here — ``keep_reconnecting`` alone never connected — so the whole
+    protocol surface is new traffic next to an ADB media-session poller the user
+    relies on. The store's fields are disjoint by design; these tests pin that, and
+    that an ADB-only entry keeps working with only the protocol surface degraded.
+    """
+
+    @staticmethod
+    def _adbreath(manager):
+        """Every field the ADB poller owns, stamped, and read back."""
+        manager.store.set(FIELD_AWAKE, True)
+        manager.store.set(FIELD_SESSIONS, ["session-data"])
+        manager.store.set(FIELD_AUDIO_PLAYBACK, ["com.audio.last"])
+        manager.store.set(FIELD_RUNNING_APPS, [{"package": "com.run.now"}])
+        manager.store.set(FIELD_FOREGROUND, "com.adb.foreground")
+        return {
+            field: manager.store.get(field)
+            for field in (
+                FIELD_AWAKE,
+                FIELD_SESSIONS,
+                FIELD_AUDIO_PLAYBACK,
+                FIELD_RUNNING_APPS,
+                FIELD_FOREGROUND,
+            )
+        }
+
+    async def test_a_protocol_push_never_overwrites_an_adb_reading(self):
+        client = FakeProtocolClient()
+        manager = _protocol_manager(client)
+        await manager._start_protocol()
+        await _connect(manager, client)
+        before = self._adbreath(manager)
+
+        client.push_is_on(True)
+        client.push_is_available(True)
+        client.push_current_app("com.protocol.foreground")
+        client.push_volume_info({"level": 5, "max": 15, "muted": False})
+        await _drain_loop()
+
+        assert {
+            field: manager.store.get(field) for field in before
+        } == before, "ADB's own fields are untouched by every protocol push"
+        # ...and the protocol wrote exactly its own four
+        assert manager.is_on is True
+        assert manager.is_available is True
+        assert manager.current_app == "com.protocol.foreground"
+        assert manager.volume_info == (5.0, 15.0, False)
+        await manager.async_stop()
+
+    async def test_the_foreground_identity_falls_back_to_adb(self):
+        """No push yet: the ADB parse answers, and nothing is invented."""
+        client = FakeProtocolClient(connect_error=CannotConnect("down"))
+        manager = _protocol_manager(client)
+        await manager._start_protocol()
+        await _connect(manager, client)
+        self._adbreath(manager)
+
+        assert manager.current_app is None, "no protocol push ever arrived"
+        assert manager.foreground_app == "com.adb.foreground"
+        assert manager.foreground_source == "adb", "the tier is named, not implied"
+        await manager.async_stop()
+
+    async def test_a_push_wins_while_fresh_and_the_adb_reading_survives_it(self):
+        """The documented priority, and no loss: an expired push re-exposes ADB."""
+        now = [1000.0]
+        store = StateStore(ttl=60.0, clock=lambda: now[0])
+        client = FakeProtocolClient()
+        manager = _protocol_manager(client, store=store)
+        await manager._start_protocol()
+        await _connect(manager, client)
+        self._adbreath(manager)
+        client.push_current_app("com.protocol.foreground")
+        await _drain_loop()
+
+        assert manager.foreground_app == "com.protocol.foreground"
+        assert manager.foreground_source == "protocol"
+        assert manager.store.get(FIELD_FOREGROUND) == "com.adb.foreground", (
+            "the push wrote its OWN field — the ADB reading was not clobbered"
+        )
+
+        now[0] += 60.1  # both fields expire on the store's own TTL, not by a push
+
+        assert manager.current_app is None
+        assert manager.store.get(FIELD_FOREGROUND) is None
+        # re-stamped (the next ADB poll), the ADB reading answers again untouched
+        manager.store.set(FIELD_FOREGROUND, "com.adb.foreground")
+        assert manager.foreground_app == "com.adb.foreground"
+        assert manager.foreground_source == "adb"
+        await manager.async_stop()
+
+    async def test_an_adb_only_entry_keeps_working(self):
+        """No protocol client at all: ADB truth refreshes, only the transport degrades."""
+        poller = FakeAdbForOutage()
+        poller.connected = True
+        poller.awake = True
+        manager = ShieldConnectionManager(hass=None, entry=None, poller=poller)
+        await manager._start_protocol()
+
+        assert manager._protocol is None
+        assert manager.protocol_state == "off"
+        assert manager.protocol_connected is False
+        assert manager.protocol_ever_authenticated is False
+
+        await manager.async_stop()
+
+
 class TestImportGuard:
     def test_missing_library_degrades_to_adb_only(self, monkeypatch, caplog):
         """F1: a protocol-enabled entry without androidtvremote2 keeps