M8-R15: fix HA discovery identities and thermal totals

app/integrations/expose.py

@@ -45,12 +45,15 @@ class DeviceInfo:
     """Grouping metadata that maps an entity to a logical HA device.
 
     ``identifiers`` corresponds to ``device.identifiers`` in HA Discovery
-    config — a stable tuple used to group entities under one device card.
+    config.  HA merges device cards if *any* identifier overlaps, so every
+    logical device deliberately has exactly one, globally namespaced value.
+    ``identity`` is an independent internal seed for MQTT topics and unique
+    ids; it must never be inferred from the HA identifier list.
     ``name`` is the human-readable device name (friendly_name).
     """
 
     identifiers: tuple[str, ...]
-    """Stable identifiers for HA device grouping (e.g. ``("modbus", "<uuid>")``).
+    """Stable HA identifiers, normally a one-item tuple.
 
     These must not change over the device lifetime; they anchor the HA device
     card even when friendly_name changes.
@@ -59,6 +62,14 @@ class DeviceInfo:
     name: str
     """Human-readable device name (may change; triggers HA discovery re-publish)."""
 
+    identity: Optional[str] = None
+    """Stable internal topic/unique-id identity, separate from HA identifiers.
+
+    The fallback preserves compatibility for third-party providers still using
+    the historic ``("kind", "uuid")`` construction while first-party providers
+    use this field explicitly.
+    """
+
     provides_availability: bool = True
     """Whether this device publishes an availability ("online"/"offline") heartbeat.
 
@@ -85,6 +96,11 @@ class DeviceInfo:
     )
     """Return whether the source behind this device is currently usable."""
 
+    @property
+    def internal_identity(self) -> str:
+        """Return the topic/unique-id seed without exposing HA grouping details."""
+        return self.identity or self.identifiers[-1]
+
 
 @dataclass
 class ExposableEntity:
@@ -276,8 +292,9 @@ def _modbus_provider(session: Session) -> list[ExposableEntity]:
 
     for device in devices:
         device_info = DeviceInfo(
-            identifiers=("modbus", device.uuid),
+            identifiers=(f"home-automation:modbus:{device.uuid}",),
             name=device.friendly_name,
+            identity=device.uuid,
         )
 
         # Load the profile to get metric metadata.
@@ -407,9 +424,10 @@ def _energy_cost_provider(session: Session) -> list[ExposableEntity]:
 
     HA device identity (换表 → 新 sensor)
     --------------------------------------
-    ``identifiers[1]`` is set to the active meter's **uuid** (not the fixed
+    The independent internal identity is set to the active meter's **uuid**.
-    string ``"energy-cost"``).  ``ha_discovery.py`` uses ``identifiers[1]`` as
+    ``ha_discovery.py`` uses that internal identity as the MQTT node_id and as
-    the MQTT node_id and as part of the ``unique_id`` for every entity.
+    part of the ``unique_id`` for every entity; HA receives one separate,
+    namespaced device identifier.
     Declaring a new active electricity meter produces a new uuid → new node_id /
     unique_id → HA creates a brand-new sensor, cleanly isolating post-swap data.
 
@@ -460,9 +478,9 @@ def _energy_cost_provider(session: Session) -> list[ExposableEntity]:
 
     DeviceInfo identifiers
     ----------------------
-    **Two-element tuple** ``("energy-cost", meter.uuid)`` so that
+    One-element namespaced tuple ``("home-automation:energy-cost:<uuid>",)``.
-    ``ha_discovery.py``'s ``entity.device.identifiers[1]`` resolves to the
+    The meter uuid used for MQTT topics and unique IDs is carried independently
-    meter uuid (used as the MQTT node_id and unique_id seed throughout).
+    in ``DeviceInfo.identity``.
     """
     from app.models.energy import EnergyCostPeriod, Meter  # local import to avoid circular
     from sqlalchemy import select
@@ -499,14 +517,15 @@ def _energy_cost_provider(session: Session) -> list[ExposableEntity]:
         currency = latest_period.currency
 
     # --- Shared DeviceInfo anchored to the active meter's uuid ---
-    # identifiers[1] = meter.uuid drives the MQTT node_id and unique_id in
+    # The internal identity is meter.uuid, while the HA identifier is a single
-    # ha_discovery.py.  Swapping the meter produces a new uuid → new HA sensor.
+    # namespaced value.  Swapping the meter produces a new HA sensor/card.
     # provides_availability=False: the energy-cost device has only sensors and no
     # online/offline heartbeat, so its entities must be "always available" in HA.
     # (Otherwise HA shows them unavailable despite state being published.)
     device_info = DeviceInfo(
-        identifiers=("energy-cost", active_meter.uuid),
+        identifiers=(f"home-automation:energy-cost:{active_meter.uuid}",),
         name=active_meter.label,
+        identity=active_meter.uuid,
         provides_availability=False,
     )
 
@@ -967,7 +986,8 @@ def _m8_energy_provider(session: Session) -> list[ExposableEntity]:
 
     for source in sources:
         source_info = DeviceInfo(
-            identifiers=("meter-source", source.uuid), name=source.name,
+            identifiers=(f"home-automation:meter-source:{source.uuid}",), name=source.name,
+            identity=source.uuid,
             availability_id=source.uuid,
             availability_getter=lambda sess, source_id=source.id: _source_online_by_id(sess, source_id),
         )
@@ -994,7 +1014,8 @@ def _m8_energy_provider(session: Session) -> list[ExposableEntity]:
             continue
         binding, channel, source = bound
         info = DeviceInfo(
-            identifiers=("meter", meter.uuid), name=meter.label,
+            identifiers=(f"home-automation:meter:{meter.uuid}",), name=meter.label,
+            identity=meter.uuid,
             # Keep this opaque and Meter-anchored.  In particular, do not use a
             # source UUID here: two channels of one source can be independently
             # stale/invalid and must not overwrite each other's availability.
@@ -1021,15 +1042,20 @@ def _m8_energy_provider(session: Session) -> list[ExposableEntity]:
 
     heating, hot_water = active_by_commodity.get("heating"), active_by_commodity.get("hot_water")
     if heating is not None and hot_water is not None:
+        # Keep the v1.6.1 canonical identity/key seed stable.  Discovery topic
+        # segments are sanitised independently by ha_discovery, so this dot is
+        # never emitted in a topic while existing toggle rows remain usable.
         identity = ".".join(sorted((heating.uuid, hot_water.uuid)))
         currency = _thermal_currency(session)
         cost_info = DeviceInfo(
-            identifiers=("thermal-cost", identity), name="Thermal Energy Cost",
+            identifiers=(f"home-automation:thermal-cost:{identity}",), name="Thermal Energy Cost",
+            identity=identity,
             provides_availability=False,
         )
         labels = {
             "heating": "Heating", "hot_water_heating": "Hot Water Heating", "water": "Water",
-            "water_tax": "Water Tax", "fixed": "Fixed", "all_in": "All-in",
+            "water_tax": "Water Tax", "hot_water_total": "Hot Water",
+            "fixed": "Fixed", "all_in": "All-in",
         }
         for suffix, window in (("total", None), ("today", "today")):
             for metric, label in labels.items():
@@ -1203,6 +1229,8 @@ def _thermal_cost_getter(metric: str, window: str | None) -> Callable[[Session],
             return result["breakdown"]["hot_water"]
         if metric == "water_tax":
             return result["breakdown"]["hot_water_tax"]
+        if metric == "hot_water_total":
+            return result["breakdown"]["hot_water_heating"] + result["breakdown"]["hot_water"]
         return result["breakdown"][metric]
     return _getter
 

app/integrations/homeassistant.py

@@ -2,10 +2,14 @@ from __future__ import annotations
 
 import json
 import logging
+from time import monotonic
 from dataclasses import dataclass, field
 from typing import Any
 from urllib import error, parse, request
 
+from websockets.exceptions import WebSocketException
+from websockets.sync.client import connect
+
 from app.config import Settings
 
 logger = logging.getLogger(__name__)
@@ -57,6 +61,81 @@ class HomeAssistantClient:
 
         self._post_json(f"/api/webhook/{webhook_id}", body, operation="trigger_webhook")
 
+    def discovery_registry_bindings(self, unique_ids: set[str]) -> dict[str, set[str]]:
+        """Return HA device identifiers currently bound to MQTT unique IDs.
+
+        This is deliberately a read-only WebSocket query.  MQTT only confirms
+        broker receipt; the entity/device registries are the authoritative HA
+        observation that a discovery unload/re-add was actually processed.
+        """
+        self._require_config()
+        if not unique_ids:
+            return {}
+        try:
+            deadline = monotonic() + self.timeout_seconds
+            with connect(self._websocket_url(), open_timeout=self.timeout_seconds,
+                         close_timeout=self.timeout_seconds) as websocket:
+                greeting = json.loads(self._websocket_recv(websocket, deadline))
+                if greeting.get("type") != "auth_required":
+                    raise HomeAssistantRequestError("Unexpected Home Assistant WebSocket greeting")
+                websocket.send(json.dumps({"type": "auth", "access_token": self.settings.home_assistant_auth_token}))
+                auth = json.loads(self._websocket_recv(websocket, deadline))
+                if auth.get("type") != "auth_ok":
+                    raise HomeAssistantRequestError("Home Assistant WebSocket authentication failed")
+                entities = self._websocket_command(websocket, 1, "config/entity_registry/list", deadline)
+                devices = self._websocket_command(websocket, 2, "config/device_registry/list", deadline)
+        except (OSError, WebSocketException, TimeoutError, ValueError, KeyError, TypeError) as exc:
+            raise HomeAssistantRequestError("Home Assistant registry query failed") from exc
+
+        devices_by_id = {
+            device["id"]: {
+                identifier[1]
+                for identifier in device.get("identifiers", [])
+                if (
+                    isinstance(identifier, (list, tuple))
+                    and len(identifier) == 2
+                    and identifier[0] == "mqtt"
+                    and isinstance(identifier[1], str)
+                    and identifier[1]
+                )
+            }
+            for device in devices
+            if isinstance(device, dict) and isinstance(device.get("id"), str)
+        }
+        return {
+            entity["unique_id"]: devices_by_id.get(entity.get("device_id"), set())
+            for entity in entities
+            if entity.get("platform") == "mqtt" and entity.get("unique_id") in unique_ids
+        }
+
+    @staticmethod
+    def _websocket_recv(websocket: Any, deadline: float) -> str:
+        remaining = deadline - monotonic()
+        if remaining <= 0:
+            raise TimeoutError("Home Assistant WebSocket registry query timed out")
+        return websocket.recv(timeout=remaining)
+
+    @classmethod
+    def _websocket_command(
+        cls, websocket: Any, message_id: int, command: str, deadline: float
+    ) -> list[dict[str, Any]]:
+        websocket.send(json.dumps({"id": message_id, "type": command}))
+        while True:
+            response = json.loads(cls._websocket_recv(websocket, deadline))
+            if response.get("id") != message_id:
+                continue
+            if not response.get("success"):
+                raise HomeAssistantRequestError(f"Home Assistant WebSocket {command} failed")
+            return response.get("result", [])
+
+    def _websocket_url(self) -> str:
+        parsed = parse.urlsplit(self.settings.home_assistant_base_url)
+        if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+            raise HomeAssistantConfigError("HOME_ASSISTANT_BASE_URL must be an HTTP(S) URL")
+        scheme = "wss" if parsed.scheme == "https" else "ws"
+        path = f"{parsed.path.rstrip('/')}/api/websocket"
+        return parse.urlunsplit((scheme, parsed.netloc, path, "", ""))
+
     def _require_config(self) -> None:
         if self.is_configured():
             return

app/integrations/meter_sources.py

@@ -98,8 +98,10 @@ def _validate_field_value(kind: str, field: SourceConfigField, value: Any) -> No
     _check_type(field, value)
     if field.name == "path" and not value.startswith("/dev/"):
         raise SourceProfileError("warmtelink_serial config path must start with '/dev/'.")
-    if field.name in {"broker_port", "sample_interval_s", "baudrate"} and value <= 0:
+    if field.name in {"broker_port", "baudrate"} and value <= 0:
         raise SourceProfileError(f"Config field {field.name!r} must be greater than zero.")
+    if field.name == "sample_interval_s" and value < 0:
+        raise SourceProfileError("Config field 'sample_interval_s' must not be negative.")
     if field.name == "data_bits" and value != 7:
         raise SourceProfileError("warmtelink_serial data_bits must be 7.")
     if field.name == "parity" and value != "N":

app/integrations/mqtt.py

@@ -174,23 +174,33 @@ class MqttManager:
         *,
         retain: bool = False,
         qos: int = 0,
-    ) -> None:
+    ) -> bool:
         """Publish a message to *topic*.
 
-        If the client is not connected the call is silently skipped.
+        Return whether paho accepted the message for publication.  If the
-        Errors are logged but do not raise.
+        client is not connected, or paho raises/rejects it, return ``False``;
+        errors still do not escape this best-effort boundary.
         """
         with self._lock:
             client = self._client
 
         if client is None or not self._connected:
             logger.debug("MQTT publish skipped — not connected (topic=%s).", topic)
