fix(media_player): the state ladder is idle/playing/buffering/off; ON means no data (t55)

The live report: the card showed "On" where the user expected Idle,
Playing, Buffering or Off, and "On" appeared to override playing states.

Root cause: `state` returned MediaPlayerState.ON for two different
situations — "powered on with an app open and nothing playing" (tier 3.5)
and "we have no idea" (the fallthrough) — while the power gate above the
media tiers returned IDLE whenever power resolved to None, which is
exactly the case in which Jellyfin a media session DO know what is
playing. Media truth was therefore lost whenever the protocol push and
the ADB wake poll had not reported yet.

The ladder is now:

  1. OFF     — the power entity says off, or the protocol is_on push is
               False. Explicit power outranks media, so a stale Jellyfin
               session cannot resurrect playback on a device that is off.
  2. intent  — a just-sent play/pause/stop.
  3. Jellyfin — playing / paused.
  4. session  — the ADB media session: buffering / playing / paused.
  5. IDLE     — ADB standby (powered but asleep), or the device is on
               (protocol is_on / ADB awake / the plug) with nothing
               playing. "Powered on, app open, nothing playing" is IDLE.
  6. ON       — nothing reported anything at all: no transport, no power
               entity, no Jellyfin, no session. ON never masks a
               playing/paused/buffering source and never appears while
               any source reported a power or media state.

Buffering also carries the playback controls: a buffering session
advertises PLAY/PAUSE/STOP exactly like a playing one, and the
now-playing dict names it "buffering" instead of collapsing it into
"playing" and contradicting the entity state.

The module docstring documents the ladder and the ON-only-with-no-data
rule, and the new TestStateModelT55 / TestCurrentAppNameEndToEndT55 pin
one test per source (Jellyfin, intent, media session, power entity,
protocol push, ADB wake) that a reporting source outranks ON/IDLE.

Red-proof against the previous tree (new tests, old source): 21 failed /
17 passed, with the three headline symptoms visible as literal asserts —
"on" where "idle" belongs, "idle" where "on" belongs (no data), and
"playing" lost because power was unknown.

tests/test_command_routing.py

@@ -9,6 +9,7 @@ layer marks which tier served each command.
 import ast
 import asyncio
 import sys
+from types import SimpleNamespace
 
 import entity_stublib
 import pytest
@@ -429,7 +430,7 @@ def _ladder_player(modules, foreground=None, audio=None):
 
 
 class TestForegroundLadderTier:
-    """P3-3a: the ladder branch between the media-session tier and ON."""
+    """P3-3a/t55: the ladder branch between the media-session tier and IDLE."""
 
     def _player(self, modules, sessions=None, foreground=None, audio=None):
         player, _s, _i = make_player(modules)
@@ -466,8 +467,9 @@ class TestForegroundLadderTier:
             audio=["ar.tvplayer.tv"],
         )
 
