diff --git a/custom_components/opendisplay/__init__.py b/custom_components/opendisplay/__init__.py
index e024a8f..4059761 100644
--- a/custom_components/opendisplay/__init__.py
+++ b/custom_components/opendisplay/__init__.py
@@ -3,7 +3,8 @@
 import asyncio
 import contextlib
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING
+import time
+from typing import TYPE_CHECKING, Any
 
 from opendisplay import (
     AuthenticationFailedError,
@@ -15,11 +16,13 @@
     OpenDisplayError,
     PartialState,
 )
+from opendisplay.models.config_json import config_from_json, config_to_json
 
 from homeassistant.components.bluetooth import (
     BluetoothReachabilityIntent,
     async_address_reachability_diagnostics,
     async_ble_device_from_address,
+    async_set_fallback_availability_interval,
 )
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import Platform
@@ -29,17 +32,29 @@
 from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH
 from homeassistant.helpers.typing import ConfigType
 
-if TYPE_CHECKING:
-    from opendisplay.models import FirmwareVersion
-
-from .const import CONF_ENCRYPTION_KEY, DOMAIN
+from .const import CONF_CACHED_STATE, CONF_ENCRYPTION_KEY, DOMAIN
 from .coordinator import OpenDisplayCoordinator
+from .delivery import DeliveryManager
 from .services import async_setup_services
+from .sleep import SleepProfile
+
+if TYPE_CHECKING:
+    from opendisplay.models import FirmwareVersion
 
 CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
 
-_BASE_PLATFORMS: list[Platform] = [Platform.IMAGE, Platform.SENSOR]
-_FLEX_PLATFORMS = [Platform.EVENT, Platform.IMAGE, Platform.SENSOR, Platform.UPDATE]
+_BASE_PLATFORMS: list[Platform] = [
+    Platform.BINARY_SENSOR,
+    Platform.IMAGE,
+    Platform.SENSOR,
+]
+_FLEX_PLATFORMS = [
+    Platform.BINARY_SENSOR,
+    Platform.EVENT,
+    Platform.IMAGE,
+    Platform.SENSOR,
+    Platform.UPDATE,
+]
 
 
 @dataclass
@@ -50,6 +65,8 @@ class OpenDisplayRuntimeData:
     firmware: FirmwareVersion
     device_config: GlobalConfig
     is_flex: bool
+    # Resolved deep-sleep behavior (options + device power config).
+    sleep_profile: SleepProfile
     upload_task: asyncio.Task | None = None
     # Serializes every BLE connection to this tag (one config entry == one MAC).
     # The device exposes a single BLE link with no per-address lock in the
@@ -60,11 +77,87 @@ class OpenDisplayRuntimeData:
     # (0x76). Replaced with a fresh instance on every full/fast refresh so the
     # next partial diffs against the frame actually on the panel.
     partial_state: PartialState = field(default_factory=PartialState)
+    # Set when the entry was set up from cache without connecting; the delivery
+    # manager re-reads firmware/config at the next wake and refreshes the cache.
+    config_resync_pending: bool = False
+    # Owns queued work delivered at the next wake (set in async_setup_entry).
+    delivery: DeliveryManager | None = None
 
 
 type OpenDisplayConfigEntry = ConfigEntry[OpenDisplayRuntimeData]
 
 
+@dataclass
+class _CachedState:
+    """Device state restored from ``entry.data`` for setup-without-connect."""
+
+    firmware: FirmwareVersion
+    is_flex: bool
+    device_config: GlobalConfig
+    landing_url: str | None
+
+
+def _load_cache(entry: OpenDisplayConfigEntry) -> _CachedState | None:
+    """Rebuild cached device state from the config entry, or None if absent."""
+    raw = entry.data.get(CONF_CACHED_STATE)
+    if not raw:
+        return None
+    try:
+        device_config = config_from_json(raw["config"])
+        return _CachedState(
+            firmware=raw["firmware"],
+            is_flex=raw["is_flex"],
+            device_config=device_config,
+            landing_url=raw.get("landing_url"),
+        )
+    except (KeyError, ValueError, TypeError):
+        return None
+
+
+def _write_cache(
+    hass: HomeAssistant,
+    entry: OpenDisplayConfigEntry,
+    device_config: GlobalConfig,
+    firmware: FirmwareVersion,
+    is_flex: bool,
+    landing_url: str | None,
+) -> None:
+    """Persist the device state needed to set up the entry without connecting.
+
+    Called after every successful interrogation (setup and, later, wake-time
+    resync). Writes only when the meaningful contents changed so we don't churn
+    the config-entry store on every reload.
+    """
+    payload: dict[str, Any] = {
+        "config": config_to_json(device_config),
+        "firmware": dict(firmware),
+        "is_flex": is_flex,
+        "landing_url": landing_url,
+    }
+    existing = entry.data.get(CONF_CACHED_STATE)
+    if existing is not None and all(
+        existing.get(key) == value for key, value in payload.items()
+    ):
+        return
+    hass.config_entries.async_update_entry(
+        entry,
+        data={**entry.data, CONF_CACHED_STATE: {**payload, "cached_at": time.time()}},
+    )
+
+
+def _cache_setup_if_sleepy(
+    entry: OpenDisplayConfigEntry,
+) -> _CachedState | None:
+    """Return cached state iff it exists and resolves to a sleepy device."""
+    cached = _load_cache(entry)
+    if cached is None:
+        return None
+    profile = SleepProfile.from_entry(entry, cached.device_config)
+    if not profile.is_sleepy:
+        return None
+    return cached
+
+
 def _get_encryption_key(entry: OpenDisplayConfigEntry) -> bytes | None:
     """Return the encryption key bytes from entry data, or None."""
     raw = entry.data.get(CONF_ENCRYPTION_KEY)
@@ -89,47 +182,73 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: OpenDisplayConfigEntry) -> bool:
-    """Set up OpenDisplay from a config entry."""
+    """Set up OpenDisplay from a config entry.
+
+    A reachable device is interrogated live (firmware + config) and the result
+    cached. When the device is dark and the cache says it is a deep-sleeping
+    tag, the entry is set up entirely from cache without connecting; the
+    delivery manager re-reads it at the next wake. Any other unreachable case
+    preserves the original ConfigEntryNotReady/ConfigEntryAuthFailed behavior.
+    """
     address = entry.unique_id
     if TYPE_CHECKING:
         assert address is not None
 
     ble_device = async_ble_device_from_address(hass, address, connectable=True)
-    if ble_device is None:
-        raise ConfigEntryNotReady(
-            translation_domain=DOMAIN,
-            translation_key="device_not_found",
-            translation_placeholders={
-                "address": address,
-                "reason": async_address_reachability_diagnostics(
-                    hass,
-                    address.upper(),
-                    BluetoothReachabilityIntent.CONNECTION,
-                ),
-            },
-        )
-    encryption_key = _get_encryption_key(entry)
-
-    try:
-        async with OpenDisplayDevice(
-            mac_address=address, ble_device=ble_device, encryption_key=encryption_key
-        ) as device:
-            fw = await device.read_firmware_version()
-            is_flex = device.is_flex
-            # Capture while connected: landing_url() reads the advertised name.
-            landing_url = device.landing_url()
-    except (AuthenticationFailedError, AuthenticationRequiredError) as err:
-        raise ConfigEntryAuthFailed(
-            f"Encryption key rejected by OpenDisplay device: {err}"
-        ) from err
-    except (BLEConnectionError, BLETimeoutError, OpenDisplayError) as err:
-        raise ConfigEntryNotReady(
-            f"Failed to connect to OpenDisplay device: {err}"
-        ) from err
-    device_config = device.config
-    if TYPE_CHECKING:
-        assert device_config is not None
+    from_cache = False
 
+    if ble_device is None:
+        cached = _cache_setup_if_sleepy(entry)
+        if cached is None:
+            raise ConfigEntryNotReady(
+                translation_domain=DOMAIN,
+                translation_key="device_not_found",
+                translation_placeholders={
+                    "address": address,
+                    "reason": async_address_reachability_diagnostics(
+                        hass,
+                        address.upper(),
+                        BluetoothReachabilityIntent.CONNECTION,
+                    ),
+                },
+            )
+        fw = cached.firmware
+        is_flex = cached.is_flex
+        device_config = cached.device_config
+        landing_url = cached.landing_url
+        from_cache = True
+    else:
+        encryption_key = _get_encryption_key(entry)
+        try:
+            async with OpenDisplayDevice(
+                mac_address=address,
+                ble_device=ble_device,
+                encryption_key=encryption_key,
+            ) as device:
+                fw = await device.read_firmware_version()
+                is_flex = device.is_flex
+                # Capture while connected: landing_url() reads the advertised name.
+                landing_url = device.landing_url()
+            device_config = device.config
+            if TYPE_CHECKING:
+                assert device_config is not None
+        except (AuthenticationFailedError, AuthenticationRequiredError) as err:
+            raise ConfigEntryAuthFailed(
+                f"Encryption key rejected by OpenDisplay device: {err}"
+            ) from err
+        except (BLEConnectionError, BLETimeoutError, OpenDisplayError) as err:
+            cached = _cache_setup_if_sleepy(entry)
+            if cached is None:
+                raise ConfigEntryNotReady(
+                    f"Failed to connect to OpenDisplay device: {err}"
+                ) from err
+            fw = cached.firmware
+            is_flex = cached.is_flex
+            device_config = cached.device_config
+            landing_url = cached.landing_url
+            from_cache = True
+
+    profile = SleepProfile.from_entry(entry, device_config)
     coordinator = OpenDisplayCoordinator(hass, address)
 
     manufacturer = device_config.manufacturer
@@ -162,17 +281,49 @@ async def async_setup_entry(hass: HomeAssistant, entry: OpenDisplayConfigEntry)
         firmware=fw,
         device_config=device_config,
         is_flex=is_flex,
+        sleep_profile=profile,
+        config_resync_pending=from_cache,
     )
 
+    # Persist a fresh interrogation so the next dark startup can set up from
+    # cache; skip when we just loaded from cache (nothing new to store).
+    if not from_cache:
+        _write_cache(hass, entry, device_config, fw, is_flex, landing_url)
+
+    # Keep entities available across sleep cycles: without this, any sleep
+    # interval longer than the ~5 min staleness horizon flaps everything
+    # unavailable once per cycle.
+    if profile.is_sleepy:
+        async_set_fallback_availability_interval(
+            hass, address, profile.availability_interval
+        )
+
+    manager = DeliveryManager(hass, entry)
+    entry.runtime_data.delivery = manager
+
     await hass.config_entries.async_forward_entry_setups(
         entry, _get_platforms(entry.runtime_data)
     )
     entry.async_on_unload(coordinator.async_start())
+    manager.async_start()
+
+    if from_cache:
+        # Re-read firmware/config on the next wake and refresh the cache.
+        manager.request_config_resync()
 
     @callback
     def _schedule_reboot_reload() -> None:
-        """Re-read firmware/config after the device signals a reboot."""
-        hass.async_create_task(_async_reload_after_reboot(hass, entry))
+        """React to the device's advertised reboot edge.
+
+        For sleepy devices the firmware's unpersisted reboot flag makes every
+        wake look like a reboot, and a reconnecting reload races the wake
+        window; route to an opportunistic config resync instead. Non-sleepy
+        devices keep the reload behavior.
+        """
+        if profile.is_sleepy:
+            manager.request_config_resync()
+        else:
+            hass.async_create_task(_async_reload_after_reboot(hass, entry))
 
     entry.async_on_unload(coordinator.async_subscribe_reboot(_schedule_reboot_reload))
 
@@ -215,9 +366,12 @@ async def async_unload_entry(
 ) -> bool:
     """Unload a config entry."""
     runtime = entry.runtime_data
-    # Abort an in-flight image upload quickly so unload is not blocked on a long
-    # BLE transfer, then wait for any other in-flight BLE op (LED, buzzer, OTA,
+    # Cancel pending deadline timers and any in-flight delivery task, then abort
+    # an in-flight image upload quickly so unload is not blocked on a long BLE
+    # transfer, then wait for any other in-flight BLE op (LED, buzzer, OTA,
     # drawcustom) to release the link before tearing the entry down.
+    if runtime.delivery is not None:
+        await runtime.delivery.async_shutdown()
     if (task := runtime.upload_task) and not task.done():
         task.cancel()
         with contextlib.suppress(asyncio.CancelledError):
diff --git a/custom_components/opendisplay/binary_sensor.py b/custom_components/opendisplay/binary_sensor.py
new file mode 100644
index 0000000..2678fee
--- /dev/null
+++ b/custom_components/opendisplay/binary_sensor.py
@@ -0,0 +1,93 @@
+"""Binary sensor platform for OpenDisplay devices."""
+
+from datetime import datetime, timezone
+from typing import Any
+
+from homeassistant.components.binary_sensor import BinarySensorEntity
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from . import OpenDisplayConfigEntry
+from .const import SIGNAL_PENDING_STATE
+from .delivery import DeliveryManager, DeliverySnapshot
+
+PARALLEL_UPDATES = 0
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: OpenDisplayConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up the OpenDisplay update-pending binary sensor."""
+    manager = entry.runtime_data.delivery
+    if manager is None:
+        return
+    async_add_entities(
+        [OpenDisplayUpdatePendingSensor(entry.unique_id or "", manager)]
+    )
+
+
+def _to_iso(epoch: float | None) -> str | None:
+    """Convert an epoch timestamp to an ISO string, or None."""
+    if epoch is None:
+        return None
+    return datetime.fromtimestamp(epoch, tz=timezone.utc).isoformat()
+
+
+class OpenDisplayUpdatePendingSensor(BinarySensorEntity):
+    """On while content is queued for delivery at the next wake.
+
+    Backed entirely by the delivery manager's state, so it stays available and
+    meaningful even while the device is dark.
+    """
+
+    _attr_has_entity_name = True
+    _attr_translation_key = "update_pending"
+
+    def __init__(self, address: str, manager: DeliveryManager) -> None:
+        """Initialize the binary sensor from the manager's current state."""
+        self._address = address
+        self._attr_unique_id = f"{address}-update_pending"
+        self._attr_device_info = DeviceInfo(
+            connections={(CONNECTION_BLUETOOTH, address)},
+        )
+        self._apply(manager.state)
+
+    @callback
+    def _apply(self, snapshot: DeliverySnapshot) -> None:
+        """Store the latest delivery snapshot on the entity."""
+        self._attr_is_on = snapshot.pending
+        self._queued_at = snapshot.queued_at
+        self._expires_at = snapshot.expires_at
+        self._attempts = snapshot.attempts
+        self._last_error = snapshot.last_error
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any]:
+        """Expose queue timing details for automations/dashboards."""
+        return {
+            "queued_at": _to_iso(self._queued_at),
+            "expires_at": _to_iso(self._expires_at),
+            "attempts": self._attempts,
+            "last_error": self._last_error,
+        }
+
+    async def async_added_to_hass(self) -> None:
+        """Subscribe to delivery state updates."""
+        await super().async_added_to_hass()
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                f"{SIGNAL_PENDING_STATE}_{self._address}",
+                self._handle_pending_state,
+            )
+        )
+
+    @callback
+    def _handle_pending_state(self, snapshot: DeliverySnapshot) -> None:
+        """Handle a delivery-state change."""
+        self._apply(snapshot)
+        self.async_write_ha_state()
diff --git a/custom_components/opendisplay/config_flow.py b/custom_components/opendisplay/config_flow.py
index 622723e..cdfe9cb 100644
--- a/custom_components/opendisplay/config_flow.py
+++ b/custom_components/opendisplay/config_flow.py
@@ -19,14 +19,86 @@
     async_ble_device_from_address,
     async_discovered_service_info,
 )
