Pace writes and retry a timed-out PUT in place

Issue #384: an AC reported `climate.set_temperature`/`set_hvac_mode`/
`set_swing_mode` failing with "command failed for <href> after reconnect:
session operation timed out" across several hrefs.

smartthings-local retransmits every block of a GET (3 attempts) but sends
a POST exactly once, so a single dropped datagram is a write failure. Two
things made that likely and then unrecoverable:

- The write path never paced. RT-OCF drops requests that arrive inside
  the rate-limit interval, and the reconnect retry is the worst case --
  `_connect_session()` reads /oic/p, /oic/d and /oic/res immediately
  before it, so the retry PUT lands milliseconds behind three GETs.
- Any exception, timeout included, was read as session death: close the
  session, downgrade OBSERVE to poll, pause, re-handshake, one more
  single-shot PUT. The poll path already reads a TimeoutError the other
  way (`_defer_reconnect_for`).

Pace before both POST sites, and retry a timed-out write once on the
session in hand before spending a reconnect on it. The reconnect ladder
is unchanged past that point; it just moves to a helper.

README.md

@@ -85,7 +85,7 @@ This repo doesn't include the needed CA bundle. For an example of how to obtain
 5. The flow sends a DTLS `ClientHello` to every port in the `49152-49160` range at once and keeps the one that answers -- a real DTLS server identifies itself in about one round trip, and the probe stops there, so nothing is left behind on the appliance. Only that port is then given a real certificate handshake: it fetches the current UUID from Samsung's cloud gateway, mints a leaf cert signed by your CA, and reads the device's identity and `/device/0`. On success it creates the config entry, already knowing the appliance's serial, model, and type.
 6. Every subsequent device only asks for the host IP. The stored CA credentials are reused, and so is the leaf cert itself -- every appliance accepts the same one -- so adding a second appliance doesn't depend on Samsung's cloud being reachable at all. If a device rejects the reused cert (the UUID behind it does rotate), the flow mints a fresh one and retries by itself.
 
-Entities appear under one HA device per appliance, named for the appliance's type and model. Rename freely: the device is keyed on its serial, not its name.
+Entities appear under one HA device per appliance, named for the appliance's type and model. Rename freely: the device is keyed on the appliance's own OCF device ID, not its name. (Some Samsung models ship the same serial number on every unit of a model, so the serial can't tell two of them apart -- the OCF device ID can.)
 
 ---
 
@@ -237,7 +237,8 @@ docker-compose.yml / ha_config/   Local HA dev environment
 If your appliance's type isn't recognized, or it exposes resources this integration doesn't model yet, a Repairs
 issue appears under Settings > System > Repairs pointing you at Settings > Devices & Services > this device >
 the menu > Download diagnostics. That download is already redacted of account/network identifiers (Bixby login
-email, access tokens, device IDs, MAC addresses, serial numbers) before it's generated, so it's safe to attach
+email, access tokens, hashed device IDs, MAC addresses, serial numbers, and the owner-set device name) before it's
+generated, so it's safe to attach
 directly to a new issue using the linked device-support template. This is the fastest way to help add or expand
 support for hardware the maintainers don't have.
 

custom_components/localthings/__init__.py

@@ -19,6 +19,7 @@ from homeassistant.helpers.typing import ConfigType
 from .const import CONF_DEVICE_TYPE, CONF_HOST, CONF_PORT, CONF_SERIAL, DOMAIN, PLATFORMS
 from .coordinator import LocalThingsCoordinator, snapshot_store
 from .registry.identity import resolve_serial
+from .rekey import rekey_entry
 from .services import async_setup_services
 
 _LOGGER = logging.getLogger(__name__)