-            return
+            return False
 
         try:
-            client.publish(topic, payload=payload, qos=qos, retain=retain)
+            result = client.publish(topic, payload=payload, qos=qos, retain=retain)
+            # paho returns MQTTMessageInfo with an integer rc.  Keep fakes and
+            # alternate clients compatible by treating an absent/non-int rc as
+            # accepted after the call itself succeeded.
+            rc = getattr(result, "rc", None)
+            if isinstance(rc, int) and rc != mqtt.MQTT_ERR_SUCCESS:
+                logger.warning("MQTT publish rejected (topic=%s, rc=%s).", topic, rc)
+                return False
+            return True
         except Exception:
             logger.exception("MQTT publish error (topic=%s).", topic)
+            return False
 
     def subscribe(self, topic: str, handler: Callable[[bytes], None]) -> None:
         """Register *handler* to be called when a message arrives on *topic*.

app/main.py

@@ -38,7 +38,7 @@ from app.services.config_page import build_runtime_settings, seed_missing_config
 from app.services.dsmr_ingest import apply_dsmr_subscription
 from app.services.public_ip import check_public_ipv4_and_notify
 from app.services.modbus_poll import poll_all_enabled_devices, BASE_POLL_TICK_SECONDS
-from app.services.ha_discovery import publish_discovery, publish_states
+from app.services.ha_discovery import initialize_legacy_thermal_cleanup, publish_discovery, publish_states
 from app.services.tibber_prices import run_tibber_refresh_best_effort
 from app.services.energy_cost import compute_closed_periods
 from app.services.meter_cost import compute_closed_periods as compute_closed_meter_cost_periods
@@ -203,6 +203,7 @@ def ensure_auth_db_ready() -> None:
         initialize_auth_schema(session, get_settings())
         seed_missing_config_from_bootstrap(session, get_settings())
         sync_app_hostname_from_bootstrap(session, get_settings())
+        initialize_legacy_thermal_cleanup(session)
     except AppDatabaseAdoptionError as exc:
         raise RuntimeError(str(exc)) from exc
     except AuthBootstrapError as exc:

app/services/ha_discovery.py

@@ -34,16 +34,30 @@ from __future__ import annotations
 
 import json
 import logging
+import re
+from datetime import UTC, datetime
 from typing import Any
 
 from sqlalchemy.orm import Session
+from sqlalchemy import inspect
 
 from app.integrations.expose import ExposableEntity, build_catalog
+from app.models.config import AppConfigEntry
 from app.integrations.mqtt import mqtt_manager
 from app.services.config_page import build_runtime_settings
 
 logger = logging.getLogger(__name__)
 
+_HA_SEGMENT_RE = re.compile(r"[^A-Za-z0-9_-]+")
+# These durable, versioned application metadata entries prevent the minute
+# publisher from repeatedly emitting already accepted known-bad v1.6.1 thermal
+# topics.  A failed publish is not acknowledged and remains retryable across
+# restarts.  They are intentionally application metadata, rather than broker
+# state.  An empty retained publish to an illegal v1.6.1 topic cannot be used
+# as an acknowledgement: HA rejects that topic before it processes its payload.
+_LEGACY_THERMAL_CLEANUP_KEY = "HA_DISCOVERY_LEGACY_THERMAL_CLEANUP_V1"
+_REGISTRY_REPAIR_KEY = "HA_DISCOVERY_REGISTRY_REPAIR_V2"
+
 
 # ---------------------------------------------------------------------------
 # Internal helpers
@@ -59,14 +73,25 @@ def _should_publish(settings: Any) -> bool:
     )
 
 
-def _node_id(device_uuid: str) -> str:
+def _safe_segment(value: str) -> str:
-    """Stable MQTT node_id derived from device uuid (replace hyphens for safety)."""
+    """Return a deterministic HA discovery topic segment.
-    return device_uuid.replace("-", "_")
+
+    Home Assistant accepts only ``[A-Za-z0-9_-]+`` for node/object ids.
+    Replacing each run of other characters keeps ordinary UUID/key output
+    readable while ensuring composite identities cannot produce invalid topics.
+    """
+    result = _HA_SEGMENT_RE.sub("_", value).strip("_")
+    return result or "home_automation"
+
+
+def _node_id(identity: str) -> str:
+    """Stable, strictly valid MQTT node_id from an internal identity."""
+    return _safe_segment(identity.replace("-", "_"))
 
 
 def _object_id(entity: ExposableEntity) -> str:
-    """Stable MQTT object_id derived from entity key (dots + hyphens → underscores)."""
+    """Stable, strictly valid MQTT object_id derived from the entity key."""
-    return entity.key.replace(".", "_").replace("-", "_")
+    return _safe_segment(entity.key.replace("-", "_"))
 
 
 def _discovery_topic(entity: ExposableEntity, prefix: str) -> str:
@@ -74,7 +99,7 @@ def _discovery_topic(entity: ExposableEntity, prefix: str) -> str:
 
     Format: ``<prefix>/<component>/<node_id>/<object_id>/config``
     """
-    node = _node_id(entity.device.identifiers[1])  # device uuid
+    node = _node_id(entity.device.internal_identity)
     obj = _object_id(entity)
     return f"{prefix}/{entity.component}/{node}/{obj}/config"
 
@@ -84,7 +109,7 @@ def _state_topic(entity: ExposableEntity, prefix: str) -> str:
 
     Format: ``<prefix>/<component>/<node_id>/<object_id>/state``
     """
-    node = _node_id(entity.device.identifiers[1])
+    node = _node_id(entity.device.internal_identity)
     obj = _object_id(entity)
     return f"{prefix}/{entity.component}/{node}/{obj}/state"
 
@@ -104,16 +129,133 @@ def _availability_id(entity: ExposableEntity) -> str:
     M8 meters deliberately retain their own UUID as HA node/unique identity,
     while their availability is supplied by a MeterSource UUID.
     """
-    return entity.device.availability_id or entity.device.identifiers[1]
+    return entity.device.availability_id or entity.device.internal_identity
 
 
 def _unique_id(entity: ExposableEntity) -> str:
     """Stable unique_id — device uuid + metric key (never from mutable fields)."""
-    device_uuid = entity.device.identifiers[1]
+    device_uuid = entity.device.internal_identity
     # entity.key is already "modbus.<uuid>.<metric_key>" — use it as the seed
     return f"{device_uuid}_{entity.key.replace('.', '_')}"
 
 
+def _migration_state(session: Session, key: str, default: str) -> str:
+    """Read a versioned discovery-repair acknowledgement from ``app_config``."""
+    entry = session.query(AppConfigEntry).filter(AppConfigEntry.key == key).one_or_none()
+    return entry.value if entry is not None else default
+
+
+def _has_repair_store(session: Session) -> bool:
+    """Whether this session uses the app schema (not a lightweight unit DB)."""
+    return inspect(session.get_bind()).has_table("app_config")
+
+
+def _set_migration_state(session: Session, key: str, value: str) -> None:
+    """Durably acknowledge a completed discovery-repair step.
+
+    The state is operational metadata only; it never alters meters, readings,
+    contracts, costs, or expose toggles.
+    """
+    entry = session.query(AppConfigEntry).filter(AppConfigEntry.key == key).one_or_none()
+    if entry is None:
+        session.add(AppConfigEntry(key=key, value=value, updated_at=datetime.now(UTC)))
+    else:
+        entry.value = value
+        entry.updated_at = datetime.now(UTC)
+    session.commit()
+
+
+def _migration_json(session: Session, key: str) -> dict[str, Any]:
+    """Read a versioned JSON migration ledger, treating corrupt values as pending."""
+    raw = _migration_state(session, key, "{}")
+    try:
+        value = json.loads(raw)
+    except json.JSONDecodeError:
+        logger.warning("Ignoring malformed HA discovery migration ledger %s", key)
+        return {}
+    return value if isinstance(value, dict) else {}
+
+
+def _set_migration_json(session: Session, key: str, value: dict[str, Any]) -> None:
+    _set_migration_state(session, key, json.dumps(value, sort_keys=True))
+
+
+def _frozen_legacy_inventory(ledger: dict[str, Any]) -> list[str] | None:
+    """Return a structurally valid immutable cleanup inventory, if present.
+
+    The ledger predates a schema migration, so it must tolerate both an absent
+    entry and old ``topics``-only values.  An inventory becomes immutable only
+    once it is a list of non-empty topic strings; anything else is compatibility
+    data to be frozen during startup, not state a publisher may reinterpret.
+    """
+    value = ledger.get("inventory")
+    if not isinstance(value, list) or not all(isinstance(topic, str) and topic for topic in value):
+        return None
+    return list(dict.fromkeys(value))
+
+
+def _publish_ok(topic: str, payload: str | bytes, *, retain: bool = True) -> bool:
+    """Publish best-effort while treating only an explicit ``False`` as failure.
+
+    Production ``MqttManager.publish`` returns ``bool``.  Accepting ``None``
+    keeps existing third-party/mock publishers best-effort compatible.
+    """
+    return mqtt_manager.publish(topic, payload, retain=retain) is not False
+
+
+def initialize_legacy_thermal_cleanup(session: Session) -> None:
+    """Create the legacy-cleanup ledger before expose toggles can be edited.
+
+    A newly installed database has no v1.6.1 retained thermal topics.  Marking
+    that case complete during startup prevents a later first-time toggle from
+    manufacturing an old, illegal topic.  Conversely, an upgraded database
+    freezes its exact enabled legacy topics before the UI can change a toggle.
+    """
+    if not _has_repair_store(session):
+        return
+
+    ledger = _migration_json(session, _LEGACY_THERMAL_CLEANUP_KEY)
+    if ledger.get("complete") is True:
+        return
+
+    inventory = _frozen_legacy_inventory(ledger)
+    if inventory is not None:
+        # A previously frozen non-empty inventory must never be expanded or
+        # recomputed from mutable toggle/prefix state.  A zero-item inventory
+        # is just the fresh-install terminal state written by an older build.
+        if not inventory:
+            _set_migration_json(
+                session,
+                _LEGACY_THERMAL_CLEANUP_KEY,
+                {"complete": True, "inventory": [], "topics": []},
+            )
+        return
+
+    # This deliberately propagates.  Startup runs before the expose UI can
+    # write a toggle, so swallowing an enumeration or durable-write error would
+    # create a window in which the cleanup scope could be changed or lost.
+    legacy_entities = _legacy_thermal_entities(session)
+    from app.config import get_settings
+
+    settings = build_runtime_settings(session, get_settings())
+    inventory = list(dict.fromkeys(
+        _legacy_discovery_topic(entity, settings.ha_discovery_prefix)
+        for entity in legacy_entities
+    ))
+    _set_migration_json(
+        session,
+        _LEGACY_THERMAL_CLEANUP_KEY,
+        {
+            "complete": not inventory,
+            "inventory": inventory,
+            "topics": sorted({
+                topic for topic in ledger.get("topics", [])
+                if isinstance(topic, str) and topic in inventory
+            }),
+        },
+    )
+
+
 # ---------------------------------------------------------------------------
 # Public: build discovery payload
 # ---------------------------------------------------------------------------
