Load a config entry offline from the last discovery snapshot (#295)

An appliance switched off at the wall used to take its whole config entry
down with it: async_setup_entry raised ConfigEntryNotReady, so the device
read as failed and its entities existed only as registry rows until the
appliance came back.

Loading the entry anyway isn't enough on its own. Entities here are the
output of discovery, discovery only runs inside a successful poll, and
platforms enumerate `bound` exactly once at forward time -- so an entry
that loads while offline loads empty, and with no listeners subscribed the
base coordinator stops rescheduling and never polls again.

Bank the resources dict each successful first cycle hands _run_discovery,
along with the subdevice candidate list and the /oic identity that route
the registry, and replay it through _run_discovery when the first refresh
fails. Storing the poll input rather than a rendered entity list keeps one
implementation of discovery instead of two: the offline entity set is
produced by the same code that produced the live one.

Three things fall out of that:

- Platforms judge entity existence against `discovery_resources`, not the
  live cache. The live cache deliberately stays empty, which is what keeps
  a restored entity `unavailable` rather than rendering a stale value for
  an appliance nobody can currently reach.
- A live discovery that disagrees with the snapshot reloads the entry --
  platforms can't adopt a changed set in place, so a firmware update or a
  sibling subdevice that starts answering needs a fresh setup.
- The entry holds one coordinator listener for its lifetime, so polling is
  scheduled regardless of how many entities are live.

An entry that has never reached the device has no snapshot, keeps raising
ConfigEntryNotReady, and closes its session on the way out as before -- no
metadata to build a device from, and it leaves room for setup flows that
need to interact with the appliance (#168).

Restores the two tests PR #303 rewrote, narrowed to that no-snapshot path.

README.md

@@ -270,6 +270,12 @@ If reconnects become persistent (more than a handful per minute), something's ac
 
 Deregistering a device in SmartThings causes a reset of its network settings as soon as it accesses Samsung's servers, dropping it off Wi-Fi until it's re-onboarded through the SmartThings app. As such, consider keeping devices registered even if egress-blocked, to avoid them resetting upon brief internet access.
 
+### Restarting while an appliance is powered off
+
+If Home Assistant restarts while an appliance is unplugged or switched off at the wall, its device and entities still load — restored from the last successful discovery, showing `unavailable` until the appliance answers again. Automations and dashboards keep referring to entities that exist, and the integration retries in the background, so the device comes back on its own within a poll cycle of being powered on. Entities read `unavailable` rather than their last known values on purpose: the integration can't verify what a disconnected appliance is doing, and recorded history is kept by the recorder either way.
+
+This only applies to an appliance the integration has reached at least once. A brand-new device that has never answered has nothing to restore from, so setting it up still requires it to be reachable.
+
 ### Multi-subdevice ("2-in-1") air conditioner systems
 
 Some Samsung installs run more than one indoor subdevice off a single outdoor unit, all reachable over the *one* IP/DTLS session your config entry connects to (a floor-standing + wall-mounted 2-in-1 is a common shape). The integration discovers any sibling subdevices automatically, once, right after the first successful poll — there's nothing to configure. Each discovered subdevice gets its own HA device (linked to the main one via "via device") and its own `climate` card, so it lands in its own room in the dashboard instead of being invisible or mixed into the master's state.

custom_components/localthings/__init__.py

@@ -13,12 +13,13 @@ from homeassistant.const import (
 )
 from homeassistant.const import EVENT_HOMEASSISTANT_STOP
 from homeassistant.core import Event, HomeAssistant, callback
+from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import device_registry as dr
 from homeassistant.helpers import entity_registry as er
 from homeassistant.helpers.typing import ConfigType
 
 from .const import CONF_DEVICE_TYPE, CONF_HOST, CONF_PORT, CONF_SERIAL, DOMAIN, PLATFORMS
-from .coordinator import LocalThingsCoordinator
+from .coordinator import LocalThingsCoordinator, snapshot_store
 from .registry.identity import resolve_serial
 from .services import async_setup_services
 
@@ -259,13 +260,30 @@ async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     hass.data.setdefault(DOMAIN, {})
     coordinator = LocalThingsCoordinator(hass, entry)
+
+    # Before the first refresh, so the coordinator keeps rescheduling even
+    # when that refresh fails and leaves nothing subscribed: the base class
+    # only re-arms its timer while it has listeners, and an offline load can
+    # legitimately have zero live entities (every one of them disabled, say).
+    # Without this the entry loads once and never polls again (issue #295).
+    entry.async_on_unload(coordinator.async_add_listener(lambda: None))
+
     try:
         await coordinator.async_config_entry_first_refresh()
-    except Exception as err:
-        _LOGGER.warning(
-            "Initial connection to device failed (%s); starting offline and retrying in background",
-            err,
-        )
+    except ConfigEntryNotReady:
+        # An entry that has polled successfully before comes up on its last
+        # known entity set and keeps retrying on the normal poll interval,
+        # rather than sitting in setup-retry with a device that reads as
+        # broken and entities that exist only as registry rows (issue #295).
+        #
+        # An entry that has never reached the device has no snapshot, so
+        # there is nothing to show and no device metadata to name it with --
+        # that case still fails, which is also what keeps the door open for
+        # setup flows that need to interact with the device (issue #168).
+        if not await coordinator.async_rehydrate():
+            await coordinator.async_close()
+            raise
+
     hass.data[DOMAIN][entry.entry_id] = coordinator
 
     # Send the DTLS close_notify on Core shutdown, not just on unload (issue
@@ -322,6 +340,16 @@ async def async_remove_config_entry_device(
     return not (device.identifiers & live)
 
 
+async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
+    """Delete the discovery snapshot this entry accumulated (issue #295).
+
+    Nothing else would: the store is keyed on entry_id, so re-adding the same
+    appliance mints a new one and the old file would linger in .storage
+    forever.
+    """
+    await snapshot_store(hass, entry).async_remove()
+
+
 async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     unloaded = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
     if unloaded:

custom_components/localthings/coordinator.py

@@ -9,6 +9,7 @@ import logging
 import threading
 import time
 import zlib
+from dataclasses import asdict
 from datetime import timedelta
 from typing import Any, cast
 
@@ -18,6 +19,7 @@ from homeassistant.core import HomeAssistant, callback
 from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
 from homeassistant.helpers import issue_registry as ir
 from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.storage import Store
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 from smartthings_local.ocf.state_cache import StateCache
 from smartthings_local.protocol.dtls_session import DtlsCoapSession
@@ -47,7 +49,7 @@ from .const import (
 from .learned import LEARNABLE, LearnedModes, persist
 from .observe import GRACE_PERIOD_S, MODE_OBSERVE, MODE_POLL, ObserveManager
 from .registry import CAPABILITIES
-from .registry.adapter import flatten
+from .registry.adapter import _key, flatten
 from .registry.batch import parse_device0_batch
 from .registry.by_type import resolve as resolve_registry
 from .registry.capabilities.common import (
@@ -83,6 +85,21 @@ _LOGGER = logging.getLogger(__name__)
 
 _SEED_PATH = ["device", "0"]
 
+# Discovery snapshot (issue #295): exactly what the last successful first
+# cycle fed _run_discovery, so a restart can register the same entities
+# while the appliance is unreachable. Kept in .storage rather than on the
+# config entry -- it's device state, not configuration, and runs to tens of
+# kilobytes.
+_SNAPSHOT_VERSION = 1
+_SNAPSHOT_SAVE_DELAY_S = 10.0
+
+
+def snapshot_store(hass: HomeAssistant, entry: ConfigEntry) -> Store[dict[str, Any]]:
+    """This entry's discovery-snapshot store. A free function so
+    `async_remove_entry` can delete the file without standing up a whole
+    coordinator to reach it."""
+    return Store(hass, _SNAPSHOT_VERSION, f"{DOMAIN}.{entry.entry_id}.discovery")
+
 
 class _NoOpDescriptor:
     """No-op: StateCache requires an on_observation hook; this integration
@@ -248,6 +265,11 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         self._identity: DeviceIdentity | None = None
         self._discovered = False
         self.bound = []
+        self._snapshot_store = snapshot_store(hass, entry)
+        # Both set only when this entry loaded from a snapshot instead of a
+        # live poll -- see async_rehydrate.
+        self._rehydrate_resources: dict[str, dict] | None = None
+        self._rehydrated_keys: frozenset[tuple[str, str]] | None = None
         # Sibling indoor subdevices on this connection (issue #177); set
         # once at first discovery, narrowed to the ones with live state (see
         # subdevices.discover_partitioned). Never includes MAIN itself.
@@ -405,6 +427,35 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         self._canonical_cache[view_key] = view
         return view
 
+    @property
+    def rehydrated(self) -> bool:
+        """True while this entry's entities came from a snapshot rather than
+        a live poll (issue #295)."""
+        return self._rehydrate_resources is not None
+
+    @property
+    def discovery_resources(self) -> dict[str, dict]:
+        """What entity._is_included should judge an entity's existence
+        against: the rehydration snapshot on an offline load, the live cache
+        otherwise.
+
+        Deliberately separate from `last_resources`, which stays empty until
+        the device answers -- that emptiness is what keeps a rehydrated
+        entity `unavailable` instead of rendering a snapshot's stale value.
+        Only read while platforms are being forwarded; nothing consults it
+        once the entities exist.
+        """
+        if self._rehydrate_resources is None:
+            return self.last_resources
+        return self._rehydrate_resources
+
+    def discovery_canonical(self, subdevice: Subdevice) -> dict[str, dict]:
+        """`discovery_resources` in `subdevice`'s canonical view -- the
+        exists_fn counterpart to canonical_resources."""
+        if self._rehydrate_resources is None:
+            return self.canonical_resources(subdevice)
+        return canonical_view(subdevice, self._rehydrate_resources, self.subdevices)
+
     # ------------------------------------------------------------------
     # Learned modes (issue #327)
     # ------------------------------------------------------------------
@@ -999,6 +1050,128 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             self._entry, data={**self._entry.data, **identity}
         )
 
+    # ------------------------------------------------------------------
+    # Discovery snapshot (issue #295)
+    # ------------------------------------------------------------------
+
+    def _bound_keys(self) -> frozenset[tuple[str, str]]:
+        """This entity set's identity: one (subdevice, state key) pair per
+        bound entity. `_key` is the unique_id suffix, so two discoveries that
+        agree here would register byte-identical entities."""
+        return frozenset((b.subdevice.key, _key(b)) for b in self.bound)
+
+    @callback
+    def _save_snapshot(self, resources: dict[str, dict], candidates: list[Subdevice]) -> None:
+        """Record what this first cycle handed `_run_discovery`, so the next
+        restart can replay it while the appliance is unreachable.
+
+        `candidates` is the pre-narrowing subdevice list (issue #177):
+        `discover_partitioned` takes candidates and returns the live ones, so
+        replaying against the narrowed list would rediscover nothing for a
+        composite appliance's siblings.
+        """
+        ident = self._identity
+        payload = {
+            "resources": dict(resources),
+            "subdevice_candidates": [asdict(su) for su in candidates],
+            "identity": asdict(ident) if ident is not None else None,
+        }
+        self._snapshot_store.async_delay_save(lambda: payload, _SNAPSHOT_SAVE_DELAY_S)
+
+    async def async_rehydrate(self) -> bool:
+        """Register the last known entity set without reaching the device.
+
+        Replays the stored snapshot through `_run_discovery`, which is what
+        makes this faithful: same code path, same registry resolution, so an
+        offline load produces the entity set the device last actually
+        reported rather than a guess reconstructed from a parallel format.
+
+        Returns False when there's nothing stored (an entry that has never
+        polled successfully) or the replay produced nothing usable -- the
+        caller raises ConfigEntryNotReady in that case, exactly as before.
+        """
+        try:
+            stored = await self._snapshot_store.async_load()
+        except Exception as e:  # corrupt or unreadable store
+            self._log.warning("could not read discovery snapshot: %s", e)
+            return False
+        if not stored or not stored.get("resources"):
+            return False
+
+        ident = stored.get("identity")
+        if ident is not None:
+            self._identity = DeviceIdentity(
+                manufacturer=ident.get("manufacturer") or "",
+                model=ident.get("model") or "",
+                name=ident.get("name") or "",
+                serial=ident.get("serial"),
+                device_types=tuple(ident.get("device_types") or ()),
+                raw=ident.get("raw") or {},
+            )
+        # JSON gives lists back where Subdevice declares tuples, and it's a
+        # frozen (hashable) dataclass used as a dict key in flatten().
+        self.subdevices = [
+            Subdevice(
+                kind=su["kind"],
+                key=su["key"],
+                seed_path=tuple(su.get("seed_path") or ()),
+                flat_hrefs=tuple(su.get("flat_hrefs") or ()),
+            )
+            for su in stored.get("subdevice_candidates") or ()
+        ]
+
+        resources = stored["resources"]
+        try:
+            self._run_discovery(resources)
+        except Exception as e:
+            # A snapshot written by an older release can outlive the registry
+            # shape it was discovered against. Fall back to the pre-#295
+            # behavior rather than failing the entry outright.
+            self._log.warning("discovery snapshot could not be replayed: %s", e, exc_info=True)
+            self.bound = []
+            return False
+
+        # _run_discovery sets _discovered; put it back. The snapshot only
+        # supplied an entity set to register -- the first live poll must
+        # still enumerate subdevices and rediscover for real.
+        self._discovered = False
+        self._rehydrate_resources = resources
+        self._rehydrated_keys = self._bound_keys()
+        if not self.bound:
+            return False
+        self._log.info(
+            "device unreachable; restored %d entities from the last discovery "
+            "snapshot and will keep retrying every %ds",
+            len(self.bound),
+            SUMMARY_INTERVAL_S,
+        )
+        return True
+
+    @callback
+    def _reconcile_rehydrated(self) -> None:
+        """Reload the entry when a live discovery disagrees with the snapshot
+        this load registered from.
+
+        Platforms enumerate `bound` exactly once, at forward time, so a
+        firmware update, a newly-answering sibling subdevice or a different
+        appliance at the same IP can't be picked up in place -- the entry has
+        to come back up against the live set.
+        """
+        if self._rehydrated_keys is None:
+            return
+        stale = self._rehydrated_keys
+        self._rehydrated_keys = None
+        live = self._bound_keys()
+        if live == stale:
+            return
+        self._log.info(
+            "live discovery differs from the snapshot this entry loaded from "
+            "(%d entities gone, %d new); reloading",
+            len(stale - live),
+            len(live - stale),
+        )
+        self.hass.config_entries.async_schedule_reload(self._entry.entry_id)
+
     def _run_discovery(self, resources: dict[str, dict]) -> None:
         # Diagnostics only -- names the firmware generation (e.g. '7.0 Air
         # conditioner' is Tizen Lite); doesn't route, since every device
@@ -1422,7 +1595,10 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             # StateCache has no eviction, so the only way to keep them out
             # is to not put them in. Safe to reorder: _run_discovery reads
             # the passed dict, never the cache.
+            candidates = list(self.subdevices)
             self._run_discovery(resources)
+            self._save_snapshot(resources, candidates)
+            self._reconcile_rehydrated()
             resources = self._live_subdevice_resources(resources)
         sweep_mismatch = False
         if self._observe.mode == MODE_OBSERVE:

custom_components/localthings/entity.py

@@ -35,12 +35,16 @@ def _is_included(bound: BoundEntity, coordinator: LocalThingsCoordinator) -> boo
     `bound`'s own subdevice's canonical view instead of the raw snapshot,
     same rule as everywhere else a whole-resources-dict scan happens --
     this is a free function, so it can't use self._resources.
+
+    Reads `discovery_resources`, not `last_resources`: on an offline load
+    (issue #295) the live cache is still empty, and judging existence
+    against it would filter every rehydrated entity away.
     """
-    rep = coordinator.last_resources.get(bound.href)
+    rep = coordinator.discovery_resources.get(bound.href)
     if rep is None:
         return False
     if bound.desc.exists_fn is not None:
-        return bound.desc.exists_fn(rep, coordinator.canonical_resources(bound.subdevice))
+        return bound.desc.exists_fn(rep, coordinator.discovery_canonical(bound.subdevice))
     if bound.desc.field:
         if not rep or is_stub_rep(rep):
             return True

docs/investigations/offline-setup.md

@@ -1,14 +1,19 @@
 # Loading a config entry while the appliance is offline
 
-Issue #295 asks for sub-30s recovery when a powered-off appliance comes back,
-instead of HA's `ConfigEntryNotReady` backoff (30s → … → 15 min). PR #303
-tries to get there by catching the first-refresh failure in
-`async_setup_entry` and loading the entry anyway.
+Issue #295 asks for faster recovery when a powered-off appliance comes back,
+instead of waiting out HA's `ConfigEntryNotReady` backoff. PR #303 tried to
+get there by catching the first-refresh failure in `async_setup_entry` and
+loading the entry anyway.
 
 That doesn't work here, and the reason is worth writing down: this
 integration has no static entity list. Every entity comes from discovery,
 and discovery only happens inside a successful poll.
 
+(The issue's "up to 15 minutes" is out of date, incidentally. Current HA
+retries on `2 ** min(tries, 4) * 5` seconds — capped at 80s, not 900. The
+backoff was never the worst part; a device card reading "Retrying setup" with
+no entities behind it is.)
+
 ## What PR #303 produces today
 
 Measured on the PR's branch — set up with `_poll_once` raising, then advance
@@ -55,51 +60,63 @@ device description* to build entities from — ESPHome keeps its entity list in
 `.storage`, Shelly caches device info. The pattern is portable; the mechanism
 underneath it is the part PR #303 is missing.
 
-## What a working version needs
+## How the implemented version works
 
-Three separable pieces, plus a gating rule.
+Three pieces, plus a gating rule.
 
 ### 1. A persisted discovery snapshot
 
-Write the discovery result back onto the config entry after each successful
-`_run_discovery`. There is already precedent for exactly this shape —
-`_persist_identity` (`coordinator.py:973`), `CONF_LEARNED_MODES`,
-`CONF_CLOUD_COURSES`.
+After each successful first cycle, `_save_snapshot` banks exactly the
+`resources` dict that cycle handed `_run_discovery`, along with the
+pre-narrowing subdevice candidate list and the `DeviceIdentity` read from
+`/oic/*`.
 
-`BoundEntity` holds live `Capability`/`SamsungEntityDescription` objects, so
-it isn't directly serializable, but it is re-derivable: persist
-`device_type_name`, the materialized `Subdevice` list (plain str/tuple
-fields), and per entity the `(subdevice key, capability, desc.key, instance,
-key_override, instance_name)` tuple. Rehydrate by re-resolving through
-`resolve_registry` + `CAPABILITIES`.
+Storing the poll input rather than a rendered entity list is the decision
+that keeps this honest. `BoundEntity` holds live
+`Capability`/`SamsungEntityDescription` objects and isn't serializable, so
+the alternative was a parallel format plus a re-resolution path — a second
+implementation of discovery that could drift from the real one. Replaying the
+input through `_run_discovery` means the same code, the same registry
+resolution, and no second source of truth.
 
-Persist the *post-`_is_included`* set, so rehydration doesn't need reps, and
-flag the coordinator as rehydrated so gate 3 above is skipped for that pass.
+Three things ride along because `_run_discovery` reads them off `self`
+rather than out of `resources`, and getting them wrong would silently resolve
+a *different* registry offline than online — which reconciliation below would
+then see as a real change and reload on every restart:
+
+- `_identity.device_types` routes `resolve_registry`.
+- `self.subdevices` is the candidate list `discover_partitioned` narrows;
+  replaying against the already-narrowed list finds no siblings at all.
+- `_identity.manufacturer`/`model` feed `device_info`.
+
+It lives in `.storage` (`Store`, keyed on entry_id) rather than on the config
+entry: it's device state, not configuration, and runs to tens of kilobytes.
+`async_remove_entry` deletes it with the entry.
 
 ### 2. Reconcile on reconnect
 
-The snapshot is a guess about a device we haven't talked to yet. When the
-first successful poll lands, `_run_discovery` computes the real set; if it
-differs from what was rehydrated, the entry has to reload, because gate 2
-means platforms can't add the difference in place. `async_schedule_reload`
-covers it (available well below the 2025.1.0 floor in `hacs.json`).
+The snapshot is a claim about a device we haven't talked to yet. When the
+first live poll lands, `_reconcile_rehydrated` compares the live entity set
+against the rehydrated one — as `(subdevice key, _key(bound))` pairs, which
+is the unique_id identity — and calls `async_schedule_reload` if they differ.
 
-This is the piece that makes the whole thing safe against a firmware update,
-a newly-appearing subdevice, or a different appliance at the same IP. PR #303
-has no equivalent.
+Gate 2 above is why this has to be a reload rather than an in-place fixup.
+It's what makes the feature safe against a firmware update, a sibling
+subdevice that starts answering, or a different appliance at the same IP.
 
 ### 3. Keep polling with no listeners
 
-Independent of the above, and worth doing on its own merits: hold one
-refresh alive for the entry's lifetime so scheduling never depends on entity
-count.
+`async_setup_entry` holds one listener for the entry's lifetime:
 
 ```python
 entry.async_on_unload(coordinator.async_add_listener(lambda: None))
 ```
 
-That alone fixes the measured "never polls again" bug, and covers the case
-where every rehydrated entity happens to be registry-disabled.
+Registered *before* the first refresh, so scheduling survives a refresh that
+fails. This alone fixes the measured "never polls again" bug, and covers a
+rehydrated set whose entities are all registry-disabled. Removing the last
+listener unschedules the timer, and HA runs `async_on_unload` callbacks when
+setup raises, so the setup-retry path doesn't leak a polling coordinator.
 
 ### Gating rule: only load offline when there's a snapshot
 
@@ -110,6 +127,10 @@ without one HA's backoff is still the right behavior. It also leaves room for
 the #168-style flows that need to interact with the device during setup: a
 device that never completed setup still blocks.
 
+It also means `async_remove_config_entry_device` is no longer reachable with
+an empty `coordinator.subdevices`, so an offline load can't offer to delete a
+real-but-unreachable subdevice.
+
 ## What this still won't do
 
 Entities will be present and `unavailable` — not showing their last values.
@@ -124,26 +145,22 @@ Worth being explicit about, because it is the gap between what PR #303
 promises in the thread ("load their previously recorded states") and what any
 correct version can deliver.
 
-## The cheaper alternative
-
-If the goal is only issue #295's — fast recovery — pieces 1 and 2 are
-unnecessary. Keep `ConfigEntryNotReady`, and add a retry that probes on a
-fixed short interval and calls `async_schedule_reload` on the first success.
-No persistence, no reconcile, no new failure modes. The device tile still
-reads "Retrying setup" while the appliance is off, which is true, and
-recovery drops from up-to-15-min to one probe interval.
+## Rejected: zeroconf
 
 The issue's other suggestion — wire zeroconf so the device's own boot
-announcement triggers the retry — is a non-starter as things stand: there is
-no `zeroconf` or `dhcp` key in `manifest.json` and the config flow is
-user-driven only, so HA has no discovery signal for this integration to hang
-a retry on. It would first need a confirmed mDNS service on the appliance.
+announcement triggers a retry, which is the genuinely idiomatic HA answer —
+is a non-starter as things stand: there is no `zeroconf` or `dhcp` key in
+`manifest.json` and the config flow is user-driven only, so HA has no
+discovery signal for this integration to hang a retry on. It would first need
+a confirmed mDNS service on the appliance. Worth revisiting if one turns up;
+it would make recovery near-instant instead of within one poll interval.
 
-## Recommendation
+## Rejected: the cheap version
 
-The cheap alternative solves the filed issue. The three-piece version solves
-the goal stated later in the PR thread, at the cost of a new persisted
-schema, a reconcile path, and a reload-on-mismatch — and it still leaves
-entities unavailable, which is the part that was actually being asked for.
-Do the cheap one first; treat offline entity materialization as a separate
-change with its own issue.
+Keeping `ConfigEntryNotReady` and adding a probe that calls
+`async_schedule_reload` on first success would have fixed the recovery *time*
+in about twenty lines, with no persistence and no reconcile. It was rejected
+because it leaves the device reading as broken for as long as the appliance
+is off, which is the half of issue #295 that actually bites — an appliance
+switched off at the wall is offline for days, not seconds, and a whole
+integration that looks failed for that entire window is the complaint.

tests/localthings/test_coordinator.py

@@ -70,8 +70,13 @@ async def test_summary_interval(hass: HomeAssistant, mock_entry, mock_coordinato
     assert coordinator.update_interval == timedelta(seconds=SUMMARY_INTERVAL_S)
 
 
-async def test_offline_initial_setup_succeeds(hass: HomeAssistant, mock_entry) -> None:
-    """ConfigEntry setup succeeds in offline mode when initial poll fails."""
+async def test_update_failed_on_persistent_poll_error(hass: HomeAssistant, mock_entry) -> None:
+    """ConfigEntryNotReady raised when poll fails even after reconnect.
+
+    An entry with no stored discovery snapshot has never reached this device,
+    so there is nothing to load offline from (issue #295) -- it stays on HA's
+    backoff rather than loading empty.
+    """
 
     with (
         patch("custom_components.localthings.coordinator.LocalThingsCoordinator._connect_session"),
@@ -84,10 +89,8 @@ async def test_offline_initial_setup_succeeds(hass: HomeAssistant, mock_entry) -
         result = await hass.config_entries.async_setup(mock_entry.entry_id)
         await hass.async_block_till_done()
 
-    # Setup succeeds — entry should be in hass.data and state is LOADED
-    assert result
-    assert mock_entry.entry_id in hass.data.get(DOMAIN, {})
-    assert mock_entry.state == ConfigEntryState.LOADED
+    # Setup fails — entry should not be in hass.data
+    assert not result or mock_entry.entry_id not in hass.data.get(DOMAIN, {})
 
 
 # ---------------------------------------------------------------------------
@@ -1823,11 +1826,14 @@ async def test_first_refresh_timeout_recovers_via_reconnect(
     assert coordinator._consecutive_poll_timeouts == 0
 
 
-async def test_first_refresh_persistent_timeout_succeeds_offline(
+async def test_first_refresh_persistent_timeout_fails_setup(
     hass: HomeAssistant, mock_entry
 ) -> None:
-    """When initial refresh times out, setup still succeeds in offline mode
-    and retries in background."""
+    """With no snapshot to load from, a reconnect that times out too must
+    fail the first refresh so HA retries on its backoff -- not load an
+    entity-less entry. The session it left open is closed on the way out
+    (`_poll_once` keeps it up on a `TimeoutError`, and the source port is
+    fixed per device)."""
     with (
         patch("custom_components.localthings.coordinator.LocalThingsCoordinator._connect_session"),
         patch(
@@ -1835,11 +1841,21 @@ async def test_first_refresh_persistent_timeout_succeeds_offline(
             side_effect=_HANDSHAKE_TIMEOUT,
         ),
         patch("custom_components.localthings.coordinator.LocalThingsCoordinator._close_session"),
+        # Asserted on `async_close` rather than `_close_session`: the
+        # reconnect path calls the latter on its own, so it is already true
+        # whether or not setup cleans up after itself.
+        patch.object(
+            LocalThingsCoordinator,
+            "async_close",
+            autospec=True,
+        ) as mock_async_close,
     ):
         await hass.config_entries.async_setup(mock_entry.entry_id)
         await hass.async_block_till_done()
 
-    assert mock_entry.state is ConfigEntryState.LOADED
+    assert mock_entry.state is ConfigEntryState.SETUP_RETRY
+    assert not hass.states.async_all()
+    mock_async_close.assert_awaited_once()
 
 
 async def test_post_discovery_timeout_is_still_deferred(

tests/localthings/test_offline_setup.py

@@ -0,0 +1,366 @@
+"""Loading a config entry while the appliance is unreachable (issue #295).
+
+The device's entity set only exists as the output of a live poll, so coming
+up offline means replaying the last successful discovery from a stored
+snapshot. These tests pin the four things that makes load-bearing: the
+snapshot gets written, it produces the same entity set offline, the
+coordinator keeps polling until the device answers, and a live discovery that
+disagrees with the snapshot reloads the entry rather than silently keeping a
+stale set.
+"""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from datetime import timedelta
+from unittest.mock import patch
+
+import pytest
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+from homeassistant.util import dt as dt_util
+from pytest_homeassistant_custom_component.common import async_fire_time_changed
+
+from custom_components.localthings.const import DOMAIN, SUMMARY_INTERVAL_S
+from custom_components.localthings.coordinator import (
+    _SNAPSHOT_SAVE_DELAY_S,
+    LocalThingsCoordinator,
+)
+from custom_components.localthings.registry.identity import DeviceIdentity
+
+from .conftest import _load_fridge_resources as _load_fridge
+
+_COORD = "custom_components.localthings.coordinator.LocalThingsCoordinator"
+
+
+@contextmanager
+def _reachable(resources: dict, identity: DeviceIdentity | None = None):
+    """A device that answers, optionally with an /oic/* identity -- which
+    `_connect_session` is what normally reads, so a test that patches it out
+    otherwise leaves `_identity` None."""
+
+    def _connect(self) -> None:
+        self._identity = identity
+
+    with (
+        patch(f"{_COORD}._connect_session", _connect),
+        patch(f"{_COORD}._poll_once", return_value=resources),
+        patch(f"{_COORD}._close_session"),
+    ):
+        yield
+
+
+@contextmanager
+def _unreachable():
+    with (
+        patch(f"{_COORD}._connect_session"),
+        patch(f"{_COORD}._poll_once", side_effect=OSError("device offline")),
+        patch(f"{_COORD}._close_session"),
+    ):
+        yield
+
+
+def _store_key(entry) -> str:
+    return f"{DOMAIN}.{entry.entry_id}.discovery"
+
+
+async def _flush_snapshot(hass: HomeAssistant) -> None:
+    """Run out `async_delay_save`'s timer without reaching the next poll."""
+    async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=_SNAPSHOT_SAVE_DELAY_S + 1))
+    await hass.async_block_till_done()
+
+
+async def _tick(hass: HomeAssistant) -> None:
+    """Advance past one summary interval so the coordinator polls again.
+
+    `wait_background_tasks` is load-bearing: DataUpdateCoordinator runs its
+    interval refresh as a background task, which a plain block_till_done
+    doesn't await -- the poll would still be in flight at the assertion.
+    """
+    async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=SUMMARY_INTERVAL_S + 1))
+    await hass.async_block_till_done(wait_background_tasks=True)
+
+
+async def _setup_online_then_unload(hass: HomeAssistant, entry, resources: dict) -> set[str]:
+    """Bring the entry up against a live device, bank the snapshot, and take
+    it back down. Returns the entity_ids that run produced."""
+    with _reachable(resources):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+        entity_ids = {s.entity_id for s in hass.states.async_all()}
+        await _flush_snapshot(hass)
+        await hass.config_entries.async_unload(entry.entry_id)
+        await hass.async_block_till_done()
+    return entity_ids
+
+
+# ---------------------------------------------------------------------------
+# Writing the snapshot
+# ---------------------------------------------------------------------------
+
+
+async def test_snapshot_written_after_first_discovery(
+    hass: HomeAssistant, mock_entry, mock_coordinator_session, hass_storage
+) -> None:
+    """A successful first cycle banks what it handed _run_discovery."""
+    await hass.config_entries.async_setup(mock_entry.entry_id)
+    await hass.async_block_till_done()
+    await _flush_snapshot(hass)
+
+    stored = hass_storage[_store_key(mock_entry)]["data"]
+    assert stored["resources"]
+    assert "/information/vs/0" in stored["resources"]
+    assert "subdevice_candidates" in stored
+
+
+async def test_snapshot_not_written_when_device_never_answers(
+    hass: HomeAssistant, mock_entry, hass_storage
+) -> None:
+    """Nothing to bank, so nothing is -- this is what keeps the no-snapshot
+    gate meaningful on a brand-new entry."""
+    with _unreachable():
+        await hass.config_entries.async_setup(mock_entry.entry_id)
+        await hass.async_block_till_done()
+        # Asserted before the flush: running the save timer out also runs out
+        # HA's own setup-retry timer, which puts the entry back into
+        # SETUP_IN_PROGRESS.
+        assert mock_entry.state is ConfigEntryState.SETUP_RETRY
+        await _flush_snapshot(hass)
+
+    assert _store_key(mock_entry) not in hass_storage
+
+
+async def test_snapshot_removed_when_entry_removed(
+    hass: HomeAssistant, mock_entry, mock_coordinator_session, hass_storage
+) -> None:
+    """The store is keyed on entry_id, so re-adding the appliance mints a new
+    one -- the old file has to go with the entry that wrote it."""
+    await hass.config_entries.async_setup(mock_entry.entry_id)
+    await hass.async_block_till_done()
+    await _flush_snapshot(hass)
+    assert _store_key(mock_entry) in hass_storage
+
+    await hass.config_entries.async_remove(mock_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert hass_storage.get(_store_key(mock_entry), {}).get("data") is None
+
+
+# ---------------------------------------------------------------------------
+# Loading from it
+# ---------------------------------------------------------------------------
+
+
+async def test_offline_load_restores_the_same_entity_set(
+    hass: HomeAssistant, mock_entry, hass_storage
+) -> None:
+    """The whole point: a restart with the appliance powered off comes up on
+    the entity set the device last actually reported."""
+    resources = _load_fridge()
+    online_ids = await _setup_online_then_unload(hass, mock_entry, resources)
+    assert online_ids  # guard: the online run must actually produce entities
+
+    with _unreachable():
+        await hass.config_entries.async_setup(mock_entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert mock_entry.state is ConfigEntryState.LOADED
+    assert {s.entity_id for s in hass.states.async_all()} == online_ids
+
+
+async def test_offline_entities_are_unavailable_not_stale(hass: HomeAssistant, mock_entry) -> None:
+    """Restored entities must not render the snapshot's values -- the
+    appliance is unreachable, so `unavailable` is the honest state and the
+    live cache stays empty to enforce it."""
+    resources = _load_fridge()
+    await _setup_online_then_unload(hass, mock_entry, resources)
+
+    with _unreachable():
+        await hass.config_entries.async_setup(mock_entry.entry_id)
+        await hass.async_block_till_done()
+
+    states = hass.states.async_all()
+    assert states
+    assert all(s.state == "unavailable" for s in states)
+
+    coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
+    assert coordinator.rehydrated
+    assert not coordinator.last_resources
+
+
+async def test_offline_load_without_snapshot_still_fails(hass: HomeAssistant, mock_entry) -> None:
+    """No snapshot means no device metadata to build anything from, so the
+    entry stays on HA's backoff rather than loading empty."""
+    with _unreachable():
+        await hass.config_entries.async_setup(mock_entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert mock_entry.state is ConfigEntryState.SETUP_RETRY
+    assert not hass.states.async_all()
+
+
+async def test_corrupt_snapshot_falls_back_to_setup_retry(
+    hass: HomeAssistant, mock_entry, hass_storage
+) -> None:
+    """A snapshot whose resources no longer replay cleanly must not take the
+    entry down with it."""
+    key = _store_key(mock_entry)
+    hass_storage[key] = {
+        "version": 1,
+        "minor_version": 1,
+        "key": key,
+        "data": {"resources": {"/information/vs/0": "not-a-rep"}, "subdevice_candidates": []},
+    }
+
+    with _unreachable():
+        await hass.config_entries.async_setup(mock_entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert mock_entry.state is ConfigEntryState.SETUP_RETRY
+
+
+async def test_snapshot_restores_identity(hass: HomeAssistant, mock_entry) -> None:
+    """`/oic/d`'s device types route the registry, so an offline load that
+    lost them could resolve a different one than the live poll did -- which
+    would show up as a spurious reconcile reload every restart."""
+    resources = _load_fridge()
+    identity = DeviceIdentity(
+        manufacturer="Samsung",
+        model="TEST-MODEL",
+        name="Fridge",
+        serial=None,
+        device_types=("oic.d.refrigerator",),
+        raw={"/oic/p": {}, "/oic/d": {}, "/oic/res": []},
+    )
+    with _reachable(resources, identity):
+        await hass.config_entries.async_setup(mock_entry.entry_id)
+        await hass.async_block_till_done()
+        await _flush_snapshot(hass)
+        await hass.config_entries.async_unload(mock_entry.entry_id)
+        await hass.async_block_till_done()
+
+    with _unreachable():
+        await hass.config_entries.async_setup(mock_entry.entry_id)
+        await hass.async_block_till_done()
+
+    coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
+    assert coordinator._identity is not None
+    assert coordinator._identity.device_types == ("oic.d.refrigerator",)
+
+
+# ---------------------------------------------------------------------------
+# Recovery and reconciliation
+# ---------------------------------------------------------------------------
+
+
+async def test_offline_load_keeps_polling_and_recovers(hass: HomeAssistant, mock_entry) -> None:
+    """The failure the PR this replaces actually shipped: with zero live
+    listeners the base coordinator stops rescheduling, and the entry never
+    polls again. Entities must go available on the next interval once the
+    appliance answers."""
+    resources = _load_fridge()
+    await _setup_online_then_unload(hass, mock_entry, resources)
+
+    with _unreachable():
+        await hass.config_entries.async_setup(mock_entry.entry_id)
+        await hass.async_block_till_done()
+
+    coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
+    assert coordinator._unsub_refresh is not None  # a poll is actually queued
+    assert not coordinator.last_update_success
+
+    with _reachable(resources):
+        await _tick(hass)
+
+    assert coordinator.last_update_success
+    assert any(s.state != "unavailable" for s in hass.states.async_all())
+
+
+async def test_reconcile_reloads_when_live_discovery_differs(
+    hass: HomeAssistant, mock_entry
+) -> None:
+    """Platforms enumerate `bound` once, so a live set that disagrees with the
+    snapshot can only be adopted by bringing the entry back up."""
+    resources = _load_fridge()
+
+    with _reachable(resources):
+        await hass.config_entries.async_setup(mock_entry.entry_id)
+        await hass.async_block_till_done()
+        coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
+        victim = coordinator.bound[0].href
+        await _flush_snapshot(hass)
+        await hass.config_entries.async_unload(mock_entry.entry_id)
+        await hass.async_block_till_done()
+
+    with _unreachable():
+        await hass.config_entries.async_setup(mock_entry.entry_id)
+        await hass.async_block_till_done()
+
+    reduced = {href: rep for href, rep in resources.items() if href != victim}
+    with (
+        _reachable(reduced),
+        patch.object(hass.config_entries, "async_schedule_reload") as reload,
+    ):
+        await _tick(hass)
+
+    reload.assert_called_once_with(mock_entry.entry_id)
+
+
+async def test_reconcile_is_quiet_when_live_discovery_agrees(
+    hass: HomeAssistant, mock_entry
+) -> None:
+    """The common case -- same appliance, same firmware -- must not reload,
+    or every offline restart would cost a second setup cycle."""
+    resources = _load_fridge()
+    await _setup_online_then_unload(hass, mock_entry, resources)
+
+    with _unreachable():
+        await hass.config_entries.async_setup(mock_entry.entry_id)
+        await hass.async_block_till_done()
+
+    with (
+        _reachable(resources),
+        patch.object(hass.config_entries, "async_schedule_reload") as reload,
+    ):
+        await _tick(hass)
+
+    reload.assert_not_called()
+
+
+async def test_live_load_never_reconciles(
+    hass: HomeAssistant, mock_entry, mock_coordinator_session
+) -> None:
+    """An entry that came up against a live device has nothing to reconcile
+    against; the reload path must stay out of the normal startup entirely."""
+    with patch.object(hass.config_entries, "async_schedule_reload") as reload:
+        await hass.config_entries.async_setup(mock_entry.entry_id)
+        await hass.async_block_till_done()
+
+    coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
+    assert not coordinator.rehydrated
+    reload.assert_not_called()
+
+
+@pytest.mark.parametrize("failures", [1, 3])
+async def test_offline_load_survives_repeated_poll_failures(
+    hass: HomeAssistant, mock_entry, failures: int
+) -> None:
+    """Recovery isn't one-shot: the entry keeps its entities and keeps
+    retrying across however many intervals the appliance stays dark."""
+    resources = _load_fridge()
+    online_ids = await _setup_online_then_unload(hass, mock_entry, resources)
+
+    with _unreachable():
+        await hass.config_entries.async_setup(mock_entry.entry_id)
+        await hass.async_block_till_done()
+        for _ in range(failures):
+            await _tick(hass)
+
+    assert mock_entry.state is ConfigEntryState.LOADED
+    assert {s.entity_id for s in hass.states.async_all()} == online_ids
+
+    with _reachable(resources):
+        await _tick(hass)
+
+    coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
+    assert coordinator.last_update_success

tests/test_airconditioner_ailp_fac_capabilities.py

@@ -36,6 +36,15 @@ class _FakeCoordinator:
     def canonical_resources(self, subdevice):
         return self.last_resources
 
+    # _is_included judges existence against the discovery view, which is the
+    # live cache for everything but an offline load (issue #295).
+    @property
+    def discovery_resources(self):
+        return self.last_resources
+
+    def discovery_canonical(self, subdevice):
+        return self.canonical_resources(subdevice)
+
 
 def _resources():
     return _load_device("airconditioner_ailp_fac")

tests/test_entity.py

@@ -29,6 +29,15 @@ class _FakeCoordinator:
         # subdevices (issue #177).
         return self.last_resources
 
+    # _is_included judges existence against the discovery view, which is the
+    # live cache for everything but an offline load (issue #295).
+    @property
+    def discovery_resources(self):
+        return self.last_resources
+
+    def discovery_canonical(self, subdevice):
+        return self.canonical_resources(subdevice)
+
 
 def _coord(last_resources) -> LocalThingsCoordinator:
     return cast(LocalThingsCoordinator, _FakeCoordinator(last_resources))

tests/test_unique_ids.py

@@ -34,6 +34,15 @@ class _FakeCoordinator:
 
         return canonical_view(subdevice, self.last_resources, self._subdevices)
 
+    # _is_included judges existence against the discovery view, which is the
+    # live cache for everything but an offline load (issue #295).
+    @property
+    def discovery_resources(self):
+        return self.last_resources
+
+    def discovery_canonical(self, subdevice):
+        return self.canonical_resources(subdevice)
+
 
 @pytest.mark.parametrize("name", _FIXTURE_NAMES)
 def test_key_is_unique_across_all_bound_entities(name):