@@ -421,6 +421,7 @@ newtok
 njpwerner
 nobat
 nocharge
+NODETAIL
 nodischarge
 nohup
 noopener
 
@@ -41,6 +41,7 @@
 import aiohttp
 from datetime import datetime
 from component_base import ComponentBase
+from coordinator import inverter_record
 from alphaess_const import (
     ALPHAESS_BASE_URL,
     ALPHAESS_ENDPOINTS,
@@ -865,6 +866,91 @@ async def automatic_config(self):
 
         await self.apply_hybrid_verdict()
 
+    def build_discovery(self):
+        """
+        Describe the discovered AlphaESS systems for the discovery catalogue.
+
+        Reads only what get_device_list() already holds - self.device_list and self.device_detail -
+        plus the EV-charger verdict _apply_live_payload() records, so this adds no API calls and
+        cannot change what AlphaESS does. Reporting is independent of self.automatic: the catalogue
+        records the hardware, and the report's own "automatic" flag says whether Predbat wired
+        apps.yaml to it.
+
+        self.device_list holds serial strings (sysSn), and only systems that passed has_battery()
+        at discovery: a battery-less system (the VT1000 family, cobat 0 or missing) is dropped in
+        get_device_list() and never reaches device_detail. automatic_config() applies no further
+        test - it registers every serial in device_list as "AlphaESSCloud" - so inverter_type is set
+        on every record, mirroring automatic_config() as the source of truth.
+
+        Ratings are getEssList's own figures: poinv (kW) as inverter_w via inverter_limit(), popv
+        (kW) as pv_w, cobat (kWh) as battery_kwh via battery_capacity(). cobat is one scalar per
+        system, so there is no list of entries to mis-sum. A system reports "solar" only when popv
+        says PV is attached.
+
+        Deliberately not reported:
+        - emsStatus: a status that changes. refresh_discovery() compares whole reports, so a
+          changing value would re-file the report every time it moved.
+        - usCapacity and surplusCobat: they fit "current SoC" and "configured usable depth" equally
+          well, which is why publish_data() never maps them to SoC; reporting either as a rating
+          would assert a meaning nobody knows.
+        - firmware: the AlphaESS Open API exposes none.
+        - account_ids: the API has no station, plant or site identifier; every endpoint is keyed by
+          sysSn alone.
+        - an AC-coupled verdict: detect_ac_coupled() infers it from live telemetry and
+          ALPHAESS_AC_COUPLED_MODELS ships empty on purpose, so there is no held fact to report.
+
+        An EV charger is reported as a flag only when _ev_present says True. That verdict comes from
+        live telemetry alone, so a charger not yet seen reports nothing either way rather than a
+        guess. No chargers record is invented for it: AlphaESS reports a charger's power, not its
+        identity.
+
+        Returns None when no system has been discovered yet, which refresh_discovery() treats as
+        "nothing to report, ask again next cycle".
+        """
+        if not self.device_list:
+            return None
+
+        inverters = []
+        for sn in self.device_list:
+            detail = self.device_detail.get(sn, {}) or {}
+            pv_kw = self._as_float(detail.get("popv"), 0.0)
+
+            functions = ["solar", "battery"] if pv_kw > 0 else ["battery"]
+
+            info = {}
+            if detail.get("minv"):
+                info["model"] = str(detail["minv"])
+            if detail.get("mbat"):
+                info["battery_model"] = str(detail["mbat"])
+
+            ratings = {}
+            inverter_w = self.inverter_limit(sn)
+            if inverter_w > 0:
+                ratings["inverter_w"] = inverter_w
+            if pv_kw > 0:
+                ratings["pv_w"] = pv_kw * 1000.0
+            battery_kwh = self.battery_capacity(sn)
+            if battery_kwh > 0:
+                ratings["battery_kwh"] = battery_kwh
+
+            flags = ["ev_charger"] if self._ev_present.get(sn) is True else []
+
+            inverters.append(
+                inverter_record(
+                    "alphaess:{}".format(sn),
+                    inverter_type="AlphaESSCloud",
+                    composition="direct",
+                    functions=functions,
+                    capabilities=["schedule", "target_soc", "discharge_target", "charge_rate_power"],
+                    flags=flags,
+                    hardware_ids={"serial": sn},
+                    info=info,
+                    ratings=ratings,
+                )
+            )
+
+        return {"automatic": self.automatic, "inverters": inverters}
+
     @staticmethod
     def _empty_schedule():
         """Return a fresh, disabled schedule shape - the single source of truth for its defaults.
@@ -2000,6 +2086,14 @@ async def run(self, seconds, first):
 
         await self.publish_data()
 
+        # Filed right after this cycle's publish and BEFORE `if first and not live_ok:` below:
+        # that branch returns False to defer startup when the first telemetry poll fails, and
+        # automatic_config() follows it, so a report filed any later would never be filed on
+        # exactly the installs whose dump most needs to say what hardware was found. (AlphaESS's
+        # automatic_config() does not raise - the early return is the reason.) refresh_discovery()
+        # owns the compare/retry/guard loop and never raises.
+        self.refresh_discovery()
+
         if first and not live_ok:
             # Startup has not really succeeded without telemetry: automatic_config() runs
             # on the first cycle ALONE, so it would map only the args backed by cached
 
@@ -185,6 +185,11 @@ class Redactor:
     key, which is rewritten by exact match only - see _substitute_key), and a value inside
     hardware_ids is only shape-flagged when it is nothing BUT digits, since a letter-prefixed
     vendor serial with a long digit tail is that container's entire declared purpose.
+
+    A hardware serial is not secret, so a serial a record declares - hardware_ids.serial or an
+    entry in the structural serials list - is clear wherever it appears as a whole token, however
+    many digits it has (Solis, Deye and Sunsynk serials are nothing but digits): in hardware_ids,
+    in a device_id built from it, in a duplicate_serial observation. See _collect_serials.
     """
 
     # Minimum length of an original before it is substituted inside other strings; below this a
@@ -206,6 +211,8 @@ def __init__(self, salt, log=None):
         # unrelated text it happens to share a substring with.
         self.substring_ok = set()
         self._substring_order = []
+        # Whole-token patterns for every serial the document's records declare - see _collect_serials
+        self._serial_patterns = []
 
     def token(self, value):
         """The stable pseudonym for one value under this installation's salt."""
@@ -290,6 +297,88 @@ def _numeric_variants(self, text):
             variants.add(sign + body[:-2])
         return variants
 
+    @staticmethod
+    def _declared_serials(record):
+        """The serials one record declares for itself: hardware_ids.serial and every entry in serials."""
+        declared = set()
+        hardware_ids = record.get("hardware_ids")
+        if isinstance(hardware_ids, dict) and hardware_ids.get("serial") not in (None, ""):
+            declared.add(str(hardware_ids["serial"]))
+        serials = record.get("serials")
+        if isinstance(serials, list):
+            declared |= {str(serial) for serial in serials if serial not in (None, "")}
+        return declared
+
+    def _collect_serials(self, node):
+        """Every serial any record in the document declares, anywhere in it.
+
+        Collected before the walk so the shape guard can leave a serial readable wherever it
+        turns up - a device_id built from it, the duplicate_serial observation that names it -
+        not only inside the record that declared it.
+        """
+        found = set()
+        if isinstance(node, dict):
+            found |= self._declared_serials(node)
+            for value in node.values():
+                found |= self._collect_serials(value)
+        elif isinstance(node, list):
+            for entry in node:
+                found |= self._collect_serials(entry)
+        return found
+
+    def _strip_serials(self, text):
+        """`text` with every declared serial removed where it stands as a whole token.
+
+        Whole-token only - bounded by a non-alphanumeric character or the string's ends, matched
+        case-insensitively as entity ids fold case - so "solis:1031260253072197" and
+        "sensor.predbat_solis_1031260253072197_soc" lose the serial, but a short serial that
+        happens to sit inside a longer number does not split that number's digit run and let a
+        misfiled MPAN through.
+        """
+        for pattern in self._serial_patterns:
+            text = pattern.sub("", text)
+        return text
+
+    def _pseudonym_values(self, node):
+        """Every value held in a pseudonym container (account_ids) anywhere in this record."""
+        values = set()
+        if isinstance(node, dict):
+            for key, value in node.items():
+                if key in PSEUDONYM_CONTAINERS and isinstance(value, dict):
+                    values |= {str(entry) for entry in value.values()}
+                else:
+                    values |= self._pseudonym_values(value)
+        elif isinstance(node, list):
+            for entry in node:
+                values |= self._pseudonym_values(entry)
+        return values
+
+    @staticmethod
+    def _whole_token(text):
+        """A case-insensitive pattern matching `text` only where it stands as a whole token.
+
+        Bounded by a non-alphanumeric character or the string's ends, so "2306178123" is found in
+        "deye:2306178123" and "sensor.predbat_solis_2306178123_soc" but not inside "inv98765432".
+        """
+        return re.compile(r"(?<![0-9A-Za-z]){}(?![0-9A-Za-z])".format(re.escape(text)), re.IGNORECASE)
+
+    def _serial_derived(self, record):
+        """Whether a record's device_id is built from its own serial rather than from an account identifier.
+
+        True only when a serial the record declares stands in the device_id as a whole token and
+        none of the record's own pseudonym values does - whole tokens both ways, as the shape
+        guard matches serials (see _strip_serials). A serial merely sitting inside a longer token
+        is coincidence, not construction; and an account id counts however short it is, since one
+        under MIN_SUBSTITUTE is not otherwise caught by the substring pass. Deye's "deye:{serial}"
+        beside a station id in account_ids is serial-derived; Octopus's "octopus:{mpan}" declares
+        no serial and a device_id that embeds an account id does, so both stay identity-derived
+        exactly as before.
+        """
+        device_id = record["device_id"]
+        if not any(self._whole_token(serial).search(device_id) for serial in self._declared_serials(record)):
+            return False
+        return not any(self._whole_token(value).search(device_id) for value in self._pseudonym_values(record) if value)
+
     def _misfiled(self, value, strict_numeric=False):
         """Whether a value looks like an identifier rather than a measurement, a vendor code, or ordinary text.
 
@@ -307,6 +396,10 @@ def _misfiled(self, value, strict_numeric=False):
         industry-standard shape: "HV2160123456") is that container's entire declared purpose, not
         a misfiling; a BARE all-digit string there is still genuinely suspicious, since an MPAN
         misfiled where a serial belongs looks exactly like one.
+
+        A serial some record declares is never a misfiling: it is removed (as a whole token - see
+        _strip_serials) before any of the above is judged, so a declared all-digit serial and a
+        device_id built from one stay readable while anything left beside them is still checked.
         """
         if isinstance(value, bool):
             return False
@@ -315,8 +408,10 @@ def _misfiled(self, value, strict_numeric=False):
                 integer_part = abs(int(value))
             except (ValueError, OverflowError):
                 return False
+            if not self._strip_serials(str(integer_part)):
+                return False
             return len(str(integer_part)) >= 10
-        text = str(value)
+        text = self._strip_serials(str(value))
         if "@" in text:
             return True
         stripped = SEPARATOR_RE.sub("", text)
@@ -414,7 +509,7 @@ def _walk(self, node, container=None):
         CONTAINER_SPEC.
         """
         if isinstance(node, dict):
-            if isinstance(node.get("device_id"), str) and self._has_pseudonym_container(node):
+            if isinstance(node.get("device_id"), str) and self._has_pseudonym_container(node) and not self._serial_derived(node):
                 self._note(node["device_id"], substring=True)
             out = {}
             for key, value in node.items():
@@ -527,6 +622,9 @@ def redact(self, catalogue):
         otherwise be free to corrupt it with.
         """
         generated = catalogue.get("generated")
+        # Longest first, so a serial that contains a shorter one is removed whole
+        serials = sorted(self._collect_serials(catalogue), key=len, reverse=True)
+        self._serial_patterns = [self._whole_token(serial) for serial in serials]
         walked = self._walk(catalogue)
         self._substring_order = sorted(self.substring_ok, key=len, reverse=True)
         substituted = self._substitute(walked)
 
@@ -24,6 +24,7 @@
 import json
 import os
 from component_base import ComponentBase
+from coordinator import inverter_record
 from mock_base import MockBase
 from oauth_mixin import OAuthMixin
 from tou_schedule import TouScheduleMixin
@@ -1216,6 +1217,74 @@ async def automatic_config(self):
         self.set_arg("scheduled_discharge_enable", [self._control_name("switch", sn, "battery_schedule_export_enable") for sn in devices])
         self.set_arg("schedule_write_button", [self._control_name("switch", sn, "battery_schedule_charge_write") for sn in devices])
 
+    def build_discovery(self):
+        """
+        Describe the discovered Deye inverters for the discovery catalogue.
+
+        Reads only state the component already holds - device_list, station_ids,
+        device_rated_power and the battery accessors - so this adds no API calls and cannot
+        change what Deye does. Reporting is independent of self.automatic.
+
+        automatic_config() registers every serial in device_list (already filtered to deviceType
+        "INVERTER") as "DeyeCloud" with no further test, and binds both PV and battery entities for
+        each. Deye's API offers no way to tell a PV-only unit from a hybrid, so neither can this:
+        every record carries inverter_type "DeyeCloud" and functions solar and battery, mirroring
+        automatic_config() as the source of truth. That is exactly what Predbat believes about the
+        device - and the fact a maintainer needs when a PV-only unit has been configured as a
+        battery inverter.
+
+        Ratings: RatedPower (W) as inverter_w; battery_capacity() (kWh) as battery_kwh; and
+        config/battery's battCapacity as battery_capacity_ah, the raw Ah the API returned, so a
+        reader can check the kWh against its inputs. derive_battery_capacity() is never called
+        here: it logs and writes device_pack_voltage/device_capacity, whereas battery_capacity()
+        only reads them. Both battery ratings follow the latest poll - device_pack_voltage is set
+        from the BMS charge-voltage request each cycle, the same derivation soc_max already relies
+        on - so a changed request or a failed battery fetch changes them and re-files the report;
+        accepted, because it is the same derivation the component uses for soc_max.
+
+        station_ids goes in account_ids only when the account has exactly one station:
+        get_device_list() queries every station at once and flattens the result, so which device
+        belongs to which station is not held, and attributing one of several would be a guess.
+
+        Deliberately not reported: model and firmware (not held - device/measurePoints and
+        station/latest are fetched for debug logging and discarded, and reading them would be a
+        new API dependency).
+
+        Returns None when no inverter has been discovered yet.
+        """
+        if not self.device_list:
+            return None
+
+        account_ids = {"station_id": self.station_ids[0]} if len(self.station_ids) == 1 else None
+
+        inverters = []
+        for sn in self.device_list:
+            ratings = {}
+            rated_w = self._as_float(self.device_rated_power.get(sn), 0.0)
+            if rated_w > 0:
+                ratings["inverter_w"] = rated_w
+            battery_kwh = self.battery_capacity(sn)
+            if battery_kwh > 0:
+                ratings["battery_kwh"] = battery_kwh
+            configured_ah = self._battery_config_value(sn, "capacity")
+            if configured_ah > 0:
+                ratings["battery_capacity_ah"] = configured_ah
+
+            inverters.append(
+                inverter_record(
+                    "deye:{}".format(sn),
+                    inverter_type="DeyeCloud",
+                    composition="direct",
+                    functions=["solar", "battery"],
+                    capabilities=["schedule", "target_soc", "discharge_target", "charge_rate_power"],
+                    hardware_ids={"serial": sn},
+                    account_ids=account_ids,
+                    ratings=ratings,
+                )
+            )
+
+        return {"automatic": self.automatic, "inverters": inverters}
+
     @staticmethod
     def _age_text(age):
         """Render a cache age in minutes for logging, tolerating an unknown age."""
@@ -1529,6 +1598,14 @@ async def run(self, seconds, first):
         for sn in self.device_list:
             await self.publish_schedule_settings_ha(sn)
 
+        # Filed right after this cycle's publish and BEFORE `if first and not live_ok:` below:
+        # that branch returns False to defer startup when the first live poll fails, and
+        # automatic_config() follows it, so a report filed any later would never be filed on
+        # exactly the installs whose dump most needs to say what hardware was found. (Deye's
+        # automatic_config() does not raise - the early return is the reason.) refresh_discovery()
+        # owns the compare/retry/guard loop and never raises.
+        self.refresh_discovery()
+
         # Drain any control orders left pending by apply_dynamic_control() every cycle (not
         # just on first run) so a write that is HTTP-accepted but then fails to apply on the
         # device doesn't stay masked forever behind the applied-payload change-detection cache.