@@ -220,34 +362,90 @@ def publish_discovery(session: Session) -> None:
         logger.exception("publish_discovery: failed to build catalog; aborting")
         return
 
-    # Meter UUIDs are intentionally identity-changing epochs.  Discovery config
+    # Clear the *illegal* v1.6.1 thermal keys once.  Startup freezes the exact
-    # is retained, so clear only the precisely enumerable old M8 identities;
+    # inventory while the legacy toggle state still describes v1.6.1.  Thus a
-    # never wildcard a provider/topic and risk removing another source's card.
+    # later UI toggle/prefix/meter change cannot erase or expand this repair.
+    # ``topics`` is only durable success progress; ``inventory`` is immutable.
+    # Startup freezes compatibility ledgers before the UI can mutate the
+    # toggle/prefix inputs.  The publisher only consumes that frozen snapshot.
+    repaired_keys: set[str] = set()
+    if _has_repair_store(session):
+        ledger = _migration_json(session, _LEGACY_THERMAL_CLEANUP_KEY)
+        if not ledger.get("complete"):
+            inventory = _frozen_legacy_inventory(ledger)
+            if inventory is None:
+                logger.error(
+                    "publish_discovery: legacy cleanup ledger was not frozen at startup; refusing cleanup"
+                )
+
+            if inventory is not None:
+                completed = {
+                    topic for topic in ledger.get("topics", [])
+                    if isinstance(topic, str) and topic in inventory
+                }
+                for topic in (topic for topic in inventory if topic not in completed):
+                    try:
+                        if _publish_ok(topic, b""):
+                            completed.add(topic)
+                            # Commit each accepted illegal-topic tombstone before
+                            # attempting another one: a crash must not replay it.
+                            _set_migration_json(
+                                session,
+                                _LEGACY_THERMAL_CLEANUP_KEY,
+                                {
+                                    "complete": False,
+                                    "inventory": inventory,
+                                    "topics": sorted(completed),
+                                },
+                            )
+                        else:
+                            logger.warning("publish_discovery: broker rejected v1.6.1 cleanup for %s", topic)
+                    except Exception:
+                        logger.exception("publish_discovery: unable to clear v1.6.1 topic %s", topic)
+                if set(inventory).issubset(completed):
+                    _set_migration_json(
+                        session,
+                        _LEGACY_THERMAL_CLEANUP_KEY,
+                        {"complete": True, "inventory": inventory, "topics": sorted(completed)},
+                    )
+
+        # Current-format stale topics are legal and deliberately remain retryable:
+        # they cover post-repair meter swaps and include all 14 thermal metrics.
         try:
             stale_entities = _stale_m8_entities(session)
         except Exception:
-        logger.exception("publish_discovery: unable to enumerate old M8 identities")
+            logger.exception("publish_discovery: unable to enumerate stale M8 identities")
             stale_entities = []
         for old_entity in stale_entities:
             try:
-            old_topic, _ = build_discovery_payload(old_entity, discovery_prefix, state_prefix)
+                if not _publish_ok(_discovery_topic(old_entity, discovery_prefix), b""):
-            mqtt_manager.publish(old_topic, b"", retain=True)
+                    logger.warning("publish_discovery: broker rejected stale cleanup for %r", old_entity.key)
             except Exception:
-            logger.exception("publish_discovery: unable to clear old identity %r", old_entity.key)
+                logger.exception("publish_discovery: unable to clear stale identity %r", old_entity.key)
+
+        # v1.6.1 used the same legal topics/unique_ids for Meter, Source, Modbus
+        # and electricity-cost entities, but attached their registry entries to
+        # wrongly merged HA devices.  HA does not move those entries when only a
+        # discovery ``device`` block changes.  One durable unload -> re-add cycle
+        # lets HA rebuild the existing unique_id entries against the new singleton
+        # identifiers, preserving entity keys, toggles and user customizations.
+        repaired_keys = _run_registry_repair(session, catalog, discovery_prefix, state_prefix, settings)
 
     for entry in catalog:
         entity = entry.entity
+        if entity.key in repaired_keys:
+            continue
         try:
             topic, config = build_discovery_payload(entity, discovery_prefix, state_prefix)
             if entry.enabled:
                 payload = json.dumps(config)
-                mqtt_manager.publish(topic, payload, retain=True)
+                _publish_ok(topic, payload)
                 logger.debug(
                     "publish_discovery: published config for %r → %s", entity.key, topic
                 )
             else:
                 # Clear retained config for disabled entities.
-                mqtt_manager.publish(topic, b"", retain=True)
+                _publish_ok(topic, b"")
                 logger.debug(
                     "publish_discovery: cleared config for disabled entity %r → %s",
                     entity.key,
@@ -259,39 +457,163 @@ def publish_discovery(session: Session) -> None:
             )
 
 
-def _stale_m8_entities(session: Session) -> list[ExposableEntity]:
+def _is_registry_repair_entity(entity: ExposableEntity) -> bool:
-    """Return synthetic discovery entries for superseded thermal identities.
+    """Whether a legal v1.6.1 config needs an unload/re-add device split."""
+    return entity.key.startswith(("modbus.", "source.", "meter.", "energy.", "thermal_cost."))
 
-    This is deliberately a narrow, best-effort cleanup: ended Meter UUIDs and
+
-    historically possible thermal combinations only; current identities are excluded.
+def _ha_registry_bindings(settings: Any, unique_ids: set[str]) -> dict[str, set[str]] | None:
+    """Read HA's registry, returning ``None`` when it is not observable."""
+    from app.integrations.homeassistant import (
+        HomeAssistantClient,
+        HomeAssistantConfigError,
+        HomeAssistantRequestError,
+    )
+
+    # Runtime settings from old callers/tests may not expose outbound HA
+    # fields.  Treat absent/non-string credentials as intentionally optional.
+    if not isinstance(getattr(settings, "home_assistant_base_url", None), str) or not isinstance(
+        getattr(settings, "home_assistant_auth_token", None), str
+    ):
+        return None
+    client = HomeAssistantClient(settings)
+    if not client.is_configured():
+        return None
+    try:
+        return client.discovery_registry_bindings(unique_ids)
+    except (HomeAssistantConfigError, HomeAssistantRequestError, TypeError, ValueError):
+        logger.warning("HA registry repair remains pending: HA registry is unavailable")
+        return None
+
+
+def _run_registry_repair(
+    session: Session,
+    catalog: list[Any],
+    discovery_prefix: str,
+    state_prefix: str,
+    settings: Any,
+) -> set[str]:
+    """Repair each v1.6.1 entity only after HA confirms every phase.
+
+    MQTT acknowledges broker receipt, not Home Assistant processing.  The
+    ledger therefore holds each target at ``unloaded`` until HA's entity
+    registry no longer contains its stable unique_id, then at ``republished``
+    until HA reports the expected singleton device identifier.  A partial
+    catalog simply adds targets on a later run; it cannot complete others.
+    When HA's registry is unavailable, ordinary discovery remains untouched.
     """
-    from app.integrations.expose import DeviceInfo
+    targets = [entry for entry in catalog if _is_registry_repair_entity(entry.entity)]
+    if not targets:
+        return set()
+    ledger = _migration_json(session, _REGISTRY_REPAIR_KEY)
+    pending_targets = [
+        entry for entry in targets if ledger.get(_unique_id(entry.entity), {}).get("phase") != "complete"
+    ]
+    if not pending_targets:
+        return set()
+    unique_ids = {_unique_id(entry.entity) for entry in pending_targets}
+    bindings = _ha_registry_bindings(settings, unique_ids)
+    if bindings is None:
+        return set()
+    blocked: set[str] = set()
+    for entry in pending_targets:
+        entity = entry.entity
+        unique_id = _unique_id(entity)
+        expected = set(entity.device.identifiers)
+        phase = ledger.get(unique_id, {}).get("phase", "pending")
+        observed = bindings.get(unique_id)
+        if phase == "unloaded":
+            if observed is not None:
+                # HA was disconnected or otherwise missed the retained
+                # tombstone. Keep it retained until HA itself confirms delete.
+                try:
+                    _publish_ok(_discovery_topic(entity, discovery_prefix), b"")
+                except Exception:
+                    logger.exception("publish_discovery: unable to repeat registry unload for %r", entity.key)
+                blocked.add(entity.key)
+                continue
+            if not entry.enabled:
+                _set_registry_phase(session, ledger, unique_id, "complete", expected)
+                continue
+            _publish_registry_config(session, ledger, entry, discovery_prefix, state_prefix, expected, blocked)
+        elif phase == "republished":
+            if observed == expected:
+                _set_registry_phase(session, ledger, unique_id, "complete", expected)
+            elif observed is None and entry.enabled:
+                _publish_registry_config(session, ledger, entry, discovery_prefix, state_prefix, expected, blocked)
+            else:
+                # A stale/wrong device binding must pass through deletion again.
+                try:
+                    if _publish_ok(_discovery_topic(entity, discovery_prefix), b""):
+                        _set_registry_phase(session, ledger, unique_id, "unloaded", expected)
+                except Exception:
+                    logger.exception("publish_discovery: unable to unload wrong registry entity %r", entity.key)
+                blocked.add(entity.key)
+        elif phase != "complete":
+            if observed == expected or (observed is None and not entry.enabled):
+                _set_registry_phase(session, ledger, unique_id, "complete", expected)
+            elif observed is None:
+                _publish_registry_config(session, ledger, entry, discovery_prefix, state_prefix, expected, blocked)
+            else:
+                try:
+                    if _publish_ok(_discovery_topic(entity, discovery_prefix), b""):
+                        _set_registry_phase(session, ledger, unique_id, "unloaded", expected)
+                        blocked.add(entity.key)
+                    else:
+                        logger.warning("publish_discovery: broker rejected registry unload for %r", entity.key)
+                except Exception:
+                    logger.exception("publish_discovery: unable to unload registry entity %r", entity.key)
+    return blocked
+
+
+def _set_registry_phase(
+    session: Session, ledger: dict[str, Any], unique_id: str, phase: str, expected: set[str]
+) -> None:
+    ledger[unique_id] = {"phase": phase, "identifier": sorted(expected)}
+    _set_migration_json(session, _REGISTRY_REPAIR_KEY, ledger)
+
+
+def _publish_registry_config(
+    session: Session, ledger: dict[str, Any], entry: Any, discovery_prefix: str,
+    state_prefix: str, expected: set[str], blocked: set[str],
+) -> None:
+    entity = entry.entity
+    topic, config = build_discovery_payload(entity, discovery_prefix, state_prefix)
+    try:
+        if _publish_ok(topic, json.dumps(config)):
+            _set_registry_phase(session, ledger, _unique_id(entity), "republished", expected)
+        else:
+            logger.warning("publish_discovery: broker rejected registry re-add for %r", entity.key)
+    except Exception:
+        logger.exception("publish_discovery: unable to re-add registry entity %r", entity.key)
+    blocked.add(entity.key)
+
+
+def _legacy_discovery_topic(entity: ExposableEntity, prefix: str) -> str:
+    """Return the exact v1.6.1 config topic for a known stale M8 entity.
+
+    This intentionally preserves the old hyphen-only conversion because the
+    point is to remove that exact retained broker key once, never to publish a
+    wildcard or manufacture a new invalid topic.
+    """
+    node = entity.device.internal_identity.replace("-", "_")
+    obj = entity.key.replace(".", "_").replace("-", "_")
+    return f"{prefix}/{entity.component}/{node}/{obj}/config"
+
+
+def _overlapping_thermal_pairs(session: Session) -> list[tuple[Any, Any]]:
+    """Return only heating/water Meter epochs that could have coexisted."""
     from app.models.energy import Meter
     from sqlalchemy import select
 
     meters = session.execute(select(Meter).where(
         Meter.commodity.in_(("electricity", "heating", "hot_water"))
     )).scalars().all()
