feat(media_player): foreground tier in the state ladder with app-source now-playing (last-known → last_channel → audio hint), sensor reads the store's foreground priority chain (P3-3a,c)

custom_components/nvidia_shield/media_player.py

@@ -131,6 +131,9 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
         # In-flight channel digit entry (play_media(channel) — §7); a new
         # play_media cancels the previous sequence (core task-cancel guard).
         self._channel_task = None
+        # Last channel entered via play_media(channel) — the foreground
+        # tier's title fallback when an app publishes no session (P3-3a).
+        self._last_channel: str | None = None
         # Entity-owned background tasks (intent retries, media-info fetch);
         # cancelled on unload. The manager owns the socket — never touch it
         # here directly.
@@ -468,6 +471,13 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
             if adb_info.paused:
                 return MediaPlayerState.PAUSED
 
+        # Tier 3.5 (P3-3): foreground fallback — an app is open (protocol
+        # current_app push, else the ADB mResumedActivity parse) but publishes
+        # no media session. The device is still ON; the now-playing tier
+        # carries the last-known/channel/audio-hint identity.
+        if self._manager is not None and self._manager.foreground_app:
+            return MediaPlayerState.ON
+
         # Tier 4: On but no media
         return MediaPlayerState.ON
 
@@ -622,6 +632,36 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
             np["album"] = info.album
         return np
 
+    def _get_app_now_playing(self) -> dict | None:
+        """Foreground-tier now-playing (P3-3a): an app is open but publishes
+        no media session (TiviMate, R2). Title = last known state, else the
+        last channel entered, else the zero-cost audio-playback hint.
+        """
+        manager = self._manager
+        if manager is None:
+            return None
+        foreground = manager.foreground_app
+        if not foreground:
+            return None
+
+        state = "idle"
+        title = None
+        info = manager.last_known_session(foreground)
+        if info is not None and (info.playing or info.paused):
+            state = "playing" if info.playing else "paused"
+            # Privacy mode: identity stays the app, never the media title
+            title = None if not self._adb_store_info else info.title
+        if title is None:
+            title = self._last_channel
+        if title is None and not self._adb_store_info:
+            title = self.app_name or self.app_id or foreground
+        if title is None and foreground in manager.audio_playback:
+            # E hint (zero round-trip cost): audio was recently played
+            title = "Media recently active"
+        if title is None:
+            return None  # bare foreground app — the state stays ON regardless
+        return {"state": state, "title": title, "source": "app"}
+
     def _get_power_data(self) -> dict | None:
         """Read current values from discovered power sensors."""
         if not self._power_sensors:
@@ -656,10 +696,13 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
 
     @property
     def extra_state_attributes(self) -> dict:
-        # Now playing: prefer Jellyfin, fall back to ADB media session
+        # Now playing: prefer Jellyfin, fall back to ADB media session,
+        # then the foreground tier (P3-3a).
         now_playing = self._get_now_playing()
         if not now_playing:
             now_playing = self._get_adb_now_playing()