-        # TiviMate publishes no session (R2) — fg tier keeps the device ON
-        assert player.state.value == "on"
+        # TiviMate publishes no session (R2) — an open app, nothing playing:
+        # the t55 ladder reports IDLE, never ON ("On" is reserved for no data)
+        assert player.state.value == "idle"
         np = player._get_app_now_playing()
         assert np == {
             "state": "idle",
@@ -478,7 +480,7 @@ class TestForegroundLadderTier:
     async def test_no_session_app_without_hint_returns_none(self, modules):
         player, _m = self._player(modules, foreground="com.spocky.projengmenu")
 
-        assert player.state.value == "on"  # launcher foreground → still ON
+        assert player.state.value == "idle"  # launcher open, nothing playing
         assert player._get_app_now_playing() is None
 
     async def test_last_known_title_beats_audio_hint(self, modules):
@@ -516,6 +518,311 @@ class TestForegroundLadderTier:
         assert np == {"state": "idle", "title": "101", "source": "app"}
 
 
+class _MpState:
+    """A HA State-shaped stand-in for the power entity / Jellyfin entity."""
+
+    def __init__(self, state, attributes=None):
+        self.state = state
+        self.attributes = attributes or {}
+
+
+def _state_player(
+    modules,
+    *,
+    awake=None,
+    sessions=None,
+    foreground=None,
+    protocol_on=None,
+    power_entity_state=None,
+    jellyfin_state=None,
+):
+    """A player whose power/media sources are exactly what the test sets.
+
+    ``None`` means "this source has not reported" — the distinction the t55
+    ladder turns on, so every test spells out which sources spoke.
+    """
+    player, _served, _intents = make_player(modules)
+    manager = player._manager
+    if awake is not None:
+        manager.store.set("awake", awake)
+    if sessions is not None:
+        manager.store.set("sessions", sessions)
+    if foreground is not None:
+        manager.store.set("foreground", foreground)
+    if protocol_on is not None:
+        manager.store.set("is_on", protocol_on)
+
+    entry = player._entry
+    entry.options = dict(entry.options or {})
+    if power_entity_state is not None:
+        entry.options["power_entity"] = "switch.shield_plug"
+    if jellyfin_state is not None:
+        entry.options["jellyfin_entity"] = "media_player.jellyfin"
+
+    def _read(entity_id):
+        if entity_id == "switch.shield_plug" and power_entity_state is not None:
+            return _MpState(power_entity_state)
+        if entity_id == "media_player.jellyfin" and jellyfin_state is not None:
+            return _MpState(jellyfin_state, {"media_title": "Movie"})
+        return None
+
+    player._read = _read
+    return player, manager
+
+
+class TestStateModelT55:
+    """t55: exactly {playing, paused, buffering, idle, off} + ON for no data.
+
+    The live defect: ``state`` returned ON for "powered on with an open app but
+    nothing playing", which users read as "On overrides my playing states" and
+    which sat ABOVE the media tiers in one branch. Every source that reports
+    playing/paused/buffering must win, and ON must only describe the case where
+    no source reported anything at all.
+    """
+
+    # ── the five states ──
+
+    async def test_no_data_at_all_is_the_only_on(self, modules):
+        player, _m = _state_player(modules)
+
+        assert player.state.value == "on"
+
+    async def test_protocol_off_is_off(self, modules):
+        player, _m = _state_player(modules, protocol_on=False)
+
+        assert player.state.value == "off"
+
+    async def test_power_entity_off_is_off_even_with_a_session(self, modules):
+        """Explicit OFF outranks a stale media source."""
+        from nsh.ha_adb_common import MediaSessionInfo
+
+        player, _m = _state_player(
+            modules,
+            awake=True,
+            power_entity_state="off",
+            sessions=[MediaSessionInfo(package="com.netflix.ninja", playing=True)],
+        )
+
+        assert player.state.value == "off"
+
+    async def test_open_app_with_nothing_playing_is_idle(self, modules):
+        """The headline fix: ON no longer masks "on but nothing playing"."""
+        player, _m = _state_player(
+            modules, awake=True, foreground="ar.tvplayer.tv", protocol_on=True
+        )
+
+        assert player.state.value == "idle"
+
+    async def test_powered_on_with_no_app_is_idle(self, modules):
+        player, _m = _state_player(modules, protocol_on=True)
+
+        assert player.state.value == "idle"
+
+    async def test_power_entity_on_is_idle_not_on(self, modules):
+        """A plug reporting 'on' IS a source — ON is reserved for no data."""
+        player, _m = _state_player(modules, power_entity_state="on")
+
+        assert player.state.value == "idle"
+
+    async def test_adb_standby_is_idle(self, modules):
+        player, _m = _state_player(modules, awake=False)
+
+        assert player.state.value == "idle"
+
+    # ── each source outranks the idle/on fallbacks ──
+
+    async def test_jellyfin_playing_outranks_the_idle_fallback(self, modules):
+        player, _m = _state_player(
+            modules, awake=True, foreground="ar.tvplayer.tv", jellyfin_state="playing"
+        )
+
+        assert player.state.value == "playing"
+
+    async def test_jellyfin_playing_outranks_no_power_data(self, modules):
+        """No transport, no power entity — Jellyfin still knows better than ON."""
+        player, _m = _state_player(modules, jellyfin_state="playing")
+
+        assert player.state.value == "playing"
+
+    async def test_media_session_buffering_outranks_the_idle_fallback(self, modules):
+        from nsh.ha_adb_common import MediaSessionInfo
+
+        player, _m = _state_player(
+            modules,
+            awake=True,
+            foreground="com.netflix.ninja",
+            sessions=[
+                MediaSessionInfo(package="com.netflix.ninja", buffering=True)
+            ],
+        )
+
+        assert player.state.value == "buffering"
+
+    async def test_media_session_playing_outranks_the_idle_fallback(self, modules):
+        from nsh.ha_adb_common import MediaSessionInfo
+
+        player, _m = _state_player(
+            modules,
+            awake=True,
+            foreground="com.netflix.ninja",
+            sessions=[MediaSessionInfo(package="com.netflix.ninja", playing=True)],
+        )
+
+        assert player.state.value == "playing"
+
+    async def test_media_session_paused_outranks_the_idle_fallback(self, modules):
+        from nsh.ha_adb_common import MediaSessionInfo
+
+        player, _m = _state_player(
+            modules,
+            awake=True,
+            foreground="com.netflix.ninja",
+            sessions=[MediaSessionInfo(package="com.netflix.ninja", paused=True)],
+        )
+
+        assert player.state.value == "paused"
+
+    async def test_intent_playing_outranks_the_idle_fallback(self, modules):
+        player, _m = _state_player(modules, awake=True, foreground="ar.tvplayer.tv")
+        player._intent = SimpleNamespace(target_state="playing")
+
+        assert player.state.value == "playing"
+
+    async def test_live_session_beats_a_stale_cached_one(self, modules):
+        """Session state 3/4 keep behaving as before: live truth, then cache."""
+        from nsh.ha_adb_common import MediaSessionInfo
+
+        player, manager = _state_player(
+            modules,
+            awake=True,
+            foreground="com.netflix.ninja",
+            sessions=[MediaSessionInfo(package="com.netflix.ninja", paused=True)],
+        )
+        info = MediaSessionInfo(package="com.netflix.ninja", playing=True, title="Live")
+        manager._last_known["com.netflix.ninja"] = (info, manager._clock())
+
+        assert player.state.value == "paused"  # the live session, not the cache
+
+    # ── vocabulary guard: no other value can be produced ──
+
+    def test_the_vocabulary_is_exactly_the_five_states_plus_on(self, modules):
+        from pathlib import Path
+
+        from nsh import media_player as mp
+
+        # Checked on the AST (the t37 idiom), not on prose: every state the
+        # module can return is in the documented vocabulary, and the
+        # media-tier values really are all used.
+        tree = ast.parse(Path(mp.__file__).read_text(encoding="utf-8"))
+        used = {
+            node.attr
+            for node in ast.walk(tree)
+            if isinstance(node, ast.Attribute)
+            and isinstance(node.value, ast.Name)
+            and node.value.id == "MediaPlayerState"
+        }
+        assert used <= {"PLAYING", "PAUSED", "BUFFERING", "IDLE", "OFF", "ON"}
+        assert {"PLAYING", "PAUSED", "BUFFERING", "IDLE", "OFF"} <= used
+        # The documented vocabulary itself, as the user reads it on the card.
+        assert {s.lower() for s in used} == {
+            "playing",
+            "paused",
+            "buffering",
+            "idle",
+            "off",
+            "on",
+        }
+
+    async def test_on_never_masks_a_reported_power_state(self, modules):
+        """Every power-reporting shape must be off or idle — never on."""
+        shapes = [
+            {"protocol_on": True},
+            {"protocol_on": False},
+            {"awake": True},
+            {"awake": False},
+            {"power_entity_state": "on"},
+            {"power_entity_state": "off"},
+        ]
+        for kwargs in shapes:
+            player, _m = _state_player(modules, **kwargs)
+            assert player.state.value != "on", kwargs
+
+    # ── BUFFERING: state, controls, now-playing ──
+
+    async def test_buffering_advertises_the_playback_controls(self, modules):
+        from entity_stublib import MediaPlayerEntityFeature
+        from nsh.ha_adb_common import MediaSessionInfo
+
+        player, _m = _state_player(
+            modules,
+            awake=True,
+            foreground="com.netflix.ninja",
+            sessions=[MediaSessionInfo(package="com.netflix.ninja", buffering=True)],
+        )
+
+        features = player.supported_features
+        assert features & MediaPlayerEntityFeature.PLAY
+        assert features & MediaPlayerEntityFeature.PAUSE
+        assert features & MediaPlayerEntityFeature.STOP
+
+    async def test_no_session_still_means_base_features_only(self, modules):
+        from entity_stublib import MediaPlayerEntityFeature
+
+        player, _m = _state_player(modules, awake=True, foreground="ar.tvplayer.tv")
+
+        assert not (player.supported_features & MediaPlayerEntityFeature.PLAY)
+
+    async def test_buffering_now_playing_is_named_buffering(self, modules):
+        from nsh.ha_adb_common import MediaSessionInfo
+
+        player, _m = _state_player(
+            modules,
+            awake=True,
+            foreground="com.netflix.ninja",
+            sessions=[
+                MediaSessionInfo(
+                    package="com.netflix.ninja", buffering=True, title="Movie"
+                )
+            ],
+        )
+
+        np = player._get_adb_now_playing()
+        assert np["state"] == "buffering"
+        assert np["title"] == "Movie"
+
+    async def test_buffering_survives_privacy_mode(self, modules):
+        from nsh.ha_adb_common import MediaSessionInfo
+
+        player, _m = _state_player(
+            modules,
+            awake=True,
+            foreground="com.netflix.ninja",
+            sessions=[
+                MediaSessionInfo(
+                    package="com.netflix.ninja", buffering=True, title="Secret"
+                )
+            ],
+        )
+        player._entry.options["adb_store_app_info"] = False
+
+        np = player._get_adb_now_playing()
+        assert np["state"] == "buffering"
+        assert np["title"] != "Secret"
+
+    async def test_last_known_buffering_reaches_the_foreground_tier(self, modules):
+        from nsh.ha_adb_common import MediaSessionInfo
+
+        player, manager = _state_player(modules, awake=True, foreground="ar.tvplayer.tv")
+        manager._clock = lambda: 0.0
+        info = MediaSessionInfo(
+            package="ar.tvplayer.tv", buffering=True, title="Sky News"
+        )
+        manager._last_known["ar.tvplayer.tv"] = (info, 0.0)
+
+        np = player._get_app_now_playing()
+        assert np == {"state": "buffering", "title": "Sky News", "source": "app"}
+
+
 class TestChannelLeakGuard:
     """F1: the channel number never leaks to a non-channel foreground app."""
 
@@ -1259,7 +1566,7 @@ class TestCurrentAppSensorFriendlyName:
     def test_cached_manifest_label_is_the_second_tier(self, modules):
         player, _served, _intents = make_player(modules)
         manager = player._manager
-        manager._app_labels["com.netflix.ninja"] = ("Netflix", manager._clock())
+        manager.cache_app_label("com.netflix.ninja", "Netflix")
         sensor = self._sensor(modules, player)
 
         manager._on_current_app("com.netflix.ninja")
@@ -1270,7 +1577,7 @@ class TestCurrentAppSensorFriendlyName:
     def test_recent_apps_fallback_resolves_through_the_label_tier(self, modules):
         player, _served, _intents = make_player(modules)
         manager = player._manager
-        manager._app_labels["ar.tvplayer.tv"] = ("TiviMate", manager._clock())
+        manager.cache_app_label("ar.tvplayer.tv", "TiviMate")
         manager.store.set("running_apps", [{"package": "ar.tvplayer.tv"}])
         sensor = self._sensor(modules, player)
 
@@ -1299,6 +1606,180 @@ class TestCurrentAppSensorFriendlyName:
         assert sensor.native_value == player.app_name == "TiviMate"
 
 
+class _LabelPoller(FakeAdbPoller):
+    """FakeAdbPoller + the surface a real watchdog cycle needs.
+
+    ``is_awake``/``poll_media_sessions``/``all_sessions``/``poll_running_apps``
+    make ``_refresh`` a harmless cycle that still reaches the
+    ``_drain_app_label_requests`` call, so a test drives exactly the
+    one-fetch-per-cycle behaviour the label tier depends on.
+    """
+
+    def __init__(self):
+        super().__init__()
+        self.pm_dumps: dict[str, str] = {}
+        self.dumps: list[str] = []
+
+    @property
+    def connected(self):
+        # FakeAdbPoller's property is always True; restated so this class reads
+        # on its own and the watchdog cycle spends no retry on connect.
+        return True
+
+    async def is_awake(self) -> bool:
+        return True
+
+    async def poll_media_sessions(self, foreground_package=None, head_lines=200):
+        return None
+
+    @property
+    def all_sessions(self) -> list:
+        return []
+
+    @property
+    def last_dump(self) -> str:
+        return ""
+
+    async def poll_running_apps(self) -> list[dict]:
+        return []
+
+    async def shell(self, cmd: str) -> str | None:
+        self.dumps.append(cmd)
+        import shlex
+
+        for package, dump in self.pm_dumps.items():
+            if shlex.quote(package) in cmd or package in cmd:
+                return dump
+        return None
+
+
+class TestCurrentAppNameEndToEndT55:
+    """t55: the label fetch is driven off the CURRENT app, and a miss retries.
+
+    The live symptom was the Current App sensor (and app_name) showing a
+    package id. These drive the real chain: sensor render → app_label cache
+    miss → the fetch the manager enqueues → the next watchdog cycle drains it
+    for the package that is actually on screen, NOT some earlier app.
+    """
+
+    def _wire(self, modules, current_app, dumps):
+        player, _served, _intents = make_player(modules)
+        manager = player._manager
+        poller = _LabelPoller()
+        poller.pm_dumps.update(dumps)
+        manager._poller = poller
+        player._entry.runtime_data = manager
+        sensor = make_app_sensor(modules, manager, player._entry)
+
+        manager._on_current_app(current_app)
+        return player, manager, sensor, poller
+
+    async def test_known_app_resolves_by_fetching_its_label(self, modules):
+        player, manager, sensor, poller = self._wire(
+            modules, "com.netflix.ninja", {"com.netflix.ninja": "application-label: 'Netflix'"}
+        )
+
+        # first render: cache miss → the sensor falls through to the package
+        assert sensor.native_value == "com.netflix.ninja"
+        assert manager.app_label_requests == {"com.netflix.ninja"}
+
+        # the watchdog drains the fetch for the CURRENT package
+        await manager._watchdog_cycle(5.0)
+        await asyncio.sleep(0)
+        assert poller.dumps and "com.netflix.ninja" in poller.dumps[0]
+
+        # next render: the resolved device name, in the sensor AND the player
+        assert sensor.native_value == "Netflix"
+        assert sensor.extra_state_attributes["app_name"] == "Netflix"
+        assert player.app_name == "Netflix"
+
+    async def test_mapped_app_never_needs_the_device_label(self, modules):
+        player, manager, sensor, poller = self._wire(
+            modules, "ar.tvplayer.tv", {"ar.tvplayer.tv": "application-label: 'TiviMate'"}
+        )
+        player._entry.options["app_name_map"] = {"ar.tvplayer.tv": "TiviMate TV"}
+
+        assert sensor.native_value == "TiviMate TV"
+        await manager._watchdog_cycle(5.0)
+        await asyncio.sleep(0)
+
+        # the explicit map wins and no pm dump was issued for it
+        assert sensor.native_value == "TiviMate TV"
+        assert not [d for d in poller.dumps if "pm dump" in d]
+
+    async def test_unknown_app_falls_through_and_the_next_poll_retries(self, modules):
+        """The requirement, end to end: no permanent miss for an unresolvable app.
+
+        A failed fetch must not be cached as an answer: the next poll tries
+        again. Only the fetch that RAN may stamp the negative.
+        """
+        player, manager, sensor, poller = self._wire(
+            modules, "com.example.unknown", {}
+        )
+        real_shell = poller.shell
+        failures = {"n": 0}
+
+        async def _flaky(cmd):
+            if "pm dump" in cmd:
+                failures["n"] += 1
+                raise RuntimeError("adb shell blew up")
+            return await real_shell(cmd)
+
+        poller.shell = _flaky
+
+        # poll 1: fetch fails → still the package id, nothing cached
+        assert sensor.native_value == "com.example.unknown"
+        await manager._watchdog_cycle(5.0)
+        await asyncio.sleep(0)
+        assert sensor.native_value == "com.example.unknown"
+        assert failures["n"] == 1
+        assert "com.example.unknown" not in manager._app_labels
+
+        # poll 2: the render after the failure RE-ENQUEUES it — the retry is
+        # driven by the next render, exactly like the live poll cycle does
+        assert manager.app_label("com.example.unknown") is None
+        assert manager.app_label_requests == {"com.example.unknown"}
+        await manager._watchdog_cycle(5.0)
+        await asyncio.sleep(0)
+        assert failures["n"] == 2
+        assert "com.example.unknown" not in manager._app_labels
+
+        # poll 3: the shell answers, the dump carries no label → a definitive
+        # negative is cached and the sensor keeps falling through to the id
+        poller.shell = real_shell
+        assert manager.app_label("com.example.unknown") is None
+        await manager._watchdog_cycle(5.0)
+        await asyncio.sleep(0)
+        assert sensor.native_value == "com.example.unknown"
+        assert "com.example.unknown" in manager._app_labels
+
+    async def test_the_fetch_targets_the_app_on_screen_not_the_previous_one(self, modules):
+        player, manager, sensor, poller = self._wire(
+            modules,
+            "ar.tvplayer.tv",
+            {
+                "ar.tvplayer.tv": "application-label: 'TiviMate'",
+                "com.google.android.youtube.tv": "application-label: 'YouTube'",
+            },
+        )
+
+        assert sensor.native_value == "ar.tvplayer.tv"  # miss for the 1st app
+        manager._on_current_app("com.google.android.youtube.tv")
+        assert sensor.native_value == "com.google.android.youtube.tv"
+
+        # both packages are queued; the watchdog drains ONE per cycle
+        await manager._watchdog_cycle(5.0)
+        await asyncio.sleep(0)
+        await manager._watchdog_cycle(5.0)
+        await asyncio.sleep(0)
+
+        # every drained fetch names a package, and the app that is CURRENT is
+        # the one the sensor now shows the label of
+        assert "ar.tvplayer.tv" in poller.dumps[0]
+        assert "com.google.android.youtube.tv" in poller.dumps[1]
+        assert sensor.native_value == "YouTube"
+
+
 class TestAvailabilityLadder:
     """t40: ADB enabled-but-unreachable + a healthy protocol = AVAILABLE.