-    current = {meter.commodity: meter for meter in meters if meter.ended_at is None}
-    old = [meter for meter in meters if meter.ended_at is not None]
-    entities: list[ExposableEntity] = []
-    for meter in old:
-        info = DeviceInfo(identifiers=("meter", meter.uuid), name=meter.label)
-        for suffix in ("total", "today"):
-            entities.append(ExposableEntity(
-                key=f"meter.{meter.uuid}.{suffix}", component="sensor", device=info,
-                device_class=None, unit="", name="obsolete",
-            ))
     heatings = [meter for meter in meters if meter.commodity == "heating"]
     waters = [meter for meter in meters if meter.commodity == "hot_water"]
-    current_identity = (
+    pairs: list[tuple[Any, Any]] = []
-        ".".join(sorted((current["heating"].uuid, current["hot_water"].uuid)))
-        if current.get("heating") is not None and current.get("hot_water") is not None else None
-    )
     for heating in heatings:
         for water in waters:
-            if heating.ended_at is None and water.ended_at is None:
-                continue
             # A thermal identity can only have been published when both Meter
             # epochs were current at the same instant.  Do not form a Cartesian
             # product of historical records: that would tombstone identities
@@ -302,11 +624,28 @@ def _stale_m8_entities(session: Session) -> list[ExposableEntity]:
                 water_end is not None and heating_start >= water_end
             ):
                 continue
+            pairs.append((heating, water))
+    return pairs
+
+
+def _thermal_cleanup_entities(
+    pairs: list[tuple[Any, Any]], *, include_hot_water_total: bool
+) -> list[ExposableEntity]:
+    """Build exact synthetic config entries for known thermal identities."""
+    from app.integrations.expose import DeviceInfo
+
+    metrics = ["heating", "hot_water_heating", "water", "water_tax", "fixed", "all_in"]
+    if include_hot_water_total:
+        metrics.insert(2, "hot_water_total")
+    entities: list[ExposableEntity] = []
+    for heating, water in pairs:
         identity = ".".join(sorted((heating.uuid, water.uuid)))