+        if not now_playing:
+            now_playing = self._get_app_now_playing()
 
         attrs = {"app_id": self.app_id, "app_name": self.app_name,
                  ATTR_MEDIA_DAG_ROLE: "source",
@@ -802,6 +845,7 @@ class NvidiaShieldMediaPlayer(MediaPlayerEntity):
             if not digits:
                 _LOGGER.warning("play_media(channel): no digits in %r", media_id)
                 return
+            self._last_channel = "".join(digits)
             self._cancel_inflight_channel()
             self._channel_task = self._track_task(
                 self._channel_entry_sequence(digits)

custom_components/nvidia_shield/sensor.py

@@ -105,9 +105,13 @@ class NvidiaShieldCurrentAppSensor(NvidiaShieldSensorBase):
         if state:
             return state.attributes.get("app_name") or state.attributes.get("app_id")
 
-        # androidtv entity unavailable — StateStore fallback (ADB recents)
+        # androidtv entity unavailable — StateStore priority chain (P3-3c):
+        # protocol current_app push → ADB mResumedActivity → ADB recents
         manager = self._manager()
         if manager is not None:
+            foreground = manager.foreground_app
+            if foreground:
+                return foreground
             running = manager.running_apps
             if running:
                 return running[0].get("package")

tests/test_command_routing.py

@@ -113,6 +113,7 @@ def make_player(modules, manager=None, jellyfin_entity=None, options=None):
     player._entry = entry
     player._manager = manager
     player._intent = None
+    player._last_channel = None
     player.hass = entity_stublib.FakeHass()
     player._entity_tasks = set()
     player._channel_task = None
@@ -380,6 +381,95 @@ class TestAppFallback:
         assert player.app_id == "com.google.android.youtube.tv"
 
 
+class TestForegroundLadderTier:
+    """P3-3a: the ladder branch between the media-session tier and ON."""
+
+    def _player(self, modules, sessions=None, foreground=None, audio=None):
+        player, _s, _i = make_player(modules)
+        manager = player._manager
+        manager.store.set("awake", True)
+        if sessions:
+            manager.store.set("sessions", sessions)
+        if foreground:
+            manager.store.set("foreground", foreground)
+        if audio:
+            manager.store.set("audio_playback", audio)
+        player._androidtv_entity = ""
+        player._read = lambda eid: None
+        return player, manager
+
+    async def test_session_app_serves_playing_via_session_tier(self, modules):
+        from nsh.ha_adb_common import MediaSessionInfo
+
+        sessions = [
+            MediaSessionInfo(package="com.netflix.ninja", playing=True, title="Movie")
+        ]
+        player, manager = self._player(
+            modules, sessions=sessions, foreground="com.netflix.ninja"
+        )
+
+        assert player.state.value == "playing"  # session tier, not the fg tier
+        np = player._get_adb_now_playing()
+        assert np["source"] == "media_session"
+        assert np["title"] == "Movie"
+
+    async def test_no_session_app_with_audio_hint(self, modules):
+        player, _m = self._player(
+            modules,
+            foreground="ar.tvplayer.tv",
+            audio=["ar.tvplayer.tv"],
+        )
+
+        # TiviMate publishes no session (R2) — fg tier keeps the device ON
+        assert player.state.value == "on"
+        np = player._get_app_now_playing()
+        assert np == {
+            "state": "idle",
+            "title": "Media recently active",
+            "source": "app",
+        }
+
+    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._get_app_now_playing() is None
+
+    async def test_last_known_title_beats_audio_hint(self, modules):
+        from nsh.ha_adb_common import MediaSessionInfo
+
+        player, manager = self._player(
+            modules, foreground="ar.tvplayer.tv", audio=["ar.tvplayer.tv"]
+        )
+        manager._clock = lambda: 0.0  # stamp 0.0, clock 0.0 → inside TTL
+        info = MediaSessionInfo(package="ar.tvplayer.tv", playing=True, title="Sky News")
+        manager._last_known["ar.tvplayer.tv"] = (info, 0.0)
+
+        np = player._get_app_now_playing()
+        assert np == {"state": "playing", "title": "Sky News", "source": "app"}
+
+    async def test_ttl_resurrect_guard_in_the_tier(self, modules):
+        from nsh.ha_adb_common import MediaSessionInfo
+
+        now = 1000.0
+        player, manager = self._player(modules, foreground="ar.tvplayer.tv")
+        player._last_channel = "101"
+
+        info = MediaSessionInfo(package="ar.tvplayer.tv", playing=True, title="Sky News")
+        manager._clock = lambda: now  # the closure reads the live `now` cell
+        manager._last_known["ar.tvplayer.tv"] = (info, now)
+
+        # inside TTL → last-known title served
+        np = player._get_app_now_playing()
+        assert np == {"state": "playing", "title": "Sky News", "source": "app"}
+
+        # past STATE_STORE_TTL → 'playing' must NOT be resurrected;
+        # the tier falls back to the last channel instead
+        now = now + 61.0
+        np = player._get_app_now_playing()
+        assert np == {"state": "idle", "title": "101", "source": "app"}
+
+
 class TestVolumeAndPowerLadder:
     async def test_volume_readout_from_protocol_push(self, modules):
         player, served, _intents = make_player(modules)