diff --git a/custom_components/nvidia_shield/connection.py b/custom_components/nvidia_shield/connection.py
new file mode 100644
index 0000000..b7688e6
--- /dev/null
+++ b/custom_components/nvidia_shield/connection.py
@@ -0,0 +1,321 @@
+"""ShieldConnectionManager — entry-scoped owner of the ADB connection (Phase 1).
+
+The ADB socket, its watchdog and all ADB-derived state live here instead of on
+the media-player entity (plan blockers B1/B2/B5):
+
+- one persistent socket; a watchdog task reconnects with exponential backoff
+  (5 s doubling to a 300 s cap, ±25 % jitter) when it drops, and pulls the
+  heartbeat data (wake + media sessions + running apps) while connected;
+- ADB-derived values live in a
+  :class:`StateStore` with per-field timestamps and TTL expiry — stale values
+  read as expired (None/empty) instead of clinging on;
+- entities are thin callback-views: they ``register_listener()`` and write HA
+  state when the store changes (the androidtv core ``entity.py`` pattern);
+- ADB is optional: when no poller factory is provided the manager is *passive*
+  and the entry still sets up — a failed ADB connect is a background retry,
+  never a ``ConfigEntryNotReady``.
+
+This module deliberately stays Home-Assistant-free at import time so it is
+unit-testable with plain pytest; every hass capability is injected (the entry
+builds the concrete poller factory). Task creation additionally works without
+hass (plain ``asyncio.create_task``) for the unit tests.
+"""
+
+import asyncio
+import contextlib
+import logging
+import random
+import time
+from collections.abc import Callable
+from typing import Any
+
+from .adb_session import AdbMediaSessionPoller
+from .const import (
+    ADB_POLL_INTERVAL,
+    ADB_RECONNECT_BACKOFF_CAP,
+    ADB_RECONNECT_BACKOFF_START,
+    ADB_RECONNECT_JITTER,
+    STATE_STORE_TTL,
+)
+from .ha_adb_common import MediaSessionInfo, parse_audio_playback
+
+_LOGGER = logging.getLogger(__name__)
+
+# StateStore field names
+FIELD_AWAKE = "awake"
+FIELD_SESSIONS = "sessions"
+FIELD_AUDIO_PLAYBACK = "audio_playback"
+FIELD_RUNNING_APPS = "running_apps"
+
+ALL_FIELDS = (FIELD_AWAKE, FIELD_SESSIONS, FIELD_AUDIO_PLAYBACK, FIELD_RUNNING_APPS)
+
+
+class StateStore:
+    """Timed per-field state cache shared by all entities of one entry.
+
+    Reading an expired (or missing) field returns the caller-supplied
+    default, so a stale "playing" session can never outlive the TTL (plan
+    blocker B5). Values are written only via :meth:`set`, which re-stamps the
+    field's freshness timestamp.
+    """
+
+    def __init__(self, ttl: float = STATE_STORE_TTL, clock: Callable[[], float] = time.monotonic):
+        self._ttl = ttl
+        self._clock = clock
+        self._values: dict[str, tuple[float, Any]] = {}
+
+    def set(self, name: str, value: Any) -> None:
+        """Store a value, re-stamping its freshness timestamp."""
+        self._values[name] = (self._clock(), value)
+
+    def get(self, name: str, default: Any = None) -> Any:
+        """Return the field's value, or ``default`` when it is missing/expired."""
+        stamped = self._values.get(name)
+        if stamped is None:
+            return default
+        at, value = stamped
+        if self._clock() - at > self._ttl:
+            return default
+        return value
+
+    def clear(self, *names: str) -> None:
+        """Drop fields immediately so they read as expired on the next get."""
+        for name in names:
+            self._values.pop(name, None)
+
+    @property
+    def ttl(self) -> float:
+        return self._ttl
+
+
+class ShieldConnectionManager:
+    """Owner of one entry's persistent ADB connection and state.
+
+    :param poller_factory: builds the ADB poller when the watchdog starts;
+        skipped entirely for entries without ADB (passive manager) or tests
+        that inject a ready-to-use ``poller`` directly.
+    """
+
+    def __init__(
+        self,
+        hass=None,
+        entry=None,
+        poller_factory: Callable[[], AdbMediaSessionPoller] | None = None,
+        poller: AdbMediaSessionPoller | None = None,
+        store: StateStore | None = None,
+    ):
+        self.hass = hass
+        self.entry = entry
+        self.store = store if store is not None else StateStore(ttl=STATE_STORE_TTL)
+        self._poller_factory = poller_factory
+        self._poller = poller
+        self._listeners: list[Callable[[], None]] = []
+        self._tasks: list[asyncio.Task] = []
+        self._stopping = False
+        self._failures = 0
+        self._down_since = False  # listeners already told the connection is down
+
+    # ── Introspection (thin views for entities) ──
+
+    @property
+    def available(self) -> bool:
+        """True when ADB is active for this entry (a poller exists)."""
+        return self._poller is not None
+
+    @property
+    def connected(self) -> bool:
+        return bool(self._poller and self._poller.connected)
+
+    @property
+    def awake(self) -> bool | None:
+        """Device wake state (exact mWakefulness parse), None = unknown."""
+        return self.store.get(FIELD_AWAKE)
+
+    @property
+    def active_sessions(self) -> list[MediaSessionInfo]:
+        """Stored media sessions, sorted playing > paused > other."""
+        return self.store.get(FIELD_SESSIONS, [])
+
+    @property
+    def audio_playback(self) -> list[str]:
+        """Packages that recently played audio, ordered last-played-first."""
+        return self.store.get(FIELD_AUDIO_PLAYBACK, [])
+
+    @property
+    def running_apps(self) -> list[dict]:
+        """Visible recent-task apps, ordered by recency."""
+        return self.store.get(FIELD_RUNNING_APPS, [])
+
+    def best_session(
+        self, foreground_package: str | None = None
+    ) -> MediaSessionInfo | None:
+        """Best media session: foreground app first, then playing > paused."""
+        sessions = self.active_sessions
+        if not sessions:
+            return None
+        if foreground_package:
+            for session in sessions:
+                if session.package == foreground_package:
+                    return session
+        return sessions[0]
+
+    # ── Listener API (entities are thin callback-views) ──
+
+    def register_listener(self, listener: Callable[[], None]) -> Callable[[], None]:
+        """Register a refresh callback; returns an unsubscribe callable."""
+        self._listeners.append(listener)
+
+        def _unregister() -> None:
+            with contextlib.suppress(ValueError):
+                self._listeners.remove(listener)
+
+        return _unregister
+
+    def _notify(self) -> None:
+        for listener in tuple(self._listeners):
+            try:
+                listener()
+            except Exception:
+                _LOGGER.exception("Shield state listener failed")
+
+    # ── Delegation (commands over the managed socket) ──
+
+    async def send_media_command(self, command: str) -> bool:
+        """Send a media key command over the persistent socket."""
+        if not self.available:
+            return False
+        return await self._poller.send_media_command(command)
+
+    async def bring_to_foreground(self, package: str) -> bool:
+        if not self.available:
+            return False
+        return await self._poller.bring_to_foreground(package)
+
+    # ── Lifecycle ──
+
+    async def async_start(self) -> None:
+        """Start the persistent connection loop.
+
+        Never validates the ADB connection here: setup must not fail because
+        of ADB (it is optional), so all failure handling lives in the
+        watchdog's background retry.
+        """
+        self._stopping = False
+        self._tasks.append(self._create_task(self._watchdog(), "shield_adb_watchdog"))
+
+    async def async_stop(self) -> None:
+        """Cancel all manager tasks and close the socket.
+
+        Called on entry unload and on EVENT_HOMEASSISTANT_STOP.
+        """
+        self._stopping = True
+        tasks, self._tasks = self._tasks, []
+        for task in tasks:
+            task.cancel()
+        self._listeners.clear()
+        if self._poller is not None:
+            with contextlib.suppress(Exception):
+                await self._poller.disconnect()
+            self._poller = None
+
+    def _create_task(self, coro, name: str) -> asyncio.Task:
+        if self.hass is not None:
+            return self.hass.async_create_background_task(coro, name)
+        return asyncio.create_task(coro, name=name)
+
+    async def _watchdog(self) -> None:
+        """Persistent loop: reconnect with backoff when down, refresh when up.
+
+        A passive manager (no poller possible) exits immediately — the entry
+        recreates it when options change (full reload).
+        """
+        try:
+            if self._poller is None:
+                if self._poller_factory is None:
+                    return
+                self._poller = self._poller_factory()
+            if self._poller is None:
+                return
+
+            backoff = ADB_RECONNECT_BACKOFF_START
+            while not self._stopping:
+                delay = await self._watchdog_cycle(backoff)
+                if self._stopping:
+                    break
+                if self.connected:
+                    backoff = ADB_RECONNECT_BACKOFF_START
+                else:
+                    backoff = min(backoff * 2, ADB_RECONNECT_BACKOFF_CAP)
+                await asyncio.sleep(delay)
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            _LOGGER.exception("Shield connection watchdog crashed")
+
+    async def _watchdog_cycle(self, backoff: float) -> float:
+        """Run one watchdog cycle; return seconds to sleep before the next.
+
+        When the connection is up this is the heartbeat: wake check + (when
+        awake) media sessions + running apps, all stored with timestamps.
+        The current backoff sleeps after failures; success resets it."""
+
+        try:
+            if not self.connected and not await self._poller.connect():
+                self._note_failure(backoff)
+                return self._jitter(backoff)
+            if self._failures:
+                self._failures = 0
+                self._down_since = False
+                _LOGGER.info("ADB connection recovered")
+            await self._refresh()
+            return ADB_POLL_INTERVAL
+        except Exception as err:
+            self._note_failure(backoff)
+            _LOGGER.debug("Watchdog cycle failed: %s", err)
+            return self._jitter(backoff)
+
+    def _note_failure(self, backoff: float) -> None:
+        """Bookkeep a failed cycle: log sparingly, drop stale store, notify."""
+        self._failures += 1
+        if self._failures == 1 or self._failures % 6 == 0:
+            _LOGGER.warning(
+                "ADB connection down (%s consecutive failures) — retrying in ~%.0fs",
+                self._failures,
+                backoff,
+            )
+        if not self._down_since:
+            self._down_since = True
+            self.store.clear(*ALL_FIELDS)
+            self._notify()
+
+    @staticmethod
+    def _jitter(delay: float) -> float:
+        """Backoff with ±jitter so many clients do not reconnect in lockstep."""
+        return max(
+            1.0,
+            delay * (1.0 + random.uniform(-ADB_RECONNECT_JITTER, ADB_RECONNECT_JITTER)),
+        )
+
+    async def _refresh(self) -> None:
+        """Heartbeat: pull all ADB-derived state into the store (with stamps)."""
+        poller = self._poller
+        awake = await poller.is_awake()
+        if awake is None:
+            # Failed poll (connection dropped / device unreachable) — surface
+            # as a cycle failure instead of clinging to the last values.
+            raise ConnectionError("ADB wake poll failed")
+        self.store.set(FIELD_AWAKE, awake)
+        if awake:
+            # Full dump, no truncation (B4) — low sessions must not be lost.
+            await poller.poll_media_sessions(head_lines=None)
+            self.store.set(FIELD_SESSIONS, list(poller.all_sessions))
+            self.store.set(
+                FIELD_AUDIO_PLAYBACK, parse_audio_playback(poller.last_dump or "")
+            )
+            self.store.set(FIELD_RUNNING_APPS, await poller.poll_running_apps())
+        else:
+            # Standby: playback data is not current — expire it.
+            self.store.set(FIELD_SESSIONS, [])
+            self.store.set(FIELD_AUDIO_PLAYBACK, [])
+            self.store.set(FIELD_RUNNING_APPS, [])
+        self._notify()