-            if identity == current_identity:
+        info = DeviceInfo(
-                continue
+            identifiers=(f"home-automation:thermal-cost:{identity}",),
-            info = DeviceInfo(identifiers=("thermal-cost", identity), name="obsolete")
+            name="obsolete",
-            for metric in ("heating", "hot_water_heating", "water", "water_tax", "fixed", "all_in"):
+            identity=identity,
+        )
+        for metric in metrics:
             for suffix in ("total", "today"):
                 entities.append(ExposableEntity(
                     key=f"thermal_cost.{identity}.{metric}_{suffix}", component="sensor", device=info,
@@ -315,6 +654,59 @@ def _stale_m8_entities(session: Session) -> list[ExposableEntity]:
     return entities
 
 
+def _legacy_thermal_entities(session: Session) -> list[ExposableEntity]:
+    """Return only v1.6.1 illegal topics that could retain a non-empty config.
+
+    v1.6.1 published an illegal thermal config only while its expose toggle was
+    enabled.  A disabled toggle published its own tombstone, so querying the
+    durable toggle state prevents a fresh install from manufacturing warnings
+    for twelve never-used illegal topics.
+    """
+    from app.models.expose import ExposedEntityToggle
+
+    entities = _thermal_cleanup_entities(_overlapping_thermal_pairs(session), include_hot_water_total=False)
+    keys = [entity.key for entity in entities]
+    enabled_keys = {
+        row.key for row in session.query(ExposedEntityToggle).filter(
+            ExposedEntityToggle.key.in_(keys), ExposedEntityToggle.enabled.is_(True)
+        )
+    }
+    return [entity for entity in entities if entity.key in enabled_keys]
+
+
+def _stale_m8_entities(session: Session) -> list[ExposableEntity]:
+    """Return *safe-format* configs that became stale after a later epoch swap.
+
+    Unlike the v1.6.1 cleanup, this intentionally excludes the active thermal
+    pair and includes the post-repair ``hot_water_total`` metric (14 configs
+    per stale pair).  All generated topics use ``_discovery_topic``.
+    """
+    from app.integrations.expose import DeviceInfo
+    from app.models.energy import Meter
+    from sqlalchemy import select
+
+    meters = session.execute(select(Meter).where(
+        Meter.commodity.in_(("electricity", "heating", "hot_water"))
+    )).scalars().all()
+    entities: list[ExposableEntity] = []
+    for meter in (meter for meter in meters if meter.ended_at is not None):
+        info = DeviceInfo(
+            identifiers=(f"home-automation:meter:{meter.uuid}",), name=meter.label, identity=meter.uuid
+        )
+        for suffix in ("total", "today"):
+            entities.append(ExposableEntity(
+                key=f"meter.{meter.uuid}.{suffix}", component="sensor", device=info,
+                device_class=None, unit="", name="obsolete",
+            ))
+    stale_pairs = [
+        (heating, water)
+        for heating, water in _overlapping_thermal_pairs(session)
+        if heating.ended_at is not None or water.ended_at is not None
+    ]
+    entities.extend(_thermal_cleanup_entities(stale_pairs, include_hot_water_total=True))
+    return entities
+
+
 # ---------------------------------------------------------------------------
 # Public: publish states
 # ---------------------------------------------------------------------------
@@ -468,7 +860,7 @@ def publish_device_state(session: Session, device: Any) -> None:
             continue
         entity = entry.entity
         # Only process entities belonging to this device.
-        if entity.device.identifiers[1] != device_uuid:
+        if entity.device.internal_identity != device_uuid:
             continue
         # Skip the online binary_sensor itself (availability handled above).
         if entity.component == "binary_sensor" and "online" in entity.key:
@@ -543,7 +935,7 @@ def clear_device_discovery(session: Session, device_uuid: str) -> None:
     for entry in catalog:
         entity = entry.entity
         # Only clear entities belonging to this device.
-        if entity.device.identifiers[1] != device_uuid:
+        if entity.device.internal_identity != device_uuid:
             continue
         try:
             topic, _config = build_discovery_payload(entity, discovery_prefix, state_prefix)

docs/design/m8-warmtelink-energy.md

@@ -1214,6 +1214,10 @@ M8 收尾的前置条件。agent 不得执行、记录为已执行，或以 mock
 
 ## 14. Post-M8 lifecycle repair（M8-R08～R10）
 
+### M8-R15 — HA Discovery identity/topic repair
+
+修复 HA 将多值 `device.identifiers` 任一匹配合并的风险：Source、Meter、Modbus 与成本 epoch 均使用单一完整 namespaced identifier，内部 MQTT identity 独立保存。thermal 的既有点号 identity/key 保持兼容，node/object topic segment 单独规范化；仅对 v1.6.1 实际发布过的 dot topic 做可重试、成功后抑制的精确 retained cleanup。新增默认关闭的 `Thermal Hot Water Total/Today`，金额为 `hot_water_heating + hot_water`，不含 `hot_water_tax`。
+
 M8 交付后的 Meter lifecycle 修复链记录在本地 `review-notes/M8-meter-lifecycle-repair-plan.md`。
 它不新增 ORM / **数据库** schema 或 Alembic revision，不做启动自动修复、一次性数据脚本或历史删除；R08 虽然
 更新了 API/Pydantic schema 及 OpenAPI/codegen，但没有变更 ORM 或数据库 schema。已有的 stranded binding 只能

docs/homeassistant-outbound.md

@@ -55,7 +55,9 @@
 Expose 框架还可以把已勾选的 Energy 实体通过 MQTT Home Assistant Discovery 发布；开关位于应用 Config 页的 HA Expose 面板，默认均为关闭。M8 增加了 source online、按 Meter UUID 锚定的累计量/today，以及 heating、hot-water-heating、water、water-tax、fixed、all-in total/today 等 thermal 实体。
 
 - source 和 Meter identity 不依赖可变 label；换表会产生新 Meter UUID identity。
-- thermal 组合成本 identity 由当前 heating/hot_water Meter UUID 的有序组合锚定，任一换表都会产生新 identity，避免不同累计域拼接。
+- thermal 组合成本 identity 由当前 heating/hot_water Meter UUID 的有序组合锚定，任一换表都会产生新 identity，避免不同累计域拼接。`Thermal Hot Water Total/Today` 精确为 `hot_water_heating + hot_water`，不包含 `hot_water_tax`。
+- 每个 HA device 只发布一个完整、`home-automation:` namespaced identifier；内部 topic/unique-id seed 与该 identifier 分离。所有 discovery node/object segment 都会转为 HA 允许的 `[A-Za-z0-9_-]+` 字符集。
+- v1.6.1 的 dot-containing thermal retained topics 会在启动、UI 尚未能修改 toggle 前，按当时实际重叠的 Meter epoch、enabled toggle 和 runtime discovery prefix 冻结为精确清单；每个成功 topic 都会持久记账，失败/未尝试项才会重试。冻结清单完成后跨重启不再枚举或发布旧 topic；fresh install 的空清单也会立即完成。没有 wildcard，也不会清理任何应用数据。旧 device 合并修复则以已配置的 Home Assistant WebSocket entity/device registry 实际观察 unload、重建和正确 device identifier；HA 不可达时普通 discovery 继续发布，repair 保持 pending。
 - availability、unit、device/state class 与 today reset 由 provider 声明；operator 应在 HA 中核对，而不应假设同名实体可跨换表连续。
 - 关闭 toggle 后 retained discovery 会被清理；关闭暴露不删除 source、Meter、合同、读数或成本历史。
 

tests/test_app.py

@@ -118,6 +118,35 @@ def test_app_start_seeds_missing_config_from_env_without_overwriting_existing_va
     reset_db_caches()
 
 
+def test_startup_initializes_fresh_legacy_discovery_cleanup_ledger(
+    tmp_path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The cleanup ledger exists before the expose UI can write its first toggle."""
+    import app.main as main
+
+    app_database_url = _prepare_app_db(tmp_path)
+    monkeypatch.setenv("APP_DATABASE_URL", app_database_url)
+    monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin")
+    monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password")
+    get_settings.cache_clear()
+    reset_db_caches()
+
+    main.ensure_auth_db_ready()
+
+    conn = sqlite3.connect(tmp_path / "app_ready.db")
+    try:
+        value = conn.execute(
+            "SELECT value FROM app_config WHERE key = ?",
+            ("HA_DISCOVERY_LEGACY_THERMAL_CLEANUP_V1",),
+        ).fetchone()[0]
+    finally:
+        conn.close()
+    assert value == '{"complete": true, "inventory": [], "topics": []}'
+
+    get_settings.cache_clear()
+    reset_db_caches()
+
+
 def test_app_start_syncs_app_hostname_from_env_even_when_db_has_old_value(
     tmp_path, monkeypatch: pytest.MonkeyPatch
 ) -> None:

tests/test_dsmr_source_migration.py

@@ -12,6 +12,8 @@ from alembic import command
 from alembic.config import Config
 from sqlalchemy import create_engine, event, inspect, text
 
+from app.integrations.meter_sources import sanitize_source_config, validate_source_config
+
 
 def _config(database_url: str) -> Config:
     config = Config("alembic_app.ini")
@@ -108,7 +110,7 @@ def test_populated_revision_14_adopts_dsmr_history_at_revision_16(tmp_path: Path
                     {"key": "DSMR_INGEST_ENABLED", "value": "true", "at": start},
                     {"key": "DSMR_MQTT_TOPIC", "value": "historic/dsmr", "at": start},
                     {"key": "DSMR_TARIFF_TOPIC", "value": "historic/tariff", "at": start},
-                    {"key": "DSMR_SAMPLE_INTERVAL_S", "value": "15", "at": start},
+                    {"key": "DSMR_SAMPLE_INTERVAL_S", "value": "0", "at": start},
                     {"key": "MQTT_BROKER_HOST", "value": "mqtt.example.invalid", "at": start},
                     {"key": "MQTT_BROKER_PORT", "value": "1884", "at": start},
                     {"key": "MQTT_USERNAME", "value": "historic-user", "at": start},
@@ -159,11 +161,14 @@ def test_populated_revision_14_adopts_dsmr_history_at_revision_16(tmp_path: Path
                 text("SELECT id, enabled, config FROM meter_source WHERE kind = 'dsmr_mqtt'")
             ).one()
             assert source.enabled == 1
-            assert json.loads(source.config) == {
+            source_config = json.loads(source.config)
+            assert source_config == {
                 "broker_host": "mqtt.example.invalid", "broker_port": 1884,
                 "username": "historic-user", "password": "historic-password", "tls_enabled": True,
-                "topic": "historic/dsmr", "tariff_topic": "historic/tariff", "sample_interval_s": 15,
+                "topic": "historic/dsmr", "tariff_topic": "historic/tariff", "sample_interval_s": 0,
             }
+            assert validate_source_config("dsmr_mqtt", source_config) == source_config
+            assert sanitize_source_config("dsmr_mqtt", source_config)["sample_interval_s"] == 0
             assert connection.execute(text("SELECT value FROM app_config WHERE key = 'DSMR_MQTT_TOPIC'")).scalar_one() == "historic/dsmr"
             assert dict(connection.execute(text("SELECT key, value FROM app_config")).all()) == config_before
             assert connection.execute(text("SELECT group_concat(telegram_id) FROM dsmr_reading")).scalar_one() == "77,78,77"

tests/test_energy_expose.py

@@ -13,7 +13,7 @@ Coverage:
   7. value_getter returns None when no non-degraded period exists.
   8. MQTT not enabled → publish_states is a no-op (no raises, no publish calls).
   9. Integration: build_discovery_payload on an energy_cost entity does NOT raise
-     IndexError (validates 2-element identifiers).
+    IndexError (validates one-item HA identifiers and independent internal identities).
  10. Keys are stable fixed strings (not derived from mutable data or DB ids).
  11. Provider registered: energy_cost entities appear alongside modbus entities
      in the full catalog.
@@ -767,9 +767,8 @@ def test_publish_states_noop_when_not_connected() -> None:
 def test_build_discovery_payload_no_index_error_for_energy_entities(energy_db) -> None:
     """build_discovery_payload must NOT raise IndexError for energy_cost entities.
 
-    FUE-T05: identifiers is now ('energy-cost', meter.uuid).
+    The HA grouping identifier and internal MQTT identity are deliberately
-    Validates that the 2-element identifiers tuple satisfies ha_discovery.py's
+    separate; a one-item HA identifier must therefore remain sufficient.
-    requirement to access identifiers[1] as node_id.
     """
     from app.integrations.expose import build_catalog
     from app.services.ha_discovery import build_discovery_payload
@@ -800,15 +799,11 @@ def test_build_discovery_payload_no_index_error_for_energy_entities(energy_db) -
     assert len(energy_entries) == 6, "Expected 6 energy_cost entities in catalog"
 
     for entry in energy_entries:
-        # identifiers[1] must be the meter uuid (not "energy-cost").
+        assert entry.entity.device.internal_identity == meter_uuid
-        assert entry.entity.device.identifiers[1] == meter_uuid, (
+        assert entry.entity.device.identifiers == (f"home-automation:energy-cost:{meter_uuid}",)
-            f"identifiers[1] must be meter uuid {meter_uuid!r}, "
-            f"got {entry.entity.device.identifiers[1]!r}"
-        )
-        # Must not raise — specifically no IndexError from identifiers[1]
         topic, config = build_discovery_payload(entry.entity, "homeassistant")
 
-        # node_id = identifiers[1] with hyphens → underscores
+        # node_id = internal meter identity with hyphens → underscores
         node_id = meter_uuid.replace("-", "_")
         assert node_id in topic, (
             f"Expected meter uuid node_id {node_id!r} in discovery topic, got {topic!r}"
@@ -817,13 +812,12 @@ def test_build_discovery_payload_no_index_error_for_energy_entities(energy_db) -
             f"Discovery topic must end with /config, got {topic!r}"
         )
         assert "unique_id" in config
-        # unique_id seed is identifiers[1] (meter uuid) + entity key
+        # unique_id seed is the internal meter identity + entity key
         assert meter_uuid in config["unique_id"], (
             f"unique_id must contain meter uuid, got {config['unique_id']!r}"
         )
         assert "device" in config
-        assert "energy-cost" in config["device"]["identifiers"]
+        assert config["device"]["identifiers"] == [f"home-automation:energy-cost:{meter_uuid}"]
-        assert meter_uuid in config["device"]["identifiers"]
 
 
 def test_energy_cost_entities_omit_availability_so_ha_shows_them(energy_db) -> None:
@@ -859,18 +853,18 @@ def test_energy_cost_entities_omit_availability_so_ha_shows_them(energy_db) -> N
 def test_energy_entity_discovery_topics_contain_correct_node_id() -> None:
     """Discovery topic node_id for energy entities must be derived from meter uuid.
 
-    FUE-T05: identifiers[1] is now the active meter's uuid.
+    The internal identity is the active meter's uuid, independent from the
-    ha_discovery._node_id() replaces hyphens with underscores in identifiers[1]
+    singleton HA identifier.  This test verifies that the topic reflects the
-    to build the MQTT node_id.  This test verifies that the topic reflects the
     meter uuid (not the old fixed 'energy-cost' string).
     """
     from app.integrations.expose import DeviceInfo, ExposableEntity
     from app.services.ha_discovery import build_discovery_payload
 
     meter_uuid = "12345678-abcd-ef00-1234-567890abcdef"
-    # identifiers[1] = meter uuid — this is what FUE-T05 sets.
+    # ``identity`` is the meter UUID; HA grouping is a separate one-item tuple.
     device = DeviceInfo(
-        identifiers=("energy-cost", meter_uuid),
+        identifiers=(f"home-automation:energy-cost:{meter_uuid}",),
+        identity=meter_uuid,
         name="Test Meter",
         provides_availability=False,
     )
@@ -886,7 +880,7 @@ def test_energy_entity_discovery_topics_contain_correct_node_id() -> None:
 
     topic, config = build_discovery_payload(entity, discovery_prefix="homeassistant")
 
-    # node_id: identifiers[1] = meter_uuid, hyphens → underscores
+    # node_id: internal identity = meter_uuid, hyphens → underscores
     expected_node = meter_uuid.replace("-", "_")
     assert f"/{expected_node}/" in topic, (
         f"Expected meter uuid node_id {expected_node!r} in topic {topic!r}"
@@ -2345,12 +2339,11 @@ def test_energy_cost_provider_returns_empty_when_no_active_meter(energy_db) -> N
     )
 
 
-def test_energy_cost_provider_identifiers_match_meter_uuid(energy_db) -> None:
+def test_energy_cost_provider_identity_matches_meter_uuid(energy_db) -> None:
-    """FUE-T05 ②: with an active meter, identifiers[1] == meter.uuid.
+    """The active meter anchors the internal identity and namespaced HA identifier.
 
-    The HA device identity is anchored to the active meter's uuid.  ha_discovery.py
+    The HA device identity and MQTT internal identity are both anchored to the
-    uses identifiers[1] as the MQTT node_id and unique_id seed; changing the active
+    active meter's uuid.  Changing the meter produces a new node_id and HA sensor.
-    meter (meter swap) produces a new uuid → new node_id → new HA sensor.
     """
     from app.integrations.expose import _energy_cost_provider
 
@@ -2367,11 +2360,8 @@ def test_energy_cost_provider_identifiers_match_meter_uuid(energy_db) -> None:
 
     assert len(entities) == 6, f"Expected 6 entities, got {len(entities)}"
     for entity in entities:
-        assert entity.device.identifiers == ("energy-cost", meter_uuid), (
+        assert entity.device.internal_identity == meter_uuid
-            f"identifiers must be ('energy-cost', meter.uuid); "
+        assert entity.device.identifiers == (f"home-automation:energy-cost:{meter_uuid}",)
-            f"expected ('energy-cost', {meter_uuid!r}), "
-            f"got {entity.device.identifiers!r}"
-        )
         assert entity.device.name == meter_label, (
             f"device.name must be exactly the meter label {meter_label!r}, "
             f"got {entity.device.name!r}"
@@ -2382,7 +2372,7 @@ def test_energy_cost_entity_keys_do_not_contain_meter_uuid(energy_db) -> None:
     """FUE-T05 ③: entity keys remain stable 'energy.*' strings (no uuid injected).
 
     Keys are the anchor for the toggle table; they must NOT change when the meter
-    changes.  Only identifiers[1] (node_id / unique_id) changes on a meter swap.
+    changes.  Only the internal identity (node_id / unique_id) changes on a meter swap.
     """
     from app.integrations.expose import _energy_cost_provider
 
@@ -2418,7 +2408,7 @@ def test_energy_cost_entity_keys_do_not_contain_meter_uuid(energy_db) -> None:
 def test_energy_cost_toggle_survives_meter_swap(energy_db) -> None:
     """FUE-T05 ③ (toggle stability): enabled toggle on 'energy.buy_price_now' survives meter swap.
 
-    After a meter swap the provider queries a new active meter → new identifiers[1] /
+    After a meter swap the provider queries a new active meter → new internal identity /
     unique_id / topic in HA.  But the entity key stays 'energy.buy_price_now', so the
     existing toggle row (keyed by 'energy.buy_price_now') is still found → enabled=True.
 
@@ -2475,11 +2465,7 @@ def test_energy_cost_toggle_survives_meter_swap(energy_db) -> None:
         (e for e in catalog if e.entity.key == "energy.buy_price_now"), None
     )
     assert buy_entry is not None, "energy.buy_price_now must be present in catalog"
-    # identifiers[1] must now be the NEW meter's uuid
+    assert buy_entry.entity.device.internal_identity == new_meter_uuid
-    assert buy_entry.entity.device.identifiers[1] == new_meter_uuid, (
-        f"After swap, identifiers[1] must be new meter uuid {new_meter_uuid!r}, "
-        f"got {buy_entry.entity.device.identifiers[1]!r}"
-    )
     # Toggle state must still be enabled (key unchanged → same toggle row found)
     assert buy_entry.enabled is True, (
         "energy.buy_price_now toggle must remain enabled after meter swap "
@@ -2488,7 +2474,7 @@ def test_energy_cost_toggle_survives_meter_swap(energy_db) -> None:
 
 
 def test_energy_cost_identifiers_change_after_meter_swap(energy_db) -> None:
-    """FUE-T05 ④: after meter swap, provider produces new identifiers[1] (new meter uuid).
+    """After a meter swap, provider produces a new internal meter identity.
 
     Old uuid's entities are no longer produced → HA sensor for old uuid is frozen.
     New uuid's entities appear → HA creates fresh sensors for the new meter.
@@ -2544,12 +2530,9 @@ def test_energy_cost_identifiers_change_after_meter_swap(energy_db) -> None:
         f"Expected 6 entities after swap, got {len(entities_after_swap)}"
     )
     for entity in entities_after_swap:
-        assert entity.device.identifiers[1] == new_uuid, (
+        assert entity.device.internal_identity == new_uuid
-            f"After swap, identifiers[1] must be new uuid {new_uuid!r}, "
-            f"got {entity.device.identifiers[1]!r}"
-        )
         # Old uuid must not appear in identifiers
-        assert entity.device.identifiers[1] != old_uuid, (
+        assert entity.device.internal_identity != old_uuid, (
             f"After swap, old uuid {old_uuid!r} must not appear in identifiers"
         )
 
@@ -2809,8 +2792,22 @@ def test_m8_catalog_has_source_meter_and_thermal_entities_disabled(energy_db) ->
     assert entries[f"meter.{electricity.uuid}.total"].entity.unit == "kWh"
     assert entries[f"meter.{electricity.uuid}.total"].entity.device_class == "energy"
     assert entries[f"meter.{electricity.uuid}.today"].entity.state_class == "total_increasing"
+    device_ids = {
+        entries[f"source.{heating_source.uuid}.online"].entity.device.identifiers[0],
+        entries[f"meter.{heating.uuid}.total"].entity.device.identifiers[0],
+        entries[f"meter.{water.uuid}.total"].entity.device.identifiers[0],
+        entries[f"meter.{electricity.uuid}.total"].entity.device.identifiers[0],
+    }
+    assert len(device_ids) == 4
+    assert all(len(entry.entity.device.identifiers) == 1 for entry in entries.values())
     thermal = [entry for key, entry in entries.items() if key.startswith("thermal_cost.")]
-    assert len(thermal) == 12
+    assert len(thermal) == 14
+    assert {entry.entity.key.rsplit(".", 1)[-1] for entry in thermal} >= {
+        "hot_water_total_total", "hot_water_total_today"
+    }
+    assert {entry.entity.name for entry in thermal if ".hot_water_total_" in entry.entity.key} == {
+        "Thermal Hot Water Total", "Thermal Hot Water Today"
+    }
     assert all(entry.enabled is False for entry in thermal)
     assert all(entry.entity.unit == "EUR" and entry.entity.device_class == "monetary" for entry in thermal)
 
@@ -2972,9 +2969,10 @@ def test_m8_thermal_today_summary_ends_at_frozen_now(energy_db) -> None:
     captured: dict[str, datetime] = {}
     result = {
         "period_count": 1, "fixed_cost": Decimal("0"), "total_cost": Decimal("0"),
-        "breakdown": {key: Decimal("0") for key in (
+        "breakdown": {
-            "heating", "hot_water_heating", "hot_water", "hot_water_tax",
+            "heating": Decimal("0"), "hot_water_heating": Decimal("1.25"),
-        )},
+            "hot_water": Decimal("2.75"), "hot_water_tax": Decimal("9.99"),
+        },
     }
 
     def summarize_spy(_session: Session, start: datetime, end: datetime, *, now: datetime) -> dict:
@@ -2994,4 +2992,7 @@ def test_m8_thermal_today_summary_ends_at_frozen_now(energy_db) -> None:
             entity = next(item.entity for item in build_catalog(session)
                           if item.entity.key.endswith(".heating_today"))
             assert entity.value_getter(session) == Decimal("0")
+            hot_water_total = next(item.entity for item in build_catalog(session)
+                                   if item.entity.key.endswith(".hot_water_total_today"))
+            assert hot_water_total.value_getter(session) == Decimal("4.00")
     assert captured["end"] == now

tests/test_expose_catalog.py

@@ -362,7 +362,21 @@ def test_build_catalog_device_grouping(expose_db):
     assert len(identifiers_set) == 1, (
         "All entities for one device must share the same DeviceInfo identifiers"
     )
-    assert identifiers_set.pop() == ("modbus", test_uuid)
+    own_identifiers = identifiers_set.pop()
+    assert own_identifiers == (f"home-automation:modbus:{test_uuid}",)
+
+    # A second Modbus device must never overlap its singleton HA identifier.
+    other_uuid = "aaaaaaaa-0000-0000-0000-000000000099"
+    with Session(expose_db) as session:
+        _make_modbus_device(session, friendly_name="SDM120 E", uuid=other_uuid)
+        session.commit()
+    with Session(expose_db) as session:
+        second_catalog = build_catalog(session)
+    other_identifiers = {
+        entry.entity.device.identifiers for entry in second_catalog if other_uuid in entry.entity.key
+    }
+    assert other_identifiers == {(f"home-automation:modbus:{other_uuid}",)}
+    assert {own_identifiers} != other_identifiers
 
 
 def test_build_catalog_metric_metadata_from_profile(expose_db):

tests/test_ha_discovery.py

@@ -15,6 +15,7 @@ Coverage:
 
 from __future__ import annotations
 
+import json
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any
@@ -443,14 +444,14 @@ def test_publish_discovery_enabled_vs_disabled_payload(disco_db) -> None:
     voltage_config_topic = f"homeassistant/sensor/{node_id}/{voltage_obj_id}/config"
 
     # The voltage entity's config topic should have non-empty JSON payload
-    voltage_call = next(
+    voltage_calls = [(t, p, r) for t, p, r in config_calls if t == voltage_config_topic]
-        ((t, p, r) for t, p, r in config_calls if t == voltage_config_topic), None
+    assert voltage_calls, (
-    )
-    assert voltage_call is not None, (
         f"Expected config publish for voltage topic {voltage_config_topic!r}. "
         f"Got topics: {[t for t, _, _ in config_calls]}"
     )
-    _t, payload, retain = voltage_call
+    # First v1.6.1 repair run deliberately unloads the old config before it
+    # re-adds the same unique_id under the corrected HA device identifier.
+    _t, payload, retain = voltage_calls[-1]
     assert payload not in (b"", "", None), "Enabled entity should get non-empty config payload"
     assert retain is True, "Discovery config must be retained"
 
@@ -1378,3 +1379,497 @@ def test_stale_m8_entities_includes_ended_electricity_meter(disco_db) -> None:
         session.commit()
         keys = {entity.key for entity in _stale_m8_entities(session)}
         assert keys == {f"meter.{old.uuid}.total", f"meter.{old.uuid}.today"}
+
+
+def test_discovery_uses_single_namespaced_identifier_and_safe_thermal_topic() -> None:
+    """HA device grouping is independent from a dot-containing internal seed."""
+    from app.integrations.expose import DeviceInfo, ExposableEntity
+    from app.services.ha_discovery import build_discovery_payload
+
+    identity = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa.bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"
+    device = DeviceInfo(
+        identifiers=(f"home-automation:thermal-cost:{identity}",),
+        identity=identity,
+        name="Thermal Energy Cost",
+        provides_availability=False,
+    )
+    entity = ExposableEntity(
+        key=f"thermal_cost.{identity}.heating_total", component="sensor", device=device,
+        device_class="monetary", unit="EUR", name="Thermal Heating Total", state_class="total",
+    )
+    topic, config = build_discovery_payload(entity, "homeassistant")
+    assert config["device"]["identifiers"] == [f"home-automation:thermal-cost:{identity}"]
+    assert all(part.replace("-", "").replace("_", "").isalnum()
+               for part in topic.split("/")[2:4])
+    assert "." not in topic
+
+
+def test_legacy_thermal_cleanup_persists_each_success_and_retries_only_failure(disco_db) -> None:
+    """Illegal v1.6.1 cleanup never re-sends a durably accepted topic."""
+    from app.integrations.expose import DeviceInfo, ExposableEntity
+    from app.services import ha_discovery
+    from app.models.config import AppConfigEntry
+
+    identity = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa.bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"
+    legacy = ExposableEntity(
+        key=f"thermal_cost.{identity}.heating_total", component="sensor",
+        device=DeviceInfo(identifiers=("thermal-cost", identity), identity=identity, name="obsolete"),
+        device_class=None, unit="", name="obsolete",
+    )
+    second_legacy = ExposableEntity(
+        key=f"thermal_cost.{identity}.heating_today", component="sensor",
+        device=legacy.device, device_class=None, unit="", name="obsolete",
+    )
+    settings = _make_settings()
+    manager = _make_mock_manager()
+    with (
+        patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
+        patch("app.services.ha_discovery.mqtt_manager", manager),
+        patch("app.services.ha_discovery.build_catalog", return_value=[]),
+        patch("app.services.ha_discovery._legacy_thermal_entities", return_value=[legacy, second_legacy]),
+        patch("app.services.ha_discovery._stale_m8_entities", return_value=[]),
+    ):
+        with Session(disco_db) as session:
+            ha_discovery.initialize_legacy_thermal_cleanup(session)
+        manager.publish.side_effect = [True, False]
+        with Session(disco_db) as session:
+            ha_discovery.publish_discovery(session)
+        assert manager.publish.call_count == 2
+        with Session(disco_db) as session:
+            progress = session.query(AppConfigEntry).filter_by(
+                key=ha_discovery._LEGACY_THERMAL_CLEANUP_KEY
+            ).one()
+            assert ha_discovery._legacy_discovery_topic(legacy, "homeassistant") in progress.value
+
+        manager.publish.reset_mock()
+        manager.publish.side_effect = None
+        manager.publish.return_value = True
+        with Session(disco_db) as session:
+            ha_discovery.publish_discovery(session)
+        assert manager.publish.call_args_list == [
+            ((ha_discovery._legacy_discovery_topic(second_legacy, "homeassistant"), b""), {"retain": True}),
+        ]
+
+        # A fresh Session simulates a process restart: no illegal topic is sent.
+        manager.publish.reset_mock()
+        with Session(disco_db) as session:
+            ha_discovery.publish_discovery(session)
+        manager.publish.assert_not_called()
+
+
+def test_legacy_thermal_enumeration_failure_does_not_advance_marker(disco_db) -> None:
+    """An inventory error is pending work, never an empty successful cleanup."""
+    from app.models.config import AppConfigEntry
+    from app.services import ha_discovery
+
+    settings = _make_settings()
+    manager = _make_mock_manager()
+    with (
+        patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
+        patch("app.services.ha_discovery.mqtt_manager", manager),
+        patch("app.services.ha_discovery.build_catalog", return_value=[]),
+        patch("app.services.ha_discovery._legacy_thermal_entities", side_effect=RuntimeError("enumeration failed")),
+        patch("app.services.ha_discovery._stale_m8_entities", return_value=[]),
+    ):
+        with Session(disco_db) as session:
+            ha_discovery.publish_discovery(session)
+        with Session(disco_db) as session:
+            assert session.query(AppConfigEntry).filter_by(
+                key=ha_discovery._LEGACY_THERMAL_CLEANUP_KEY
+            ).one_or_none() is None
+
+
+def test_legacy_cleanup_compatibility_freeze_keeps_previous_success_progress(disco_db) -> None:
+    """Startup freezes a pre-inventory ledger and preserves its success progress."""
+    from app.integrations.expose import DeviceInfo, ExposableEntity
+    from app.models.config import AppConfigEntry
+    from app.services import ha_discovery
+
+    device = DeviceInfo(identifiers=("legacy",), identity="old", name="obsolete")
+    first = ExposableEntity(key="thermal_cost.old.heating_total", component="sensor", device=device,
+                            device_class=None, unit="", name="obsolete")
+    second = ExposableEntity(key="thermal_cost.old.heating_today", component="sensor", device=device,
+                             device_class=None, unit="", name="obsolete")
+    first_topic = ha_discovery._legacy_discovery_topic(first, "homeassistant")
+    with Session(disco_db) as session:
+        session.add(AppConfigEntry(
+            key=ha_discovery._LEGACY_THERMAL_CLEANUP_KEY,
+            value=json.dumps({"complete": False, "topics": [first_topic]}),
+            updated_at=datetime.now(timezone.utc),
+        ))
+        session.commit()
+
+    with (
+        patch("app.services.ha_discovery.build_runtime_settings", return_value=_make_settings()),
+        patch("app.services.ha_discovery._legacy_thermal_entities", return_value=[first, second]),
+    ):
+        with Session(disco_db) as session:
+            ha_discovery.initialize_legacy_thermal_cleanup(session)
+
+    manager = _make_mock_manager()
+    with (
+        patch("app.services.ha_discovery.build_runtime_settings", return_value=_make_settings()),
+        patch("app.services.ha_discovery.mqtt_manager", manager),
+        patch("app.services.ha_discovery.build_catalog", return_value=[]),
+        patch("app.services.ha_discovery._stale_m8_entities", return_value=[]),
+        patch("app.services.ha_discovery._ha_registry_bindings", return_value=None),
+    ):
+        with Session(disco_db) as session:
+            ha_discovery.publish_discovery(session)
+    assert manager.publish.call_args_list == [
+        ((ha_discovery._legacy_discovery_topic(second, "homeassistant"), b""), {"retain": True}),
+    ]
+    with Session(disco_db) as session:
+        ledger = ha_discovery._migration_json(session, ha_discovery._LEGACY_THERMAL_CLEANUP_KEY)
+    assert ledger == {
+        "complete": True,
+        "inventory": [first_topic, ha_discovery._legacy_discovery_topic(second, "homeassistant")],
+        "topics": sorted((first_topic, ha_discovery._legacy_discovery_topic(second, "homeassistant"))),
+    }
+
+
+def test_legacy_cleanup_startup_marks_fresh_install_complete_and_short_circuits(disco_db) -> None:
+    """A later first toggle cannot make a v1.6.1 topic after fresh startup."""
+    from app.models.config import AppConfigEntry
+    from app.services import ha_discovery
+
+    manager = _make_mock_manager()
+    settings = _make_settings()
+    with patch("app.services.ha_discovery._legacy_thermal_entities", return_value=[]):
+        with Session(disco_db) as session:
+            ha_discovery.initialize_legacy_thermal_cleanup(session)
+    with Session(disco_db) as session:
+        ledger = session.query(AppConfigEntry).filter_by(
+            key=ha_discovery._LEGACY_THERMAL_CLEANUP_KEY
+        ).one()
+        assert ledger.value == '{"complete": true, "inventory": [], "topics": []}'
+
+    # A fresh Session is equivalent to a restarted process.  If a user now
+    # enables a thermal entity, the completed ledger prevents re-enumeration.
+    with (
+        patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
+        patch("app.services.ha_discovery.mqtt_manager", manager),
+        patch("app.services.ha_discovery.build_catalog", return_value=[]),
+        patch("app.services.ha_discovery._legacy_thermal_entities") as legacy,
+        patch("app.services.ha_discovery._stale_m8_entities", return_value=[]),
+        patch("app.services.ha_discovery._ha_registry_bindings", return_value=None),
+    ):
+        with Session(disco_db) as session:
+            ha_discovery.publish_discovery(session)
+    legacy.assert_not_called()
+    manager.publish.assert_not_called()
+
+
+def test_legacy_cleanup_startup_completes_empty_compat_ledger_before_later_enable(disco_db) -> None:
+    """An Alembic-head empty compat ledger cannot manufacture a later old topic."""
+    from app.models.config import AppConfigEntry
+    from app.models.energy import Meter
+    from app.services import ha_discovery
+
+    now = datetime.now(timezone.utc)
+    with Session(disco_db) as session:
+        heating = Meter(label="heating", commodity="heating", started_at=now, ended_at=None,
+                        reason="initial", note=None, created_at=now)
+        water = Meter(label="water", commodity="hot_water", started_at=now, ended_at=None,
+                      reason="initial", note=None, created_at=now)
+        session.add_all((heating, water))
+        session.add(AppConfigEntry(
+            key=ha_discovery._LEGACY_THERMAL_CLEANUP_KEY,
+            value=json.dumps({"complete": False, "topics": []}),
+            updated_at=now,
+        ))
+        session.commit()
+
+    settings = _make_settings(ha_discovery_prefix="startup_prefix")
+    with patch("app.services.ha_discovery.build_runtime_settings", return_value=settings):
+        with Session(disco_db) as session:
+            ha_discovery.initialize_legacy_thermal_cleanup(session)
+    with Session(disco_db) as session:
+        ledger = ha_discovery._migration_json(session, ha_discovery._LEGACY_THERMAL_CLEANUP_KEY)
+        assert ledger == {"complete": True, "inventory": [], "topics": []}
+        active_meters = session.query(Meter).filter(Meter.ended_at.is_(None)).all()
+        pair = ha_discovery._thermal_cleanup_entities(
+            [(next(meter for meter in active_meters if meter.commodity == "heating"),
+              next(meter for meter in active_meters if meter.commodity == "hot_water"))],
+            include_hot_water_total=False,
+        )
+        _enable_entity(session, pair[0].key)
+        session.commit()
+
+    manager = _make_mock_manager()
+    with (
+        patch("app.services.ha_discovery.build_runtime_settings",
+              return_value=_make_settings(ha_discovery_prefix="changed_prefix")),
+        patch("app.services.ha_discovery.mqtt_manager", manager),
+        patch("app.services.ha_discovery.build_catalog", return_value=[]),
+        patch("app.services.ha_discovery._stale_m8_entities", return_value=[]),
+        patch("app.services.ha_discovery._ha_registry_bindings", return_value=None),
+    ):
+        with Session(disco_db) as session:
+            ha_discovery.publish_discovery(session)
+    manager.publish.assert_not_called()
+
+
+def test_legacy_cleanup_startup_propagates_inventory_failure(disco_db) -> None:
+    """Startup must fail closed instead of exposing a mutable cleanup window."""
+    from app.services import ha_discovery
+
+    with patch("app.services.ha_discovery._legacy_thermal_entities", side_effect=RuntimeError("boom")):
+        with Session(disco_db) as session:
+            with pytest.raises(RuntimeError, match="boom"):
+                ha_discovery.initialize_legacy_thermal_cleanup(session)
+
+
+def test_legacy_cleanup_startup_propagates_durable_write_failure(disco_db) -> None:
+    """A failed freeze write is fatal; UI must not open with an unfrozen ledger."""
+    from app.services import ha_discovery
+
+    with (
+        patch("app.services.ha_discovery._legacy_thermal_entities", return_value=[]),
+        patch("app.services.ha_discovery._set_migration_json", side_effect=RuntimeError("disk full")),
+    ):
+        with Session(disco_db) as session:
+            with pytest.raises(RuntimeError, match="disk full"):
+                ha_discovery.initialize_legacy_thermal_cleanup(session)
+
+
+def test_legacy_cleanup_startup_keeps_upgrade_with_enabled_topic_pending(disco_db) -> None:
+    """Existing enabled v1.6.1 inventory is not mistaken for a fresh install."""
+    from app.integrations.expose import DeviceInfo, ExposableEntity
+    from app.models.config import AppConfigEntry
+    from app.services import ha_discovery
+
+    legacy = ExposableEntity(
+        key="thermal_cost.old.heating_total", component="sensor",
+        device=DeviceInfo(identifiers=("legacy",), identity="old", name="obsolete"),
+        device_class=None, unit="", name="obsolete",
+    )
+    with patch("app.services.ha_discovery._legacy_thermal_entities", return_value=[legacy]):
+        with Session(disco_db) as session:
+            ha_discovery.initialize_legacy_thermal_cleanup(session)
+    with Session(disco_db) as session:
+        ledger = session.query(AppConfigEntry).filter_by(
+            key=ha_discovery._LEGACY_THERMAL_CLEANUP_KEY
+        ).one()
+        assert ledger.value == (
+            '{"complete": false, "inventory": ["homeassistant/sensor/old/'
+            'thermal_cost_old_heating_total/config"], "topics": []}'
+        )
+
+
+def test_legacy_cleanup_freezes_startup_upgrade_inventory_across_toggle_changes(disco_db) -> None:
+    """Alembic-head compat ledger survives UI disable and runtime prefix changes."""
+    from app.models.config import AppConfigEntry
+    from app.models.energy import Meter
+    from app.models.expose import ExposedEntityToggle
+    from app.services import ha_discovery
+
+    now = datetime.now(timezone.utc)
+    settings = _make_settings(ha_discovery_prefix="frozen_prefix")
+    with Session(disco_db) as session:
+        heating = Meter(label="heating", commodity="heating", started_at=now, ended_at=None,
+                        reason="initial", note=None, created_at=now)
+        water = Meter(label="water", commodity="hot_water", started_at=now, ended_at=None,
+                      reason="initial", note=None, created_at=now)
+        session.add_all((heating, water))
+        session.flush()
+        enabled = ha_discovery._thermal_cleanup_entities(
+            [(heating, water)], include_hot_water_total=False
+        )[:2]
+        for entity in enabled:
+            _enable_entity(session, entity.key)
+        first_topic = ha_discovery._legacy_discovery_topic(enabled[0], "frozen_prefix")
+        session.add(AppConfigEntry(
+            key=ha_discovery._LEGACY_THERMAL_CLEANUP_KEY,
+            value=json.dumps({"complete": False, "topics": [first_topic]}),
+            updated_at=now,
+        ))
+        session.commit()
+
+    with patch("app.services.ha_discovery.build_runtime_settings", return_value=settings):
+        with Session(disco_db) as session:
+            ha_discovery.initialize_legacy_thermal_cleanup(session)
+
+    with Session(disco_db) as session:
+        ledger = ha_discovery._migration_json(session, ha_discovery._LEGACY_THERMAL_CLEANUP_KEY)
+        inventory = ledger["inventory"]
+        assert len(inventory) == 2
+        assert all(topic.startswith("frozen_prefix/") for topic in inventory)
+        assert ledger["topics"] == [first_topic]
+        # This mirrors PUT /api/expose: persist the UI change before it invokes
+        # publish_discovery in the same request.
+        toggle = session.query(ExposedEntityToggle).filter_by(key=enabled[0].key).one()
+        toggle.enabled = False
+        session.commit()
+
+    manager = _make_mock_manager()
+    failed_topic = inventory[-1]
+    published: list[str] = []
+    manager.publish.side_effect = lambda topic, _payload, **_kwargs: (
+        published.append(topic) or topic != failed_topic
+    )
+    with (
+        patch("app.services.ha_discovery.build_runtime_settings",
+              return_value=_make_settings(ha_discovery_prefix="changed_prefix")),
+        patch("app.services.ha_discovery.mqtt_manager", manager),
+        patch("app.services.ha_discovery.build_catalog", return_value=[]),
+        patch("app.services.ha_discovery._stale_m8_entities", return_value=[]),
+        patch("app.services.ha_discovery._ha_registry_bindings", return_value=None),
+    ):
+        with Session(disco_db) as session:
+            ha_discovery.publish_discovery(session)
+    assert published == [failed_topic]
+    with Session(disco_db) as session:
+        pending = ha_discovery._migration_json(session, ha_discovery._LEGACY_THERMAL_CLEANUP_KEY)
+        assert pending["complete"] is False
+        assert pending["inventory"] == inventory
+        assert pending["topics"] == [first_topic]
+
+    manager.publish.reset_mock()
+    manager.publish.side_effect = lambda topic, _payload, **_kwargs: published.append(topic) or True
+    published.clear()
+    with (
+        patch("app.services.ha_discovery.build_runtime_settings",
+              return_value=_make_settings(ha_discovery_prefix="changed_prefix")),
+        patch("app.services.ha_discovery.mqtt_manager", manager),
+        patch("app.services.ha_discovery.build_catalog", return_value=[]),
+        patch("app.services.ha_discovery._legacy_thermal_entities") as enumerate_legacy,
+        patch("app.services.ha_discovery._stale_m8_entities", return_value=[]),
+        patch("app.services.ha_discovery._ha_registry_bindings", return_value=None),
+    ):
+        with Session(disco_db) as session:
+            ha_discovery.publish_discovery(session)
+    assert published == [failed_topic]
+    enumerate_legacy.assert_not_called()
+
+    manager.publish.reset_mock()
+    with (
+        patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
+        patch("app.services.ha_discovery.mqtt_manager", manager),
+        patch("app.services.ha_discovery.build_catalog", return_value=[]),
+        patch("app.services.ha_discovery._legacy_thermal_entities") as enumerate_legacy,
+        patch("app.services.ha_discovery._stale_m8_entities", return_value=[]),
+        patch("app.services.ha_discovery._ha_registry_bindings", return_value=None),
+    ):
+        with Session(disco_db) as session:
+            ha_discovery.publish_discovery(session)
+    enumerate_legacy.assert_not_called()
+    manager.publish.assert_not_called()
+
+
+def test_registry_repair_waits_for_registry_observation_and_partial_catalog_recovers(disco_db) -> None:
+    """The V2 repair is per entity and never mistakes broker ACK for HA ACK."""
+    from app.integrations.expose import CatalogEntry, DeviceInfo, ExposableEntity
+    from app.services import ha_discovery
+
+    def entity(kind: str, identity: str, metric: str) -> ExposableEntity:
+        return ExposableEntity(
+            key=f"{kind}.{identity}.{metric}" if kind != "energy" else f"energy.{metric}",
+            component="sensor",
+            device=DeviceInfo(
+                identifiers=(f"home-automation:{kind}:{identity}",), identity=identity, name=identity,
+                provides_availability=False,
+            ),
+            device_class=None, unit="", name=metric,
+        )
+
+    entities = [
+        *(entity("meter", f"meter-{number}", "total") for number in range(3)),
+        *(entity("source", f"source-{number}", "online") for number in range(2)),
+        *(entity("modbus", f"modbus-{number}", "voltage") for number in range(2)),
+        entity("energy", "electricity-epoch", "import_cost_total"),
+    ]
+    catalog = [CatalogEntry(entity=item, enabled=True) for item in entities]
+    manager = _make_mock_manager()
+    manager.publish.return_value = True
+    settings = _make_settings()
+    calls: list[tuple[str, object]] = []
+    manager.publish.side_effect = lambda topic, payload, **_kwargs: calls.append((topic, payload)) or True
+
+    bindings: dict[str, set[str]] = {ha_discovery._unique_id(item): {"old"} for item in entities}
+    with (
+        patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
+        patch("app.services.ha_discovery.mqtt_manager", manager),
+        patch("app.services.ha_discovery.build_catalog", return_value=catalog),
+        patch("app.services.ha_discovery._legacy_thermal_entities", return_value=[]),
+        patch("app.services.ha_discovery._stale_m8_entities", return_value=[]),
+        patch("app.services.ha_discovery._ha_registry_bindings", side_effect=lambda _settings, ids: {
+            key: value for key, value in bindings.items() if key in ids
+        }),
+    ):
+        with Session(disco_db) as session:
+            ha_discovery.publish_discovery(session)
+
+    topics = [ha_discovery._discovery_topic(item, "homeassistant") for item in entities]
+    assert [payload for _topic, payload in calls[:len(entities)]] == [b""] * len(entities)
+    assert [topic for topic, _payload in calls[:len(entities)]] == topics
+    assert len({ha_discovery._unique_id(item) for item in entities}) == len(entities)
+
+    # HA confirms every tombstone; only then is each target re-added.
+    calls.clear()
+    bindings.clear()
+    with (
+        patch("app.services.ha_discovery.build_runtime_settings", return_value=settings),
+        patch("app.services.ha_discovery.mqtt_manager", manager),
+        patch("app.services.ha_discovery.build_catalog", return_value=catalog),
+        patch("app.services.ha_discovery._stale_m8_entities", return_value=[]),
+        patch("app.services.ha_discovery._ha_registry_bindings", side_effect=lambda _settings, _ids: dict(bindings)),
+    ):
+        with Session(disco_db) as session:
+            ha_discovery.publish_discovery(session)
+    assert len(calls) == len(entities)
+    assert all(payload not in (b"", "", None) for _topic, payload in calls)
+
+
+def test_registry_repair_unavailable_keeps_normal_discovery_publishing(disco_db) -> None:
+    """A broken optional HA WS link cannot leave legal configs tombstoned."""
+    from app.integrations.expose import CatalogEntry, DeviceInfo, ExposableEntity
+    from app.services import ha_discovery
+
+    entity = ExposableEntity(
+        key="meter.meter-1.total", component="sensor",
+        device=DeviceInfo(identifiers=("home-automation:meter:meter-1",), identity="meter-1", name="m1"),
+        device_class=None, unit="", name="total",
+    )
+    manager = _make_mock_manager()
+    with (
+        patch("app.services.ha_discovery.build_runtime_settings", return_value=_make_settings()),
+        patch("app.services.ha_discovery.mqtt_manager", manager),
+        patch("app.services.ha_discovery.build_catalog", return_value=[CatalogEntry(entity, True)]),
+        patch("app.services.ha_discovery._legacy_thermal_entities", return_value=[]),
+        patch("app.services.ha_discovery._stale_m8_entities", return_value=[]),
+        patch("app.services.ha_discovery._ha_registry_bindings", return_value=None),
+    ):
+        with Session(disco_db) as session:
+            ha_discovery.publish_discovery(session)
+    assert manager.publish.call_args.args[1] != b""
+
+def test_stale_thermal_cleanup_uses_safe_topics_and_all_fourteen_metrics(disco_db) -> None:
+    """Later thermal meter swaps clear only ended safe-format pair configs."""
+    from app.models.energy import Meter
+    from app.services import ha_discovery
+
+    now = datetime.now(timezone.utc)
+    with Session(disco_db) as session:
+        old_heating = Meter(label="old heat", commodity="heating", started_at=now - timedelta(days=2),
+                            ended_at=now - timedelta(days=1), reason="meter_swap", note=None, created_at=now)
+        old_water = Meter(label="old water", commodity="hot_water", started_at=now - timedelta(days=2),
+                          ended_at=now - timedelta(days=1), reason="meter_swap", note=None, created_at=now)
+        active_heating = Meter(label="new heat", commodity="heating", started_at=now - timedelta(days=1),
+                               ended_at=None, reason="meter_swap", note=None, created_at=now)
+        active_water = Meter(label="new water", commodity="hot_water", started_at=now - timedelta(days=1),
+                             ended_at=None, reason="meter_swap", note=None, created_at=now)
+        session.add_all((old_heating, old_water, active_heating, active_water))
+        session.commit()
+        stale = ha_discovery._stale_m8_entities(session)
+
+    old_identity = ".".join(sorted((old_heating.uuid, old_water.uuid)))
+    thermal = [item for item in stale if item.key.startswith("thermal_cost.")]
+    assert len(thermal) == 14
+    assert {item.key for item in thermal} == {
+        f"thermal_cost.{old_identity}.{metric}_{suffix}"
+        for metric in ("heating", "hot_water_heating", "hot_water_total", "water", "water_tax", "fixed", "all_in")
+        for suffix in ("total", "today")
+    }
+    assert all("." not in ha_discovery._discovery_topic(item, "homeassistant") for item in thermal)
+    assert not any(active_heating.uuid in item.key and active_water.uuid in item.key for item in thermal)

tests/test_homeassistant.py

@@ -111,3 +111,87 @@ def test_homeassistant_client_raises_on_invalid_arguments() -> None:
 
     with pytest.raises(ValueError, match="webhook_id"):
         client.trigger_webhook(webhook_id="", body={})
+
+
+def test_discovery_registry_bindings_reads_entity_and_device_registry(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The repair confirmation reads HA's authoritative registry bindings."""
+    sent: list[dict] = []
+
+    class _Socket:
+        replies = iter((
+            '{"type":"auth_required"}',
+            '{"type":"auth_ok"}',
+            '{"id":1,"success":true,"result":[{"platform":"mqtt","unique_id":"u1","device_id":"d1"}]}',
+            '{"id":2,"success":true,"result":[{"id":"d1","identifiers":[["mqtt","home-automation:meter:m1"]]}]}',
+        ))
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_args):
+            return None
+
+        def recv(self, *, timeout=None):
+            assert timeout is not None
+            assert timeout <= 1.5
+            return next(self.replies)
+
+        def send(self, payload):
+            sent.append(json.loads(payload))
+
+    monkeypatch.setattr("app.integrations.homeassistant.connect", lambda *_args, **_kwargs: _Socket())
+    bindings = HomeAssistantClient(_configured_settings()).discovery_registry_bindings({"u1", "missing"})
+
+    assert bindings == {"u1": {"home-automation:meter:m1"}}
+    assert [message.get("type") for message in sent] == [
+        "auth", "config/entity_registry/list", "config/device_registry/list"
+    ]
+
+
+def test_discovery_registry_bindings_uses_one_deadline_and_ignores_non_mqtt_identifiers(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Only official ["mqtt", value] pairs participate in repair matching."""
+    received_timeouts: list[float] = []
+
+    class _Socket:
+        replies = iter((
+            '{"type":"auth_required"}',
+            '{"type":"auth_ok"}',
+            '{"id":1,"success":true,"result":['
+            '{"platform":"mqtt","unique_id":"u1","device_id":"d1"},'
+            '{"platform":"mqtt","unique_id":"u2","device_id":"d2"},'
+            '{"platform":"mqtt","unique_id":"u3","device_id":"d3"}]}',
+            '{"id":2,"success":true,"result":['
+            '{"id":"d1","identifiers":[["mqtt","expected"],["esphome","aux"]]},'
+            '{"id":"d2","identifiers":[["esphome","expected"],"expected",["mqtt",3],[]]},'
+            '{"id":"d3","identifiers":[["mqtt","expected"],["mqtt","old"]]}]}',
+        ))
+
+        def __enter__(self): return self
+        def __exit__(self, *_args): return None
+        def send(self, _payload): return None
+        def recv(self, *, timeout=None):
+            received_timeouts.append(timeout)
+            return next(self.replies)
+
+    monkeypatch.setattr("app.integrations.homeassistant.connect", lambda *_args, **_kwargs: _Socket())
+    bindings = HomeAssistantClient(_configured_settings()).discovery_registry_bindings({"u1", "u2", "u3"})
+
+    assert bindings == {"u1": {"expected"}, "u2": set(), "u3": {"expected", "old"}}
+    assert len(received_timeouts) == 4
+    assert all(timeout is not None and 0 < timeout <= 1.5 for timeout in received_timeouts)
+
+
+def test_discovery_registry_bindings_silent_socket_times_out(monkeypatch: pytest.MonkeyPatch) -> None:
+    class _Socket:
+        def __enter__(self): return self
+        def __exit__(self, *_args): return None
+        def send(self, _payload): return None
+        def recv(self, *, timeout=None):
+            assert timeout is not None and timeout > 0
+            raise TimeoutError("silent")
+
+    monkeypatch.setattr("app.integrations.homeassistant.connect", lambda *_args, **_kwargs: _Socket())
+    with pytest.raises(HomeAssistantRequestError, match="registry query failed"):
+        HomeAssistantClient(_configured_settings()).discovery_registry_bindings({"u1"})

tests/test_meter_source_services.py

@@ -68,6 +68,32 @@ def test_secret_sanitize_and_mask_merge_keep_old_value():
     assert merged["topic"] == "new/topic"
 
 
+def test_dsmr_zero_interval_validates_sanitizes_and_merges_unchanged():
+    config = validate_source_config("dsmr_mqtt", {"sample_interval_s": 0})
+
+    assert config["sample_interval_s"] == 0
+    assert sanitize_source_config("dsmr_mqtt", config)["sample_interval_s"] == 0
+    assert merge_source_config("dsmr_mqtt", config, {"topic": "new/topic"})["sample_interval_s"] == 0
+
+
+@pytest.mark.parametrize(
+    ("kind", "config"),
+    [
+        ("dsmr_mqtt", {"sample_interval_s": -1}),
+        ("dsmr_mqtt", {"sample_interval_s": False}),
+        ("dsmr_mqtt", {"broker_port": 0}),
+        ("dsmr_mqtt", {"broker_port": -1}),
+        ("dsmr_mqtt", {"broker_port": False}),
+        ("warmtelink_serial", {"path": "/dev/warmtelink", "baudrate": 0}),
+        ("warmtelink_serial", {"path": "/dev/warmtelink", "baudrate": -1}),
+        ("warmtelink_serial", {"path": "/dev/warmtelink", "baudrate": False}),
+    ],
+)
+def test_numeric_source_profile_constraints_still_reject_invalid_values(kind, config):
+    with pytest.raises(SourceProfileError):
+        validate_source_config(kind, config)
+
+
 @pytest.fixture()
 def session(tmp_path):
     engine = create_engine(f"sqlite:///{tmp_path / 'source_services.db'}")

tests/test_mqtt_client.py

@@ -257,7 +257,7 @@ def test_publish_passes_topic_payload_retain_to_paho() -> None:
         manager._connected = True
         manager._client = mock_client
 
-    manager.publish("test/topic", '{"key": "value"}', retain=True, qos=1)
+    assert manager.publish("test/topic", '{"key": "value"}', retain=True, qos=1) is True
 
     mock_client.publish.assert_called_once_with(
         "test/topic", payload='{"key": "value"}', qos=1, retain=True
@@ -267,8 +267,7 @@ def test_publish_passes_topic_payload_retain_to_paho() -> None:
 def test_publish_is_noop_when_not_connected() -> None:
     manager = MqttManager()
     # No connect — should silently skip
-    manager.publish("topic", "payload", retain=False)
+    assert manager.publish("topic", "payload", retain=False) is False
-    # No exception raised
 
 
 def test_publish_does_not_raise_on_paho_error() -> None:
@@ -279,7 +278,17 @@ def test_publish_does_not_raise_on_paho_error() -> None:
     manager._connected = True
 
     # Must not raise
-    manager.publish("topic", "payload")
+    assert manager.publish("topic", "payload") is False
+
+
+def test_publish_returns_false_when_paho_rejects_message() -> None:
+    manager = MqttManager()
+    mock_client = MagicMock()
+    mock_client.publish.return_value.rc = 1
+    manager._client = mock_client
+    manager._connected = True
+
+    assert manager.publish("topic", "payload") is False
 
 
 # ---------------------------------------------------------------------------