-from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.config_entries import (
+    ConfigEntry,
+    ConfigFlow,
+    ConfigFlowResult,
+    OptionsFlow,
+    OptionsFlowWithReload,
+)
 from homeassistant.const import CONF_ADDRESS
+from homeassistant.core import callback
+from homeassistant.helpers.selector import (
+    BooleanSelector,
+    NumberSelector,
+    NumberSelectorConfig,
+    NumberSelectorMode,
+    SelectSelector,
+    SelectSelectorConfig,
+    SelectSelectorMode,
+)
 
-from .const import CONF_ENCRYPTION_KEY, DOMAIN
+from .const import (
+    CONF_ENCRYPTION_KEY,
+    CONF_MISSED_CYCLES,
+    CONF_PROBE_BEFORE_QUEUE,
+    CONF_QUEUE_TIMEOUT_HOURS,
+    CONF_SLEEP_MODE,
+    DEFAULT_MISSED_CYCLES,
+    DEFAULT_PROBE_BEFORE_QUEUE,
+    DEFAULT_QUEUE_TIMEOUT_HOURS,
+    DEFAULT_SLEEP_MODE,
+    DOMAIN,
+    SLEEP_MODE_AUTO,
+    SLEEP_MODE_OFF,
+    SLEEP_MODE_ON,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
 
+def _options_schema() -> vol.Schema:
+    """Return the options-flow schema."""
+    return vol.Schema(
+        {
+            vol.Required(
+                CONF_SLEEP_MODE, default=DEFAULT_SLEEP_MODE
+            ): SelectSelector(
+                SelectSelectorConfig(
+                    options=[SLEEP_MODE_AUTO, SLEEP_MODE_ON, SLEEP_MODE_OFF],
+                    translation_key="sleep_mode",
+                    mode=SelectSelectorMode.DROPDOWN,
+                )
+            ),
+            vol.Required(CONF_MISSED_CYCLES, default=DEFAULT_MISSED_CYCLES): vol.All(
+                NumberSelector(
+                    NumberSelectorConfig(
+                        min=1, max=100, step=1, mode=NumberSelectorMode.BOX
+                    )
+                ),
+                vol.Coerce(int),
+            ),
+            vol.Required(
+                CONF_QUEUE_TIMEOUT_HOURS, default=DEFAULT_QUEUE_TIMEOUT_HOURS
+            ): vol.All(
+                NumberSelector(
+                    NumberSelectorConfig(
+                        min=1,
+                        max=168,
+                        step=1,
+                        mode=NumberSelectorMode.BOX,
+                        unit_of_measurement="h",
+                    )
+                ),
+                vol.Coerce(int),
+            ),
+            vol.Required(
+                CONF_PROBE_BEFORE_QUEUE, default=DEFAULT_PROBE_BEFORE_QUEUE
+            ): BooleanSelector(),
+        }
+    )
+
+
 _ENCRYPTION_KEY_VALIDATOR = vol.All(str.strip, str.lower, vol.Match(r"^[0-9a-f]{32}$"))
 
 
@@ -38,6 +110,12 @@ def __init__(self) -> None:
         self._discovery_info: BluetoothServiceInfoBleak | None = None
         self._discovered_devices: dict[str, BluetoothServiceInfoBleak] = {}
 
+    @staticmethod
+    @callback
+    def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
+        """Return the options flow handler."""
+        return OpenDisplayOptionsFlow()
+
     async def _async_test_connection(
         self, address: str, encryption_key: bytes | None = None
     ) -> None:
@@ -241,3 +319,27 @@ async def async_step_reauth_confirm(
             description_placeholders={"name": reauth_entry.title},
             errors=errors,
         )
+
+
+class OpenDisplayOptionsFlow(OptionsFlowWithReload):
+    """Handle deep-sleep options for an OpenDisplay device.
+
+    Extends ``OptionsFlowWithReload`` so saving changed options automatically
+    reloads the entry, re-resolving the sleep profile and re-applying the
+    availability interval. No manual update listener is registered (mixing the
+    two is disallowed).
+    """
+
+    async def async_step_init(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Manage the deep-sleep options."""
+        if user_input is not None:
+            return self.async_create_entry(data=user_input)
+
+        return self.async_show_form(
+            step_id="init",
+            data_schema=self.add_suggested_values_to_schema(
+                _options_schema(), self.config_entry.options
+            ),
+        )
diff --git a/custom_components/opendisplay/const.py b/custom_components/opendisplay/const.py
index 779b74c..ee9143f 100644
--- a/custom_components/opendisplay/const.py
+++ b/custom_components/opendisplay/const.py
@@ -2,4 +2,44 @@
 
 DOMAIN = "opendisplay"
 CONF_ENCRYPTION_KEY = "encryption_key"
+
+# Dispatcher signals (suffixed with the device address at send/subscribe time).
+# Carries the JPEG preview of the frame shown by the image entity.
 SIGNAL_IMAGE_UPDATED = f"{DOMAIN}_image_updated"
+# Carries a DeliverySnapshot describing the delivery manager's pending state
+# (see delivery.py). Consumed by the image entity and the "update pending"
+# binary sensor.
+SIGNAL_PENDING_STATE = f"{DOMAIN}_pending_state"
+
+# --- Options (options flow) -------------------------------------------------
+# Deep-sleep handling mode: follow the device power config, or force on/off.
+CONF_SLEEP_MODE = "sleep_mode"
+SLEEP_MODE_AUTO = "auto"
+SLEEP_MODE_ON = "on"
+SLEEP_MODE_OFF = "off"
+DEFAULT_SLEEP_MODE = SLEEP_MODE_AUTO
+
+# Consecutive expected wakes that may be missed before entities go unavailable.
+CONF_MISSED_CYCLES = "missed_cycles"
+DEFAULT_MISSED_CYCLES = 3
+
+# Hours a queued upload survives before expiring (safely above the 18 h max
+# firmware sleep interval).
+CONF_QUEUE_TIMEOUT_HOURS = "queue_timeout_hours"
+DEFAULT_QUEUE_TIMEOUT_HOURS = 24
+
+# Probe a probably-asleep device with one short connect attempt before queuing
+# an image send. Catches wake adverts the scanner missed and Silabs tags that
+# advertise continuously despite a battery+deep-sleep power config. Bounded
+# cost: one attempt with a short timeout (see services.PROBE_CONNECT_TIMEOUT_S).
+CONF_PROBE_BEFORE_QUEUE = "probe_before_queue"
+DEFAULT_PROBE_BEFORE_QUEUE = True
+
+# --- Cache ------------------------------------------------------------------
+# entry.data key holding the serialized device state used to set up the entry
+# without connecting when a sleepy device is dark at startup.
+CONF_CACHED_STATE = "cached_state"
+
+# --- Bus events -------------------------------------------------------------
+EVENT_CONTENT_DELIVERED = f"{DOMAIN}_content_delivered"
+EVENT_CONTENT_EXPIRED = f"{DOMAIN}_content_expired"
diff --git a/custom_components/opendisplay/coordinator.py b/custom_components/opendisplay/coordinator.py
index ec743a0..2858ff4 100644
--- a/custom_components/opendisplay/coordinator.py
+++ b/custom_components/opendisplay/coordinator.py
@@ -55,6 +55,10 @@ def __init__(self, hass: HomeAssistant, address: str) -> None:
         # Subscribers notified once when the advertised reboot flag goes
         # False -> True (the device rebooted since we last talked to it).
         self._reboot_callbacks: set[CALLBACK_TYPE] = set()
+        # Subscribers notified on every parsed advertisement (the device is
+        # awake and reachable right now). The delivery manager uses this as the
+        # wake rendezvous to drain queued work.
+        self._device_seen_callbacks: set[CALLBACK_TYPE] = set()
         # Reboot-flag edge detection: the device sets the advertised reboot flag
         # on boot and clears it on first connect. None until the first v1 advert.
         self._last_reboot_flag: bool | None = None
@@ -73,6 +77,21 @@ def _unsubscribe() -> None:
 
         return _unsubscribe
 
+    @callback
+    def async_subscribe_device_seen(self, callback_: CALLBACK_TYPE) -> CALLBACK_TYPE:
+        """Subscribe to "device seen" events (every parsed advertisement).
+
+        Fired after each successfully parsed advertisement, i.e. whenever the
+        device is awake and reachable. Returns an unsubscribe callback.
+        """
+        self._device_seen_callbacks.add(callback_)
+
+        @callback
+        def _unsubscribe() -> None:
+            self._device_seen_callbacks.discard(callback_)
+
+        return _unsubscribe
+
     @callback
     def _async_handle_unavailable(
         self, service_info: BluetoothServiceInfoBleak
@@ -123,6 +142,8 @@ def _async_handle_bluetooth_event(
                 button_events=button_events,
                 touch_events=touch_events,
             )
+            for device_seen_callback in list(self._device_seen_callbacks):
+                device_seen_callback()
 
         super()._async_handle_bluetooth_event(service_info, change)
 
diff --git a/custom_components/opendisplay/delivery.py b/custom_components/opendisplay/delivery.py
new file mode 100644
index 0000000..b7ffb62
--- /dev/null
+++ b/custom_components/opendisplay/delivery.py
@@ -0,0 +1,488 @@
+"""Wake-triggered delivery of queued work to deep-sleeping OpenDisplay devices.
+
+A sleeping ESP32 tag is dark most of the time and reachable only during a short
+advertising window after each timer wake. The `DeliveryManager` owns the work
+that could not be delivered immediately (a prepared image, a config resync) and
+drains it over a single BLE connection the moment the device is seen advertising
+again.
+
+Design (see docs/DEEP_SLEEP_IMPLEMENTATION_PLAN_2026-07-06.md, D4):
+
+* Latest-wins per work type — only the newest image survives, matching the
+  existing live-upload semantics.
+* One connection per wake — the device stays awake while connected, so all
+  pending work drains over a single session in priority order (upload first).
+* Transport-agnostic trigger — `notify_device_seen(source)` is the entry point;
+  today the BLE coordinator calls it, a future WiFi presence tracker could too.
+* Memory-only in v1 — a HA restart drops a pending upload (documented).
+"""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass
+import logging
+import time
+from typing import TYPE_CHECKING, Any
+
+from opendisplay import (
+    AuthenticationFailedError,
+    AuthenticationRequiredError,
+    BLEConnectionError,
+    BLETimeoutError,
+    OpenDisplayDevice,
+    OpenDisplayError,
+    PartialState,
+    RefreshMode,
+)
+
+from homeassistant.components.bluetooth import async_ble_device_from_address
+from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH
+from homeassistant.helpers.dispatcher import async_dispatcher_send
+from homeassistant.helpers.event import async_call_later
+
+from .const import (
+    CONF_ENCRYPTION_KEY,
+    EVENT_CONTENT_DELIVERED,
+    EVENT_CONTENT_EXPIRED,
+    SIGNAL_IMAGE_UPDATED,
+    SIGNAL_PENDING_STATE,
+)
+
+if TYPE_CHECKING:
+    from . import OpenDisplayConfigEntry
+
+_LOGGER = logging.getLogger(__name__)
+
+# Upper bound on a single wake's delivery attempt: connection establishment plus
+# the queued work. Bounds one bad wake so it cannot overlap the next. The device
+# stays awake while the BLE link is held, so this is sized to let a full image
+# stream over a slow link within one wake rather than to cap a healthy transfer.
+# Exceeding it is therefore treated as a genuine failure (payload too large / link
+# too slow) and surfaced loudly by ``_deliver`` instead of silently retried.
+DELIVERY_DEADLINE_S = 600.0
+
+# Give up a queued upload after this many failed wake attempts. Without this cap a
+# frame that can never be delivered (unsupported payload, link too slow) would
+# retry on every wake until ``queue_timeout`` hours later; instead it is dropped
+# with a ``content_expired`` event so the failure surfaces promptly.
+MAX_DELIVERY_ATTEMPTS = 5
+
+# Sentinel distinguishing "no key" (None) from "malformed key" during resolve.
+_KEY_INVALID = object()
+
+
+@dataclass
+class PendingUpload:
+    """A prepared image queued for delivery at the next wake."""
+
+    # (uncompressed_data, compressed_data or None, processed_image) from
+    # prepare_image() — the heavy CPU work is done once, at queue time.
+    prepared: tuple[bytes, bytes | None, Any]
+    refresh_mode: RefreshMode
+    partial_state: PartialState
+    use_measured_palettes: bool
+    preview_jpeg: bytes
+    device_id: str | None
+    queued_at: float
+    expires_at: float
+    attempts: int = 0
+    # Set when a delivery hit an auth failure; suppresses further doomed
+    # attempts until the entry reloads after reauth.
+    paused: bool = False
+    cancel_deadline: CALLBACK_TYPE | None = None
+
+
+@dataclass(frozen=True)
+class DeliveryReceipt:
+    """Result of submitting content: whether it was delivered or queued."""
+
+    status: str  # "delivered" | "queued"
+    expires_at: float | None
+
+
+@dataclass(frozen=True)
+class DeliverySnapshot:
+    """Public delivery state consumed by entities (image, binary sensor)."""
+
+    pending: bool
+    queued_at: float | None
+    expires_at: float | None
+    attempts: int
+    last_error: str | None
+
+
+class DeliveryManager:
+    """Owns queued work for one config entry and delivers it at the next wake."""
+
+    def __init__(self, hass: HomeAssistant, entry: OpenDisplayConfigEntry) -> None:
+        """Initialize the delivery manager."""
+        self._hass = hass
+        self._entry = entry
+        runtime = entry.runtime_data
+        self._coordinator = runtime.coordinator
+        self._profile = runtime.sleep_profile
+        self._address: str = entry.unique_id or ""
+
+        self._pending_upload: PendingUpload | None = None
+        self._pending_config_resync: bool = False
+        self._last_error: str | None = None
+
+        self._delivering: bool = False
+        self._delivery_task: asyncio.Task[None] | None = None
+        self._unsub_device_seen: CALLBACK_TYPE | None = None
+
+    # -- lifecycle ----------------------------------------------------------
+
+    @callback
+    def async_start(self) -> None:
+        """Subscribe to wake advertisements."""
+        self._unsub_device_seen = self._coordinator.async_subscribe_device_seen(
+            self._on_device_seen
+        )
+
+    async def async_shutdown(self) -> None:
+        """Cancel timers and any in-flight delivery, and unsubscribe."""
+        if self._unsub_device_seen is not None:
+            self._unsub_device_seen()
+            self._unsub_device_seen = None
+        if self._pending_upload is not None and self._pending_upload.cancel_deadline:
+            self._pending_upload.cancel_deadline()
+            self._pending_upload.cancel_deadline = None
+        if self._delivery_task is not None and not self._delivery_task.done():
+            self._delivery_task.cancel()
+            try:
+                await self._delivery_task
+            except asyncio.CancelledError:
+                pass
+            except Exception:  # noqa: BLE001 - shutdown must not raise
+                _LOGGER.debug("Delivery task raised during shutdown", exc_info=True)
+        self._delivery_task = None
+        self._delivering = False
+
+    # -- public state -------------------------------------------------------
+
+    @property
+    def state(self) -> DeliverySnapshot:
+        """Return the current pending-delivery state for entities."""
+        upload = self._pending_upload
+        return DeliverySnapshot(
+            pending=upload is not None,
+            queued_at=upload.queued_at if upload else None,
+            expires_at=upload.expires_at if upload else None,
+            attempts=upload.attempts if upload else 0,
+            last_error=self._last_error,
+        )
+
+    # -- submission ---------------------------------------------------------
+
+    @callback
+    def submit_upload(
+        self,
+        *,
+        prepared: tuple[bytes, bytes | None, Any],
+        refresh_mode: RefreshMode,
+        partial_state: PartialState,
+        use_measured_palettes: bool,
+        preview_jpeg: bytes,
+        device_id: str | None,
+    ) -> DeliveryReceipt:
+        """Queue a prepared image for delivery at the next wake (latest-wins)."""
+        now = time.time()
+        # Latest-wins: drop any previously queued image and its deadline timer.
+        if self._pending_upload is not None and self._pending_upload.cancel_deadline:
+            self._pending_upload.cancel_deadline()
+        expires_at = now + self._profile.queue_timeout_s
+        slot = PendingUpload(
+            prepared=prepared,
+            refresh_mode=refresh_mode,
+            partial_state=partial_state,
+            use_measured_palettes=use_measured_palettes,
+            preview_jpeg=preview_jpeg,
+            device_id=device_id,
+            queued_at=now,
+            expires_at=expires_at,
+        )
+        self._schedule_expiry(slot)
+        self._pending_upload = slot
+        self._last_error = None
+        # Show the intended frame immediately (the image entity now reflects
+        # what will be delivered, not what is on the panel).
+        async_dispatcher_send(
+            self._hass, f"{SIGNAL_IMAGE_UPDATED}_{self._address}", preview_jpeg
+        )
+        self._notify_state()
+        return DeliveryReceipt(status="queued", expires_at=expires_at)
+
+    @callback
+    def request_config_resync(self) -> None:
+        """Request a firmware/config re-read at the next wake."""
+        self._pending_config_resync = True
+        self._entry.runtime_data.config_resync_pending = True
+
+    # -- wake handling ------------------------------------------------------
+
+    @callback
+    def _on_device_seen(self) -> None:
+        """Coordinator callback: the device advertised (it is awake now)."""
+        self.notify_device_seen("ble")
+
+    @callback
+    def notify_device_seen(self, source: str = "ble") -> None:
+        """Start a delivery drain if there is pending work and none in flight."""
+        if self._delivering or not self._has_pending_work():
+            return
+        _LOGGER.debug("%s: device seen (%s); starting delivery", self._address, source)
+        self._delivering = True
+        self._delivery_task = self._entry.async_create_background_task(
+            self._hass, self._deliver(), f"opendisplay_delivery_{self._address}"
+        )
+
+    def _has_pending_work(self) -> bool:
+        """Return True if there is deliverable work queued."""
+        upload_ready = self._pending_upload is not None and not self._pending_upload.paused
+        return upload_ready or self._pending_config_resync
+
+    # -- delivery drain -----------------------------------------------------
+
+    async def _deliver(self) -> None:
+        """Open one connection and drain all pending slots in priority order."""
+        try:
+            await self._drain_once()
+        except asyncio.CancelledError:
+            raise
+        except TimeoutError as err:
+            # The whole drain exceeded DELIVERY_DEADLINE_S. Because the device
+            # stays awake while connected, hitting this ceiling is not a missed
+            # wake but a genuine failure: the payload is too large or the link too
+            # slow to complete in one wake. Record it, log loudly, and raise so it
+            # is not swallowed as a routine retry. (py-opendisplay wraps its own
+            # read/connect timeouts as BLETimeoutError, so a bare TimeoutError here
+            # can only be the asyncio.timeout(DELIVERY_DEADLINE_S) firing.)
+            self._register_attempt_failure(
+                f"delivery deadline exceeded ({DELIVERY_DEADLINE_S:.0f}s)"
+            )
+            _LOGGER.error(
+                "%s: delivery aborted after exceeding the %.0f s deadline; the "
+                "image is likely too large to stream within a single wake",
+                self._address,
+                DELIVERY_DEADLINE_S,
+            )
+            raise HomeAssistantError(
+                f"OpenDisplay {self._address}: delivery timed out after "
+                f"{DELIVERY_DEADLINE_S:.0f}s (image too large or BLE link too slow "
+                "to finish in one wake)"
+            ) from err
+        except (BLEConnectionError, BLETimeoutError) as err:
+            # Device slept mid-connect or the wake window was missed; keep the
+            # work queued and try again on the next wake.
+            self._register_attempt_failure(str(err))
+        except (AuthenticationFailedError, AuthenticationRequiredError):
+            # Bad/rotated key: pause the upload and prompt the user to reauth.
+            _LOGGER.warning("%s: delivery auth failed; starting reauth", self._address)
+            if self._pending_upload is not None:
+                self._pending_upload.paused = True
+            self._last_error = "auth"
+            self._entry.async_start_reauth(self._hass)
+            self._notify_state()
+        except OpenDisplayError as err:
+            _LOGGER.warning("%s: delivery failed: %s", self._address, err)
+            self._register_attempt_failure(str(err))
+        finally:
+            self._delivering = False
+            self._delivery_task = None
+
+    async def _drain_once(self) -> None:
+        """The connection + drain body (see `_deliver` for error handling)."""
+        key = self._resolve_key()
+        if key is _KEY_INVALID:
+            # Malformed stored key; reauth was already started.
+            return
+
+        runtime = self._entry.runtime_data
+        async with runtime.ble_lock:
+            ble_device = async_ble_device_from_address(
+                self._hass, self._address, connectable=True
+            )
+            if ble_device is None:
+                # The advertisement that woke us has already aged out; retry next.
+                self._register_attempt_failure("device not connectable")
+                return
+
+            upload = self._pending_upload
+            use_measured = upload.use_measured_palettes if upload else True
+            async with (
+                asyncio.timeout(DELIVERY_DEADLINE_S),
+                OpenDisplayDevice(
+                    mac_address=self._address,
+                    ble_device=ble_device,
+                    config=runtime.device_config,
+                    use_measured_palettes=use_measured,
+                    encryption_key=key,  # type: ignore[arg-type]
+                ) as device,
+            ):
+                if upload is not None and not upload.paused:
+                    await self._drain_upload(device, upload)
+                if self._pending_config_resync:
+                    await self._drain_resync(device)
+
+    async def _drain_upload(
+        self, device: OpenDisplayDevice, upload: PendingUpload
+    ) -> None:
+        """Send the queued prepared image and fire the delivered event."""
+        await device.upload_prepared_image(
+            upload.prepared,
+            refresh_mode=upload.refresh_mode,
+            state=upload.partial_state,
+        )
+        if upload.cancel_deadline:
+            upload.cancel_deadline()
+        self._pending_upload = None
+        self._last_error = None
+        # Bump the image entity's timestamp to when the frame actually landed.
+        async_dispatcher_send(
+            self._hass, f"{SIGNAL_IMAGE_UPDATED}_{self._address}", upload.preview_jpeg
+        )
+        self._fire_content_event(EVENT_CONTENT_DELIVERED, upload)
+        self._notify_state()
+        _LOGGER.info("%s: queued content delivered", self._address)
+
+    async def _drain_resync(self, device: OpenDisplayDevice) -> None:
+        """Re-read firmware/config over the open link and refresh the cache."""
+        # Local import avoids an import cycle (__init__ imports this module).
+        from . import _write_cache  # noqa: PLC0415
+
+        fw = await device.read_firmware_version()
+        is_flex = device.is_flex
+        landing_url = device.landing_url()
+        device_config = device.config
+        if device_config is None:
+            return
+        runtime = self._entry.runtime_data
+        runtime.firmware = fw
+        runtime.device_config = device_config
+        runtime.is_flex = is_flex
+        runtime.config_resync_pending = False
+        self._pending_config_resync = False
+        _write_cache(self._hass, self._entry, device_config, fw, is_flex, landing_url)
+        _LOGGER.debug("%s: config resync complete", self._address)
+
+    # -- expiry -------------------------------------------------------------
+
+    @callback
+    def _schedule_expiry(self, slot: PendingUpload) -> None:
+        """Arm the deadline timer that expires this queued upload."""
+
+        @callback
+        def _expired(_now: Any) -> None:
+            # Guard against a stale timer firing after latest-wins replaced it.
+            if self._pending_upload is slot:
+                self._expire_upload(slot)
+
+        slot.cancel_deadline = async_call_later(
+            self._hass, self._profile.queue_timeout_s, _expired
+        )
+
+    @callback
+    def _expire_upload(self, slot: PendingUpload) -> None:
+        """Drop an expired queued upload and fire the expired event."""
+        self._pending_upload = None
+        self._last_error = "expired"
+        _LOGGER.warning(
+            "%s: queued content expired after %s h without a wake",
+            self._address,
+            self._profile.queue_timeout_hours,
+        )
+        self._fire_content_event(EVENT_CONTENT_EXPIRED, slot)
+        self._notify_state()
+
+    @callback
+    def _give_up_upload(self, slot: PendingUpload, reason: str) -> None:
+        """Drop an upload that has exhausted ``MAX_DELIVERY_ATTEMPTS``.
+
+        Cancels the pending expiry timer, clears the slot, and fires the same
+        ``content_expired`` event as time-based expiry — the payload's
+        ``attempts == MAX_DELIVERY_ATTEMPTS`` distinguishes this cause, and the
+        entity's ``last_error`` reads ``"failed"`` rather than ``"expired"``.
+        """
+        if slot.cancel_deadline:
+            slot.cancel_deadline()
+            slot.cancel_deadline = None
+        self._pending_upload = None
+        self._last_error = "failed"
+        _LOGGER.error(
+            "%s: giving up queued content after %s failed delivery attempts (%s)",
+            self._address,
+            slot.attempts,
+            reason,
+        )
+        self._fire_content_event(EVENT_CONTENT_EXPIRED, slot)
+        self._notify_state()
+
+    # -- helpers ------------------------------------------------------------
+
+    def _register_attempt_failure(self, reason: str) -> None:
+        """Record a failed wake attempt and keep the work queued.
+
+        After ``MAX_DELIVERY_ATTEMPTS`` failures the upload is given up and
+        dropped (``_give_up_upload``) so a frame that can never be delivered does
+        not retry on every wake until ``queue_timeout``.
+        """
+        upload = self._pending_upload
+        if upload is not None:
+            upload.attempts += 1
+            if upload.attempts >= MAX_DELIVERY_ATTEMPTS:
+                self._give_up_upload(upload, reason)
+                return
+        self._last_error = reason
+        _LOGGER.debug(
+            "%s: delivery attempt %s/%s failed (%s); will retry next wake",
+            self._address,
+            upload.attempts if upload is not None else 0,
+            MAX_DELIVERY_ATTEMPTS,
+            reason,
+        )
+        self._notify_state()
+
+    def _resolve_key(self) -> bytes | None | object:
+        """Return the encryption key bytes, None, or a sentinel on bad format."""
+        raw = self._entry.data.get(CONF_ENCRYPTION_KEY)
+        if raw is None:
+            return None
+        if len(raw) != 32:
+            self._entry.async_start_reauth(self._hass)
+            return _KEY_INVALID
+        try:
+            return bytes.fromhex(raw)
+        except ValueError:
+            self._entry.async_start_reauth(self._hass)
+            return _KEY_INVALID
+
+    @callback
+    def _notify_state(self) -> None:
+        """Push the current pending state to entities."""
+        async_dispatcher_send(
+            self._hass, f"{SIGNAL_PENDING_STATE}_{self._address}", self.state
+        )
+
+    def _fire_content_event(self, event: str, slot: PendingUpload) -> None:
+        """Fire a content delivered/expired bus event for automations."""
+        device_id = slot.device_id or self._device_id()
+        self._hass.bus.async_fire(
+            event,
+            {
+                "device_id": device_id,
+                "queued_at": slot.queued_at,
+                "attempts": slot.attempts,
+            },
+        )
+
+    def _device_id(self) -> str | None:
+        """Resolve this entry's device registry id, if any."""
+        device = dr.async_get(self._hass).async_get_device(
+            connections={(CONNECTION_BLUETOOTH, self._address)}
+        )
+        return device.id if device else None
diff --git a/custom_components/opendisplay/diagnostics.py b/custom_components/opendisplay/diagnostics.py
index 0fd89c3..14a6b57 100644
--- a/custom_components/opendisplay/diagnostics.py
+++ b/custom_components/opendisplay/diagnostics.py
@@ -28,6 +28,29 @@ async def async_get_config_entry_diagnostics(
     """Return diagnostics for a config entry."""
     runtime = entry.runtime_data
     fw = runtime.firmware
+    profile = runtime.sleep_profile
+
+    sleep_diag: dict[str, Any] = {
+        "is_sleepy": profile.is_sleepy,
+        "deep_sleep_enabled": profile.deep_sleep_enabled,
+        "deep_sleep_time_seconds": profile.deep_sleep_time_seconds,
+        "sleep_timeout_ms": profile.sleep_timeout_ms,
+        "missed_cycles": profile.missed_cycles,
+        "queue_timeout_hours": profile.queue_timeout_hours,
+        "availability_interval_s": profile.availability_interval,
+        "config_resync_pending": runtime.config_resync_pending,
+    }
+
+    # Delivery slot state only: timestamps/attempts, never queued image bytes.
+    if runtime.delivery is not None:
+        state = runtime.delivery.state
+        sleep_diag["delivery"] = {
+            "pending": state.pending,
+            "queued_at": state.queued_at,
+            "expires_at": state.expires_at,
+            "attempts": state.attempts,
+            "last_error": state.last_error,
+        }
 
     return {
         "firmware": {
@@ -36,5 +59,6 @@ async def async_get_config_entry_diagnostics(
             "sha": fw["sha"],
         },
         "is_flex": runtime.is_flex,
+        "sleep": sleep_diag,
         "device_config": async_redact_data(_asdict(runtime.device_config), TO_REDACT),
     }
diff --git a/custom_components/opendisplay/icons.json b/custom_components/opendisplay/icons.json
index 6150e04..da60480 100644
--- a/custom_components/opendisplay/icons.json
+++ b/custom_components/opendisplay/icons.json
@@ -1,5 +1,13 @@
 {
   "entity": {
+    "binary_sensor": {
+      "update_pending": {
+        "default": "mdi:image-sync",
+        "state": {
+          "off": "mdi:image-check"
+        }
+      }
+    },
     "update": {
       "firmware": {
         "default": "mdi:chip"
diff --git a/custom_components/opendisplay/image.py b/custom_components/opendisplay/image.py
index 3d32703..fbb0428 100644
--- a/custom_components/opendisplay/image.py
+++ b/custom_components/opendisplay/image.py
@@ -1,5 +1,8 @@
 """Image entity for OpenDisplay devices."""
 
+from datetime import datetime, timezone
+from typing import Any
+
 from homeassistant.components.image import ImageEntity
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo
@@ -8,8 +11,9 @@
 import homeassistant.util.dt as dt_util
 
 from . import OpenDisplayConfigEntry
-from .const import SIGNAL_IMAGE_UPDATED
+from .const import SIGNAL_IMAGE_UPDATED, SIGNAL_PENDING_STATE
 from .coordinator import OpenDisplayCoordinator
+from .delivery import DeliverySnapshot
 
 PARALLEL_UPDATES = 0
 
@@ -20,11 +24,20 @@ async def async_setup_entry(
     async_add_entities: AddConfigEntryEntitiesCallback,
 ) -> None:
     """Set up the OpenDisplay image entity."""
-    async_add_entities([OpenDisplayImageEntity(hass, entry.runtime_data.coordinator)])
+    async_add_entities(
+        [OpenDisplayImageEntity(hass, entry.runtime_data.coordinator)]
+    )
+
+
+def _to_iso(epoch: float | None) -> str | None:
+    """Convert an epoch timestamp to an ISO string, or None."""
+    if epoch is None:
+        return None
+    return datetime.fromtimestamp(epoch, tz=timezone.utc).isoformat()
 
 
 class OpenDisplayImageEntity(ImageEntity):
-    """Shows the last image sent to an OpenDisplay device."""
+    """Shows the last image sent to (or queued for) an OpenDisplay device."""
 
     _attr_has_entity_name = True
     _attr_translation_key = "content"
@@ -39,13 +52,27 @@ def __init__(self, hass: HomeAssistant, coordinator: OpenDisplayCoordinator) ->
             connections={(CONNECTION_BLUETOOTH, coordinator.address)},
         )
         self._image_bytes: bytes | None = None
+        # When True the shown frame is queued for the next wake, not yet on the
+        # panel (D6).
+        self._pending: bool = False
+        self._queued_at: float | None = None
+        self._last_error: str | None = None
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any]:
+        """Expose whether the shown frame is still waiting to be delivered."""
+        return {
+            "pending": self._pending,
+            "queued_at": _to_iso(self._queued_at),
+            "last_error": self._last_error,
+        }
 
     async def async_image(self) -> bytes | None:
-        """Return the last uploaded image bytes."""
+        """Return the last uploaded (or queued) image bytes."""
         return self._image_bytes
 
     async def async_added_to_hass(self) -> None:
-        """Subscribe to image update signals."""
+        """Subscribe to image and pending-state update signals."""
         await super().async_added_to_hass()
         self.async_on_remove(
             async_dispatcher_connect(
@@ -54,10 +81,25 @@ async def async_added_to_hass(self) -> None:
                 self._handle_image_update,
             )
         )
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                f"{SIGNAL_PENDING_STATE}_{self._coordinator.address}",
+                self._handle_pending_state,
+            )
+        )
 
     @callback
     def _handle_image_update(self, image_bytes: bytes) -> None:
-        """Handle a new image from a completed upload."""
+        """Handle a new image from a completed or queued upload."""
         self._image_bytes = image_bytes
         self._attr_image_last_updated = dt_util.utcnow()
         self.async_write_ha_state()
+
+    @callback
+    def _handle_pending_state(self, snapshot: DeliverySnapshot) -> None:
+        """Reflect the delivery manager's pending state on the entity."""
+        self._pending = snapshot.pending
+        self._queued_at = snapshot.queued_at
+        self._last_error = snapshot.last_error
+        self.async_write_ha_state()
diff --git a/custom_components/opendisplay/manifest.json b/custom_components/opendisplay/manifest.json
index ade3dee..ebfe322 100644
--- a/custom_components/opendisplay/manifest.json
+++ b/custom_components/opendisplay/manifest.json
@@ -15,6 +15,6 @@
   "iot_class": "local_push",
   "issue_tracker": "https://github.com/OpenDisplay/Home_Assistant_Integration/issues",
   "loggers": ["opendisplay"],
-  "requirements": ["py-opendisplay[silabs-ota]==7.11.1", "odl-renderer==0.5.10"],
+  "requirements": ["py-opendisplay[silabs-ota]@git+https://github.com/davelee98/py-opendisplay.git@feat/ble-speed", "odl-renderer==0.5.10"],
   "version": "3.0.0-beta.7"
 }
diff --git a/custom_components/opendisplay/sensor.py b/custom_components/opendisplay/sensor.py
index 335f0a0..8107ed0 100644
--- a/custom_components/opendisplay/sensor.py
+++ b/custom_components/opendisplay/sensor.py
@@ -3,10 +3,12 @@
 from collections.abc import Callable
 from dataclasses import dataclass
 from datetime import datetime, timezone
+import time
 
 from opendisplay import voltage_to_percent
 from opendisplay.models.enums import CapacityEstimator, PowerMode
 
+from homeassistant.components.bluetooth import async_last_service_info
 from homeassistant.components.sensor import (
     SensorDeviceClass,
     SensorEntity,
@@ -77,11 +79,9 @@ class OpenDisplaySensorEntityDescription(SensorEntityDescription):
     device_class=SensorDeviceClass.TIMESTAMP,
     entity_category=EntityCategory.DIAGNOSTIC,
     entity_registry_enabled_default=False,
-    value_fn=lambda upd: (
-        datetime.fromtimestamp(upd.last_seen, tz=timezone.utc)
-        if upd.last_seen is not None
-        else None
-    ),
+    # native_value is overridden by OpenDisplayLastSeenSensor, so this value_fn
+    # is dead code; value_fn is a required field, hence the no-op.
+    value_fn=lambda _upd: None,
 )
 
 
@@ -116,7 +116,11 @@ async def async_setup_entry(
         ]
 
     async_add_entities(
-        OpenDisplaySensorEntity(coordinator, description)
+        (
+            OpenDisplayLastSeenSensor(coordinator, description)
+            if description.key == "last_seen"
+            else OpenDisplaySensorEntity(coordinator, description)
+        )
         for description in descriptions
     )
 
@@ -132,3 +136,22 @@ def native_value(self) -> float | int | str | datetime | None:
         if self.coordinator.data is None:
             return None
         return self.entity_description.value_fn(self.coordinator.data)
+
+
+class OpenDisplayLastSeenSensor(OpenDisplaySensorEntity):
+    """last_seen sourced from the bluetooth stack, not the gated callback."""
+
+    @property
+    def native_value(self) -> datetime | None:
+        # connectable=False matches the Bluetooth advertisement monitor's
+        # "updated" field: _all_history, refreshed on every received advert
+        # before core's connectable/de-dup gates.
+        info = async_last_service_info(
+            self.hass, self.coordinator.address, connectable=False
+        )
+        if info is None:
+            return None
+        # info.time is a monotonic clock (monotonic_time_coarse); convert to
+        # wall time with the same offset the advertisement monitor uses.
+        wall = info.time + (time.time() - time.monotonic())
+        return datetime.fromtimestamp(wall, tz=timezone.utc)
diff --git a/custom_components/opendisplay/services.py b/custom_components/opendisplay/services.py
index f347997..d2dfcca 100644
--- a/custom_components/opendisplay/services.py
+++ b/custom_components/opendisplay/services.py
@@ -3,7 +3,7 @@
 import asyncio
 from collections.abc import Awaitable, Callable
 import contextlib
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
 from enum import IntEnum
 import functools
 import io
@@ -16,6 +16,8 @@
 from opendisplay import (
     AuthenticationFailedError,
     AuthenticationRequiredError,
+    BLEConnectionError,
+    BLETimeoutError,
     ColorScheme,
     DitherMode,
     FitMode,
@@ -41,7 +43,13 @@
 from homeassistant.components.media_source import async_resolve_media
 from homeassistant.config_entries import ConfigEntryState
 from homeassistant.const import ATTR_DEVICE_ID
-from homeassistant.core import HomeAssistant, ServiceCall, callback
+from homeassistant.core import (
+    HomeAssistant,
+    ServiceCall,
+    ServiceResponse,
+    SupportsResponse,
+    callback,
+)
 from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
 from homeassistant.helpers import config_validation as cv, device_registry as dr
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
@@ -54,6 +62,7 @@
     from . import OpenDisplayConfigEntry
 
 from .const import CONF_ENCRYPTION_KEY, DOMAIN, SIGNAL_IMAGE_UPDATED
+from .delivery import DeliveryReceipt
 
 ATTR_IMAGE = "image"
 ATTR_ROTATION = "rotation"
@@ -280,17 +289,52 @@ async def _async_download_image(hass: HomeAssistant, url: str) -> PILImage.Image
     return await hass.async_add_executor_job(_load_image_from_bytes, data)
 
 
+class _DeviceUnavailable(Exception):
+    """Signals the tag was dark or dropped the link during a live send.
+
+    Raised only when ``reraise_ble=True`` so the caller can queue the content
+    for the next wake instead of surfacing an ``upload_error``.
+    """
+
+
+# Probe budget for a probably-asleep tag: one connect attempt, short timeout.
+# 5 s is >2x the observed 1-3 s connect-during-window latency (plus proxy
+# slack), so a genuinely awake tag connects comfortably, while a dark ESP32
+# (radio fully off in timer deep sleep) costs at most ~5 s before queuing —
+# vs the ~40 s default budget (4 attempts x 10 s). Deliberately below both the
+# 10 s wake window and the 15 s freshness horizon (wake_window_s +
+# FRESHNESS_SLACK_S), so a probe triggered by a just-missed advert still lands
+# inside the window it is betting on.
+PROBE_CONNECT_TIMEOUT_S = 5.0
+PROBE_MAX_ATTEMPTS = 1
+
+
 async def _async_connect_and_run(
     hass: HomeAssistant,
     entry: "OpenDisplayConfigEntry",
     action: Callable[[OpenDisplayDevice], Awaitable[None]],
     use_measured_palettes: bool = False,
+    reraise_ble: bool = False,
+    *,
+    connect_timeout: float | None = None,
+    max_attempts: int | None = None,
 ) -> None:
-    """Resolve BLE device, open a connection, run action, handle auth errors."""
+    """Resolve BLE device, open a connection, run action, handle auth errors.
+
+    When ``reraise_ble`` is set, a missing connectable device or a BLE
+    connect/timeout failure raises ``_DeviceUnavailable`` instead of a
+    translated ``device_not_found``/``upload_error``, so the caller can defer
+    the work to the delivery queue. ``connect_timeout``/``max_attempts``
+    override the library's connect budget (10 s x 4 attempts) — used by the
+    probe path to bound a likely-doomed attempt; ``None`` keeps the library
+    defaults. All other behavior is unchanged.
+    """
     address = entry.unique_id
     assert address is not None
     ble_device = async_ble_device_from_address(hass, address, connectable=True)
     if ble_device is None:
+        if reraise_ble:
+            raise _DeviceUnavailable
         raise HomeAssistantError(
             translation_domain=DOMAIN,
             translation_key="device_not_found",
@@ -318,6 +362,12 @@ async def _async_connect_and_run(
             translation_domain=DOMAIN, translation_key="authentication_error"
         ) from err
 
+    device_kwargs: dict[str, Any] = {}
+    if connect_timeout is not None:
+        device_kwargs["timeout"] = connect_timeout
+    if max_attempts is not None:
+        device_kwargs["max_attempts"] = max_attempts
+
     try:
         # Serialize all BLE access to this tag: the device has a single BLE link
         # and the library has no per-address lock, so overlapping connections
@@ -331,6 +381,7 @@ async def _async_connect_and_run(
             config=entry.runtime_data.device_config,
             use_measured_palettes=use_measured_palettes,
             encryption_key=encryption_key,
+            **device_kwargs,
         ) as device:
             await action(device)
     except (AuthenticationFailedError, AuthenticationRequiredError) as err:
@@ -338,6 +389,14 @@ async def _async_connect_and_run(
         raise HomeAssistantError(
             translation_domain=DOMAIN, translation_key="authentication_error"
         ) from err
+    except (BLEConnectionError, BLETimeoutError) as err:
+        if reraise_ble:
+            raise _DeviceUnavailable from err
+        raise HomeAssistantError(
+            translation_domain=DOMAIN,
+            translation_key="upload_error",
+            translation_placeholders={"error": str(err)},
+        ) from err
     except OpenDisplayError as err:
         raise HomeAssistantError(
             translation_domain=DOMAIN,
@@ -351,14 +410,20 @@ async def _async_send_image(
     entry: "OpenDisplayConfigEntry",
     img: PILImage.Image,
     *,
+    device_id: str | None = None,
     dither_mode: DitherMode,
     refresh_mode: RefreshMode,
     fit: FitMode = FitMode.CONTAIN,
     tone: float | str = "auto",
     rotate: Rotation = Rotation.ROTATE_0,
     use_measured_palettes: bool = False,
-) -> None:
-    """Upload a PIL image to the device."""
+) -> DeliveryReceipt:
+    """Upload a PIL image, delivering live or queuing it for the next wake.
+
+    Returns a receipt describing whether the frame was delivered immediately or
+    queued (for a sleeping device). Non-sleepy devices always deliver live and
+    keep the original strict-failure behavior.
+    """
     # Split the upload into its heavy CPU half and its BLE-I/O half. The CPU
     # work (rotate + fit + dither + encode + zlib on a full frame) is offloaded
     # to an executor thread so it never blocks the event loop; only the BLE
@@ -399,20 +464,98 @@ async def _async_send_image(
         runtime.partial_state = PartialState()
     state = runtime.partial_state
 
+    # The preview JPEG is built up-front so a queued frame can be shown on the
+    # image entity immediately (D6), not only after a successful delivery.
+    jpeg = await hass.async_add_executor_job(_pil_to_jpeg, img)
+
+    profile = runtime.sleep_profile
+    manager = runtime.delivery
+    sleepy = manager is not None and profile.is_sleepy
+
+    def _queue() -> DeliveryReceipt:
+        assert manager is not None
+        return manager.submit_upload(
+            prepared=prepared,
+            refresh_mode=refresh_mode,
+            partial_state=state,
+            use_measured_palettes=use_measured_palettes,
+            preview_jpeg=jpeg,
+            device_id=device_id,
+        )
+
     async def _upload(device: OpenDisplayDevice) -> None:
-        await device.upload_prepared_image(prepared, refresh_mode=refresh_mode, state=state)
+        await device.upload_prepared_image(
+            prepared, refresh_mode=refresh_mode, state=state
+        )
+
+    # Freshness gate: a probably-asleep tag will not usually answer a live
+    # connect. Instead of queuing blind, spend one short connect attempt (the
+    # "probe") in case the tag is actually awake: its wake adverts may have
+    # been missed by the scanner, and Silabs tags advertise continuously even
+    # when their power config looks sleepy. A dark ESP32 costs at most
+    # ~PROBE_CONNECT_TIMEOUT_S before falling back to the queue. HA drops the
+    # connectable BLEDevice ~3-5 min after the last advert, so a long-dark or
+    # never-seen tag short-circuits to the queue at near-zero cost (no
+    # BLEDevice -> _DeviceUnavailable without any radio traffic).
+    probing = False
+    connect_kwargs: dict[str, Any] = {}
+    if sleepy:
+        last_seen = (
+            runtime.coordinator.data.last_seen if runtime.coordinator.data else None
+        )
+        if profile.probably_asleep(last_seen):
+            if not profile.probe_before_queue:
+                return _queue()
+            probing = True
+            connect_kwargs = {
+                "connect_timeout": PROBE_CONNECT_TIMEOUT_S,
+                "max_attempts": PROBE_MAX_ATTEMPTS,
+            }
+
+    try:
+        await _async_connect_and_run(
+            hass,
+            entry,
+            _upload,
+            use_measured_palettes=use_measured_palettes,
+            reraise_ble=sleepy,
+            **connect_kwargs,
+        )
+    except _DeviceUnavailable:
+        # The device was dark, dropped, or refused the link; defer.
+        receipt = _queue()
+        if probing:
+            # Race: a wake advert arriving DURING the failed probe found no
+            # pending work (nothing queued yet), so no drain started. If the
+            # tag now looks fresh, kick a drain instead of waiting out a full
+            # sleep cycle. notify_device_seen is a no-op while one is running.
+            last_seen = (
+                runtime.coordinator.data.last_seen if runtime.coordinator.data else None
+            )
+            if not profile.probably_asleep(last_seen):
+                assert manager is not None
+                manager.notify_device_seen("post-probe")
+        return receipt
 
-    await _async_connect_and_run(
-        hass, entry, _upload, use_measured_palettes=use_measured_palettes
-    )
-    jpeg = await hass.async_add_executor_job(_pil_to_jpeg, img)
     async_dispatcher_send(hass, f"{SIGNAL_IMAGE_UPDATED}_{entry.unique_id}", jpeg)
+    return DeliveryReceipt(status="delivered", expires_at=None)
 
 
-async def _async_upload_image(call: ServiceCall) -> None:
+def _receipt_response(receipt: DeliveryReceipt) -> ServiceResponse:
+    """Build the service response payload from a delivery receipt."""
+    expires_at = (
+        datetime.fromtimestamp(receipt.expires_at, tz=timezone.utc).isoformat()
+        if receipt.expires_at is not None
+        else None
+    )
+    return {"status": receipt.status, "expires_at": expires_at}
+
+
+async def _async_upload_image(call: ServiceCall) -> ServiceResponse:
     """Handle the upload_image service call."""
     entry = _get_entry_for_device(call)
 
+    device_id: str = call.data[ATTR_DEVICE_ID]
     image_data: dict[str, Any] | str = call.data[ATTR_IMAGE]
     rotation: Rotation = call.data[ATTR_ROTATION]
     dither_mode: DitherMode = call.data[ATTR_DITHER_MODE]
@@ -463,10 +606,11 @@ async def _async_upload_image(call: ServiceCall) -> None:
             else:
                 pil_image = await _async_download_image(call.hass, media.url)
 
-        await _async_send_image(
+        receipt = await _async_send_image(
             call.hass,
             entry,
             pil_image,
+            device_id=device_id,
             dither_mode=dither_mode,
             refresh_mode=refresh_mode,
             fit=fit_mode,
@@ -474,8 +618,11 @@ async def _async_upload_image(call: ServiceCall) -> None:
             rotate=rotation,
             use_measured_palettes=use_measured_palettes,
         )
+        return _receipt_response(receipt)
     except asyncio.CancelledError:
-        return
+        # Superseded by a newer upload (latest-wins); report it honestly rather
+        # than surfacing the cancellation.
+        return {"status": "superseded", "expires_at": None}
     finally:
         if entry.runtime_data.upload_task is current:
             entry.runtime_data.upload_task = None
@@ -577,7 +724,7 @@ async def _get_device_ids_from_area(hass: HomeAssistant, area_id: str) -> list[s
     ]
 
 
-async def _async_drawcustom(call: ServiceCall) -> None:
+async def _async_drawcustom(call: ServiceCall) -> ServiceResponse:
     """Handle the drawcustom service call."""
     hass = call.hass
 
@@ -596,11 +743,16 @@ async def _async_drawcustom(call: ServiceCall) -> None:
         )
 
     errors: list[str] = []
+    receipts: list[DeliveryReceipt] = []
+    results: dict[str, Any] = {}
     for device_id in unique_ids:
         try:
-            await _drawcustom_for_device(hass, device_id, call)
+            receipt = await _drawcustom_for_device(hass, device_id, call)
         except (HomeAssistantError, ServiceValidationError) as err:
             errors.append(f"{device_id}: {err}")
+        else:
+            receipts.append(receipt)
+            results[device_id] = _receipt_response(receipt)
     if errors:
         raise ServiceValidationError(
             translation_domain=DOMAIN,
@@ -608,6 +760,25 @@ async def _async_drawcustom(call: ServiceCall) -> None:
             translation_placeholders={"errors": "\n".join(errors)},
         )
 
+    # Summarize across all targets: any queued device makes the batch "queued"
+    # (with the soonest expiry); otherwise delivered (or dry_run).
+    queued = [r for r in receipts if r.status == "queued" and r.expires_at is not None]
+    if queued:
+        status = "queued"
+        expires_epoch: float | None = min(r.expires_at for r in queued)  # type: ignore[type-var]
+    elif receipts and all(r.status == "dry_run" for r in receipts):
+        status = "dry_run"
+        expires_epoch = None
+    else:
+        status = "delivered"
+        expires_epoch = None
+    expires_at = (
+        datetime.fromtimestamp(expires_epoch, tz=timezone.utc).isoformat()
+        if expires_epoch is not None
+        else None
+    )
+    return {"status": status, "expires_at": expires_at, "results": results}
+
 
 def _font_search_dirs(hass: HomeAssistant) -> list[str]:
     """Return font search directories in priority order."""
@@ -621,7 +792,7 @@ def _font_search_dirs(hass: HomeAssistant) -> list[str]:
 
 async def _drawcustom_for_device(
     hass: HomeAssistant, device_id: str, call: ServiceCall
-) -> None:
+) -> DeliveryReceipt:
     entry = _get_entry_for_device_id(hass, device_id)
     display = entry.runtime_data.device_config.displays[0]
     cs = display.color_scheme_enum
@@ -656,7 +827,7 @@ async def _drawcustom_for_device(
         _LOGGER.info("Drawcustom dry run for device %s", device_id)
         jpeg = await hass.async_add_executor_job(_pil_to_jpeg, img)
         async_dispatcher_send(hass, f"{SIGNAL_IMAGE_UPDATED}_{entry.unique_id}", jpeg)
-        return
+        return DeliveryReceipt(status="dry_run", expires_at=None)
 
     dither_mode: DitherMode = call.data["dither"]
     refresh_mode: RefreshMode = call.data["refresh_type"]
@@ -666,10 +837,11 @@ async def _drawcustom_for_device(
     )
     use_measured_palettes: bool = call.data[ATTR_USE_MEASURED_PALETTES]
 
-    await _async_send_image(
+    return await _async_send_image(
         hass,
         entry,
         img,
+        device_id=device_id,
         dither_mode=dither_mode,
         refresh_mode=refresh_mode,
         tone=tone_compression,
@@ -678,6 +850,25 @@ async def _drawcustom_for_device(
     )
 
 
+def _raise_if_sleeping(entry: "OpenDisplayConfigEntry", device_id: str) -> None:
+    """Fail fast for an immediate-only action when the tag is provably asleep.
+
+    LED/buzzer notifications that fire hours late are worse than an error, so a
+    sleeping device rejects them immediately rather than queuing.
+    """
+    runtime = entry.runtime_data
+    profile = runtime.sleep_profile
+    if not profile.is_sleepy:
+        return
+    last_seen = runtime.coordinator.data.last_seen if runtime.coordinator.data else None
+    if profile.probably_asleep(last_seen):
+        raise HomeAssistantError(
+            translation_domain=DOMAIN,
+            translation_key="device_sleeping",
+            translation_placeholders={"device_id": device_id},
+        )
+
+
 async def _async_activate_led(call: ServiceCall) -> None:
     """Handle the activate_led service call."""
     entry = _get_entry_for_device(call)
@@ -687,6 +878,7 @@ async def _async_activate_led(call: ServiceCall) -> None:
             translation_key="no_leds",
             translation_placeholders={"device_id": call.data[ATTR_DEVICE_ID]},
         )
+    _raise_if_sleeping(entry, call.data[ATTR_DEVICE_ID])
     repeats: int = call.data["repeats"]
     flash_config = LedFlashConfig(
         mode=1,
@@ -728,6 +920,7 @@ async def _async_activate_buzzer(call: ServiceCall) -> None:
             translation_key="no_buzzers",
             translation_placeholders={"device_id": call.data[ATTR_DEVICE_ID]},
         )
+    _raise_if_sleeping(entry, call.data[ATTR_DEVICE_ID])
     buzz_config = BuzzerActivateConfig.single_tone(
         frequency_hz=call.data["frequency_hz"],
         duration_ms=call.data["duration_ms"],
@@ -749,7 +942,14 @@ def async_setup_services(hass: HomeAssistant) -> None:
         "upload_image",
         _async_upload_image,
         schema=SCHEMA_UPLOAD_IMAGE,
+        supports_response=SupportsResponse.OPTIONAL,
+    )
+    hass.services.async_register(
+        DOMAIN,
+        "drawcustom",
+        _async_drawcustom,
+        schema=SCHEMA_DRAWCUSTOM,
+        supports_response=SupportsResponse.OPTIONAL,
     )
-    hass.services.async_register(DOMAIN, "drawcustom", _async_drawcustom, schema=SCHEMA_DRAWCUSTOM)
     hass.services.async_register(DOMAIN, "activate_led", _async_activate_led, schema=SCHEMA_ACTIVATE_LED)
     hass.services.async_register(DOMAIN, "activate_buzzer", _async_activate_buzzer, schema=SCHEMA_ACTIVATE_BUZZER)
\ No newline at end of file
diff --git a/custom_components/opendisplay/sleep.py b/custom_components/opendisplay/sleep.py
new file mode 100644
index 0000000..81628d0
--- /dev/null
+++ b/custom_components/opendisplay/sleep.py
@@ -0,0 +1,161 @@
+"""Deep-sleep awareness for OpenDisplay devices.
+
+`SleepProfile` resolves the integration options and the device's own power
+configuration into the three facts the rest of the integration needs:
+
+* ``is_sleepy`` — should this device be treated as a deep-sleeping tag (dark
+  most of the time, reachable only during a short post-wake advertising
+  window)?
+* ``availability_interval`` — how long HA should keep entities available
+  between wakes before flagging the device as gone.
+* ``probably_asleep(last_seen)`` — given the last advertisement time, is the
+  device almost certainly back asleep right now?
+
+The predicate for "sleepy" mirrors the firmware's own deep-sleep entry
+condition (``power_mode == BATTERY and deep_sleep_time_seconds > 0``, see
+``Firmware/src/main.cpp``) so the integration and the device can never disagree,
+with an options override for edge cases. All computation here is pure and
+side-effect free; the values are consumed by ``__init__``, ``services`` and the
+delivery manager.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import time
+from typing import TYPE_CHECKING
+
+from opendisplay.models.enums import PowerMode
+
+from .const import (
+    CONF_MISSED_CYCLES,
+    CONF_PROBE_BEFORE_QUEUE,
+    CONF_QUEUE_TIMEOUT_HOURS,
+    CONF_SLEEP_MODE,
+    DEFAULT_MISSED_CYCLES,
+    DEFAULT_PROBE_BEFORE_QUEUE,
+    DEFAULT_QUEUE_TIMEOUT_HOURS,
+    DEFAULT_SLEEP_MODE,
+    SLEEP_MODE_OFF,
+    SLEEP_MODE_ON,
+)
+
+if TYPE_CHECKING:
+    from opendisplay import GlobalConfig
+
+    from . import OpenDisplayConfigEntry
+
+# Firmware default post-wake advertising window when sleep_timeout_ms is 0
+# (Firmware/src/main.cpp:94-96). Expressed in milliseconds to match the config.
+DEFAULT_WAKE_WINDOW_MS = 10_000
+# Extra slack added on top of the wake window when deciding the device is back
+# asleep, to absorb scanner/proxy advertisement latency.
+FRESHNESS_SLACK_S = 5.0
+# Fixed slack added to the availability horizon so a marginally late wake does
+# not immediately flap entities unavailable.
+AVAILABILITY_SLACK_S = 60.0
+
+
+@dataclass(frozen=True)
+class SleepProfile:
+    """Resolved deep-sleep behavior for one config entry."""
+
+    is_sleepy: bool
+    deep_sleep_enabled: bool
+    deep_sleep_time_seconds: int
+    sleep_timeout_ms: int
+    missed_cycles: int
+    queue_timeout_hours: int
+    probe_before_queue: bool
+
+    @property
+    def wake_window_s(self) -> float:
+        """Post-wake advertising window in seconds (firmware default 10 s)."""
+        return (self.sleep_timeout_ms or DEFAULT_WAKE_WINDOW_MS) / 1000.0
+
+    @property
+    def availability_interval(self) -> float:
+        """Seconds of silence tolerated before entities go unavailable.
+
+        ``deep_sleep_time_seconds * missed_cycles`` (the device may skip a few
+        expected wakes) plus one wake window and a fixed slack.
+        """
+        return (
+            self.deep_sleep_time_seconds * self.missed_cycles
+            + self.wake_window_s
+            + AVAILABILITY_SLACK_S
+        )
+
+    @property
+    def queue_timeout_s(self) -> float:
+        """Seconds a queued upload survives before expiring."""
+        return self.queue_timeout_hours * 3600
+
+    def probably_asleep(self, last_seen: float | None, now: float | None = None) -> bool:
+        """Return True if the device is almost certainly back asleep.
+
+        The device advertises only during its wake window. If the most recent
+        advertisement is older than one wake window plus slack, it has returned
+        to deep sleep and connecting would just burn a doomed retry cycle. A
+        device never seen (``last_seen is None``) is treated as asleep.
+
+        This is a pure freshness test independent of ``is_sleepy``; callers
+        combine it with ``is_sleepy`` where the distinction matters.
+        """
+        if last_seen is None:
+            return True
+        current = time.time() if now is None else now
+        return (current - last_seen) > (self.wake_window_s + FRESHNESS_SLACK_S)
+
+    @classmethod
+    def create(
+        cls,
+        *,
+        sleep_mode: str,
+        power_mode: int,
+        sleep_timeout_ms: int,
+        deep_sleep_time_seconds: int,
+        missed_cycles: int,
+        queue_timeout_hours: int,
+        probe_before_queue: bool = DEFAULT_PROBE_BEFORE_QUEUE,
+    ) -> SleepProfile:
+        """Build a profile from primitive values (pure; unit-testable)."""
+        deep_sleep_enabled = (
+            power_mode == PowerMode.BATTERY and deep_sleep_time_seconds > 0
+        )
+        if sleep_mode == SLEEP_MODE_ON:
+            is_sleepy = True
+        elif sleep_mode == SLEEP_MODE_OFF:
+            is_sleepy = False
+        else:  # SLEEP_MODE_AUTO
+            is_sleepy = deep_sleep_enabled
+        return cls(
+            is_sleepy=is_sleepy,
+            deep_sleep_enabled=deep_sleep_enabled,
+            deep_sleep_time_seconds=deep_sleep_time_seconds,
+            sleep_timeout_ms=sleep_timeout_ms,
+            missed_cycles=missed_cycles,
+            queue_timeout_hours=queue_timeout_hours,
+            probe_before_queue=probe_before_queue,
+        )
+
+    @classmethod
+    def from_entry(
+        cls, entry: OpenDisplayConfigEntry, config: GlobalConfig
+    ) -> SleepProfile:
+        """Build a profile from a config entry's options and device config."""
+        options = entry.options
+        power = config.power
+        return cls.create(
+            sleep_mode=options.get(CONF_SLEEP_MODE, DEFAULT_SLEEP_MODE),
+            power_mode=power.power_mode,
+            sleep_timeout_ms=power.sleep_timeout_ms,
+            deep_sleep_time_seconds=power.deep_sleep_time_seconds,
+            missed_cycles=options.get(CONF_MISSED_CYCLES, DEFAULT_MISSED_CYCLES),
+            queue_timeout_hours=options.get(
+                CONF_QUEUE_TIMEOUT_HOURS, DEFAULT_QUEUE_TIMEOUT_HOURS
+            ),
+            probe_before_queue=options.get(
+                CONF_PROBE_BEFORE_QUEUE, DEFAULT_PROBE_BEFORE_QUEUE
+            ),
+        )
diff --git a/custom_components/opendisplay/strings.json b/custom_components/opendisplay/strings.json
index 582c954..5d5fe25 100644
--- a/custom_components/opendisplay/strings.json
+++ b/custom_components/opendisplay/strings.json
@@ -50,7 +50,30 @@
       }
     }
   },
+  "options": {
+    "step": {
+      "init": {
+        "data": {
+          "sleep_mode": "Deep sleep handling",
+          "missed_cycles": "Missed wake cycles before unavailable",
+          "queue_timeout_hours": "Queued content timeout (hours)",
+          "probe_before_queue": "Try connecting before queueing"
+        },
+        "data_description": {
+          "sleep_mode": "Automatic follows the device's own power configuration (battery power with deep sleep enabled). Force on or off to override.",
+          "missed_cycles": "How many expected wake-ups the device may miss before its entities are marked unavailable.",
+          "queue_timeout_hours": "How long content queued for a sleeping device waits for a wake before it expires.",
+          "probe_before_queue": "Before queueing an image for a sleeping device, make one quick connection attempt in case it is actually awake (missed advertisements, always-on radios). Adds at most a few seconds when the device really is asleep."
+        }
+      }
+    }
+  },
   "entity": {
+    "binary_sensor": {
+      "update_pending": {
+        "name": "Update pending"
+      }
+    },
     "image": {
       "content": {
         "name": "Display content"
@@ -108,6 +131,12 @@
     "device_not_found": {
       "message": "Could not find Bluetooth device with address `{address}`. Reason: {reason}"
     },
+    "device_sleeping": {
+      "message": "Device `{device_id}` is asleep and cannot be reached immediately. Wake it and try again."
+    },
+    "device_sleeping_ota": {
+      "message": "The device is asleep and cannot be updated right now. Wake it (press its button or wait for its next scheduled check-in) and start the update again."
+    },
     "invalid_device_id": {
       "message": "Device `{device_id}` is not a valid OpenDisplay device."
     },
@@ -161,6 +190,13 @@
         "full": "Full",
         "partial": "Partial"
       }
+    },
+    "sleep_mode": {
+      "options": {
+        "auto": "Automatic (follow device)",
+        "off": "Force off",
+        "on": "Force on"
+      }
     }
   },
   "services": {
diff --git a/custom_components/opendisplay/translations/en.json b/custom_components/opendisplay/translations/en.json
index b14c078..9243cbf 100644
--- a/custom_components/opendisplay/translations/en.json
+++ b/custom_components/opendisplay/translations/en.json
@@ -50,7 +50,30 @@
             }
         }
     },
+    "options": {
+        "step": {
+            "init": {
+                "data": {
+                    "sleep_mode": "Deep sleep handling",
+                    "missed_cycles": "Missed wake cycles before unavailable",
+                    "queue_timeout_hours": "Queued content timeout (hours)",
+                    "probe_before_queue": "Try connecting before queueing"
+                },
+                "data_description": {
+                    "sleep_mode": "Automatic follows the device's own power configuration (battery power with deep sleep enabled). Force on or off to override.",
+                    "missed_cycles": "How many expected wake-ups the device may miss before its entities are marked unavailable.",
+                    "queue_timeout_hours": "How long content queued for a sleeping device waits for a wake before it expires.",
+                    "probe_before_queue": "Before queueing an image for a sleeping device, make one quick connection attempt in case it is actually awake (missed advertisements, always-on radios). Adds at most a few seconds when the device really is asleep."
+                }
+            }
+        }
+    },
     "entity": {
+        "binary_sensor": {
+            "update_pending": {
+                "name": "Update pending"
+            }
+        },
         "image": {
             "content": {
                 "name": "Display content"
@@ -105,6 +128,12 @@
         "device_not_found": {
             "message": "Could not find Bluetooth device with address `{address}`."
         },
+        "device_sleeping": {
+            "message": "Device `{device_id}` is asleep and cannot be reached immediately. Wake it and try again."
+        },
+        "device_sleeping_ota": {
+            "message": "The device is asleep and cannot be updated right now. Wake it (press its button or wait for its next scheduled check-in) and start the update again."
+        },
         "invalid_device_id": {
             "message": "Device `{device_id}` is not a valid OpenDisplay device."
         },
@@ -158,6 +187,13 @@
                 "full": "Full",
                 "partial": "Partial"
             }
+        },
+        "sleep_mode": {
+            "options": {
+                "auto": "Automatic (follow device)",
+                "off": "Force off",
+                "on": "Force on"
+            }
         }
     },
     "services": {
diff --git a/custom_components/opendisplay/update.py b/custom_components/opendisplay/update.py
index daa5f92..e603132 100644
--- a/custom_components/opendisplay/update.py
+++ b/custom_components/opendisplay/update.py
@@ -26,6 +26,7 @@
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
 from . import OpenDisplayConfigEntry, _get_encryption_key
+from .const import DOMAIN
 from .entity import OpenDisplayEntity
 
 _LOGGER = logging.getLogger(__name__)
@@ -166,6 +167,24 @@ async def async_install(self, version: str | None, backup: bool, **kwargs: Any)
         if not tag:
             raise HomeAssistantError("No firmware version available to install")
 
+        # A deep-sleeping tag is dark most of the time and the multi-connection
+        # AppLoader flash (DFU trigger -> reconnect -> OTA) cannot be driven
+        # reliably inside a ~10 s wake window. Rather than start an install that
+        # will strand mid-flash, fail fast with clear guidance to wake it first.
+        runtime = self._entry.runtime_data
+        profile = runtime.sleep_profile
+        if profile.is_sleepy:
+            last_seen = (
+                runtime.coordinator.data.last_seen
+                if runtime.coordinator.data
+                else None
+            )
+            if profile.probably_asleep(last_seen):
+                raise HomeAssistantError(
+                    translation_domain=DOMAIN,
+                    translation_key="device_sleeping_ota",
+                )
+
         asset_name = firmware_ota_asset(self._ic_type, tag)
         if asset_name is None:
             raise HomeAssistantError(
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..67f149e
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,6 @@
+"""Test bootstrap: make ``custom_components`` importable from the repo root."""
+
+import pathlib
+import sys
+
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
diff --git a/tests/test_delivery.py b/tests/test_delivery.py
new file mode 100644
index 0000000..7d7437d
--- /dev/null
+++ b/tests/test_delivery.py
@@ -0,0 +1,374 @@
+"""Unit tests for DeliveryManager with mocked hass/coordinator/device.
+
+These avoid the full Home Assistant test harness: the manager's HA touchpoints
+(``async_call_later``, ``async_dispatcher_send``, ``async_ble_device_from_address``
+and ``OpenDisplayDevice``) are patched in the delivery module namespace.
+"""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from opendisplay import AuthenticationFailedError, BLEConnectionError, RefreshMode
+import pytest
+
+from homeassistant.exceptions import HomeAssistantError
+
+from custom_components.opendisplay import delivery as delivery_mod
+from custom_components.opendisplay.const import (
+    EVENT_CONTENT_DELIVERED,
+    EVENT_CONTENT_EXPIRED,
+)
+from custom_components.opendisplay.delivery import DeliveryManager
+from custom_components.opendisplay.sleep import SleepProfile
+
+ADDRESS = "AA:BB:CC:DD:EE:FF"
+
+
+def _profile(**overrides):
+    params = {
+        "sleep_mode": "on",
+        "power_mode": 1,
+        "sleep_timeout_ms": 0,
+        "deep_sleep_time_seconds": 300,
+        "missed_cycles": 3,
+        "queue_timeout_hours": 24,
+    }
+    params.update(overrides)
+    return SleepProfile.create(**params)
+
+
+def _make_env(profile=None, entry_data=None, last_seen=None):
+    """Return (hass, entry, coordinator) with a fake runtime for the manager."""
+    profile = profile or _profile()
+    coordinator = MagicMock()
+    coordinator.data = SimpleNamespace(last_seen=last_seen)
+    coordinator.async_subscribe_device_seen = MagicMock(return_value=MagicMock())
+    runtime = SimpleNamespace(
+        coordinator=coordinator,
+        sleep_profile=profile,
+        device_config=MagicMock(),
+        ble_lock=asyncio.Lock(),
+        config_resync_pending=False,
+        firmware=None,
+        is_flex=False,
+    )
+    entry = MagicMock()
+    entry.unique_id = ADDRESS
+    entry.runtime_data = runtime
+    entry.data = entry_data if entry_data is not None else {}
+    entry.async_start_reauth = MagicMock()
+    hass = MagicMock()
+    return hass, entry, coordinator
+
+
+def _submit(mgr, device_id="dev1"):
+    return mgr.submit_upload(
+        prepared=(b"img", None, object()),
+        refresh_mode=RefreshMode.FULL,
+        partial_state=MagicMock(),
+        use_measured_palettes=False,
+        preview_jpeg=b"jpeg",
+        device_id=device_id,
+    )
+
+
+def _fake_device_ctx(device):
+    """Return a factory producing an async-context-manager wrapping device."""
+
+    class _Ctx:
+        async def __aenter__(self):
+            return device
+
+        async def __aexit__(self, *exc):
+            return False
+
+    return lambda **kwargs: _Ctx()
+
+
+def _raising_device_ctx(exc):
+    class _Ctx:
+        async def __aenter__(self):
+            raise exc
+
+        async def __aexit__(self, *exc_info):
+            return False
+
+    return lambda **kwargs: _Ctx()
+
+
+def test_submit_upload_queues_and_reports():
+    hass, entry, _ = _make_env()
+    with (
+        patch.object(delivery_mod, "async_call_later", return_value=MagicMock()) as later,
+        patch.object(delivery_mod, "async_dispatcher_send") as dispatch,
+    ):
+        mgr = DeliveryManager(hass, entry)
+        receipt = _submit(mgr)
+
+    assert receipt.status == "queued"
+    assert receipt.expires_at is not None
+    snap = mgr.state
+    assert snap.pending is True
+    assert snap.queued_at is not None
+    assert snap.attempts == 0
+    later.assert_called_once()  # deadline armed
+    # Both the image preview and the pending-state signals were dispatched.
+    assert dispatch.call_count == 2
+
+
+def test_latest_wins_cancels_previous_deadline():
+    hass, entry, _ = _make_env()
+    cancels = [MagicMock(), MagicMock()]
+    with (
+        patch.object(delivery_mod, "async_call_later", side_effect=cancels),
+        patch.object(delivery_mod, "async_dispatcher_send"),
+    ):
+        mgr = DeliveryManager(hass, entry)
+        _submit(mgr, device_id="a")
+        _submit(mgr, device_id="b")
+
+    cancels[0].assert_called_once()  # the first deadline timer was cancelled
+    assert mgr.state.pending is True
+
+
+def test_expiry_clears_slot_and_fires_event():
+    hass, entry, _ = _make_env()
+    captured = {}
+
+    def _fake_later(_hass, _delay, callback):
+        captured["cb"] = callback
+        return MagicMock()
+
+    with (
+        patch.object(delivery_mod, "async_call_later", side_effect=_fake_later),
+        patch.object(delivery_mod, "async_dispatcher_send"),
+    ):
+        mgr = DeliveryManager(hass, entry)
+        _submit(mgr, device_id="dev1")
+        captured["cb"](None)  # simulate the deadline firing
+
+    assert mgr.state.pending is False
+    assert mgr.state.last_error == "expired"
+    hass.bus.async_fire.assert_called_once()
+    event, payload = hass.bus.async_fire.call_args[0]
+    assert event == EVENT_CONTENT_EXPIRED
+    assert payload["device_id"] == "dev1"
+    assert "queued_at" in payload
+
+
+def test_request_config_resync_sets_flags():
+    hass, entry, _ = _make_env()
+    mgr = DeliveryManager(hass, entry)
+    assert mgr._has_pending_work() is False
+    mgr.request_config_resync()
+    assert entry.runtime_data.config_resync_pending is True
+    assert mgr._has_pending_work() is True
+
+
+def test_notify_device_seen_starts_one_delivery():
+    hass, entry, _ = _make_env()
+
+    def _capture(_hass, coro, _name):
+        coro.close()  # don't actually run the drain
+        return MagicMock()
+
+    entry.async_create_background_task = MagicMock(side_effect=_capture)
+    with (
+        patch.object(delivery_mod, "async_call_later", return_value=MagicMock()),
+        patch.object(delivery_mod, "async_dispatcher_send"),
+    ):
+        mgr = DeliveryManager(hass, entry)
+        mgr.notify_device_seen()  # nothing queued -> no delivery
+        entry.async_create_background_task.assert_not_called()
+
+        _submit(mgr)
+        mgr.notify_device_seen()  # work queued -> one delivery
+        entry.async_create_background_task.assert_called_once()
+
+        mgr.notify_device_seen()  # already delivering -> still one
+        entry.async_create_background_task.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_drain_delivers_upload():
+    hass, entry, _ = _make_env()
+    device = MagicMock()
+    device.upload_prepared_image = AsyncMock()
+    with (
+        patch.object(delivery_mod, "async_call_later", return_value=MagicMock()),
+        patch.object(delivery_mod, "async_dispatcher_send"),
+        patch.object(delivery_mod, "async_ble_device_from_address", return_value=MagicMock()),
+        patch.object(delivery_mod, "OpenDisplayDevice", side_effect=_fake_device_ctx(device)),
+    ):
+        mgr = DeliveryManager(hass, entry)
+        _submit(mgr, device_id="dev1")
+        await mgr._deliver()
+
+    device.upload_prepared_image.assert_awaited_once()
+    assert mgr.state.pending is False
+    fired = [call.args[0] for call in hass.bus.async_fire.call_args_list]
+    assert EVENT_CONTENT_DELIVERED in fired
+
+
+@pytest.mark.asyncio
+async def test_drain_ble_failure_keeps_slot_and_counts_attempt():
+    hass, entry, _ = _make_env()
+    with (
+        patch.object(delivery_mod, "async_call_later", return_value=MagicMock()),
+        patch.object(delivery_mod, "async_dispatcher_send"),
+        patch.object(delivery_mod, "async_ble_device_from_address", return_value=MagicMock()),
+        patch.object(
+            delivery_mod,
+            "OpenDisplayDevice",
+            side_effect=_raising_device_ctx(BLEConnectionError("boom")),
+        ),
+    ):
+        mgr = DeliveryManager(hass, entry)
+        _submit(mgr)
+        await mgr._deliver()
+
+    assert mgr.state.pending is True
+    assert mgr.state.attempts == 1
+
+
+@pytest.mark.asyncio
+async def test_drain_deadline_timeout_raises_and_counts_attempt():
+    """Exceeding DELIVERY_DEADLINE_S raises loudly but keeps the slot queued."""
+    hass, entry, _ = _make_env()
+    with (
+        patch.object(delivery_mod, "async_call_later", return_value=MagicMock()),
+        patch.object(delivery_mod, "async_dispatcher_send"),
+        patch.object(delivery_mod, "async_ble_device_from_address", return_value=MagicMock()),
+        patch.object(
+            delivery_mod,
+            "OpenDisplayDevice",
+            side_effect=_raising_device_ctx(TimeoutError()),
+        ),
+    ):
+        mgr = DeliveryManager(hass, entry)
+        _submit(mgr)
+        with pytest.raises(HomeAssistantError):
+            await mgr._deliver()
+
+    # The slot is retained for the next wake and the failed attempt is recorded.
+    assert mgr.state.pending is True
+    assert mgr.state.attempts == 1
+    assert "deadline exceeded" in (mgr.state.last_error or "")
+    # Background-task bookkeeping is still cleaned up despite the raise.
+    assert mgr._delivering is False
+    assert mgr._delivery_task is None
+
+
+@pytest.mark.asyncio
+async def test_drain_gives_up_after_max_attempts():
+    """After MAX_DELIVERY_ATTEMPTS failures the slot is dropped with an expired event."""
+    hass, entry, _ = _make_env()
+    deadline_cancel = MagicMock()
+    with (
+        patch.object(delivery_mod, "async_call_later", return_value=deadline_cancel),
+        patch.object(delivery_mod, "async_dispatcher_send"),
+        patch.object(delivery_mod, "async_ble_device_from_address", return_value=MagicMock()),
+        patch.object(
+            delivery_mod,
+            "OpenDisplayDevice",
+            side_effect=_raising_device_ctx(BLEConnectionError("boom")),
+        ),
+    ):
+        mgr = DeliveryManager(hass, entry)
+        _submit(mgr)
+        for attempt in range(1, delivery_mod.MAX_DELIVERY_ATTEMPTS + 1):
+            await mgr._deliver()
+            if attempt < delivery_mod.MAX_DELIVERY_ATTEMPTS:
+                # Still queued and retriable until the cap is reached.
+                assert mgr.state.pending is True
+                assert mgr.state.attempts == attempt
+
+    # On the capped attempt the slot is dropped and the expiry timer cancelled.
+    assert mgr.state.pending is False
+    assert mgr.state.last_error == "failed"
+    deadline_cancel.assert_called_once()
+
+    # A content_expired event fired once, carrying attempts == the cap.
+    expired = [
+        c.args[1]
+        for c in hass.bus.async_fire.call_args_list
+        if c.args[0] == EVENT_CONTENT_EXPIRED
+    ]
+    assert len(expired) == 1
+    assert expired[0]["attempts"] == delivery_mod.MAX_DELIVERY_ATTEMPTS
+
+
+@pytest.mark.asyncio
+async def test_drain_auth_failure_pauses_and_starts_reauth():
+    hass, entry, _ = _make_env()
+    with (
+        patch.object(delivery_mod, "async_call_later", return_value=MagicMock()),
+        patch.object(delivery_mod, "async_dispatcher_send"),
+        patch.object(delivery_mod, "async_ble_device_from_address", return_value=MagicMock()),
+        patch.object(
+            delivery_mod,
+            "OpenDisplayDevice",
+            side_effect=_raising_device_ctx(AuthenticationFailedError("bad key")),
+        ),
+    ):
+        mgr = DeliveryManager(hass, entry)
+        _submit(mgr)
+        await mgr._deliver()
+
+    assert mgr._pending_upload is not None
+    assert mgr._pending_upload.paused is True
+    assert mgr.state.last_error == "auth"
+    entry.async_start_reauth.assert_called_once()
+    # A paused slot is not retried on the next wake.
+    assert mgr._has_pending_work() is False
+
+
+@pytest.mark.asyncio
+async def test_drain_config_resync_updates_runtime_and_cache():
+    hass, entry, _ = _make_env()
+    device = MagicMock()
+    device.read_firmware_version = AsyncMock(
+        return_value={"major": 1, "minor": 2, "sha": "abc"}
+    )
+    device.is_flex = False
+    device.landing_url = MagicMock(return_value="http://landing")
+    device.config = MagicMock()
+    with (
+        patch.object(delivery_mod, "async_call_later", return_value=MagicMock()),
+        patch.object(delivery_mod, "async_dispatcher_send"),
+        patch.object(delivery_mod, "async_ble_device_from_address", return_value=MagicMock()),
+        patch.object(delivery_mod, "OpenDisplayDevice", side_effect=_fake_device_ctx(device)),
+        patch("custom_components.opendisplay._write_cache") as write_cache,
+    ):
+        mgr = DeliveryManager(hass, entry)
+        mgr.request_config_resync()
+        await mgr._deliver()
+
+    assert entry.runtime_data.config_resync_pending is False
+    assert mgr._pending_config_resync is False
+    assert entry.runtime_data.firmware == {"major": 1, "minor": 2, "sha": "abc"}
+    write_cache.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_shutdown_unsubscribes_and_cancels_deadline():
+    hass, entry, coordinator = _make_env()
+    unsub = MagicMock()
+    coordinator.async_subscribe_device_seen = MagicMock(return_value=unsub)
+    deadline_cancel = MagicMock()
+    with (
+        patch.object(delivery_mod, "async_call_later", return_value=deadline_cancel),
+        patch.object(delivery_mod, "async_dispatcher_send"),
+    ):
+        mgr = DeliveryManager(hass, entry)
+        mgr.async_start()
+        _submit(mgr)
+        await mgr.async_shutdown()
+
+    unsub.assert_called_once()
+    deadline_cancel.assert_called_once()
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-v"]))
diff --git a/tests/test_last_seen.py b/tests/test_last_seen.py
new file mode 100644
index 0000000..94af071
--- /dev/null
+++ b/tests/test_last_seen.py
@@ -0,0 +1,65 @@
+"""Unit tests for the last_seen sensor with a mocked bluetooth stack.
+
+These avoid the full Home Assistant test harness: ``async_last_service_info``
+(the sensor's only HA touchpoint) is patched in the sensor module namespace and
+the entity's ``native_value`` property is asserted directly.
+"""
+
+import time
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from custom_components.opendisplay import sensor as sensor_mod
+from custom_components.opendisplay.sensor import (
+    _LAST_SEEN_DESCRIPTION,
+    OpenDisplayLastSeenSensor,
+)
+
+ADDRESS = "AA:BB:CC:DD:EE:FF"
+
+
+def _make_sensor():
+    """Return an OpenDisplayLastSeenSensor wired to mock coordinator/hass."""
+    coordinator = MagicMock()
+    coordinator.address = ADDRESS
+    entity = OpenDisplayLastSeenSensor(coordinator, _LAST_SEEN_DESCRIPTION)
+    entity.hass = MagicMock()
+    return entity
+
+
+def test_native_value_converts_monotonic_to_wall_time():
+    entity = _make_sensor()
+    # info.time is a monotonic-clock reading, like habluetooth's _all_history.
+    mono = time.monotonic()
+    with patch.object(
+        sensor_mod,
+        "async_last_service_info",
+        return_value=SimpleNamespace(time=mono),
+    ) as mock_info:
+        value = entity.native_value
+
+    # A monotonic "now" maps to wall-clock "now"; must be tz-aware for TIMESTAMP.
+    assert isinstance(value, datetime)
+    assert value.tzinfo is not None
+    expected = datetime.now(tz=timezone.utc)
+    assert abs((value - expected).total_seconds()) < 2
+
+    # The stack is queried with connectable=False to match the advertisement
+    # monitor's "updated" column (pre-gate _all_history), and for the coordinator
+    # address.
+    mock_info.assert_called_once()
+    assert mock_info.call_args.kwargs["connectable"] is False
+    assert ADDRESS in mock_info.call_args.args
+
+
+def test_native_value_none_when_no_service_info():
+    entity = _make_sensor()
+    with patch.object(sensor_mod, "async_last_service_info", return_value=None):
+        assert entity.native_value is None
+
+
+if __name__ == "__main__":
+    import pytest
+
+    raise SystemExit(pytest.main([__file__, "-v"]))
diff --git a/tests/test_services.py b/tests/test_services.py
new file mode 100644
index 0000000..2947838
--- /dev/null
+++ b/tests/test_services.py
@@ -0,0 +1,254 @@
+"""Unit tests for the _async_send_image sleep gate and probe-before-queue.
+
+Mirrors the test_delivery.py approach: no Home Assistant test harness; the
+service module's HA/BLE touchpoints (``prepare_image``, ``_pil_to_jpeg``,
+``async_dispatcher_send``, ``async_ble_device_from_address`` and
+``OpenDisplayDevice``) are patched in the services module namespace.
+"""
+
+import asyncio
+import time
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from opendisplay import BLEConnectionError, RefreshMode, DitherMode
+from PIL import Image as PILImage
+import pytest
+
+from custom_components.opendisplay import services as services_mod
+from custom_components.opendisplay.delivery import DeliveryReceipt
+from custom_components.opendisplay.services import (
+    PROBE_CONNECT_TIMEOUT_S,
+    PROBE_MAX_ATTEMPTS,
+    _async_send_image,
+)
+from custom_components.opendisplay.sleep import SleepProfile
+
+ADDRESS = "AA:BB:CC:DD:EE:FF"
+
+
+def _profile(**overrides):
+    params = {
+        "sleep_mode": "on",
+        "power_mode": 1,
+        "sleep_timeout_ms": 0,
+        "deep_sleep_time_seconds": 300,
+        "missed_cycles": 3,
+        "queue_timeout_hours": 24,
+    }
+    params.update(overrides)
+    return SleepProfile.create(**params)
+
+
+def _make_env(profile=None, last_seen=None):
+    """Return (hass, entry, coordinator, manager) with a fake runtime."""
+    profile = profile or _profile()
+    coordinator = MagicMock()
+    coordinator.data = SimpleNamespace(last_seen=last_seen)
+    manager = MagicMock()
+    manager.submit_upload = MagicMock(
+        return_value=DeliveryReceipt(status="queued", expires_at=123.0)
+    )
+    manager.notify_device_seen = MagicMock()
+    runtime = SimpleNamespace(
+        coordinator=coordinator,
+        sleep_profile=profile,
+        delivery=manager,
+        device_config=None,
+        ble_lock=asyncio.Lock(),
+        partial_state=MagicMock(),
+    )
+    entry = MagicMock()
+    entry.unique_id = ADDRESS
+    entry.runtime_data = runtime
+    entry.data = {}  # no encryption key
+    hass = MagicMock()
+
+    async def _executor_job(fn, *args):
+        return fn(*args)
+
+    hass.async_add_executor_job = _executor_job
+    return hass, entry, coordinator, manager
+
+
+def _device_ctx_factory(device=None, exc=None, on_enter=None):
+    """MagicMock construct-recording factory for OpenDisplayDevice.
+
+    Produces an async context manager that yields ``device``, or raises
+    ``exc`` from ``__aenter__`` (after calling ``on_enter`` if given).
+    """
+
+    class _Ctx:
+        async def __aenter__(self):
+            if on_enter is not None:
+                on_enter()
+            if exc is not None:
+                raise exc
+            return device
+
+        async def __aexit__(self, *exc_info):
+            return False
+
+    return MagicMock(side_effect=lambda **kwargs: _Ctx())
+
+
+def _send(hass, entry):
+    img = PILImage.new("RGB", (1, 1))
+    return _async_send_image(
+        hass,
+        entry,
+        img,
+        dither_mode=DitherMode.NONE,
+        refresh_mode=RefreshMode.FULL,
+    )
+
+
+def _patches(od_factory, ble_device="ble-device"):
+    """Common patch set for a _async_send_image drive."""
+    return (
+        patch.object(
+            services_mod, "prepare_image", return_value=(b"img", None, object())
+        ),
+        patch.object(services_mod, "_pil_to_jpeg", return_value=b"jpeg"),
+        patch.object(services_mod, "async_dispatcher_send"),
+        patch.object(
+            services_mod, "async_ble_device_from_address", return_value=ble_device
+        ),
+        patch.object(services_mod, "OpenDisplayDevice", od_factory),
+    )
+
+
+@pytest.mark.asyncio
+async def test_probe_success_delivers():
+    """Stale device + probe on: one reduced-budget attempt that succeeds."""
+    hass, entry, _, manager = _make_env(last_seen=None)  # never seen -> asleep
+    device = MagicMock()
+    device.upload_prepared_image = AsyncMock()
+    od = _device_ctx_factory(device=device)
+    p1, p2, p3, p4, p5 = _patches(od)
+    with p1, p2, p3, p4, p5:
+        receipt = await _send(hass, entry)
+
+    assert receipt.status == "delivered"
+    od.assert_called_once()
+    kwargs = od.call_args.kwargs
+    assert kwargs["timeout"] == PROBE_CONNECT_TIMEOUT_S
+    assert kwargs["max_attempts"] == PROBE_MAX_ATTEMPTS
+    manager.submit_upload.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_probe_failure_queues():
+    """Stale device + probe on: the failed attempt falls back to the queue."""
+    hass, entry, _, manager = _make_env(last_seen=None)
+    od = _device_ctx_factory(exc=BLEConnectionError("dark"))
+    p1, p2, p3, p4, p5 = _patches(od)
+    with p1, p2, p3, p4, p5:
+        receipt = await _send(hass, entry)
+
+    assert receipt.status == "queued"
+    od.assert_called_once()
+    manager.submit_upload.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_probe_disabled_queues_immediately():
+    """probe_before_queue=False restores the old queue-without-connect gate."""
+    hass, entry, _, manager = _make_env(
+        profile=_profile(probe_before_queue=False), last_seen=None
+    )
+    od = _device_ctx_factory()
+    p1, p2, p3, p4, p5 = _patches(od)
+    with p1, p2, p3, p4 as ble_lookup, p5:
+        receipt = await _send(hass, entry)
+
+    assert receipt.status == "queued"
+    ble_lookup.assert_not_called()
+    od.assert_not_called()
+    manager.submit_upload.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_not_sleepy_full_budget():
+    """Non-sleepy devices keep the library's default connect budget."""
+    hass, entry, _, manager = _make_env(
+        profile=_profile(sleep_mode="off"), last_seen=None
+    )
+    device = MagicMock()
+    device.upload_prepared_image = AsyncMock()
+    od = _device_ctx_factory(device=device)
+    p1, p2, p3, p4, p5 = _patches(od)
+    with p1, p2, p3, p4, p5:
+        receipt = await _send(hass, entry)
+
+    assert receipt.status == "delivered"
+    kwargs = od.call_args.kwargs
+    assert "timeout" not in kwargs
+    assert "max_attempts" not in kwargs
+    manager.submit_upload.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_probe_no_ble_device_queues_cheaply():
+    """No connectable BLEDevice: the probe short-circuits to the queue."""
+    hass, entry, _, manager = _make_env(last_seen=None)
+    od = _device_ctx_factory()
+    p1, p2, p3, p4, p5 = _patches(od, ble_device=None)
+    with p1, p2, p3, p4, p5:
+        receipt = await _send(hass, entry)
+
+    assert receipt.status == "queued"
+    od.assert_not_called()
+    manager.submit_upload.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_post_probe_fresh_advert_triggers_drain():
+    """A wake advert landing during the failed probe kicks an immediate drain."""
+    hass, entry, coordinator, manager = _make_env(last_seen=None)
+
+    def _advert_arrives():
+        coordinator.data = SimpleNamespace(last_seen=time.time())
+
+    od = _device_ctx_factory(
+        exc=BLEConnectionError("dropped"), on_enter=_advert_arrives
+    )
+    p1, p2, p3, p4, p5 = _patches(od)
+    with p1, p2, p3, p4, p5:
+        receipt = await _send(hass, entry)
+
+    assert receipt.status == "queued"
+    manager.notify_device_seen.assert_called_once_with("post-probe")
+
+
+@pytest.mark.asyncio
+async def test_post_probe_still_stale_no_drain_kick():
+    """No advert during the failed probe: wait for the natural next wake."""
+    hass, entry, _, manager = _make_env(last_seen=None)
+    od = _device_ctx_factory(exc=BLEConnectionError("dark"))
+    p1, p2, p3, p4, p5 = _patches(od)
+    with p1, p2, p3, p4, p5:
+        receipt = await _send(hass, entry)
+
+    assert receipt.status == "queued"
+    manager.notify_device_seen.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_fresh_device_failure_no_probe_kwargs():
+    """Sleepy but recently seen: full budget, queue on failure, no drain kick."""
+    hass, entry, _, manager = _make_env(last_seen=time.time())
+    od = _device_ctx_factory(exc=BLEConnectionError("dropped"))
+    p1, p2, p3, p4, p5 = _patches(od)
+    with p1, p2, p3, p4, p5:
+        receipt = await _send(hass, entry)
+
+    assert receipt.status == "queued"
+    kwargs = od.call_args.kwargs
+    assert "timeout" not in kwargs
+    assert "max_attempts" not in kwargs
+    manager.notify_device_seen.assert_not_called()
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-v"]))
diff --git a/tests/test_sleep.py b/tests/test_sleep.py
new file mode 100644
index 0000000..91dc28c
--- /dev/null
+++ b/tests/test_sleep.py
@@ -0,0 +1,109 @@
+"""Unit tests for the pure SleepProfile logic."""
+
+import pytest
+
+from custom_components.opendisplay.sleep import (
+    AVAILABILITY_SLACK_S,
+    DEFAULT_WAKE_WINDOW_MS,
+    FRESHNESS_SLACK_S,
+    SleepProfile,
+)
+
+
+def _profile(**overrides):
+    """Build a SleepProfile from sensible defaults overridden per test."""
+    params = {
+        "sleep_mode": "auto",
+        "power_mode": 1,  # BATTERY
+        "sleep_timeout_ms": 0,
+        "deep_sleep_time_seconds": 300,
+        "missed_cycles": 3,
+        "queue_timeout_hours": 24,
+    }
+    params.update(overrides)
+    return SleepProfile.create(**params)
+
+
+def test_deep_sleep_enabled_requires_battery_and_interval():
+    assert _profile(power_mode=1, deep_sleep_time_seconds=300).deep_sleep_enabled
+    # USB power → not a deep-sleeper even with an interval set.
+    assert not _profile(power_mode=2, deep_sleep_time_seconds=300).deep_sleep_enabled
+    # Battery but no interval → not a deep-sleeper.
+    assert not _profile(power_mode=1, deep_sleep_time_seconds=0).deep_sleep_enabled
+
+
+def test_sleep_mode_auto_follows_device():
+    assert _profile(sleep_mode="auto", power_mode=1, deep_sleep_time_seconds=300).is_sleepy
+    assert not _profile(
+        sleep_mode="auto", power_mode=2, deep_sleep_time_seconds=300
+    ).is_sleepy
+
+
+def test_sleep_mode_force_on_and_off_override_device():
+    # Force on even for a USB device with no deep sleep.
+    assert _profile(sleep_mode="on", power_mode=2, deep_sleep_time_seconds=0).is_sleepy
+    # Force off even for a battery device configured for deep sleep.
+    forced_off = _profile(sleep_mode="off", power_mode=1, deep_sleep_time_seconds=300)
+    assert not forced_off.is_sleepy
+    # deep_sleep_enabled still reflects the device, independent of the override.
+    assert forced_off.deep_sleep_enabled
+
+
+def test_wake_window_uses_firmware_default_when_zero():
+    assert _profile(sleep_timeout_ms=0).wake_window_s == DEFAULT_WAKE_WINDOW_MS / 1000.0
+    assert _profile(sleep_timeout_ms=5000).wake_window_s == 5.0
+
+
+def test_availability_interval_formula():
+    profile = _profile(
+        sleep_timeout_ms=0, deep_sleep_time_seconds=300, missed_cycles=3
+    )
+    # 300 * 3 + 10 (default window) + 60 slack = 970
+    expected = 300 * 3 + DEFAULT_WAKE_WINDOW_MS / 1000.0 + AVAILABILITY_SLACK_S
+    assert profile.availability_interval == expected == 970.0
+
+
+def test_queue_timeout_seconds():
+    assert _profile(queue_timeout_hours=24).queue_timeout_s == 86400
+    assert _profile(queue_timeout_hours=1).queue_timeout_s == 3600
+
+
+def test_probably_asleep_never_seen_is_true():
+    assert _profile().probably_asleep(None) is True
+
+
+def test_probably_asleep_recent_advert_is_false():
+    profile = _profile(sleep_timeout_ms=10000)  # 10 s window
+    now = 1_000_000.0
+    # Seen 1 s ago: still inside the wake window -> may be awake.
+    assert profile.probably_asleep(now - 1.0, now=now) is False
+
+
+def test_probably_asleep_stale_advert_is_true():
+    profile = _profile(sleep_timeout_ms=10000)
+    now = 1_000_000.0
+    # Seen well beyond window + slack -> back asleep.
+    stale = now - (10.0 + FRESHNESS_SLACK_S + 1.0)
+    assert profile.probably_asleep(stale, now=now) is True
+
+
+def test_probe_before_queue_defaults_true():
+    # Default also proves create() keeps working for callers that omit it.
+    assert _profile().probe_before_queue is True
+
+
+def test_probe_before_queue_override():
+    assert _profile(probe_before_queue=False).probe_before_queue is False
+
+
+def test_probably_asleep_boundary():
+    profile = _profile(sleep_timeout_ms=10000)
+    now = 1_000_000.0
+    threshold = 10.0 + FRESHNESS_SLACK_S
+    # Exactly at the threshold is not yet "asleep" (strict greater-than).
+    assert profile.probably_asleep(now - threshold, now=now) is False
+    assert profile.probably_asleep(now - threshold - 0.01, now=now) is True
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-v"]))