@@ -82,67 +83,20 @@ def _repair_placeholder_keys(hass: HomeAssistant, entry: ConfigEntry, serial: st
     """Re-key registry entries this entry minted from the placeholder identity.
 
     Before the identity moved onto the config entry, the coordinator seeded
-    `device_serial` with the host and only replaced it after the first poll.
+    its device key with the host and only replaced it after the first poll.
     Anything that registered in between -- the connection-mode sensor
     especially, added unconditionally rather than from `bound` -- was
     written into the registry keyed on the IP permanently, orphaned the
     moment the serial-keyed identity appeared (issue #236). Deleting the
     orphans by hand didn't help: the next restart that lost the same race
     recreated them.
-
-    Rewriting beats deleting where possible -- an entity keeps its
-    entity_id, name, area and automations. Only possible when the
-    serial-keyed key is still free; where both exist the placeholder-keyed
-    one is the dead duplicate (unavailable since the restart that created
-    it), so it goes.
     """
     host = entry.data[CONF_HOST]
     if serial == host:
         # A board with no usable serial resolves to the host, so its keys
         # were never placeholders.
         return
-
-    ent_reg = er.async_get(hass)
-    stale_prefix = f"{DOMAIN}_{host}_"
-    for entity in list(er.async_entries_for_config_entry(ent_reg, entry.entry_id)):
-        if not entity.unique_id.startswith(stale_prefix):
-            continue
-        new_unique_id = f"{DOMAIN}_{serial}_{entity.unique_id[len(stale_prefix) :]}"
-        if ent_reg.async_get_entity_id(entity.domain, DOMAIN, new_unique_id):
-            _LOGGER.debug("removing orphaned entity %s", entity.entity_id)
-            ent_reg.async_remove(entity.entity_id)
-        else:
-            _LOGGER.debug("re-keying entity %s to %s", entity.entity_id, new_unique_id)
-            ent_reg.async_update_entity(entity.entity_id, new_unique_id=new_unique_id)
-
-    dev_reg = dr.async_get(hass)
-    for device in list(dr.async_entries_for_config_entry(dev_reg, entry.entry_id)):
-        # `host` for the master, `host_<key>` for a subdevice (device_info_for).
-        stale = {
-            ident
-            for ident in device.identifiers
-            if ident[0] == DOMAIN and (ident[1] == host or ident[1].startswith(f"{host}_"))
-        }
-        if not stale:
-            continue
-        fresh = {(DOMAIN, f"{serial}{ident[1][len(host) :]}") for ident in stale}
-        existing = dev_reg.async_get_device(identifiers=fresh)
-        if existing is not None and existing.id != device.id:
-            # Removing a device takes its entities with it. Anything still
-            # attached here was re-keyed rather than removed above -- the
-            # surviving copy, not a duplicate -- so move it onto the device
-            # it now belongs to before the removal destroys it too.
-            for entity in er.async_entries_for_device(
-                ent_reg, device.id, include_disabled_entities=True
-            ):
-                ent_reg.async_update_entity(entity.entity_id, device_id=existing.id)
-            _LOGGER.debug("removing orphaned device %s", device.id)
-            dev_reg.async_remove_device(device.id)
-        else:
-            _LOGGER.debug("re-keying device %s to %s", device.id, fresh)
-            dev_reg.async_update_device(
-                device.id, new_identifiers=(device.identifiers - stale) | fresh
-            )
+    rekey_entry(hass, entry, host, serial)
 
 
 # Registries whose Dust/FineDust/SuperFineDust sensors gained pm10/pm25/pm1
@@ -249,8 +203,22 @@ async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
     v2 -> v3 relabels the recorded statistics for the particulate sensors,
     which gained a device_class/unit in the same release (issue #325).
+
+    v3 -> v4 moves the entry off the serialNum as its identity and onto the
+    OCF device UUID (issue #381). Deliberately almost a no-op here: the
+    UUID lives on the device, and this runs before any I/O -- and before
+    the device is even known to be reachable, since an entry can load
+    entirely from its snapshot while the appliance is off (issue #295).
+    So the migration only guarantees CONF_SERIAL is populated, which is
+    what the coordinator rewrites *from* once a live poll finally hands it
+    a UUID to rewrite *to*.
+
+    The version is still bumped now rather than at that point, because the
+    bump's real job is the `entry.version > 4` downgrade guard below: an
+    entry re-keyed onto a UUID and then loaded by a release that reads
+    CONF_SERIAL as the key would silently orphan every entity it has.
     """
-    if entry.version > 3:
+    if entry.version > 4:
         return False  # downgrade: this release doesn't know the newer shape
 
     if entry.version == 1:
@@ -268,6 +236,28 @@ async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         hass.config_entries.async_update_entry(entry, version=3)
         _LOGGER.debug("migrated entry %s to version 3", entry.entry_id)
 
+    if entry.version == 3:
+        # `or host` mirrors what the coordinator has always fallen back to,
+        # so the recorded legacy key is the one this entry's registry
+        # entries were actually minted under even if CONF_SERIAL never got
+        # written (a v1 entry whose migration predates it). Both being
+        # absent shouldn't happen for an entry the config flow created, but
+        # an exception raised here fails the whole entry -- so it bumps the
+        # version and leaves the data alone rather than taking that risk
+        # for a value the coordinator re-derives on its next poll anyway.
+        legacy_key = entry.data.get(CONF_SERIAL) or entry.data.get(CONF_HOST)
+        hass.config_entries.async_update_entry(
+            entry,
+            data={**entry.data, CONF_SERIAL: legacy_key} if legacy_key else entry.data,
+            version=4,
+        )
+        _LOGGER.debug(
+            "migrated entry %s to version 4 (legacy key=%s, awaiting a poll to adopt "
+            "the OCF device id)",
+            entry.entry_id,
+            legacy_key,
+        )
+
     return True
 
 

custom_components/localthings/config_flow.py

@@ -43,6 +43,7 @@ from .const import (
     CONF_CA_CERT_PEM,
     CONF_CA_KEY_PEM,
     CONF_CLOUD_COURSES_ENABLED,
+    CONF_DEVICE_KEY,
     CONF_DEVICE_TYPE,
     CONF_FINISH_TIME_HYSTERESIS_MINUTES,
     CONF_HOST,
@@ -662,7 +663,12 @@ def _read_device(sess, host: str, port: int) -> dict:
 
     from .registry.batch import parse_device0_batch
     from .registry.by_type import resolve as resolve_registry
-    from .registry.identity import read_identity, resolve_model, resolve_serial
+    from .registry.identity import (
+        read_identity,
+        resolve_device_key,
+        resolve_model,
+        resolve_serial,
+    )
 
     identity = read_identity(sess, None)
 
@@ -680,12 +686,19 @@ def _read_device(sess, host: str, port: int) -> dict:
 
     info = resources.get("/information/vs/0", {})
     registry = resolve_registry(resources, device_types=identity.device_types)
+    raw_serial = info.get("x.com.samsung.da.serialNum")
     return {
         "port": port,
         # Resolved through the same helpers _run_discovery uses, so the device
         # the coordinator registers up front is the one discovery would have
         # produced -- no rename, and no re-key, once the first poll lands.
-        "serial": resolve_serial(info.get("x.com.samsung.da.serialNum"), host),
+        #
+        # `device_key` is what the entry is actually keyed on; the serial is
+        # kept alongside it because the coordinator corroborates a later
+        # change of key against it (issue #381). read_identity has already
+        # fetched /oic/p and /oic/d above, so this costs no extra round trip.
+        "device_key": resolve_device_key(identity, raw_serial, host),
+        "serial": resolve_serial(raw_serial, host),
         "model": resolve_model(info.get("x.com.samsung.da.modelNum", ""), identity),
         "manufacturer": identity.manufacturer or "Samsung",
         "device_type_name": registry.name if registry is not None else None,
@@ -798,7 +811,10 @@ class LocalThingsConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
     # v3 relabels the particulate sensors' recorded statistics; a freshly
     # created entry has none to relabel, so it starts at the migrated
     # version rather than walking through v2 (see async_migrate_entry).
-    VERSION = 3
+    # v4 keys the entry on the OCF device UUID (issue #381), which the probe
+    # below resolves up front -- so a new entry is already on the v4 shape
+    # and has nothing to re-key either.
+    VERSION = 4
 
     def __init__(self) -> None:
         self._host: str = ""
@@ -817,7 +833,7 @@ class LocalThingsConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
         """Persist everything the probe resolved, identity included.
 
         The identity fields aren't decoration: the coordinator seeds
-        `device_serial` and its DeviceInfo from them at construction time,
+        `device_key` and its DeviceInfo from them at construction time,
         so entity unique_ids are correct from the first entity that
         registers, even if the first poll is slow or fails (issue #236).
         """
@@ -832,6 +848,7 @@ class LocalThingsConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
                 CONF_CA_KEY_PEM: self._ca_key_pem,
                 CONF_LEAF_CERT_PEM: info["leaf_cert_pem"],
                 CONF_LEAF_KEY_PEM: info["leaf_key_pem"],
+                CONF_DEVICE_KEY: info["device_key"],
                 CONF_SERIAL: info["serial"],
                 CONF_MODEL: info["model"],
                 CONF_MANUFACTURER: info["manufacturer"],
@@ -878,7 +895,33 @@ class LocalThingsConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
                 _LOGGER.exception("Unexpected error during device probe")
                 errors["base"] = "unknown"
             else:
-                await self.async_set_unique_id(f"localthings_{info['serial']}")
+                # An entry created before v4 still carries the serial-keyed
+                # unique_id until its first *live* poll adopts the UUID
+                # (coordinator._resolve_identity) -- which can be a long
+                # while for an appliance that is off, since an entry loads
+                # from its snapshot in the meantime (issue #295). The UUID
+                # check below can't see such an entry, so re-adding this
+                # very appliance during that window would be waved through
+                # as a second entry; the two would then collide the moment
+                # the older one re-keyed, and rekey_entry resolves a
+                # collision by *deleting* the duplicate rows -- taking the
+                # original entry's entity_ids, history and automations with
+                # them. Matched on the legacy key together with the host, so
+                # issue #381's two units (same serial, different addresses)
+                # stay separable.
+                legacy_unique_id = f"localthings_{info['serial']}"
+                if any(
+                    other.unique_id == legacy_unique_id
+                    and other.data.get(CONF_HOST) == self._host
+                    and CONF_DEVICE_KEY not in other.data
+                    for other in existing
+                ):
+                    return self.async_abort(reason="already_configured")
+                # Keyed on the OCF device UUID rather than the serialNum
+                # (issue #381): two units of a model that ship the same
+                # well-formed serial are indistinguishable here otherwise,
+                # and the second one is turned away as already configured.
+                await self.async_set_unique_id(f"localthings_{info['device_key']}")
                 self._abort_if_unique_id_configured()
                 if info["device_type_recognized"]:
                     return self._create_entry(info)

custom_components/localthings/const.py

@@ -30,6 +30,11 @@ CONF_LEAF_KEY_PEM = "leaf_key_pem"
 # output, the host itself for a placeholder-serial board -- issues
 # #83/#189), so it matches what _run_discovery computes on the first poll.
 CONF_SERIAL = "serial"
+# What this entry's devices and entities are keyed on -- normally the OCF
+# device UUID (issue #381). CONF_SERIAL stays alongside it as the pre-v4
+# key to re-key from, and as what corroborates a later change of UUID.
+# Absent until the first live poll, since only the device can report it.
+CONF_DEVICE_KEY = "device_key"
 CONF_MODEL = "model"
 CONF_MANUFACTURER = "manufacturer"
 CONF_DEVICE_TYPE = "device_type"

custom_components/localthings/coordinator.py

@@ -9,6 +9,7 @@ import logging
 import threading
 import time
 import zlib
+from collections.abc import Callable
 from dataclasses import asdict
 from datetime import timedelta
 from typing import Any, cast
@@ -31,6 +32,7 @@ from .const import (
     CONF_BYPASS_REMOTE_CONTROL,
     CONF_CLOUD_COURSES,
     CONF_CLOUD_COURSES_ENABLED,
+    CONF_DEVICE_KEY,
     CONF_DEVICE_TYPE,
     CONF_HOST,
     CONF_LEAF_CERT_PEM,
@@ -66,6 +68,7 @@ from .registry.entities import ClimateDesc
 from .registry.identity import (
     DeviceIdentity,
     device_display_name,
+    ocf_device_key,
     read_identity,
     resolve_model,
     resolve_serial,
@@ -78,6 +81,7 @@ from .registry.subdevices import (
     enumerate_subdevices,
     normalize_seed_batch,
 )
+from .rekey import rekey_entry
 
 # Sentinel for apply_cloud_courses: "leave this field as it is",
 # distinct from None which means "clear it".
@@ -205,7 +209,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
 
     bound: list[BoundEntity]
     device_info: DeviceInfo
-    device_serial: str
+    device_key: str
 
     # Class-level so tests can shrink these via patch.object() without
     # touching the production defaults.
@@ -310,15 +314,17 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         self._push_pending = False
         self._push_pending_lock = threading.Lock()
         # Identity is resolved once by the config flow's probe (issue #236).
-        # device_serial mints permanent registry keys, so it must be correct
+        # device_key mints permanent registry keys, so it must be correct
         # before the first entity registers -- a placeholder corrected once
         # the first poll lands orphans the first device/entity pair instead.
-        # The host fallback covers a pre-migration entry and matches what
-        # resolve_serial itself returns for a placeholder-serial board
-        # (issues #83/#189).
-        self.device_serial = entry.data.get(CONF_SERIAL) or entry.data[CONF_HOST]
+        # Ordered so an entry that has not polled since upgrading still
+        # loads under the key its registry rows already carry: the v4 UUID
+        # (issue #381), else the pre-v4 serial, else the host (#83/#189).
+        self.device_key = (
+            entry.data.get(CONF_DEVICE_KEY) or entry.data.get(CONF_SERIAL) or entry.data[CONF_HOST]
+        )
         self.device_info = DeviceInfo(
-            identifiers={(DOMAIN, self.device_serial)},
+            identifiers={(DOMAIN, self.device_key)},
             name=device_display_name(
                 entry.data.get(CONF_DEVICE_TYPE), entry.data.get(CONF_MODEL) or ""
             ),
@@ -768,8 +774,8 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                 else "Secondary Subdevice"
             )
         return DeviceInfo(
-            identifiers={(DOMAIN, f"{self.device_serial}_{subdevice.key}")},
-            via_device=(DOMAIN, self.device_serial),
+            identifiers={(DOMAIN, f"{self.device_key}_{subdevice.key}")},
+            via_device=(DOMAIN, self.device_key),
             name=f"{base_name} {label}",
             manufacturer=self.device_info.get("manufacturer") or "Samsung",
             model=model or None,
@@ -1093,8 +1099,87 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         self._skipped_subdevice_resources = skipped
         return kept
 
+    def _resolve_identity(self, polled_serial: str, *, from_snapshot: bool) -> tuple[str, str]:
+        """The (key, serial) this entry should be registered under, re-keying
+        the registries if the key has moved.
+
+        Key and serial are returned together because they have to agree: the
+        serial corroborates a later change of key, so adopting it while
+        *defending* the stored key would hand a different appliance the
+        corroboration it needs to win the next poll.
+
+        Nothing changes a key without calling `rekey_entry` in the same
+        breath, or the existing registry rows are orphaned (issue #236).
+        Two things are therefore never treated as a change of identity: a
+        snapshot replay (issue #295), which never reached the device, and a
+        poll that read no UUID, since the device saying nothing is not the
+        device saying something different.
+
+        A genuine difference is either the registered appliance under a new
+        identity -- the move onto the OCF device UUID (issue #381), or a
+        `di` regenerated by a hard reset -- or a different appliance on this
+        address, and the serialNum is what tells them apart. That test is
+        deliberately the same before and after an entry has adopted a UUID:
+        a pre-v4 entry has been running longest, so it is the last one that
+        should lose its rows to whatever now answers at its address.
+        """
+        host = self._entry.data[CONF_HOST]
+        stored_serial = self._entry.data.get(CONF_SERIAL)
+        if from_snapshot:
+            return self.device_key, stored_serial or polled_serial
+
+        polled_ocf = ocf_device_key(self._identity)
+        stored_key = self._entry.data.get(CONF_DEVICE_KEY)
+        # What this entry's rows carry today: a pre-v4 entry has no
+        # CONF_DEVICE_KEY, so it is whatever __init__ fell back to.
+        current_key = stored_key if stored_key is not None else (stored_serial or host)
+        polled_key = polled_ocf or polled_serial
+
+        if polled_key == current_key:
+            # Returned rather than short-circuited so a pre-v4 entry records
+            # the key it has always had, and the next poll sees no change.
+            return current_key, polled_serial
+
+        if stored_key is not None and polled_ocf is None:
+            return stored_key, stored_serial or polled_serial
+
+        # Excluding the host answer keeps this from firing on two *different*
+        # placeholder-serial units: an address is not an identity and can't
+        # corroborate anything.
+        same_unit = polled_serial == stored_serial and polled_serial != host
+        # A host-keyed entry (issues #83/#189) is registered against whatever
+        # answers at this address, so it has no claim to defend and needs no
+        # corroboration -- demanding it would strand exactly the
+        # placeholder-serial boards this exists to rescue. Same for an entry
+        # with no stored serial to compare against.
+        if not (current_key == host or same_unit or stored_serial is None):
+            self._log.warning(
+                "device at %s identifies as %r but this entry is registered as %r "
+                "(serial %r, registered %r); keeping the registered identity",
+                host,
+                polled_key,
+                current_key,
+                polled_serial,
+                stored_serial,
+            )
+            return current_key, stored_serial or polled_serial
+
+        # Corroborated: a pre-v4 entry moving onto its UUID, the same unit
+        # with a regenerated `di`, or a host-keyed entry learning a real
+        # identity. Following it keeps the user's entity_ids and history.
+        self._log.info(
+            "device %s (serial %r) changed key from %r to %r; following it",
+            host,
+            polled_serial,
+            current_key,
+            polled_key,
+        )
+        rekey_entry(self.hass, self._entry, current_key, polled_key)
+        return polled_key, polled_serial
+
     def _persist_identity(
         self,
+        device_key: str | None,
         serial: str,
         model: str,
         manufacturer: str,
@@ -1108,9 +1193,17 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         means the next restart names the device fully instead of renaming it
         again once a poll lands.
 
+        `device_key` is what the registries are keyed on; `serial` is stored
+        alongside it rather than replaced by it, because it is what
+        `_resolve_identity` corroborates a changed UUID against on a later
+        poll.
+        None leaves whatever key is already stored untouched -- see the
+        caller for why a snapshot replay must not write one.
+
         Runs on the event loop, which async_update_entry requires.
         """
         identity = {
+            **({CONF_DEVICE_KEY: device_key} if device_key is not None else {}),
             CONF_SERIAL: serial,
             CONF_MODEL: model,
             CONF_MANUFACTURER: manufacturer,
@@ -1195,6 +1288,12 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                     model=ident.get("model") or "",
                     name=ident.get("name") or "",
                     serial=ident.get("serial"),
+                    # Absent from a snapshot written before these fields
+                    # existed; _resolve_identity never re-keys from a
+                    # replay, so a missing UUID here costs nothing beyond
+                    # diagnostics.
+                    device_id=ident.get("device_id"),
+                    platform_id=ident.get("platform_id"),
                     device_types=tuple(ident.get("device_types") or ()),
                     raw=ident.get("raw") or {},
                 )
@@ -1346,26 +1445,11 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         self._unbound_hrefs = unbound
         self._refresh_learnable_hrefs()
 
-        # The entry's stored identity wins; this poll's answer is only
-        # adopted when nothing is stored (a legacy migration couldn't
-        # recover it), then written back. Re-keying an entry with existing
-        # registry entries orphans them (issue #236).
         polled_serial = resolve_serial(
             info.get("x.com.samsung.da.serialNum"), self._entry.data[CONF_HOST]
         )
-        serial = self._entry.data.get(CONF_SERIAL) or polled_serial
-        if serial != polled_serial:
-            # Same IP, different appliance (or firmware that changed what it
-            # reports) -- keep the registered identity; re-adding is the
-            # user's call.
-            self._log.warning(
-                "device at %s reports serial %r but this entry is registered "
-                "as %r; keeping the registered identity",
-                self._entry.data[CONF_HOST],
-                polled_serial,
-                serial,
-            )
-        self.device_serial = serial
+        key, serial = self._resolve_identity(polled_serial, from_snapshot=from_snapshot)
+        self.device_key = key
 
         ident = self._identity
         model = resolve_model(model_num, ident)
@@ -1373,12 +1457,16 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         mfr = (ident.manufacturer if ident else "") or "Samsung"
 
         self.device_info = DeviceInfo(
-            identifiers={(DOMAIN, serial)},
+            identifiers={(DOMAIN, key)},
             name=name,
             manufacturer=mfr,
             model=model,
         )
-        self._persist_identity(serial, model, mfr, device_type_name)
+        # A snapshot replay passes None: it never reached the device, so
+        # writing a key here would freeze a pre-v4 entry's legacy key in as
+        # if a poll had confirmed it, and the real UUID would later look
+        # like an identity to defend against rather than one to adopt.
+        self._persist_identity(None if from_snapshot else key, serial, model, mfr, device_type_name)
         if not from_snapshot:
             # A coverage gap is a claim about what the device reports, so only
             # a live poll gets to make it. Replaying a snapshot would restate
@@ -1392,9 +1480,9 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
 
         self._discovered = True
         self._log.info(
-            "discovered %d entities (serial=%s) hot=%s warm=%s subdevices=%s",
+            "discovered %d entities (key=%s) hot=%s warm=%s subdevices=%s",
             len(bound),
-            serial,
+            key,
             self._hot_hrefs,
             self._warm_hrefs,
             [su.key for su in self.subdevices],
@@ -1866,48 +1954,78 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             sess = self._session
             if sess is None:
                 raise RuntimeError("no session")
+            # Pace like every other send site (issue #384): RT-OCF drops a
+            # request that lands inside the rate-limit interval, and nothing
+            # retransmits a POST, so an unpaced write can only surface as a
+            # timeout. Worst on a reconnect retry, where _connect_session()
+            # has just read three identity resources.
+            sess.pace()
             code, _ = sess.post(path_segs, cbor2.dumps(body), timeout=self._POST_TIMEOUT_S)
             self._log.info("PUT %s → code %#04x", write_href, code)
 
-        # Mirrors the poll path's reconnect-and-retry (issue #294): a PUT
-        # landing on a session Samsung's firmware closed between polls used
-        # to be silently lost -- no retry, no user-facing error.
         async with self._session_lock:
             try:
                 await self.hass.async_add_executor_job(_do_put)
-            except Exception as e:
-                self._log.warning("command failed for %s, reconnecting: %s", write_href, e)
-                await self.hass.async_add_executor_job(self._close_session)
-                # The session is dead the moment it's closed, so any OBSERVE
-                # subscriptions on it are too -- downgrade here, before the
-                # retry, so a retry that also fails doesn't leave mode
-                # claiming "Push" on a session that no longer exists
-                # (issue #294; the poll path handles the same fact for its
-                # own reconnect the same way, unconditionally on close).
-                if self._observe.mode == MODE_OBSERVE:
-                    self._observe.downgrade_to_poll()
-                    self._resubscribe_due = True
-                await asyncio.sleep(self._RECONNECT_PAUSE_S)
+            except TimeoutError as e:
+                # A timeout is a late ACK, not a dead session -- the same
+                # reading _defer_reconnect_for takes on the poll path. A
+                # POST is sent exactly once (smartthings-local retransmits
+                # every block of a GET, never a write), so one dropped
+                # datagram lands here on a healthy session, and reconnecting
+                # over it costs the pause, a fresh handshake, and this
+                # device's OBSERVE subscriptions (issue #384).
+                self._log.info("command timed out for %s, retrying: %s", write_href, e)
                 try:
                     await self.hass.async_add_executor_job(_do_put)
                 except Exception as e2:
-                    self._log.error("command failed for %s after reconnect: %s", write_href, e2)
-                    raise HomeAssistantError(
-                        translation_domain=DOMAIN,
-                        translation_key="command_failed",
-                        translation_placeholders={"href": write_href, "error": str(e2)},
-                    ) from e2
-                # The retry's own reconnect pause + second PUT can eat well
-                # into the settle window armed above, leaving too little of
-                # it for the confirming poll below and reviving the
-                # revert-then-reapply symptom settle_s exists to prevent
-                # (issue #9). Re-arm it fresh now that the write actually
-                # landed.
-                self._observe.mark_write_pending(
-                    write_href, settle_s=self._POST_TIMEOUT_S + self._POLL_TIMEOUT_S
-                )
+                    await self._reconnect_and_retry_write(_do_put, write_href, e2)
+                else:
+                    self._rearm_write_settle(write_href)
+            except Exception as e:
+                await self._reconnect_and_retry_write(_do_put, write_href, e)
         await self.async_request_refresh()
 
+    def _rearm_write_settle(self, write_href: str) -> None:
+        """Re-arm the settle guard once a retried write finally lands.
+
+        The retries eat into the window armed before the first PUT, leaving
+        too little of it for the confirming poll and reviving the
+        revert-then-reapply symptom settle_s exists to prevent (issue #9)."""
+        self._observe.mark_write_pending(
+            write_href, settle_s=self._POST_TIMEOUT_S + self._POLL_TIMEOUT_S
+        )
+
+    async def _reconnect_and_retry_write(
+        self, do_put: Callable[[], None], write_href: str, e: Exception
+    ) -> None:
+        """Reconnect and retry a failed write once, raising if it fails again.
+
+        Mirrors the poll path's reconnect-and-retry (issue #294): a PUT
+        landing on a session Samsung's firmware closed between polls used to
+        be silently lost -- no retry, no user-facing error."""
+        self._log.warning("command failed for %s, reconnecting: %s", write_href, e)
+        await self.hass.async_add_executor_job(self._close_session)
+        # The session is dead the moment it's closed, so any OBSERVE
+        # subscriptions on it are too -- downgrade here, before the retry, so
+        # a retry that also fails doesn't leave mode claiming "Push" on a
+        # session that no longer exists (issue #294; the poll path handles
+        # the same fact for its own reconnect the same way, unconditionally
+        # on close).
+        if self._observe.mode == MODE_OBSERVE:
+            self._observe.downgrade_to_poll()
+            self._resubscribe_due = True
+        await asyncio.sleep(self._RECONNECT_PAUSE_S)
+        try:
+            await self.hass.async_add_executor_job(do_put)
+        except Exception as e2:
+            self._log.error("command failed for %s after reconnect: %s", write_href, e2)
+            raise HomeAssistantError(
+                translation_domain=DOMAIN,
+                translation_key="command_failed",
+                translation_placeholders={"href": write_href, "error": str(e2)},
+            ) from e2
+        self._rearm_write_settle(write_href)
+
     # ------------------------------------------------------------------
     # Debug raw write/read (issue #54, extended for issue #300): a
     # power-user escape hatch shared by the options-flow debug panel and
@@ -1925,6 +2043,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         sess = self._session
         if sess is None:
             raise RuntimeError("no session")
+        sess.pace()
         code, _ = sess.post(path_segs, cbor2.dumps(body), timeout=self._POST_TIMEOUT_S)
         self._log.warning("DEBUG raw write POST %s %r → code %#04x", href, body, code)
         new_rep: dict = {}

custom_components/localthings/entity.py

@@ -88,7 +88,7 @@ class LocalThingsEntity(CoordinatorEntity[LocalThingsCoordinator]):
         super().__init__(coordinator)
         self._bound = bound
         self._state_key = _key(bound)
-        self._attr_unique_id = f"{DOMAIN}_{coordinator.device_serial}_{self._state_key}"
+        self._attr_unique_id = f"{DOMAIN}_{coordinator.device_key}_{self._state_key}"
         if bound.desc.translation_placeholders is not None:
             self._attr_translation_placeholders = dict(bound.desc.translation_placeholders)
         elif bound.desc.use_instance_name:

custom_components/localthings/manifest.json

@@ -13,5 +13,5 @@
     "pyOpenSSL>=23.0",
     "smartthings-local>=0.1.8"
   ],
-  "version": "0.23.0"
+  "version": "0.24.0-beta.1"
 }

custom_components/localthings/registry/capabilities/dishwasher.py

@@ -10,7 +10,12 @@ wash, auto release dry) are read locally here.
 from ..capability import Capability
 from ..entities import ButtonDesc, SelectDesc, SensorDesc, SwitchDesc
 from .common import diagnosis_status
-from .laundry import bool_option_switch, cycle_select
+from .laundry import (
+    bool_option_switch,
+    cycle_select,
+    drum_clean_cycles_remaining,
+    drum_clean_last_cleaned,
+)
 
 # ---------------------------------------------------------------------------
 # /dishwasher/vs/0 — cycle wash/dry settings
@@ -52,6 +57,14 @@ DISHWASHER_SETTINGS = Capability(
 # kind of adjacent pair a manual screenshot transcription slips on. The
 # reporter's live confirmation (selecting 'Normal' ran the physical Express
 # 60 program and vice versa) settled it: '86' is Express 60, '83' is Normal.
+#
+# Drum Clean+ maintenance tracking reuses washer.py/dryer.py's (issues #9,
+# #258) DrumCleanProposal_/WashingTimes_/DrumCleanLog_ tokens riding on this
+# same options[] array -- a live dump confirmed the dishwasher reports the
+# identical trio (WashingTimes_18/DrumCleanProposal_20, plus a '|'-joined
+# DrumCleanLog_ history matching the dryer's multi-entry shape), so the
+# shared laundry.drum_clean_cycles_remaining/drum_clean_last_cleaned readers
+# apply unchanged; see laundry.py for the field semantics.
 
 CYCLE_OPTIONS = Capability(
     href="/course/vs/0",
@@ -61,6 +74,22 @@ CYCLE_OPTIONS = Capability(
         bool_option_switch(
             "auto_release_dry", "mdi:door-open", "AutoDoorRelease", gate_on_presence=True
         ),
+        SensorDesc(
+            key="drum_clean_cycles_remaining",
+            unit="cycles",
+            icon="mdi:dishwasher-alert",
+            state_class="measurement",
+            exists_fn=lambda rep, resources: drum_clean_cycles_remaining(rep) is not None,
+            rep_fn=drum_clean_cycles_remaining,
+        ),
+        SensorDesc(
+            key="drum_clean_last_cleaned",
+            device_class="timestamp",
+            icon="mdi:calendar-clock",
+            entity_category="diagnostic",
+            exists_fn=lambda rep, resources: drum_clean_last_cleaned(rep) is not None,
+            rep_fn=drum_clean_last_cleaned,
+        ),
     ),
 )
 

custom_components/localthings/registry/identity.py

@@ -13,6 +13,12 @@ class DeviceIdentity:
     model: str
     name: str
     serial: str | None
+    # /oic/d's `di` and /oic/p's `pi` -- OCF's own device and platform
+    # UUIDs. Promoted out of `raw` into named fields because
+    # resolve_device_key mints permanent registry keys from them; see its
+    # docstring for why `di` leads.
+    device_id: str | None = None
+    platform_id: str | None = None
     device_types: tuple[str, ...] = ()
     raw: dict[str, dict | list] = field(default_factory=dict)
 
@@ -57,6 +63,63 @@ def resolve_serial(raw_serial: str | None, host: str) -> str:
     return s
 
 
+def is_usable_device_id(value: str | None) -> bool:
+    """True for an OCF `di`/`pi` that actually identifies one unit.
+
+    Rejects OCF's nil UUID, which firmware that never had one assigned
+    reports on every unit of the family -- the #189 failure mode on a new
+    field, and one `is_placeholder_serial`'s repeated-digit rule misses
+    because the dashes make more than one distinct character. Past that the
+    same known-junk rules apply: a board flashed with 'Nothing(SVC)' in one
+    identity field is not one to trust in another.
+    """
+    s = (value or "").strip()
+    if not s:
+        return False
+    if not set(s) - {"0", "-"}:
+        return False
+    return not is_placeholder_serial(s)
+
+
+def ocf_device_key(identity: DeviceIdentity | None) -> str | None:
+    """The OCF-derived half of resolve_device_key's chain, or None when the
+    device reported no usable UUID.
+
+    Split out because "no UUID" and "this UUID" are different answers to a
+    caller holding an existing key: the coordinator must never demote an
+    entry off its UUID just because one poll couldn't read /oic/d.
+    Normalized so firmware that changes case between reads doesn't look
+    like a different appliance.
+    """
+    if identity is None:
+        return None
+    for candidate in (identity.device_id, identity.platform_id):
+        if candidate is not None and is_usable_device_id(candidate):
+            return candidate.strip().lower()
+    return None
+
+
+def resolve_device_key(identity: DeviceIdentity | None, raw_serial: str | None, host: str) -> str:
+    """The identity to mint this device's permanent registry keys from.
+
+    Tried in order: /oic/d's `di`, /oic/p's `pi`, the serialNum, the host.
+
+    `di` leads because it is what the protocol already uses to address this
+    endpoint: if it were wrong or shared, OCF discovery and the DTLS
+    association would not work at all. serialNum is a vendor-populated
+    string nothing depends on, which is why three firmware families have
+    shipped it unusable -- 'Nothing(SVC)' (#83), a flash-unset sentinel
+    (#189), and a well-formed serial duplicated across every unit (#381).
+
+    `pi` is only the fallback despite the spec calling it immutable: it is
+    *platform*-scoped, so a board hosting several logical OCF devices
+    shares one across all of them. `di` is device-scoped, the granularity
+    of a config entry. The serial and host stay below both so a board
+    answering neither resource lands where it always did.
+    """
+    return ocf_device_key(identity) or resolve_serial(raw_serial, host)
+
+
 def resolve_model(model_num: str, identity: DeviceIdentity | None) -> str:
     """The model string to name and register a device under.
 
@@ -143,6 +206,8 @@ def read_identity(sess, serial: str | None) -> DeviceIdentity:
         model=p.get("mnmo") or "",
         name=d.get("n") or "",
         serial=serial,
+        device_id=d.get("di") if isinstance(d.get("di"), str) else None,
+        platform_id=p.get("pi") if isinstance(p.get("pi"), str) else None,
         device_types=_device_types(d),
         # Kept whole rather than field-by-field: outside the /device/0 dump
         # diagnostics already captures, and we don't yet know which fields

custom_components/localthings/registry/redact.py

@@ -30,13 +30,18 @@ _SENSITIVE_SUBSTRINGS = (
     "secret",
 )
 
-# Matched whole, not as substrings: OCF's /oic/d and /oic/p identify the
-# unit with bare one/two-letter keys too short for the substring rules above
-# ('di' is a substring of 'condition', 'display', ...). 'di'/'pi' are the
-# device/platform UUIDs; 'n' is /oic/d's free-text device name, which may
-# carry a person's name -- the device-type signal we actually want from
-# that resource is `rt`, which is not redacted.
-_SENSITIVE_EXACT = frozenset({"di", "pi", "n"})
+# Matched whole, not as substrings: these are bare one/two-letter keys too
+# short for the substring rules above ('n' is a substring of very nearly
+# everything). 'n' is /oic/d's free-text device name, which the owner sets
+# from the SmartThings app and can carry a person's name -- the device-type
+# signal we actually want from that resource is `rt`, which is not redacted.
+#
+# /oic/d's `di` and /oic/p's `pi` are deliberately not redacted: they're
+# randomly-assigned per-unit UUIDs rather than account data, and they are
+# what registry keys are minted from (issue #381), so blanking them hides
+# the identity every entity in a report is named after -- which is exactly
+# what made #381's first diagnostics download unable to answer it.
+_SENSITIVE_EXACT = frozenset({"n"})
 
 
 def _is_sensitive_key(key: str) -> bool:

custom_components/localthings/rekey.py

@@ -0,0 +1,90 @@
+"""Move an entry's registry entries from one device key to another.
+
+The key (registry.identity.resolve_device_key) is permanent in three
+places, so changing it means rewriting the registries rather than storing
+a new value -- anything left behind is orphaned. Rewriting rather than
+recreating is what keeps an entity's entity_id, and with it its history,
+area and automations.
+
+Its own module because both callers need it: the v1 -> v2 migration in
+__init__.py and the coordinator's first-poll adoption (issue #381).
+"""
+
+from __future__ import annotations
+
+import logging
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers import entity_registry as er
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@callback
+def rekey_entry(hass: HomeAssistant, entry: ConfigEntry, old_key: str, new_key: str) -> None:
+    """Rewrite everything this entry registered under `old_key` to `new_key`.
+
+    All three permanent places move together: entity unique_ids
+    (f"{DOMAIN}_{key}_{state_key}"), device identifiers ((DOMAIN, key), plus
+    (DOMAIN, f"{key}_{subdevice}") per subdevice -- see device_info_for),
+    and the entry's own unique_id. Leaving that last one behind would let
+    the config flow's duplicate check wave through a re-add of this very
+    appliance.
+
+    Idempotent, so it is safe to attempt on every poll rather than tracking
+    whether it has run. Where both keys already exist the `old_key` copy is
+    the dead one, so it is removed rather than rewritten over the live entry.
+
+    Must run on the event loop; the registry helpers require it.
+    """
+    if old_key == new_key:
+        return
+
+    new_entry_unique_id = f"{DOMAIN}_{new_key}"
+    if entry.unique_id != new_entry_unique_id:
+        hass.config_entries.async_update_entry(entry, unique_id=new_entry_unique_id)
+
+    ent_reg = er.async_get(hass)
+    stale_prefix = f"{DOMAIN}_{old_key}_"
+    for entity in list(er.async_entries_for_config_entry(ent_reg, entry.entry_id)):
+        if not entity.unique_id.startswith(stale_prefix):
+            continue
+        new_unique_id = f"{DOMAIN}_{new_key}_{entity.unique_id[len(stale_prefix) :]}"
+        if ent_reg.async_get_entity_id(entity.domain, DOMAIN, new_unique_id):
+            _LOGGER.debug("removing orphaned entity %s", entity.entity_id)
+            ent_reg.async_remove(entity.entity_id)
+        else:
+            _LOGGER.debug("re-keying entity %s to %s", entity.entity_id, new_unique_id)
+            ent_reg.async_update_entity(entity.entity_id, new_unique_id=new_unique_id)
+
+    dev_reg = dr.async_get(hass)
+    for device in list(dr.async_entries_for_config_entry(dev_reg, entry.entry_id)):
+        stale = {
+            ident
+            for ident in device.identifiers
+            if ident[0] == DOMAIN and (ident[1] == old_key or ident[1].startswith(f"{old_key}_"))
+        }
+        if not stale:
+            continue
+        fresh = {(DOMAIN, f"{new_key}{ident[1][len(old_key) :]}") for ident in stale}
+        existing = dev_reg.async_get_device(identifiers=fresh)
+        if existing is not None and existing.id != device.id:
+            # Removing a device takes its entities with it. Anything still
+            # attached here was re-keyed rather than removed above -- the
+            # surviving copy, not a duplicate -- so move it onto the device
+            # it now belongs to before the removal destroys it too.
+            for entity in er.async_entries_for_device(
+                ent_reg, device.id, include_disabled_entities=True
+            ):
+                ent_reg.async_update_entity(entity.entity_id, device_id=existing.id)
+            _LOGGER.debug("removing orphaned device %s", device.id)
+            dev_reg.async_remove_device(device.id)
+        else:
+            _LOGGER.debug("re-keying device %s to %s", device.id, fresh)
+            dev_reg.async_update_device(
+                device.id, new_identifiers=(device.identifiers - stale) | fresh
+            )

custom_components/localthings/sensor.py

@@ -187,7 +187,7 @@ class LocalThingsConnectionModeSensor(CoordinatorEntity[LocalThingsCoordinator],
 
     def __init__(self, coordinator: LocalThingsCoordinator) -> None:
         super().__init__(coordinator)
-        self._attr_unique_id = f"{DOMAIN}_{coordinator.device_serial}_connection_mode"
+        self._attr_unique_id = f"{DOMAIN}_{coordinator.device_key}_connection_mode"
 
     @property
     def device_info(self) -> DeviceInfo:

tests/fixtures/golden/dishwasher.json

@@ -8,6 +8,8 @@
     "cycle_active",
     "delay_start_hours",
     "diagnosis_status",
+    "drum_clean_cycles_remaining",
+    "drum_clean_last_cleaned",
     "energy_kwh",
     "energy_saved_kwh",
     "finish_time",
@@ -40,6 +42,8 @@
     "samsung_dishwasher_delay_start_hours",
     "samsung_dishwasher_diagnosis_start",
     "samsung_dishwasher_diagnosis_status",
+    "samsung_dishwasher_drum_clean_cycles_remaining",
+    "samsung_dishwasher_drum_clean_last_cleaned",
     "samsung_dishwasher_energy_kwh",
     "samsung_dishwasher_energy_saved_kwh",
     "samsung_dishwasher_finish_time",

tests/localthings/conftest.py

@@ -14,6 +14,7 @@ from pytest_homeassistant_custom_component.common import MockConfigEntry
 from custom_components.localthings.const import (
     CONF_CA_CERT_PEM,
     CONF_CA_KEY_PEM,
+    CONF_DEVICE_KEY,
     CONF_DEVICE_TYPE,
     CONF_HOST,
     CONF_LEAF_CERT_PEM,
@@ -82,6 +83,10 @@ MOCK_PORT = 49154
 # what mock_coordinator_session polls -- so an entry built from ENTRY_DATA and
 # the device it "reaches" agree on who they are, the same as in production.
 MOCK_SERIAL = "TEST-SERIAL-0000"
+# The OCF device UUID (/oic/d's `di`) the probe resolves the entry's key from
+# (issue #381). Distinct from MOCK_SERIAL so a test that confuses the two
+# fails rather than passing by coincidence.
+MOCK_DEVICE_KEY = "7b1f0c9e-2a44-4d6b-9f10-4c8e2b5a0d31"
 MOCK_MODEL = "TEST-MODEL"
 MOCK_DEVICE_TYPE = "refrigerator"
 MOCK_CA_CERT_PEM = "-----BEGIN CERTIFICATE-----\nTEST-CA\n-----END CERTIFICATE-----"
@@ -98,6 +103,7 @@ ENTRY_DATA = {
     CONF_LEAF_KEY_PEM: MOCK_LEAF_KEY_PEM,
     # Identity the config flow's probe resolved (issue #236) -- what the
     # coordinator keys its devices and entities on from construction.
+    CONF_DEVICE_KEY: MOCK_DEVICE_KEY,
     CONF_SERIAL: MOCK_SERIAL,
     CONF_MODEL: MOCK_MODEL,
     CONF_MANUFACTURER: "Samsung",
@@ -131,6 +137,7 @@ def fridge_resources():
 def _probe_result(*, recognized: bool) -> dict:
     return {
         "port": MOCK_PORT,
+        "device_key": MOCK_DEVICE_KEY,
         "serial": MOCK_SERIAL,
         "model": MOCK_MODEL,
         "manufacturer": "Samsung",
@@ -209,6 +216,10 @@ class FakeObserveSession:
     def refresh_observes(self, paths):
         return None
 
+    def pace(self):
+        """No-op stand-in for the real session's rate limiter, which the
+        write path calls before every POST (issue #384)."""
+
     def close(self):
         self.closed = True
 
@@ -244,8 +255,8 @@ def mock_entry(hass):
     entry = MockConfigEntry(
         domain=DOMAIN,
         data=ENTRY_DATA,
-        unique_id=f"localthings_{MOCK_SERIAL}",
-        version=2,
+        unique_id=f"localthings_{MOCK_DEVICE_KEY}",
+        version=4,
     )
     entry.add_to_hass(hass)
     return entry

tests/localthings/test_config_flow.py

@@ -16,11 +16,13 @@ from custom_components.localthings.const import (
     CONF_CA_CERT_PEM,
     CONF_CA_KEY_PEM,
     CONF_CLOUD_COURSES_ENABLED,
+    CONF_DEVICE_KEY,
     CONF_HOST,
     CONF_LEAF_CERT_PEM,
     CONF_LEARN_MODES,
     CONF_LEARNED_MODES,
     CONF_PORT,
+    CONF_SERIAL,
     DOMAIN,
 )
 
@@ -28,11 +30,13 @@ from .conftest import (
     ENTRY_DATA,
     MOCK_CA_CERT_PEM,
     MOCK_CA_KEY_PEM,
+    MOCK_DEVICE_KEY,
     MOCK_HOST,
     MOCK_LEAF_CERT_PEM,
     MOCK_MODEL,
     MOCK_PORT,
     MOCK_SERIAL,
+    _probe_result,
 )
 
 
@@ -1047,7 +1051,11 @@ async def test_unknown_type_step_description_makes_no_version_claim(
 
 
 async def test_duplicate_device_aborted(hass: HomeAssistant, mock_probe) -> None:
-    """Second add of same serial: flow aborts.
+    """Second add of the same *device key*: flow aborts.
+
+    Keyed on the OCF device UUID rather than the serialNum (issue #381), so
+    this is now the check that two genuinely distinct units can no longer
+    trip -- see test_same_serial_on_two_units_is_not_a_duplicate.
 
     When a device already exists the form only asks for host (CA creds are
     reused), so we only submit CONF_HOST in the second configure call.
@@ -1055,7 +1063,7 @@ async def test_duplicate_device_aborted(hass: HomeAssistant, mock_probe) -> None
     existing = MockConfigEntry(
         domain=DOMAIN,
         data=ENTRY_DATA,
-        unique_id=f"localthings_{MOCK_SERIAL}",
+        unique_id=f"localthings_{MOCK_DEVICE_KEY}",
     )
     existing.add_to_hass(hass)
 
@@ -1069,6 +1077,154 @@ async def test_duplicate_device_aborted(hass: HomeAssistant, mock_probe) -> None
     assert result["reason"] == "already_configured"
 
 
+async def test_same_serial_on_two_units_is_not_a_duplicate(hass: HomeAssistant, mock_probe) -> None:
+    """Issue #381: two Samsung air purifiers of one model ship the identical,
+    well-formed serialNum, so keying the entry on it turned the second one
+    away as already configured. The OCF device UUID differs between them,
+    and it is what the entry is keyed on now, so both can be added.
+
+    Deliberately holds the serial *constant* across the two probes and varies
+    only the device key -- the exact shape of the bug report.
+    """
+    first = MockConfigEntry(
+        domain=DOMAIN,
+        data={**ENTRY_DATA, CONF_HOST: "192.168.0.3"},
+        unique_id=f"localthings_{MOCK_DEVICE_KEY}",
+    )
+    first.add_to_hass(hass)
+
+    second_probe = {
+        **_probe_result(recognized=True),
+        "device_key": "3771f8bf-c184-3a2d-d885-e4c9818736d2",
+        "serial": MOCK_SERIAL,
+    }
+    with patch(
+        "custom_components.localthings.config_flow._probe_and_validate",
+        return_value=second_probe,
+    ):
+        result = await hass.config_entries.flow.async_init(DOMAIN, context={"source": "user"})
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {CONF_HOST: "192.168.0.14"}
+        )
+
+    assert result["type"] == FlowResultType.CREATE_ENTRY
+    assert result["data"][CONF_DEVICE_KEY] == "3771f8bf-c184-3a2d-d885-e4c9818736d2"
+    # Both entries exist, and the shared serial is still recorded on each --
+    # it is what corroborates a later change of key.
+    assert len(hass.config_entries.async_entries(DOMAIN)) == 2
+    assert result["data"][CONF_SERIAL] == first.data[CONF_SERIAL]
+
+
+async def test_re_adding_during_the_migration_window_is_still_a_duplicate(
+    hass: HomeAssistant, mock_probe
+) -> None:
+    """An entry created before v4 keeps its serial-keyed unique_id until its
+    first *live* poll adopts the UUID, which can be a long while for an
+    appliance that is off (it loads from its snapshot meanwhile, issue
+    #295). The UUID check can't see such an entry, so without a second
+    check on the legacy key, re-adding this very appliance in that window
+    would be waved through -- and the two entries would collide the moment
+    the older one re-keyed, with rekey_entry resolving the collision by
+    deleting the duplicate rows and taking the original's entity_ids,
+    history and automations with them.
+    """
+    existing = MockConfigEntry(
+        domain=DOMAIN,
+        data={k: v for k, v in ENTRY_DATA.items() if k != CONF_DEVICE_KEY},
+        unique_id=f"localthings_{MOCK_SERIAL}",
+        version=3,
+    )
+    existing.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(DOMAIN, context={"source": "user"})
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], {CONF_HOST: ENTRY_DATA[CONF_HOST]}
+    )
+
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "already_configured"
+
+
+async def test_the_migration_window_check_still_separates_two_same_serial_units(
+    hass: HomeAssistant, mock_probe
+) -> None:
+    """The legacy-key check above matches on the host as well as the serial,
+    so it cannot undo the fix: issue #381's two units share a serial but sit
+    at different addresses, and the second must still be addable while the
+    first is mid-migration."""
+    existing = MockConfigEntry(
+        domain=DOMAIN,
+        data={
+            **{k: v for k, v in ENTRY_DATA.items() if k != CONF_DEVICE_KEY},
+            CONF_HOST: "192.168.0.3",
+        },
+        unique_id=f"localthings_{MOCK_SERIAL}",
+        version=3,
+    )
+    existing.add_to_hass(hass)
+
+    second_probe = {
+        **_probe_result(recognized=True),
+        "device_key": "3771f8bf-c184-3a2d-d885-e4c9818736d2",
+        "serial": MOCK_SERIAL,
+    }
+    with patch(
+        "custom_components.localthings.config_flow._probe_and_validate",
+        return_value=second_probe,
+    ):
+        result = await hass.config_entries.flow.async_init(DOMAIN, context={"source": "user"})
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {CONF_HOST: "192.168.0.14"}
+        )
+
+    assert result["type"] == FlowResultType.CREATE_ENTRY
+    assert result["data"][CONF_DEVICE_KEY] == "3771f8bf-c184-3a2d-d885-e4c9818736d2"
+
+
+def test_probe_reads_the_device_key_from_oic_d_without_an_extra_round_trip(monkeypatch):
+    """`_read_device` already fetches /oic/p and /oic/d for the device-type
+    signal, so keying on the OCF UUID costs no additional GET -- it reads
+    the identity that call already returned."""
+    from custom_components.localthings.config_flow import _read_device
+
+    device0 = [
+        {"rt": ["x.com.samsung.devcol"]},
+        {
+            "href": "/information/vs/0",
+            "rep": {
+                "x.com.samsung.da.modelNum": "AVT-WW-TP1-23-AXX500|10251941",
+                "x.com.samsung.da.serialNum": "BS7SP9AW400114A",
+            },
+        },
+    ]
+
+    class _Session:
+        def __init__(self):
+            self.paths = []
+
+        def get(self, path, timeout=10.0):
+            self.paths.append(tuple(path))
+            table = {
+                ("oic", "p"): {"mnmn": "Samsung Electronics", "pi": "PLATFORM-UUID"},
+                ("oic", "d"): {"di": "CCFD73B3-AEB4-792A-1100-68F06F5D603B"},
+                ("device", "0"): device0,
+            }
+            body = table.get(tuple(path))
+            if body is None:
+                return 0x84, b""
+            import cbor2
+
+            return 0x45, cbor2.dumps(body)
+
+    sess = _Session()
+    info = _read_device(sess, "192.168.0.3", MOCK_PORT)
+
+    assert info["device_key"] == "ccfd73b3-aeb4-792a-1100-68f06f5d603b"
+    assert info["serial"] == "BS7SP9AW400114A"
+    # Exactly the three reads the probe already made before this change.
+    assert sess.paths == [("oic", "p"), ("oic", "d"), ("oic", "res"), ("device", "0")]
+
+
 def test_probe_marks_washer_as_recognized(monkeypatch):
     """A washer reports no oneUiVersion at all -- its consumer-model code
     must still resolve so setup doesn't warn about an unrecognized type."""

tests/localthings/test_coordinator.py

@@ -33,7 +33,13 @@ from custom_components.localthings.registry.capabilities.common import (
     remote_control_required_for_write,
 )
 
-from .conftest import ENTRY_DATA, MOCK_MODEL, MOCK_SERIAL, FakeObserveSession
+from .conftest import (
+    ENTRY_DATA,
+    MOCK_DEVICE_KEY,
+    MOCK_MODEL,
+    MOCK_SERIAL,
+    FakeObserveSession,
+)
 from .conftest import _load_fridge_resources as _load_fridge
 
 
@@ -150,7 +156,7 @@ def test_run_discovery_falls_back_to_host_for_placeholder_serial(
 ) -> None:
     """Issue #83: the ARTIK051_DONGLE_REF firmware family reports the
     literal string 'Nothing(SVC)' as serialNum on every unit. Left as-is,
-    two such units get the same device_serial (which feeds both the HA
+    two such units get the same device_key (which feeds both the HA
     device-registry identifier and every entity's unique_id), so the
     second one's entities silently collide and get dropped. It must be
     treated the same as an empty serial and fall back to the host."""
@@ -164,7 +170,7 @@ def test_run_discovery_falls_back_to_host_for_placeholder_serial(
     }
     coordinator = LocalThingsCoordinator(hass, legacy_entry)
     coordinator._run_discovery(resources)
-    assert coordinator.device_serial == legacy_entry.data[CONF_HOST]
+    assert coordinator.device_key == legacy_entry.data[CONF_HOST]
 
 
 def test_run_discovery_falls_back_to_host_for_all_f_placeholder_serial(
@@ -175,7 +181,7 @@ def test_run_discovery_falls_back_to_host_for_all_f_placeholder_serial(
     character the same repeated hex digit. A washer and a dryer, two
     different physical units, both reported the literal serialNum
     'FFFFFFFFFFFFFFF', so without this fallback they'd collide on
-    device_serial exactly like the #83 case above."""
+    device_key exactly like the #83 case above."""
     resources = {
         "/information/vs/0": {
             "x.com.samsung.da.modelNum": "DA_WM_A51_20_COMMON|20221341|30010102001211000103000000000000",  # noqa: E501
@@ -186,7 +192,7 @@ def test_run_discovery_falls_back_to_host_for_all_f_placeholder_serial(
     }
     coordinator = LocalThingsCoordinator(hass, legacy_entry)
     coordinator._run_discovery(resources)
-    assert coordinator.device_serial == legacy_entry.data[CONF_HOST]
+    assert coordinator.device_key == legacy_entry.data[CONF_HOST]
 
 
 # ---------------------------------------------------------------------------
@@ -198,7 +204,7 @@ def test_identity_is_resolved_before_any_poll(hass: HomeAssistant, mock_entry) -
     """The coordinator mints registry keys from the entry's stored identity at
     construction time.
 
-    `device_serial` is what entity unique_ids and device identifiers are built
+    `device_key` is what entity unique_ids and device identifiers are built
     from, and those are permanent. Seeding it with the host meant anything that
     registered before the first poll returned -- the connection-mode sensor
     especially, added unconditionally rather than from `bound` -- was written
@@ -207,8 +213,8 @@ def test_identity_is_resolved_before_any_poll(hass: HomeAssistant, mock_entry) -
     """
     coordinator = LocalThingsCoordinator(hass, mock_entry)
 
-    assert coordinator.device_serial == MOCK_SERIAL
-    assert coordinator.device_info["identifiers"] == {(DOMAIN, MOCK_SERIAL)}
+    assert coordinator.device_key == MOCK_DEVICE_KEY
+    assert coordinator.device_info["identifiers"] == {(DOMAIN, MOCK_DEVICE_KEY)}
     assert coordinator.device_info["model"] == MOCK_MODEL
     assert coordinator.device_info["name"] == f"Samsung Refrigerator ({MOCK_MODEL})"
     assert mock_entry.data[CONF_HOST] not in str(coordinator.device_info["identifiers"])
@@ -239,7 +245,7 @@ def test_discovery_keeps_the_registered_identity(hass: HomeAssistant, mock_entry
     coordinator = LocalThingsCoordinator(hass, mock_entry)
     coordinator._run_discovery(resources)
 
-    assert coordinator.device_serial == MOCK_SERIAL
+    assert coordinator.device_key == MOCK_DEVICE_KEY
 
 
 def test_discovery_backfills_a_legacy_entry_identity(hass: HomeAssistant, legacy_entry) -> None:
@@ -1343,6 +1349,151 @@ async def test_send_command_raises_after_reconnect_retry_also_fails(
     assert coordinator.observe_mode == MODE_POLL
 
 
+async def test_send_command_timeout_retries_in_place_without_reconnecting(
+    hass: HomeAssistant,
+    mock_entry,
+    mock_coordinator_observe_session,
+) -> None:
+    """Issue #384: a write timeout is a dropped datagram, not a dead session.
+
+    smartthings-local retransmits every block of a GET but sends a POST
+    exactly once, so a single loss surfaces here as a timeout on a session
+    that is still fine. Retrying in place has to be tried before spending a
+    reconnect -- which costs the pause, a fresh handshake, and every OBSERVE
+    subscription this device has."""
+    from custom_components.localthings.registry.discovery import BoundEntity
+    from custom_components.localthings.registry.entities import NumberDesc
+
+    fake = mock_coordinator_observe_session
+    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]
+    hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs
+
+    fake.notify_on_subscribe = {"notified": True}
+    entered = await hass.async_add_executor_job(
+        coordinator._observe.try_enter_observe_mode, fake, hrefs, 0.02, 0.8
+    )
+    assert entered is True
+
+    def _write_fn(payload, rep, href=None):
+        return (["test", "vs", "0"], {"value": payload})
+
+    desc = NumberDesc(key="test", field="value", write_fn=_write_fn)
+    bound = BoundEntity(href="/test/vs/0", capability=coordinator.bound[0].capability, desc=desc)
+
+    calls = {"n": 0}
+
+    def _post(*args, **kwargs):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise SessionTimeoutError()
+        return (0x44, b"")
+
+    with (
+        patch.object(coordinator, "_close_session") as mock_close,
+        patch.object(coordinator, "_connect_session") as mock_connect,
+        patch(
+            "custom_components.localthings.coordinator.asyncio.sleep",
+            new=AsyncMock(),
+        ) as mock_sleep,
+    ):
+        fake.post = _post
+        await coordinator.async_send_command(bound, 5)
+
+    assert calls["n"] == 2
+    assert not mock_close.called
+    assert not mock_connect.called
+    # No _RECONNECT_PAUSE_S burned on a session that never needed replacing.
+    assert not mock_sleep.called
+    # And the push subscriptions the reconnect would have thrown away survive.
+    assert coordinator.observe_mode == MODE_OBSERVE
+    assert coordinator._cache.get("/test/vs/0") == {"value": 5}
+
+
+async def test_send_command_timeout_falls_back_to_reconnect(
+    hass: HomeAssistant,
+    mock_entry,
+    mock_coordinator_observe_session,
+) -> None:
+    """The in-place retry (issue #384) is an extra rung on the ladder, not a
+    replacement: a href that keeps timing out must still reach the reconnect
+    retry and, failing that, the user (issue #294)."""
+    from homeassistant.exceptions import HomeAssistantError
+
+    from custom_components.localthings.registry.discovery import BoundEntity
+    from custom_components.localthings.registry.entities import NumberDesc
+
+    fake = mock_coordinator_observe_session
+    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]
+
+    def _write_fn(payload, rep, href=None):
+        return (["test", "vs", "0"], {"value": payload})
+
+    desc = NumberDesc(key="test", field="value", write_fn=_write_fn)
+    bound = BoundEntity(href="/test/vs/0", capability=coordinator.bound[0].capability, desc=desc)
+
+    calls = {"n": 0}
+
+    def _post(*args, **kwargs):
+        calls["n"] += 1
+        raise SessionTimeoutError()
+
+    with (
+        patch.object(coordinator, "_close_session") as mock_close,
+        patch(
+            "custom_components.localthings.coordinator.asyncio.sleep",
+            new=AsyncMock(),
+        ),
+        pytest.raises(HomeAssistantError),
+    ):
+        fake.post = _post
+        await coordinator.async_send_command(bound, 5)
+
+    # First attempt, in-place retry, then the reconnect's own retry.
+    assert calls["n"] == 3
+    assert mock_close.called
+
+
+async def test_write_paces_before_posting(
+    hass: HomeAssistant,
+    mock_entry,
+    mock_coordinator_observe_session,
+) -> None:
+    """Issue #384: RT-OCF silently drops a request that lands inside the
+    rate-limit interval, and nothing retransmits a POST -- so an unpaced
+    write can only ever surface as a timeout. Every other send site paces;
+    the write path has to as well."""
+    from custom_components.localthings.registry.discovery import BoundEntity
+    from custom_components.localthings.registry.entities import NumberDesc
+
+    fake = mock_coordinator_observe_session
+    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]
+
+    def _write_fn(payload, rep, href=None):
+        return (["test", "vs", "0"], {"value": payload})
+
+    desc = NumberDesc(key="test", field="value", write_fn=_write_fn)
+    bound = BoundEntity(href="/test/vs/0", capability=coordinator.bound[0].capability, desc=desc)
+
+    order: list[str] = []
+
+    fake.pace = lambda: order.append("pace")
+
+    def _post(*args, **kwargs):
+        order.append("post")
+        return (0x44, b"")
+
+    fake.post = _post
+    await coordinator.async_send_command(bound, 5)
+
+    assert order == ["pace", "post"]
+
+
 async def test_send_command_reconnect_downgrades_observe_mode(
     hass: HomeAssistant,
     mock_entry,

tests/localthings/test_device_removal.py

@@ -39,7 +39,7 @@ async def test_stale_device_can_be_removed(
     stale = _device(
         hass,
         mock_entry,
-        {(DOMAIN, f"{coordinator.device_serial}_1")},
+        {(DOMAIN, f"{coordinator.device_key}_1")},
     )
 
     assert await async_remove_config_entry_device(hass, mock_entry, stale) is True

tests/localthings/test_diagnostics.py

@@ -76,8 +76,11 @@ async def test_diagnostics_include_ocf_identity(
     # fields identify a device type, so nothing is dropped up front beyond
     # what redaction takes out.
     assert identity["resources"]["/oic/p"]["mnmn"] == "Samsung Electronics"
-    assert identity["resources"]["/oic/d"]["di"] == REDACTED
-    assert identity["resources"]["/oic/p"]["pi"] == REDACTED
+    # The OCF UUIDs are reported rather than redacted: they are what this
+    # entry is keyed on (issue #381), and blanking them is what made the
+    # first duplicate-serial report unanswerable.
+    assert identity["resources"]["/oic/d"]["di"] == "ab-cd-ef"
+    assert identity["resources"]["/oic/p"]["pi"] == "12-34-56"
     # The owner-settable device name is redacted; `rt` -- the reason this
     # block exists -- is not.
     assert identity["resources"]["/oic/d"]["n"] == REDACTED
@@ -125,9 +128,9 @@ async def test_diagnostics_include_oic_res_links(
 
     links = diag["identity"]["resources"]["/oic/res"]
     assert len(links) == 2
-    assert links[0]["di"] == REDACTED
+    assert links[0]["di"] == "aaaa-1111"
     assert links[0]["href"] == "/device/0"
-    assert links[1]["di"] == REDACTED
+    assert links[1]["di"] == "bbbb-2222"
     assert links[1]["href"] == "/device/1"
     assert links[1]["rt"] == ["x.com.samsung.devcol", "oic.wk.col"]
 

tests/localthings/test_identity_migration.py

@@ -0,0 +1,773 @@
+"""Moving an existing install onto the OCF device UUID (issue #381).
+
+This migration can't finish inside `async_migrate_entry` -- the UUID is
+only readable from the appliance, and an entry can load from its snapshot
+while that appliance is off (issue #295) -- so the coordinator adopts it
+on the first live poll, rewriting both registries and the entry's
+unique_id together.
+
+That makes it the riskiest migration here: it rewrites the identity of
+rows a user's automations, history and areas hang off. These tests are
+organised around what must not break rather than around the functions
+involved. Statistics are covered separately, against a real recorder, in
+tests/test_rekey_statistics_end_to_end.py.
+"""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from unittest.mock import patch
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers import entity_registry as er
+from pytest_homeassistant_custom_component.common import MockConfigEntry
+
+from custom_components.localthings.const import (
+    CONF_DEVICE_KEY,
+    CONF_HOST,
+    CONF_SERIAL,
+    DOMAIN,
+)
+from custom_components.localthings.registry.identity import DeviceIdentity
+
+from .conftest import LEGACY_ENTRY_DATA, MOCK_HOST, MOCK_SERIAL
+
+_COORD = "custom_components.localthings.coordinator.LocalThingsCoordinator"
+
+# The two purifiers from issue #381: one serialNum, two device UUIDs.
+SHARED_SERIAL = "BS7SP9AW400114A"
+UUID_A = "ccfd73b3-aeb4-792a-1100-68f06f5d603b"
+UUID_B = "3771f8bf-c184-3a2d-d885-e4c9818736d2"
+
+
+@contextmanager
+def _reachable(resources: dict, device_id: str | None):
+    """A device that answers a poll, reporting `device_id` as its /oic/d
+    `di` -- which `_connect_session` is what normally reads, so a test that
+    patches it out otherwise leaves `_identity` None (indistinguishable
+    from firmware that reports no UUID at all)."""
+
+    def _connect(self) -> None:
+        self._identity = (
+            None
+            if device_id is None
+            else DeviceIdentity(
+                manufacturer="Samsung Electronics",
+                model="AVT-WW-TP1-23-AXX500",
+                name="Samsung AirPurifier",
+                serial=None,
+                device_id=device_id,
+            )
+        )
+
+    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 _entry(
+    hass: HomeAssistant,
+    *,
+    version: int,
+    key: str,
+    serial: str | None = None,
+    device_key: str | None = None,
+    host: str = MOCK_HOST,
+) -> MockConfigEntry:
+    """An entry as it sits on disk at `version`. Pre-v4 entries carry no
+    CONF_DEVICE_KEY at all -- that absence is what tells the coordinator it
+    is looking at an entry that has never adopted a UUID."""
+    data = {**LEGACY_ENTRY_DATA, CONF_HOST: host}
+    if version >= 2:
+        data[CONF_SERIAL] = serial if serial is not None else key
+    if device_key is not None:
+        data[CONF_DEVICE_KEY] = device_key
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        data=data,
+        unique_id=f"{DOMAIN}_{key}",
+        version=version,
+    )
+    entry.add_to_hass(hass)
+    return entry
+
+
+def _reporting_serial(resources: dict, serial: str) -> dict:
+    """`resources` with the serialNum the device reports swapped out, so a
+    test can pair a fixture with the identity its scenario implies."""
+    info = dict(resources["/information/vs/0"])
+    info["x.com.samsung.da.serialNum"] = serial
+    return {**resources, "/information/vs/0": info}
+
+
+def _device_identifiers(hass: HomeAssistant, device_id: str) -> set[tuple[str, str]]:
+    """This device row's identifiers, asserting the row still exists.
+
+    `dev_reg.async_get` returns `DeviceEntry | None`, so reading through it
+    directly would crash with an AttributeError on a row the re-key
+    wrongly removed instead of failing the assertion that says so.
+    """
+    row = dr.async_get(hass).async_get(device_id)
+    assert row is not None
+    return row.identifiers
+
+
+def _entity_unique_id(hass: HomeAssistant, entity_id: str) -> str:
+    """This entity row's unique_id, asserting the row still exists."""
+    row = er.async_get(hass).async_get(entity_id)
+    assert row is not None
+    return row.unique_id
+
+
+def _seed_registry(hass: HomeAssistant, entry: MockConfigEntry, key: str, **entity_kwargs):
+    """A device and one entity keyed on `key`, as a running install has."""
+    dev_reg = dr.async_get(hass)
+    ent_reg = er.async_get(hass)
+    device = dev_reg.async_get_or_create(
+        config_entry_id=entry.entry_id,
+        identifiers={(DOMAIN, key)},
+        name=f"Samsung Air Purifier ({key})",
+    )
+    entity = ent_reg.async_get_or_create(
+        "sensor",
+        DOMAIN,
+        f"{DOMAIN}_{key}_connection_mode",
+        config_entry=entry,
+        device_id=device.id,
+        **entity_kwargs,
+    )
+    return device, entity
+
+
+# ---------------------------------------------------------------------------
+# The upgrade itself
+# ---------------------------------------------------------------------------
+
+
+async def test_v3_entry_moves_onto_the_device_uuid_keeping_its_entity_ids(
+    hass: HomeAssistant, fridge_resources
+) -> None:
+    """The migration promise for an existing user, asserted end to end: all
+    three permanent places move together, and the entity_id doesn't."""
+    entry = _entry(hass, version=3, key=MOCK_SERIAL)
+    device, existing = _seed_registry(
+        hass, entry, MOCK_SERIAL, suggested_object_id="kitchen_purifier_connection"
+    )
+
+    with _reachable(fridge_resources, UUID_A):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert entry.version == 4
+    assert entry.data[CONF_DEVICE_KEY] == UUID_A
+    # The serial is kept alongside the key, not replaced by it: it is what
+    # corroborates a later change of UUID.
+    assert entry.data[CONF_SERIAL] == MOCK_SERIAL
+    assert entry.unique_id == f"{DOMAIN}_{UUID_A}"
+
+    rekeyed_device = dr.async_get(hass).async_get(device.id)
+    assert rekeyed_device is not None
+    assert rekeyed_device.identifiers == {(DOMAIN, UUID_A)}
+    kept = er.async_get(hass).async_get(existing.entity_id)
+    assert kept is not None
+    assert kept.entity_id == "sensor.kitchen_purifier_connection"
+    assert kept.unique_id == f"{DOMAIN}_{UUID_A}_connection_mode"
+    # Nothing left behind on the old key.
+    assert dr.async_get(hass).async_get_device(identifiers={(DOMAIN, MOCK_SERIAL)}) is None
+
+
+async def test_the_oldest_install_walks_all_the_way_from_v1(
+    hass: HomeAssistant, fridge_resources
+) -> None:
+    """A v1 entry -- no stored identity at all, from before issue #236 --
+    walks v1 -> v2 -> v3 -> v4 and then adopts the UUID on its first poll.
+
+    The oldest installs take the longest path, and each step rewrites what
+    the next one reads, so the chain is worth pinning as one journey rather
+    than trusting the individual steps to compose.
+    """
+    entry = _entry(hass, version=1, key=MOCK_SERIAL)
+    device, existing = _seed_registry(
+        hass, entry, MOCK_SERIAL, suggested_object_id="old_install_connection"
+    )
+
+    with _reachable(fridge_resources, UUID_A):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert entry.version == 4
+    assert entry.data[CONF_DEVICE_KEY] == UUID_A
+    assert entry.unique_id == f"{DOMAIN}_{UUID_A}"
+    kept = er.async_get(hass).async_get(existing.entity_id)
+    assert kept is not None
+    assert kept.entity_id == "sensor.old_install_connection"
+    assert kept.unique_id == f"{DOMAIN}_{UUID_A}_connection_mode"
+    assert _device_identifiers(hass, device.id) == {(DOMAIN, UUID_A)}
+
+
+async def test_a_host_keyed_entry_adopts_a_real_identity(
+    hass: HomeAssistant, fridge_resources
+) -> None:
+    """A placeholder-serial board (issues #83/#189) was keyed on its IP,
+    which is an address rather than an identity -- a new DHCP lease silently
+    makes it someone else's. Such an entry never made an identity claim to
+    defend, so a real UUID is adopted without needing the serial to
+    corroborate it; requiring corroboration would strand exactly these
+    boards, since their serial resolves to the host and can never match."""
+    entry = _entry(hass, version=3, key=MOCK_HOST)
+    device, _ = _seed_registry(hass, entry, MOCK_HOST)
+
+    with _reachable(fridge_resources, UUID_B):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert entry.data[CONF_DEVICE_KEY] == UUID_B
+    assert _device_identifiers(hass, device.id) == {(DOMAIN, UUID_B)}
+
+
+async def test_the_two_units_from_the_issue_migrate_to_separate_identities(
+    hass: HomeAssistant, fridge_resources
+) -> None:
+    """Issue #381's actual install: two entries whose stored identity is the
+    same shared serial. Before this, they could not both exist. Migrating
+    them must give each its own key rather than collapsing them again --
+    including their entity unique_ids, which is where issue #83's Bug 4
+    silently dropped the second unit's entities even once its entry existed.
+    """
+    # Both units report the shared serial, as the real ones do -- so the
+    # entry's stored identity still matches what the device says, and only
+    # the UUID separates them.
+    resources = _reporting_serial(fridge_resources, SHARED_SERIAL)
+    first = _entry(hass, version=3, key=SHARED_SERIAL, host="192.168.0.3")
+    _, first_entity = _seed_registry(hass, first, SHARED_SERIAL, suggested_object_id="purifier_a")
+
+    with _reachable(resources, UUID_A):
+        await hass.config_entries.async_setup(first.entry_id)
+        await hass.async_block_till_done()
+
+    # Added only now: setting up the first entry loads the integration, which
+    # brings up every entry already registered -- so a second one created up
+    # front would come up inside the first one's patched identity and adopt
+    # its UUID, which is the collision this test exists to disprove.
+    second = _entry(hass, version=3, key=SHARED_SERIAL, host="192.168.0.14")
+    _, second_entity = _seed_registry(hass, second, SHARED_SERIAL, suggested_object_id="purifier_b")
+
+    with _reachable(resources, UUID_B):
+        await hass.config_entries.async_setup(second.entry_id)
+        await hass.async_block_till_done()
+
+    assert first.data[CONF_DEVICE_KEY] == UUID_A
+    assert second.data[CONF_DEVICE_KEY] == UUID_B
+    assert first.unique_id != second.unique_id
+
+    assert _entity_unique_id(hass, first_entity.entity_id) == (f"{DOMAIN}_{UUID_A}_connection_mode")
+    assert _entity_unique_id(hass, second_entity.entity_id) == (
+        f"{DOMAIN}_{UUID_B}_connection_mode"
+    )
+
+
+# ---------------------------------------------------------------------------
+# What the user must not lose
+# ---------------------------------------------------------------------------
+
+
+async def test_every_user_customization_on_the_row_survives(
+    hass: HomeAssistant, fridge_resources
+) -> None:
+    """Re-keying rewrites the registry row in place rather than replacing
+    it, which is the whole reason to do it this way -- so everything the
+    user attached to that row rides along. A rename, an area, an icon
+    override and a deliberate hide are each things they would have to redo
+    by hand if the row were recreated instead.
+    """
+    entry = _entry(hass, version=3, key=MOCK_SERIAL)
+    ent_reg = er.async_get(hass)
+    device, existing = _seed_registry(
+        hass, entry, MOCK_SERIAL, suggested_object_id="kitchen_purifier_connection"
+    )
+    dr.async_get(hass).async_update_device(device.id, area_id="kitchen")
+    ent_reg.async_update_entity(
+        existing.entity_id,
+        name="Purifier link",
+        icon="mdi:air-filter",
+        area_id="kitchen",
+        hidden_by=er.RegistryEntryHider.USER,
+    )
+
+    with _reachable(fridge_resources, UUID_A):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    kept = ent_reg.async_get(existing.entity_id)
+    assert kept is not None
+    assert kept.unique_id == f"{DOMAIN}_{UUID_A}_connection_mode"
+    assert kept.name == "Purifier link"
+    assert kept.icon == "mdi:air-filter"
+    assert kept.area_id == "kitchen"
+    assert kept.hidden_by is er.RegistryEntryHider.USER
+    rekeyed_device = dr.async_get(hass).async_get(device.id)
+    assert rekeyed_device is not None
+    assert rekeyed_device.area_id == "kitchen"
+
+
+async def test_a_composite_appliance_keeps_its_subdevice_links(
+    hass: HomeAssistant, fridge_resources
+) -> None:
+    """A composite appliance (issue #177) registers one device per logical
+    subdevice, keyed f"{key}_{subdevice}" and linked via_device to the
+    master's bare key. Rewriting only the exact-match identifier would
+    strand every sibling under a via_device pointing at a device that no
+    longer exists, collapsing the user's device tree."""
+    entry = _entry(hass, version=3, key=MOCK_SERIAL)
+    dev_reg = dr.async_get(hass)
+    master, _ = _seed_registry(hass, entry, MOCK_SERIAL)
+    sub = dev_reg.async_get_or_create(
+        config_entry_id=entry.entry_id,
+        identifiers={(DOMAIN, f"{MOCK_SERIAL}_subdevice_1")},
+        via_device=(DOMAIN, MOCK_SERIAL),
+    )
+    assert sub.via_device_id == master.id
+
+    with _reachable(fridge_resources, UUID_A):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert _device_identifiers(hass, master.id) == {(DOMAIN, UUID_A)}
+    rekeyed_sub = dev_reg.async_get(sub.id)
+    assert rekeyed_sub is not None
+    assert rekeyed_sub.identifiers == {(DOMAIN, f"{UUID_A}_subdevice_1")}
+    # Still the same parent row, so the device tree the user sees is intact.
+    assert rekeyed_sub.via_device_id == master.id
+
+
+async def test_a_re_key_never_touches_another_entrys_rows(hass: HomeAssistant) -> None:
+    """The rewrite is scoped to one config entry's own registry rows.
+
+    Issue #381's install is the case that makes this sharp: *both* entries
+    are keyed on the same shared serial, so both have registry rows under
+    the identical old key, and they migrate one at a time. An unscoped
+    rewrite -- matching on the key prefix alone -- would sweep up the other
+    appliance's device and entities and hand them to the first one to
+    migrate, which is the worst outcome this change could have.
+
+    The two entries hold *different* entities under that one shared prefix
+    (HA's registry won't let two rows share a unique_id, which is issue
+    #83's Bug 4 in the first place), so only the config-entry scoping can
+    tell them apart -- prefix matching alone cannot.
+
+    Calls rekey_entry directly so the scoping is what's under test, rather
+    than the coordinator's decision about whether to call it at all.
+    """
+    from custom_components.localthings.rekey import rekey_entry
+
+    migrating = _entry(hass, version=3, key=SHARED_SERIAL, host="192.168.0.3")
+    bystander = _entry(hass, version=3, key=SHARED_SERIAL, host="192.168.0.14")
+    ent_reg = er.async_get(hass)
+    moved = ent_reg.async_get_or_create(
+        "sensor", DOMAIN, f"{DOMAIN}_{SHARED_SERIAL}_connection_mode", config_entry=migrating
+    )
+    stays = ent_reg.async_get_or_create(
+        "sensor", DOMAIN, f"{DOMAIN}_{SHARED_SERIAL}_power", config_entry=bystander
+    )
+
+    rekey_entry(hass, migrating, SHARED_SERIAL, UUID_A)
+
+    assert _entity_unique_id(hass, moved.entity_id) == f"{DOMAIN}_{UUID_A}_connection_mode"
+    # The other appliance is still on the shared serial, waiting its turn.
+    assert _entity_unique_id(hass, stays.entity_id) == f"{DOMAIN}_{SHARED_SERIAL}_power"
+    assert bystander.unique_id == f"{DOMAIN}_{SHARED_SERIAL}"
+
+
+async def test_a_re_key_stops_at_the_key_boundary(hass: HomeAssistant) -> None:
+    """Matching is on the whole key or the key plus a separator, never a
+    bare prefix. Serial numbers of one model are routinely prefixes of each
+    other, so a naive startswith would drag a *different* appliance's rows
+    along -- and the identifiers it would rewrite them to are nonsense."""
+    from custom_components.localthings.rekey import rekey_entry
+
+    entry = _entry(hass, version=3, key="TEST-SERIAL")
+    dev_reg = dr.async_get(hass)
+    ent_reg = er.async_get(hass)
+    target = dev_reg.async_get_or_create(
+        config_entry_id=entry.entry_id, identifiers={(DOMAIN, "TEST-SERIAL")}
+    )
+    # Same config entry, so scoping can't save this one -- only the boundary.
+    neighbour = dev_reg.async_get_or_create(
+        config_entry_id=entry.entry_id, identifiers={(DOMAIN, "TEST-SERIAL-0000")}
+    )
+    neighbour_entity = ent_reg.async_get_or_create(
+        "sensor", DOMAIN, f"{DOMAIN}_TEST-SERIAL-0000_connection_mode", config_entry=entry
+    )
+
+    rekey_entry(hass, entry, "TEST-SERIAL", UUID_A)
+
+    assert _device_identifiers(hass, target.id) == {(DOMAIN, UUID_A)}
+    assert _device_identifiers(hass, neighbour.id) == {(DOMAIN, "TEST-SERIAL-0000")}
+    assert _entity_unique_id(hass, neighbour_entity.entity_id) == (
+        f"{DOMAIN}_TEST-SERIAL-0000_connection_mode"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Stability: upgrading twice, or offline, must not churn
+# ---------------------------------------------------------------------------
+
+
+async def test_upgrading_while_the_appliance_is_off_changes_nothing(
+    hass: HomeAssistant, fridge_resources, hass_storage
+) -> None:
+    """The case a large share of users will actually hit: HA restarts onto
+    the new release while the appliance is unplugged or asleep.
+
+    The entry comes up from its snapshot (issue #295), which never reached
+    the device -- so it has no standing to claim an identity. It must load
+    under the key its registry rows already carry and write nothing, or the
+    real UUID would later look like a *changed* identity to defend against
+    rather than the one-time adoption it is.
+    """
+    entry = _entry(hass, version=3, key=MOCK_SERIAL)
+    _, existing = _seed_registry(hass, entry, MOCK_SERIAL)
+
+    # Bank a snapshot from a run on the old release, then take it down.
+    with _reachable(fridge_resources, None):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+        await hass.config_entries.async_unload(entry.entry_id)
+        await hass.async_block_till_done()
+    assert entry.data[CONF_DEVICE_KEY] == MOCK_SERIAL
+
+    # Now the appliance is off. Simulate the pre-v4 shape the upgrade finds.
+    hass.config_entries.async_update_entry(
+        entry, data={k: v for k, v in entry.data.items() if k != CONF_DEVICE_KEY}, version=3
+    )
+
+    with _unreachable():
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    coordinator = hass.data[DOMAIN][entry.entry_id]
+    assert coordinator.device_key == MOCK_SERIAL
+    # No key claimed, and the registry is exactly as it was.
+    assert CONF_DEVICE_KEY not in entry.data
+    assert entry.unique_id == f"{DOMAIN}_{MOCK_SERIAL}"
+    assert _entity_unique_id(hass, existing.entity_id) == (
+        f"{DOMAIN}_{MOCK_SERIAL}_connection_mode"
+    )
+
+
+async def test_an_offline_load_never_rewrites_the_registry(
+    hass: HomeAssistant, fridge_resources, hass_storage
+) -> None:
+    """A snapshot replay must not re-key on the strength of what the
+    snapshot says, because that is last run's answer rather than the
+    device's.
+
+    Modelled on a placeholder-serial board (issues #83/#189), which is
+    where the two can genuinely disagree: the entry is keyed on its address
+    because the board reports no usable serial, while the snapshot banked
+    whatever serial the polled resources carried. A replay that trusted the
+    snapshot would rewrite every registry row onto that serial -- without
+    the appliance having been reachable at any point -- and would then
+    freeze the answer into CONF_DEVICE_KEY, so the real UUID could never be
+    adopted afterwards.
+    """
+    entry = _entry(hass, version=3, key=MOCK_HOST)
+
+    with _reachable(fridge_resources, None):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+        await hass.config_entries.async_unload(entry.entry_id)
+        await hass.async_block_till_done()
+
+    # Back to the pre-v4 shape the upgrade finds, still keyed on the address.
+    hass.config_entries.async_update_entry(
+        entry,
+        data={
+            **{k: v for k, v in entry.data.items() if k != CONF_DEVICE_KEY},
+            CONF_SERIAL: MOCK_HOST,
+        },
+        unique_id=f"{DOMAIN}_{MOCK_HOST}",
+        version=3,
+    )
+    device, existing = _seed_registry(hass, entry, MOCK_HOST)
+
+    with _unreachable():
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    coordinator = hass.data[DOMAIN][entry.entry_id]
+    assert coordinator.device_key == MOCK_HOST
+    assert CONF_DEVICE_KEY not in entry.data
+    assert entry.unique_id == f"{DOMAIN}_{MOCK_HOST}"
+    assert _device_identifiers(hass, device.id) == {(DOMAIN, MOCK_HOST)}
+    assert _entity_unique_id(hass, existing.entity_id) == (f"{DOMAIN}_{MOCK_HOST}_connection_mode")
+
+    # And the deferred adoption still works once the appliance answers --
+    # the offline load left nothing frozen behind it.
+    with _reachable(fridge_resources, UUID_A):
+        coordinator._connect_session()
+        coordinator._run_discovery(fridge_resources)
+
+    assert coordinator.device_key == UUID_A
+    assert entry.data[CONF_DEVICE_KEY] == UUID_A
+    assert _entity_unique_id(hass, existing.entity_id) == (f"{DOMAIN}_{UUID_A}_connection_mode")
+
+
+async def test_the_appliance_coming_back_completes_the_upgrade(
+    hass: HomeAssistant, fridge_resources
+) -> None:
+    """The other half of the offline case: the deferred adoption is not
+    abandoned, it just waits. Once the device answers, the same re-key runs
+    and the entry finishes its upgrade."""
+    entry = _entry(hass, version=3, key=MOCK_SERIAL)
+    _, existing = _seed_registry(hass, entry, MOCK_SERIAL)
+
+    with _reachable(fridge_resources, None):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+    assert entry.data[CONF_DEVICE_KEY] == MOCK_SERIAL
+
+    # The device is reachable again, now reporting its UUID.
+    coordinator = hass.data[DOMAIN][entry.entry_id]
+    with _reachable(fridge_resources, UUID_A):
+        coordinator._connect_session()
+        coordinator._run_discovery(fridge_resources)
+
+    assert coordinator.device_key == UUID_A
+    assert entry.data[CONF_DEVICE_KEY] == UUID_A
+    assert entry.unique_id == f"{DOMAIN}_{UUID_A}"
+    assert _entity_unique_id(hass, existing.entity_id) == (f"{DOMAIN}_{UUID_A}_connection_mode")
+
+
+async def test_restarting_after_the_upgrade_is_a_no_op(
+    hass: HomeAssistant, fridge_resources
+) -> None:
+    """Every restart re-runs discovery, so the adoption path runs again on
+    an entry that has already moved. It must recognise its own work and do
+    nothing -- a re-key that fired every boot would churn the registry
+    forever."""
+    entry = _entry(hass, version=3, key=MOCK_SERIAL)
+    _, existing = _seed_registry(hass, entry, MOCK_SERIAL)
+
+    with _reachable(fridge_resources, UUID_A):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+        after_first = dict(entry.data)
+        await hass.config_entries.async_unload(entry.entry_id)
+        await hass.async_block_till_done()
+
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert dict(entry.data) == after_first
+    assert entry.unique_id == f"{DOMAIN}_{UUID_A}"
+    kept = er.async_get(hass).async_get(existing.entity_id)
+    assert kept is not None
+    assert kept.unique_id == f"{DOMAIN}_{UUID_A}_connection_mode"
+    # Exactly one device, not a duplicate alongside it.
+    assert len(dr.async_entries_for_config_entry(dr.async_get(hass), entry.entry_id)) == 1
+
+
+async def test_an_already_migrated_entry_is_left_alone(
+    hass: HomeAssistant, fridge_resources
+) -> None:
+    """A v4 entry created by the current config flow has nothing to migrate
+    and nothing to re-key -- it was minted on its UUID."""
+    entry = _entry(hass, version=4, key=UUID_A, serial=MOCK_SERIAL, device_key=UUID_A)
+    device, existing = _seed_registry(hass, entry, UUID_A)
+
+    with _reachable(fridge_resources, UUID_A):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert entry.version == 4
+    assert entry.data[CONF_DEVICE_KEY] == UUID_A
+    assert _device_identifiers(hass, device.id) == {(DOMAIN, UUID_A)}
+    assert _entity_unique_id(hass, existing.entity_id) == (f"{DOMAIN}_{UUID_A}_connection_mode")
+
+
+# ---------------------------------------------------------------------------
+# Guarding the identity once it has moved
+# ---------------------------------------------------------------------------
+
+
+async def test_a_poll_that_reads_no_uuid_does_not_demote_a_keyed_entry(
+    hass: HomeAssistant, fridge_resources
+) -> None:
+    """The device saying nothing is not the device saying something
+    different. A reconnect that can't read /oic/d (a timeout, a firmware
+    hiccup) must leave the key alone -- demoting back onto the serial would
+    re-key every entity the user has for the duration of an outage, and
+    re-key them all back afterwards."""
+    entry = _entry(hass, version=4, key=UUID_A, serial=MOCK_SERIAL, device_key=UUID_A)
+
+    with _reachable(fridge_resources, None):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    coordinator = hass.data[DOMAIN][entry.entry_id]
+    assert coordinator.device_key == UUID_A
+    assert entry.data[CONF_DEVICE_KEY] == UUID_A
+
+
+async def test_a_rotated_uuid_on_the_same_serial_is_followed(
+    hass: HomeAssistant, fridge_resources
+) -> None:
+    """OCF permits a hard factory reset to regenerate `di`. The serialNum is
+    what tells that apart from a different appliance moving onto the
+    address, and following it keeps the user's history rather than stranding
+    it on a UUID the device will never report again."""
+    entry = _entry(hass, version=4, key=UUID_A, serial=MOCK_SERIAL, device_key=UUID_A)
+    device, existing = _seed_registry(hass, entry, UUID_A)
+
+    # fridge_resources reports MOCK_SERIAL, matching what the entry stored.
+    with _reachable(fridge_resources, UUID_B):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert entry.data[CONF_DEVICE_KEY] == UUID_B
+    assert _device_identifiers(hass, device.id) == {(DOMAIN, UUID_B)}
+    assert _entity_unique_id(hass, existing.entity_id) == (f"{DOMAIN}_{UUID_B}_connection_mode")
+
+
+async def test_a_different_appliance_on_the_same_address_keeps_the_registered_identity(
+    hass: HomeAssistant, fridge_resources
+) -> None:
+    """Neither the UUID nor the serial matches what this entry was
+    registered with, so this is a different appliance answering at this
+    address -- not a reset of the registered one. Re-keying here would hand
+    one appliance's entities, history and automations to another; re-adding
+    is the user's call."""
+    entry = _entry(hass, version=4, key=UUID_A, serial="SOME-OTHER-APPLIANCE", device_key=UUID_A)
+    device, _ = _seed_registry(hass, entry, UUID_A)
+
+    with _reachable(fridge_resources, UUID_B):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+        assert entry.data[CONF_DEVICE_KEY] == UUID_A
+        assert _device_identifiers(hass, device.id) == {(DOMAIN, UUID_A)}
+        # The rejected appliance's serial is not written either. The serial
+        # is what corroborates a later change of key, so adopting it here
+        # would hand the intruder exactly the corroboration it needs to win
+        # the *next* poll -- defending the identity once and then
+        # surrendering it on the following cycle.
+        assert entry.data[CONF_SERIAL] == "SOME-OTHER-APPLIANCE"
+
+        coordinator = hass.data[DOMAIN][entry.entry_id]
+        coordinator._run_discovery(fridge_resources)
+
+        assert coordinator.device_key == UUID_A
+        assert entry.data[CONF_DEVICE_KEY] == UUID_A
+        assert entry.data[CONF_SERIAL] == "SOME-OTHER-APPLIANCE"
+
+
+async def test_a_pre_v4_entry_defends_itself_against_a_different_appliance(
+    hass: HomeAssistant, fridge_resources
+) -> None:
+    """The "same IP, different appliance" guard applies to an entry that has
+    not migrated yet, exactly as it does to one that has.
+
+    A pre-v4 entry is the population this change exists to move, but it is
+    also the population that has been running longest -- so it is the last
+    one that should hand its entity_ids, history and automations to an
+    appliance that merely happens to have taken over its address. Adoption
+    is the migration's job only when the identity is corroborated.
+    """
+    entry = _entry(hass, version=3, key=MOCK_SERIAL)
+    device, existing = _seed_registry(hass, entry, MOCK_SERIAL)
+
+    # Neither the serial nor (therefore) the UUID belongs to the registered
+    # appliance.
+    resources = _reporting_serial(fridge_resources, "SOME-OTHER-APPLIANCE")
+    with _reachable(resources, UUID_B):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    coordinator = hass.data[DOMAIN][entry.entry_id]
+    assert coordinator.device_key == MOCK_SERIAL
+    assert entry.data[CONF_DEVICE_KEY] == MOCK_SERIAL
+    assert entry.data[CONF_SERIAL] == MOCK_SERIAL
+    assert entry.unique_id == f"{DOMAIN}_{MOCK_SERIAL}"
+    assert _device_identifiers(hass, device.id) == {(DOMAIN, MOCK_SERIAL)}
+    assert _entity_unique_id(hass, existing.entity_id) == (
+        f"{DOMAIN}_{MOCK_SERIAL}_connection_mode"
+    )
+
+
+# ---------------------------------------------------------------------------
+# rekey_entry's own contract
+# ---------------------------------------------------------------------------
+
+
+async def test_rekey_is_idempotent(hass: HomeAssistant) -> None:
+    """Safe to attempt on every poll rather than having to track whether it
+    has already run -- a second call finds nothing under the old key.
+
+    Calls rekey_entry directly: what it leaves behind is the contract, and
+    going through a setup would let the platforms re-adding their entities
+    hide a row that had in fact been orphaned.
+    """
+    from custom_components.localthings.rekey import rekey_entry
+
+    entry = _entry(hass, version=3, key=MOCK_SERIAL)
+    _, existing = _seed_registry(hass, entry, MOCK_SERIAL)
+
+    rekey_entry(hass, entry, MOCK_SERIAL, UUID_A)
+    rekey_entry(hass, entry, MOCK_SERIAL, UUID_A)
+
+    kept = er.async_get(hass).async_get(existing.entity_id)
+    assert kept is not None
+    assert kept.unique_id == f"{DOMAIN}_{UUID_A}_connection_mode"
+    assert entry.unique_id == f"{DOMAIN}_{UUID_A}"
+
+
+async def test_rekey_removes_a_stale_row_rather_than_colliding(hass: HomeAssistant) -> None:
+    """Where the destination key is already taken, the old-key row is the
+    dead one -- unavailable since whichever restart created the split -- so
+    it goes rather than being rewritten onto a key that exists. Same rule
+    the #236 repair has always applied, now on the identity move."""
+    from custom_components.localthings.rekey import rekey_entry
+
+    entry = _entry(hass, version=3, key=MOCK_SERIAL)
+    ent_reg = er.async_get(hass)
+    live = ent_reg.async_get_or_create(
+        "sensor", DOMAIN, f"{DOMAIN}_{UUID_A}_connection_mode", config_entry=entry
+    )
+    stale = ent_reg.async_get_or_create(
+        "sensor", DOMAIN, f"{DOMAIN}_{MOCK_SERIAL}_connection_mode", config_entry=entry
+    )
+    assert stale.entity_id != live.entity_id
+
+    rekey_entry(hass, entry, MOCK_SERIAL, UUID_A)
+
+    assert ent_reg.async_get(stale.entity_id) is None
+    assert ent_reg.async_get(live.entity_id) is not None
+
+
+async def test_rekey_to_the_same_key_does_nothing(hass: HomeAssistant) -> None:
+    """The no-op guard that lets callers pass whatever they resolved without
+    checking first."""
+    from custom_components.localthings.rekey import rekey_entry
+
+    entry = _entry(hass, version=4, key=UUID_A, device_key=UUID_A)
+    _, existing = _seed_registry(hass, entry, UUID_A)
+
+    rekey_entry(hass, entry, UUID_A, UUID_A)
+
+    assert _entity_unique_id(hass, existing.entity_id) == (f"{DOMAIN}_{UUID_A}_connection_mode")
+    assert entry.unique_id == f"{DOMAIN}_{UUID_A}"

tests/localthings/test_migration.py

@@ -1,4 +1,9 @@
-"""Config-entry migration and the placeholder-identity repair (issue #236)."""
+"""Config-entry migration and the placeholder-identity repair (issue #236).
+
+The move onto the OCF device UUID (issue #381) has its own suite in
+test_identity_migration.py -- it is the one migration step that can't
+finish inside async_migrate_entry, so it needs a device to talk to.
+"""
 
 from __future__ import annotations
 
@@ -39,36 +44,37 @@ async def test_migration_recovers_serial_from_unique_id(
     await hass.async_block_till_done()
 
     # Straight through to the current version: v2 -> v3 is a statistics
-    # relabel that no-ops for a family without particulate sensors.
-    assert entry.version == 3
+    # relabel that no-ops for a family without particulate sensors, and
+    # v3 -> v4 only records the legacy key for the coordinator to re-key
+    # from once a poll produces an OCF device id.
+    assert entry.version == 4
     assert entry.data[CONF_SERIAL] == MOCK_SERIAL
 
 
-async def test_migration_collapses_the_host_port_unique_id(
-    hass: HomeAssistant, mock_coordinator_session
-) -> None:
+async def test_migration_collapses_the_host_port_unique_id(hass: HomeAssistant) -> None:
     """A board with no usable serial (issues #83/#189) used to be keyed two
     different ways at once: `host:port` on the config entry, `host` in the
     device and entity registries. Migration collapses the entry onto the
     registry's form, so the two finally name the same thing."""
+    from custom_components.localthings import async_migrate_entry
+
     entry = _legacy_entry(hass, f"{DOMAIN}_{MOCK_HOST}:{MOCK_PORT}")
 
-    await hass.config_entries.async_setup(entry.entry_id)
-    await hass.async_block_till_done()
+    assert await async_migrate_entry(hass, entry) is True
 
     assert entry.data[CONF_SERIAL] == MOCK_HOST
     assert entry.unique_id == f"{DOMAIN}_{MOCK_HOST}"
 
 
-async def test_migration_resolves_a_placeholder_serial_unique_id(
-    hass: HomeAssistant, mock_coordinator_session
-) -> None:
+async def test_migration_resolves_a_placeholder_serial_unique_id(hass: HomeAssistant) -> None:
     """An entry created before the placeholder rules landed was keyed on the
     placeholder itself (issues #83/#189), while the coordinator has been
     resolving those boards to the host ever since. The unique_id records what
     the flow believed then, not what the registry holds -- taking it at face
     value would re-key working devices back onto a string every unit of the
     family reports, which is the collision those issues are about."""
+    from custom_components.localthings import async_migrate_entry
+
     entry = _legacy_entry(hass, f"{DOMAIN}_Nothing(SVC)")
     dev_reg = dr.async_get(hass)
     device = dev_reg.async_get_or_create(
@@ -76,8 +82,7 @@ async def test_migration_resolves_a_placeholder_serial_unique_id(
         identifiers={(DOMAIN, MOCK_HOST)},
     )
 
-    await hass.config_entries.async_setup(entry.entry_id)
-    await hass.async_block_till_done()
+    assert await async_migrate_entry(hass, entry) is True
 
     assert entry.data[CONF_SERIAL] == MOCK_HOST
     assert entry.unique_id == f"{DOMAIN}_{MOCK_HOST}"
@@ -86,14 +91,13 @@ async def test_migration_resolves_a_placeholder_serial_unique_id(
     assert unchanged.identifiers == {(DOMAIN, MOCK_HOST)}
 
 
-async def test_migration_resolves_an_all_hex_placeholder_unique_id(
-    hass: HomeAssistant, mock_coordinator_session
-) -> None:
+async def test_migration_resolves_an_all_hex_placeholder_unique_id(hass: HomeAssistant) -> None:
     """The issue #189 flash-unset sentinel, same reasoning."""
+    from custom_components.localthings import async_migrate_entry
+
     entry = _legacy_entry(hass, f"{DOMAIN}_FFFFFFFFFFFFFFF")
 
-    await hass.config_entries.async_setup(entry.entry_id)
-    await hass.async_block_till_done()
+    assert await async_migrate_entry(hass, entry) is True
 
     assert entry.data[CONF_SERIAL] == MOCK_HOST
 
@@ -231,13 +235,13 @@ async def test_migration_removes_an_orphan_that_is_already_duplicated(
     assert dev_reg.async_get(real_device.id) is not None
 
 
-async def test_migration_leaves_a_host_identity_device_alone(
-    hass: HomeAssistant, mock_coordinator_session
-) -> None:
+async def test_migration_leaves_a_host_identity_device_alone(hass: HomeAssistant) -> None:
     """A board whose serial resolves *to* the host was never keyed on a
     placeholder -- its host-keyed device is the real one, and re-keying or
     removing it would orphan a working device to fix a problem it doesn't
     have."""
+    from custom_components.localthings import async_migrate_entry
+
     entry = _legacy_entry(hass, f"{DOMAIN}_{MOCK_HOST}")
     dev_reg = dr.async_get(hass)
     device = dev_reg.async_get_or_create(
@@ -245,8 +249,7 @@ async def test_migration_leaves_a_host_identity_device_alone(
         identifiers={(DOMAIN, MOCK_HOST)},
     )
 
-    await hass.config_entries.async_setup(entry.entry_id)
-    await hass.async_block_till_done()
+    assert await async_migrate_entry(hass, entry) is True
 
     assert entry.data[CONF_SERIAL] == MOCK_HOST
     unchanged = dev_reg.async_get(device.id)
@@ -259,7 +262,7 @@ async def test_migration_rejects_a_future_entry_version(hass: HomeAssistant) ->
     written by a newer release."""
     from custom_components.localthings import async_migrate_entry
 
-    entry = MockConfigEntry(domain=DOMAIN, data=LEGACY_ENTRY_DATA, version=4)
+    entry = MockConfigEntry(domain=DOMAIN, data=LEGACY_ENTRY_DATA, version=5)
     entry.add_to_hass(hass)
 
     assert await async_migrate_entry(hass, entry) is False

tests/localthings/test_statistics_migration.py

@@ -72,7 +72,7 @@ async def test_relabels_every_particulate_sensor(hass: HomeAssistant) -> None:
         # µg/m³ has a converter, so the class must be named, not None --
         # passing neither is deprecated and breaks in HA Core 2026.11.
         assert call.kwargs["new_unit_class"] == "concentration"
-    assert entry.version == 3
+    assert entry.version == 4
 
 
 async def test_leaves_other_sensors_on_the_same_device_alone(hass: HomeAssistant) -> None:
@@ -102,7 +102,7 @@ async def test_skips_families_that_did_not_gain_the_unit(hass: HomeAssistant) ->
             assert await async_migrate_entry(hass, entry) is True
 
         assert relabel.call_args_list == [], device_type
-        assert entry.version == 3
+        assert entry.version == 4
 
 
 async def test_defers_rather_than_consuming_the_migration_without_the_recorder(
@@ -127,7 +127,7 @@ async def test_defers_rather_than_consuming_the_migration_without_the_recorder(
         assert await async_migrate_entry(hass, entry) is True
 
     assert len(relabel.call_args_list) == 1
-    assert entry.version == 3
+    assert entry.version == 4
 
 
 async def test_omits_unit_class_on_an_older_home_assistant(hass: HomeAssistant) -> None:
@@ -151,7 +151,7 @@ async def test_omits_unit_class_on_an_older_home_assistant(hass: HomeAssistant)
         assert await async_migrate_entry(hass, entry) is True
 
     assert seen == [{"new_unit_of_measurement": CONCENTRATION_MICROGRAMS_PER_CUBIC_METER}]
-    assert entry.version == 3
+    assert entry.version == 4
 
 
 async def test_a_relabel_failure_never_fails_the_entry(hass: HomeAssistant) -> None:
@@ -165,7 +165,7 @@ async def test_a_relabel_failure_never_fails_the_entry(hass: HomeAssistant) -> N
     with patch(RELABEL, autospec=True, side_effect=TypeError("older HA signature")):
         assert await async_migrate_entry(hass, entry) is True
 
-    assert entry.version == 3
+    assert entry.version == 4
 
 
 async def test_follows_a_renamed_entity_rather_than_rebuilding_its_id(
@@ -214,8 +214,16 @@ async def test_matches_subdevice_prefixed_and_instanced_keys(hass: HomeAssistant
 
 
 async def test_a_fresh_entry_starts_at_the_migrated_version(hass: HomeAssistant) -> None:
-    """A newly created entry has no statistics to relabel, so the config flow
-    mints v3 directly rather than walking through the migration."""
+    """A newly created entry has nothing either migration step needs to do --
+    no statistics to relabel, and the probe already resolved its device key
+    (issue #381) -- so the config flow mints the current version directly
+    rather than walking through them.
+
+    Pinned rather than compared to a constant on purpose: the two must be
+    bumped together, and a migration step added without moving the flow's
+    VERSION never runs at all, because Home Assistant only calls
+    async_migrate_entry for an entry *behind* the flow's version.
+    """
     from custom_components.localthings.config_flow import LocalThingsConfigFlow
 
-    assert LocalThingsConfigFlow.VERSION == 3
+    assert LocalThingsConfigFlow.VERSION == 4

tests/test_air_purifier_airflow_fan.py

@@ -12,7 +12,7 @@ from tests.conftest import _load_device
 
 
 class _FakeCoordinator:
-    device_serial = "TEST-AIRFLOW-SERIAL"
+    device_key = "TEST-AIRFLOW-SERIAL"
     device_info: ClassVar[dict] = {}
     data: ClassVar[dict] = {}
 

tests/test_air_purifier_vtww_fan.py

@@ -21,7 +21,7 @@ from tests.conftest import _load_device
 
 
 class _FakeCoordinator:
-    device_serial = "TEST-VTWW-SERIAL"
+    device_key = "TEST-VTWW-SERIAL"
     device_info: ClassVar[dict] = {}
     data: ClassVar[dict] = {}
 

tests/test_airconditioner_artik051_krac.py

@@ -44,7 +44,7 @@ MODEL = "ARTIK051_KRAC_18K|10193441|60010119001111010100000000000000"
 
 
 class _FakeCoordinator:
-    device_serial = "TEST-KRAC-SERIAL"
+    device_key = "TEST-KRAC-SERIAL"
     device_info: ClassVar[dict] = {}
     data: ClassVar[dict] = {}
 

tests/test_airconditioner_tp1x_rac_01001_fan.py

@@ -28,7 +28,7 @@ FIXTURE = "airconditioner_tp1x_rac_01001"
 
 
 class _FakeCoordinator:
-    device_serial = "TEST-RAC-01001-SERIAL"
+    device_key = "TEST-RAC-01001-SERIAL"
     device_info: ClassVar[dict] = {}
     data: ClassVar[dict] = {}
 

tests/test_climate_ac_modes.py

@@ -104,7 +104,7 @@ def test_fac_bora_wind_strength_codes_fit_the_standard_scale():
     from tests.conftest import _load_device
 
     class _FakeCoordinator:
-        device_serial = "TEST-FAC-BORA-SERIAL"
+        device_key = "TEST-FAC-BORA-SERIAL"
         device_info: ClassVar[dict] = {}
         data: ClassVar[dict] = {}
 

tests/test_dishwasher_capabilities.py

@@ -5,6 +5,8 @@ cycle_options, cycle_write) is tested in test_laundry_capabilities.py; here we
 check the dishwasher wiring and its device-specific options.
 """
 
+from datetime import UTC, datetime
+
 from custom_components.localthings.registry.capabilities import dishwasher
 from custom_components.localthings.registry.entities import SensorDesc, SwitchDesc
 
@@ -62,6 +64,53 @@ class TestDishwasherOptions:
         assert desc.exists_fn({"x.com.samsung.da.options": ["AutoDoorRelease_On"]}, {}) is True
 
 
+class TestDrumClean:
+    """Drum Clean+ maintenance tracking shares washer.py's (issue #9) /
+    dryer.py's (issue #258) options[]-array readers -- a live dishwasher
+    dump confirmed the same WashingTimes_/DrumCleanProposal_/DrumCleanLog_
+    trio, so these entities are wired the same way here."""
+
+    def test_cycles_remaining(self):
+        desc = next(
+            e for e in dishwasher.CYCLE_OPTIONS.entities if e.key == "drum_clean_cycles_remaining"
+        )
+        assert desc.rep_fn is not None
+        rep = {"x.com.samsung.da.options": ["WashingTimes_18", "DrumCleanProposal_20"]}
+        assert desc.rep_fn(rep) == 2
+
+    def test_cycles_remaining_exists_only_when_computable(self):
+        desc = next(
+            e for e in dishwasher.CYCLE_OPTIONS.entities if e.key == "drum_clean_cycles_remaining"
+        )
+        assert desc.exists_fn is not None
+        assert desc.exists_fn({"x.com.samsung.da.options": []}, {}) is False
+        rep = {"x.com.samsung.da.options": ["WashingTimes_18", "DrumCleanProposal_20"]}
+        assert desc.exists_fn(rep, {}) is True
+
+    def test_last_cleaned(self):
+        """A live dump's DrumCleanLog_ is a '|'-joined history (the dryer
+        shape, not the washer's single-entry one) -- the last entry wins."""
+        desc = next(
+            e for e in dishwasher.CYCLE_OPTIONS.entities if e.key == "drum_clean_last_cleaned"
+        )
+        assert desc.rep_fn is not None
+        rep = {
+            "x.com.samsung.da.options": [
+                "DrumCleanLog_2026-06-26T04:18:58|2026-06-28T00:36:34",
+            ]
+        }
+        assert desc.rep_fn(rep) == datetime(2026, 6, 28, 0, 36, 34, tzinfo=UTC)
+
+    def test_last_cleaned_missing(self):
+        desc = next(
+            e for e in dishwasher.CYCLE_OPTIONS.entities if e.key == "drum_clean_last_cleaned"
+        )
+        assert desc.rep_fn is not None
+        assert desc.rep_fn({"x.com.samsung.da.options": []}) is None
+        assert desc.exists_fn is not None
+        assert desc.exists_fn({"x.com.samsung.da.options": []}, {}) is False
+
+
 def test_diagnosis_status_is_a_translatable_enum():
     desc = next(
         e

tests/test_entity_naming.py

@@ -10,7 +10,7 @@ from custom_components.localthings.registry.entities import BinarySensorDesc
 
 
 class _FakeCoordinator:
-    device_serial = "TEST-SERIAL"
+    device_key = "TEST-SERIAL"
 
     def __init__(self, last_resources=None):
         self.last_resources = last_resources or {}

tests/test_fridge_capabilities.py

@@ -463,7 +463,7 @@ class TestKimchiZone:
         )
 
         class _FakeCoordinator:
-            device_serial = "TEST-SERIAL"
+            device_key = "TEST-SERIAL"
 
             def __init__(self, resources, data):
                 self.last_resources = resources

tests/test_identity.py

@@ -1,6 +1,12 @@
 import cbor2
 
-from custom_components.localthings.registry.identity import read_identity
+from custom_components.localthings.registry.identity import (
+    DeviceIdentity,
+    is_usable_device_id,
+    ocf_device_key,
+    read_identity,
+    resolve_device_key,
+)
 
 
 class FakeSession:
@@ -124,3 +130,125 @@ def test_read_identity_tolerates_malformed_oic_res():
     _device_types' handling of a malformed /oic/d rt."""
     ident = read_identity(FakeSession({("oic", "res"): {"not": "a list"}}), serial=None)
     assert ident.raw["/oic/res"] == []
+
+
+# ---------------------------------------------------------------------------
+# The device key: which identity field registry keys are minted from (#381)
+# ---------------------------------------------------------------------------
+
+
+def _identity(
+    *,
+    serial: str | None = None,
+    device_id: str | None = None,
+    platform_id: str | None = None,
+) -> DeviceIdentity:
+    """A DeviceIdentity carrying only the fields the key chain reads."""
+    return DeviceIdentity(
+        manufacturer="Samsung",
+        model="M",
+        name="N",
+        serial=serial,
+        device_id=device_id,
+        platform_id=platform_id,
+    )
+
+
+def test_read_identity_captures_the_ocf_uuids_as_named_fields():
+    """`di`/`pi` are what resolve_device_key mints keys from, so they are
+    lifted out of `raw` rather than dug back out of it at every call site."""
+    sess = FakeSession(
+        {
+            ("oic", "p"): {"pi": "ccfd73b3-aeb4-792a-1100-68f06f5d603b"},
+            ("oic", "d"): {"di": "3771f8bf-c184-3a2d-d885-e4c9818736d2"},
+        }
+    )
+    ident = read_identity(sess, serial=None)
+    assert ident.device_id == "3771f8bf-c184-3a2d-d885-e4c9818736d2"
+    assert ident.platform_id == "ccfd73b3-aeb4-792a-1100-68f06f5d603b"
+
+
+def test_read_identity_ignores_non_string_uuids():
+    """Firmware answering with a number or a map must not put a non-string
+    into a field that goes on to be string-formatted into a unique_id."""
+    ident = read_identity(FakeSession({("oic", "d"): {"di": 42}, ("oic", "p"): {"pi": {}}}), None)
+    assert ident.device_id is None
+    assert ident.platform_id is None
+
+
+def test_two_units_sharing_a_serial_get_distinct_keys():
+    """Issue #381 exactly: two Samsung air purifiers of the same model ship
+    the identical, well-formed serialNum 'BS7SP9AW400114A', so keying on it
+    collapsed them onto one identity and the second was refused as already
+    configured. Their `di` differs, which is what makes them separable."""
+    shared_serial = "BS7SP9AW400114A"
+    first = resolve_device_key(
+        _identity(device_id="ccfd73b3-aeb4-792a-1100-68f06f5d603b"), shared_serial, "192.168.0.3"
+    )
+    second = resolve_device_key(
+        _identity(device_id="3771f8bf-c184-3a2d-d885-e4c9818736d2"), shared_serial, "192.168.0.14"
+    )
+    assert first != second
+    assert shared_serial not in (first, second)
+
+
+def test_platform_id_is_the_fallback_when_oic_d_is_unreadable():
+    ident = _identity(device_id=None, platform_id="ccfd73b3-aeb4-792a-1100-68f06f5d603b")
+    assert resolve_device_key(ident, "REAL-SERIAL", "10.0.0.1") == (
+        "ccfd73b3-aeb4-792a-1100-68f06f5d603b"
+    )
+
+
+def test_device_id_wins_over_platform_id():
+    """`pi` is platform-scoped, so a board hosting more than one logical OCF
+    device shares it -- the collision this exists to prevent."""
+    ident = _identity(device_id="dddddddd-0000-1111-2222-333333333333", platform_id="shared-plat")
+    assert resolve_device_key(ident, "REAL-SERIAL", "10.0.0.1") == (
+        "dddddddd-0000-1111-2222-333333333333"
+    )
+
+
+def test_falls_back_to_the_serial_then_the_host():
+    """A board that answers neither OCF resource lands exactly where it did
+    before any of this existed -- no regression for existing hardware."""
+    assert resolve_device_key(None, "REAL-SERIAL", "10.0.0.1") == "REAL-SERIAL"
+    assert resolve_device_key(_identity(), "REAL-SERIAL", "10.0.0.1") == "REAL-SERIAL"
+    # ...and a placeholder serial still resolves to the host (#83/#189).
+    assert resolve_device_key(_identity(), "Nothing(SVC)", "10.0.0.1") == "10.0.0.1"
+
+
+def test_the_key_is_case_normalized():
+    """The stored key is compared against a freshly polled one on every
+    poll; firmware that changed case between reads would otherwise look
+    like a different appliance every time."""
+    ident = _identity(device_id="CCFD73B3-AEB4-792A-1100-68F06F5D603B")
+    assert resolve_device_key(ident, None, "10.0.0.1") == "ccfd73b3-aeb4-792a-1100-68f06f5d603b"
+
+
+def test_the_nil_uuid_is_not_an_identity():
+    """OCF's unset UUID is identical on every unit that never had one
+    assigned -- the #189 failure mode on a new field. Its dashes stop
+    is_placeholder_serial's repeated-digit rule from seeing it, so it needs
+    its own check; falling through to the serial is the right answer."""
+    ident = _identity(device_id="00000000-0000-0000-0000-000000000000")
+    assert resolve_device_key(ident, "REAL-SERIAL", "10.0.0.1") == "REAL-SERIAL"
+    assert not is_usable_device_id("00000000-0000-0000-0000-000000000000")
+
+
+def test_known_junk_disqualifies_a_uuid_the_same_way_it_does_a_serial():
+    """A board firmware-flashed with 'Nothing(SVC)' in one identity field is
+    not a board to trust in another."""
+    assert not is_usable_device_id("Nothing(SVC)")
+    assert not is_usable_device_id("FFFFFFFFFFFFFFF")
+    assert not is_usable_device_id("")
+    assert not is_usable_device_id(None)
+    assert is_usable_device_id("3771f8bf-c184-3a2d-d885-e4c9818736d2")
+
+
+def test_ocf_device_key_reports_absence_rather_than_collapsing_to_the_serial():
+    """The coordinator needs "the device said nothing" and "the device said
+    this" to be different answers, so it never demotes a UUID-keyed entry
+    onto a serial because one poll couldn't read /oic/d."""
+    assert ocf_device_key(None) is None
+    assert ocf_device_key(_identity(serial="REAL-SERIAL")) is None
+    assert ocf_device_key(_identity(device_id="abc-123")) == "abc-123"

tests/test_range_hood_fan.py

@@ -11,7 +11,7 @@ from tests.conftest import _load_device
 
 
 class _FakeCoordinator:
-    device_serial = "TEST-HOOD-SERIAL"
+    device_key = "TEST-HOOD-SERIAL"
     device_info: ClassVar[dict] = {}
     data: ClassVar[dict] = {}
 

tests/test_redact.py

@@ -75,9 +75,18 @@ def test_redact_resources_does_not_mutate_input():
     assert resources["/information/vs/0"]["x.com.samsung.da.serialNum"] == original_serial
 
 
-def test_redacts_bare_ocf_identity_keys():
-    """/oic/d and /oic/p identify the unit with two-letter keys ('di', 'pi')
-    that the substring rules can't see."""
+def test_keeps_ocf_identity_uuids_but_redacts_the_owner_set_name():
+    """/oic/d's `di` and /oic/p's `pi` survive redaction.
+
+    They are randomly-assigned per-unit UUIDs, not account data, and they
+    are what the entry's registry keys are minted from (issue #381) -- a
+    report that blanks them hides the identity every entity in it is named
+    after, and can't answer the one question a duplicate-serial report
+    exists to ask: whether two units differ here at all.
+
+    `n` is the opposite case and stays redacted: free text the owner sets
+    from the SmartThings app, so it can carry a person's name.
+    """
     redacted = redact_resources(
         {
             "/oic/d": {
@@ -89,12 +98,10 @@ def test_redacts_bare_ocf_identity_keys():
         }
     )
 
-    assert redacted["/oic/d"]["di"] == REDACTED
-    assert redacted["/oic/p"]["pi"] == REDACTED
-    # 'n' is free text the owner sets from the SmartThings app, so it can
-    # carry a person's name -- redacted too. `rt`, the device-type signal
-    # we actually want out of /oic/d, is not.
+    assert redacted["/oic/d"]["di"] == "ab-cd-ef"
+    assert redacted["/oic/p"]["pi"] == "12-34-56"
     assert redacted["/oic/d"]["n"] == REDACTED
+    # `rt`, the device-type signal we actually want out of /oic/d, is kept.
     assert redacted["/oic/d"]["rt"] == ["oic.wk.d", "oic.d.refrigerator"]
     assert redacted["/oic/p"]["mnmo"] == "RF9000B"
 

tests/test_rekey_statistics_end_to_end.py

@@ -0,0 +1,171 @@
+"""Long-term statistics survive the move onto the OCF device UUID (#381).
+
+The identity migration rewrites entity registry rows in place rather than
+letting Home Assistant create replacements. The reason that matters most to
+an existing user is history: statistics are keyed by `statistic_id`, which
+for a sensor *is* its entity_id, so an entity that came back as
+`sensor.foo_2` would leave years of recorded data stranded under a name
+nothing writes to any more -- silently, with no error and no repair.
+
+tests/localthings/test_identity_migration.py proves the entity_id is
+preserved. It cannot prove that preserving it preserves the history,
+because no recorder is running there. This drives a real in-memory recorder
+end to end: statistics recorded, entry re-keyed, values read back.
+
+Lives here rather than under tests/localthings/ for the same reason
+test_statistics_migration_end_to_end.py does: that package's autouse
+`enable_custom_integrations` fixture depends on `hass`, which starts Home
+Assistant before `recorder_mock` can claim its database URL. Nothing here
+loads the integration -- `rekey_entry` is called directly.
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from functools import partial
+from typing import cast
+
+import pytest
+from homeassistant.components.recorder.models import StatisticMeanType
+from homeassistant.components.recorder.statistics import (
+    async_import_statistics,
+    get_metadata,
+    statistics_during_period,
+)
+from homeassistant.components.recorder.util import get_instance
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers import entity_registry as er
+from homeassistant.util import dt as dt_util
+from pytest_homeassistant_custom_component.common import MockConfigEntry
+from pytest_homeassistant_custom_component.components.recorder.common import (
+    async_wait_recording_done,
+)
+
+from custom_components.localthings.const import CONF_HOST, CONF_SERIAL, DOMAIN
+from custom_components.localthings.rekey import rekey_entry
+
+OLD_KEY = "BS7SP9AW400114A"  # the shared serial from issue #381
+NEW_KEY = "ccfd73b3-aeb4-792a-1100-68f06f5d603b"  # that unit's own /oic/d `di`
+RECORDED = [11.0, 9.0, 14.0]
+
+
+async def _seed_statistics(hass: HomeAssistant, entity_id: str) -> None:
+    start = dt_util.utcnow().replace(minute=0, second=0, microsecond=0) - timedelta(hours=4)
+    async_import_statistics(
+        hass,
+        {
+            "mean_type": StatisticMeanType.ARITHMETIC,
+            "has_sum": False,
+            "name": None,
+            "source": "recorder",
+            "statistic_id": entity_id,
+            "unit_class": None,
+            "unit_of_measurement": None,
+        },
+        [
+            {"start": start + timedelta(hours=i), "mean": value, "min": value, "max": value}
+            for i, value in enumerate(RECORDED)
+        ],
+    )
+    await async_wait_recording_done(hass)
+
+
+async def _means(hass: HomeAssistant, entity_id: str) -> list[float]:
+    rows = await get_instance(hass).async_add_executor_job(
+        statistics_during_period,
+        hass,
+        dt_util.utcnow() - timedelta(days=1),
+        None,
+        {entity_id},
+        "hour",
+        None,
+        {"mean"},
+    )
+    return [cast(float, row["mean"]) for row in rows.get(entity_id, [])]
+
+
+@pytest.fixture
+def purifier_entry(hass: HomeAssistant) -> MockConfigEntry:
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_HOST: "192.168.0.3", CONF_SERIAL: OLD_KEY},
+        unique_id=f"{DOMAIN}_{OLD_KEY}",
+        version=3,
+    )
+    entry.add_to_hass(hass)
+    return entry
+
+
+async def test_recorded_history_survives_the_re_key(
+    recorder_mock, hass: HomeAssistant, purifier_entry: MockConfigEntry
+) -> None:
+    """The claim the whole in-place rewrite exists to make good on."""
+    dev_reg = dr.async_get(hass)
+    ent_reg = er.async_get(hass)
+    device = dev_reg.async_get_or_create(
+        config_entry_id=purifier_entry.entry_id, identifiers={(DOMAIN, OLD_KEY)}
+    )
+    dust = ent_reg.async_get_or_create(
+        "sensor",
+        DOMAIN,
+        f"{DOMAIN}_{OLD_KEY}_dust",
+        config_entry=purifier_entry,
+        device_id=device.id,
+        suggested_object_id="bedroom_purifier_dust",
+    )
+    await _seed_statistics(hass, dust.entity_id)
+    assert await _means(hass, dust.entity_id) == RECORDED
+
+    rekey_entry(hass, purifier_entry, OLD_KEY, NEW_KEY)
+    await async_wait_recording_done(hass)
+
+    # The row moved to the new identity...
+    moved = ent_reg.async_get(dust.entity_id)
+    assert moved is not None
+    assert moved.unique_id == f"{DOMAIN}_{NEW_KEY}_dust"
+    # ...without moving the entity_id, which is what the statistics are
+    # filed under -- so the history is still there, unchanged and still
+    # attached to the entity the user sees.
+    assert moved.entity_id == "sensor.bedroom_purifier_dust"
+    assert await _means(hass, dust.entity_id) == RECORDED
+
+    # The metadata row is still filed under this entity_id too, so the
+    # recorder has not quietly started a second series alongside it.
+    metadata = await get_instance(hass).async_add_executor_job(
+        partial(get_metadata, hass, statistic_ids={dust.entity_id})
+    )
+    assert set(metadata) == {dust.entity_id}
+
+
+async def test_a_removed_duplicate_does_not_take_the_surviving_history_with_it(
+    recorder_mock, hass: HomeAssistant, purifier_entry: MockConfigEntry
+) -> None:
+    """Where the destination key is already taken, the old-key row is deleted
+    rather than rewritten. The history belongs to whichever entity_id the
+    user has been looking at all along -- the surviving row -- so deleting
+    the dead duplicate must not disturb it.
+
+    This is the one path in the re-key that destroys a registry row, so it
+    is the one worth proving keeps its hands off the recorder.
+    """
+    ent_reg = er.async_get(hass)
+    live = ent_reg.async_get_or_create(
+        "sensor",
+        DOMAIN,
+        f"{DOMAIN}_{NEW_KEY}_dust",
+        config_entry=purifier_entry,
+        suggested_object_id="bedroom_purifier_dust",
+    )
+    stale = ent_reg.async_get_or_create(
+        "sensor", DOMAIN, f"{DOMAIN}_{OLD_KEY}_dust", config_entry=purifier_entry
+    )
+    assert stale.entity_id != live.entity_id
+    await _seed_statistics(hass, live.entity_id)
+
+    rekey_entry(hass, purifier_entry, OLD_KEY, NEW_KEY)
+    await async_wait_recording_done(hass)
+
+    assert ent_reg.async_get(stale.entity_id) is None
+    assert ent_reg.async_get(live.entity_id) is not None
+    assert await _means(hass, live.entity_id) == RECORDED

tests/test_select_options.py

@@ -18,7 +18,7 @@ from custom_components.localthings.select import LocalThingsSelect
 
 
 class _FakeCoordinator:
-    device_serial = "TEST-SERIAL"
+    device_key = "TEST-SERIAL"
 
     def __init__(self, last_resources):
         self.last_resources = last_resources

tests/test_sensor_enum_options.py

@@ -46,7 +46,7 @@ class _FakeConfigEntry:
 
 class _FakeCoordinator:
     def __init__(self):
-        self.device_serial = "TEST-SERIAL"
+        self.device_key = "TEST-SERIAL"
         self.config_entry = _FakeConfigEntry()
         self.resources: dict[str, dict] = {}
 

tests/test_sensor_hysteresis.py

@@ -23,7 +23,7 @@ class _FakeCoordinator:
     """Just enough surface for LocalThingsEntity/LocalThingsSensor."""
 
     def __init__(self, threshold_minutes):
-        self.device_serial = "TEST-SERIAL"
+        self.device_key = "TEST-SERIAL"
         self.config_entry = _FakeConfigEntry(
             {
                 CONF_FINISH_TIME_HYSTERESIS_MINUTES: threshold_minutes,

tests/test_sensor_sticky.py

@@ -50,7 +50,7 @@ class _FakeCoordinator:
     """
 
     def __init__(self):
-        self.device_serial = "TEST-SERIAL"
+        self.device_key = "TEST-SERIAL"
         self.config_entry = _FakeConfigEntry()
         self.resources: dict[str, dict] = {}
 

tests/test_subdevice_discovery.py

@@ -147,7 +147,7 @@ async def test_pattern_a_sub1_device_info_links_via_device_to_master(hass: HomeA
     sub1 = next(su for su in coordinator.subdevices if su.key == "1")
     info = coordinator.device_info_for(sub1)
 
-    master_serial = coordinator.device_serial
+    master_serial = coordinator.device_key
     assert info["identifiers"] == {(DOMAIN, f"{master_serial}_1")}
     assert info["via_device"] == (DOMAIN, master_serial)
     # The subdevice's own /information/vs/1 (real, ARTIK051_DONGLE_FAC_RAC_18K)
@@ -245,7 +245,7 @@ async def test_fac_bora_2in1_subdevice_device_info(hass: HomeAssistant):
     subdevice = coordinator.subdevices[0]
     info = coordinator.device_info_for(subdevice)
 
-    master_serial = coordinator.device_serial
+    master_serial = coordinator.device_key
     assert info["identifiers"] == {(DOMAIN, f"{master_serial}_{_SUB_UUID}")}
     assert info["via_device"] == (DOMAIN, master_serial)
     # Confirmed live by the reporter (DESIGN-177.md section 1): the wall
@@ -266,7 +266,7 @@ async def test_fac_bora_2in1_unique_ids_include_subdevice_prefix(hass: HomeAssis
     entity = LocalThingsEntity(coordinator, sub_climate)
     expected_slug = _SUB_UUID.replace("-", "")
     assert entity._attr_unique_id == (
-        f"{DOMAIN}_{coordinator.device_serial}_subdevice_{expected_slug}_climate"
+        f"{DOMAIN}_{coordinator.device_key}_subdevice_{expected_slug}_climate"
     )
 
 

tests/test_water_heater_ehs.py

@@ -20,7 +20,7 @@ from tests.conftest import _load_device
 
 
 class _FakeCoordinator:
-    device_serial = "TEST-EHS-SERIAL"
+    device_key = "TEST-EHS-SERIAL"
     device_info: ClassVar[dict] = {}
     data: ClassVar[dict] = {}