SMA-2 round 2: keep the driver keys out of the schema, translations and labels

Owner decision 2026-11-07: clients see nothing new until SMA-4. config.yaml,
translations, labels and the page layout are back to 23ff9d0's. The validation
of the keys stays for callers that skip the schema.

Co-Authored-By: Claude <noreply@anthropic.com>

goodwe_controller/CHANGELOG.md

@@ -2,8 +2,11 @@
 
 ### Not in a release yet
 
+- Each inverter is now reached through a driver, so another make can be added later. Nothing changes for a GoodWe site ([[SMA-2]]).
 - Home now draws today's charge from the grid for every hour the add-on will really charge, not only the run under way ([[UX-77]]).
 - When Home Assistant passes readings on late, control no longer swings on them: it waits until the battery shows its last power request, and the peak limit no longer asks for full power on them. The page says so, and if the readings stay too late every inverter goes to 0 W ([[SAFETY-128]]).
+- With a 5 s meter, readings 30 s late are now noticed within 100 seconds, not 3 minutes ([[SAFETY-129]]).
+- If one of the tomorrow solar forecast sensors stops updating in the afternoon, tomorrow's forecast now stays on the Forecast tab instead of vanishing until 20:00 ([[OPT-38]]).
 
 ## 0.20.1
 

goodwe_controller/DECISIONS/OPT-27-last-fresh-forecast-for-the-night.md

@@ -44,3 +44,5 @@ tonight." (`home_words.solar.kept`, EN and NL). The event log says
 **Not changed.** Control with modes off: it does not use the forecast. The
 screens still show the line, so it says "kept", not "used for the plan". A
 forecast that stays fresh. The hold does not look at the weather, only at the clock and the day.
+
+**OPT-38 (2026-11-07).** A kept TOMORROW reading (at most `KEPT_MAX_AGE_S` old, same `day0`) now stands in at any hour, not only from `BELL_TO_H`: one sensor that stopped at 14:54 had dropped the whole sum (OPT-17) from 17:54 to 20:00. Today readings and the OPT-34 morning rules are unchanged; `test_opt38.py`.

goodwe_controller/DECISIONS/SAFETY-128-late-readings.md

@@ -31,7 +31,12 @@ inverter limit in its last seconds (OPT-34 rig F-2).
    - **Meter** (`MeterFit`, round 1's fit): `meter(t) + request(t - d)` is the
      house alone at the right `d`. Quick, but confounded by the house. Needs
      16 new values in 60 s, so only a meter of about 3.5 s or faster: at 5 s
-     it gave false delays. Late over 8 s.
+     it gave false delays. Late over 8 s. At a slower meter (a 5 s P1) the
+     same fit runs over 120 s and calls only a delay over 16 s late, on 2 fits
+     in a row, and never ends a verdict (SAFETY-129). Alone it reads 35-90 s on healthy
+     runs, a few times in a row; only the slow-start and battery-fit check
+     below keeps those from a verdict. 10 s late is not found: it cannot be
+     told from a 10 s slow battery, and it does not swing at 5 s.
    - A meter fit says late only with evidence from the batteries' side too:
      a request more than 100 W from the battery reading took 3 s or more to
      move a reading towards it (`SLOW_START_S`; on time they start within
@@ -64,9 +69,12 @@ house at 1, 2 and 5 s ticks, so the commands are the same as without the gate.
 The fits run all the time; the pull request has the counts.
 
 **What it costs.**
-- Found 36-68 s after a lag starts at 1-2 s meters, about 3 minutes at a 5 s
-  meter (battery fit only; the meter fit needs a faster meter). Until then control is as before. At 5 s and 30 s
-  late the export is cut to about a third, not removed.
+- Found 36-68 s after a lag starts at 1-2 s meters; at a 5 s meter 30 s late
+  is found within 100 s after it starts (88 s measured) (the loop first moves 100 W after 60 s,
+  SAFETY-129) with the export above -200 W, and 10 s late is not found. Until
+  then control is as before. In a quiet house the "readings late" row stays
+  about 2 minutes longer after the lag ends (126 s against 32 s on base); no
+  harm measured, and it clears as early as on base when the load changes.
 - A battery that answers slowly is read as late readings and held: a dead
   time of 10 s or more (8 s reads 7-8; 10 s reads 10 at 1-2 s meters, 9 at
   5 s), or a 15-20 s lag (review round 2). The GoodWe measured settles in 6 s

goodwe_controller/DECISIONS/SMA-2-driver-seam.md

@@ -0,0 +1,55 @@
+# SMA-2 - each inverter is reached through a driver
+
+**Status: for review.** Tests: `test_sma2_driver.py`. Source: SMA-9 report
+section 1.1. A pure refactor: no behaviour change for GoodWe, no change to a
+command, a log line or a sentence the add-on shows.
+
+Every place that read or steered an inverter used the GoodWe ESPHome entities
+directly, so a second make meant touching all of them. `app/drivers.py` now
+holds one `Driver` per roster entry (`Inverter.driver`): `configured()`,
+`actuator_id()`, `limits()`, `read()`, `write()`, `release()`, `probe()`, and
+`link_stale_s`, `batt_bound`, `settle_s`, `brand`, `words`. `GoodWeEsphomeDriver`
+makes the same calls in the same order as the code it replaced; its
+`release()` is `write(0)`.
+
+**Moved to the driver:** the SoC and power read (`_read_unit`), the write
+(`_write_unit`, still under the SAFETY-28 gate, the clamp, shadow mode and the
+write budget), the device range (`run_control`, `amain`, the settings panel),
+liveness and its window (`has_writer`, `is_live`, `link_note`), the freshness
+bound, the status rows' "is a source set" test, rule 2's per-unit range check
+and SAFETY-28's census. **Not moved:** the clamp, the gate, the heartbeat and
+`command_site_zero()`; stop is still 0 W for every unit, and nothing calls
+`release()` yet (SMA-10). `settle_s` and the control constants are not read
+from the driver yet either: with only GoodWe they would change nothing.
+
+**The seam is not TCP-shaped.** The census asks each entry's driver for its
+actuator id: the setpoint entity for GoodWe, `host:port:unit_id` for a driver
+reached over a socket, so two units behind one gateway are two actuators and
+one unit named twice is the SAFETY-28 refusal. The census still counts every
+`setpoint_entity` key as before and adds the driver's own actuator to it (parity
+with 23ff9d0 on 349 option shapes is a test). The driver binds its Home
+Assistant on first use by the Controller, so a unit added later is wired too.
+
+**Schema, and why the new keys are not in it.** The brief asks for
+`driver`, `host`, `port`, `unit_id`, `maintenance` per entry and for
+`setpoint_entity` to become optional. Owner decision 2026-11-07: clients see
+nothing new until SMA-4/SMA-7 use the keys. The Supervisor form shows every
+schema key (and marks required ones), and a nested entry with a key the schema
+lacks has that key silently dropped (measured, shape probe section K), so the
+only way to hide them is to keep them out of config.yaml. So for now config.yaml, the
+translations and the labels are exactly 23ff9d0's, and `setpoint_entity` stays
+required there: until a driver without one is selectable, no entry can lack it.
+The code underneath is the brief's: the driver defaults to `goodwe_esphome`,
+rule 9(d) is per driver (a GoodWe entry without a setpoint entity is refused,
+same sentence), and rule 9(h) checks `driver`, `host`, `port`, `unit_id`,
+`maintenance` when a caller that skips the schema (the panel's save path,
+sim/inject, the checks) supplies them; those sentences name the key, having no
+label. When SMA-4 needs them: add them to the schema as optional with
+translations, make `setpoint_entity` `str?`, and add them to `uilayout` as card
+fields or a hidden list.
+
+**Checked against 23ff9d0:** replay digests and a log of every Home Assistant
+call are byte-identical at N=1 and N=2; the TEL-07 announce diff is 0/0/0; a
+fake driver beside a GoodWe unit allocates, grades, drops and publishes per
+unit; a call site that goes round the seam is caught.
+

goodwe_controller/app/drivers.py

@@ -0,0 +1,299 @@
+"""The inverter driver seam (SMA-2): how this add-on reads and steers ONE inverter.
+
+Before this, every place that read or steered an inverter talked to the GoodWe
+ESPHome entities itself (`hass.limits`, `hass.number`, `hass.set_number`, the
+SAFETY-28 census, selfconfig, validate), so a second brand meant touching all of
+them. Now each `Inverter` holds a `Driver`, and those places ask it. This card
+is a pure refactor: `GoodWeEsphomeDriver` does exactly what the code it replaced
+did, in the same order, with the same arguments. DECISIONS/SMA-2-driver-seam.md.
+
+THE SEAM IS NOT TCP-SHAPED. Nothing here assumes a connection, or one connection
+per host: a later FoxESS unit is RTU framing over a TCP socket, and its
+`actuator_id()` is host:port:unit_id, so two units behind one gateway are two
+actuators and one unit named twice is the SAFETY-28 refusal.
+
+WHAT STAYS OUTSIDE THE DRIVER, on purpose. The gate (SAFETY-28), the clamp
+(SAFETY-03), shadow mode and the write budget all sit above `write()` in
+`Controller._write_unit`, and the freshness gate sits above `read()`; a driver
+never sees a value those did not pass. A driver moves one number to or from one
+inverter and says whether it worked.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from .control import PLANT_SETTLE_S
+from .freshness import BOUNDS
+
+# How long every write to one GoodWe inverter must have been REJECTED before
+# this controller stops claiming it can command it. Layer 1's own window: the
+# ESP32 commands 0 W after ~30 s without a refresh (`validate.
+# FIRMWARE_INGEST_VETO_S`). `inverter.LINK_STALE_S` is this number, re-exported.
+GOODWE_LINK_STALE_S = 30.0
+
+# `driver` when an entry does not say: every install that exists today.
+DEFAULT_DRIVER = "goodwe_esphome"
+
+# The `driver` values the options accept. Only the GoodWe one for now; the
+# registry below may hold more (a test registers a fake), but this tuple is
+# what validate.py rule 9 lets through and what config.yaml's schema lists.
+SELECTABLE = (DEFAULT_DRIVER,)
+
+
+class DriverNotBound(RuntimeError):
+    """A driver was asked to talk to the world before it was given a Home
+    Assistant. A wiring bug, so it is loud rather than a quiet "no answer"."""
+
+
+@dataclass(frozen=True)
+class Probe:
+    """What a read-only look at one inverter found. None means "did not answer"."""
+
+    dev_min: float | None = None
+    dev_max: float | None = None
+    soc: float | None = None
+    batt_w: float | None = None
+    extra: dict = field(default_factory=dict)
+
+
+def census_key(actuator_id: str) -> str:
+    """The form SAFETY-28 compares: stripped and lowercase, so a config cannot
+    name one actuator twice by changing its case (`Number.X` vs `number.x`)."""
+    return str(actuator_id or "").strip().lower()
+
+
+class Driver:
+    """One inverter's I/O. Subclasses set the class attributes and the methods.
+
+    Constructed once per roster entry, with no Home Assistant: `bind()` gives
+    it one (the Controller does, and so does the settings panel, which has no
+    Controller). Everything that talks to the world is a coroutine.
+    """
+
+    #: the `driver` option value that selects this class
+    name = ""
+    #: the option key that names the actuator in an entry (the census reads it)
+    writer_key = ""
+    #: the brand word the per-driver sentences and device names use
+    brand = ""
+    #: per-driver sentences (SMA-10 and later). Empty until a caller needs one.
+    words: dict = {}
+    #: how long all-rejected writes last before the unit stops being live
+    link_stale_s = GOODWE_LINK_STALE_S
+    #: this driver's `freshness.Bound` for the battery power reading
+    batt_bound = BOUNDS["batt"]
+    #: the plant's settling time after a step, in s (control.PLANT_SETTLE_S)
+    settle_s = PLANT_SETTLE_S
+
+    # -- from the options ----------------------------------------------------
+    @classmethod
+    def from_entry(cls, entry: dict) -> "Driver":
+        raise NotImplementedError
+
+    @classmethod
+    def actuator_id_of(cls, entry: dict) -> str:
+        """The actuator an options entry names, or "" for none. Pure: the
+        SAFETY-28 census runs on options, before any driver exists."""
+        raise NotImplementedError
+
+    # -- wiring --------------------------------------------------------------
+    def bind(self, hass) -> "Driver":
+        """Give this driver its Home Assistant (or whatever it talks through)."""
+        return self
+
+    def bind_getter(self, getter) -> "Driver":
+        """Like `bind()`, but looked up at each use: the Controller passes
+        `lambda: self.hass`, so replacing `controller.hass` is seen at once."""
+        return self
+
+    def bound(self) -> bool:
+        """Has it been given what it talks through? A driver that talks
+        through nothing the Controller owns is always bound."""
+        return True
+
+    # -- what it is ----------------------------------------------------------
+    def configured(self) -> bool:
+        """Is there any way to reach this inverter at all?"""
+        raise NotImplementedError
+
+    def actuator_id(self) -> str:
+        """What this driver writes to. Two units with one id are one actuator."""
+        raise NotImplementedError
+
+    def soc_source(self) -> str:
+        """Where the pack level is read from; "" when nowhere. Status rows only."""
+        return ""
+
+    def batt_source(self) -> str:
+        """Where the battery power is read from; "" when nowhere."""
+        return ""
+
+    # -- what it does --------------------------------------------------------
+    async def limits(self) -> tuple:
+        """`(dev_min, dev_max)` in W the device accepts; (None, None) unread."""
+        raise NotImplementedError
+
+    async def read(self) -> tuple:
+        """`(soc, batt_w, extra)`. A value that did not arrive is None. `batt_w`
+        is the RAW power, in the sign the controller uses; the freshness gate
+        is the caller's."""
+        raise NotImplementedError
+
+    async def write(self, w: float) -> bool:
+        """Put `w` on the inverter. True only when it took it."""
+        raise NotImplementedError
+
+    async def release(self) -> bool:
+        """Stop steering this inverter. For a driver with nothing better, 0 W."""
+        return await self.write(0.0)
+
+    async def probe(self) -> Probe:
+        """A read-only look, usable without a Controller (panel, wizard)."""
+        raise NotImplementedError
+
+
+class GoodWeEsphomeDriver(Driver):
+    """The GoodWe behind the ESP32: three Home Assistant entities.
+
+    A transcription of what `main.py` and `selfconfig.py` did inline: `limits()`
+    is `hass.limits(setpoint)`, `read()` is `hass.number(soc)` then
+    `hass.number(batt, invert)`, `write()` is `hass.set_number(setpoint, w)`.
+    `release()` is `write(0)`: the board's own watchdog is what hands back.
+    """
+
+    name = DEFAULT_DRIVER
+    writer_key = "setpoint_entity"
+    brand = "GoodWe"
+
+    def __init__(self, setpoint_entity="", soc_entity="", batt_entity="",
+                 batt_invert=False):
+        self.setpoint_entity = str(setpoint_entity or "")
+        self.soc_entity = str(soc_entity or "")
+        self.batt_entity = str(batt_entity or "")
+        self.batt_invert = bool(batt_invert)
+        self._hass = None
+        self._getter = None
+
+    @classmethod
+    def from_entry(cls, entry: dict) -> "GoodWeEsphomeDriver":
+        return cls(setpoint_entity=entry.get("setpoint_entity", ""),
+                   soc_entity=entry.get("soc_entity", ""),
+                   batt_entity=entry.get("batt_entity", ""),
+                   batt_invert=bool(entry.get("batt_invert")))
+
+    @classmethod
+    def actuator_id_of(cls, entry: dict) -> str:
+        return str(entry.get("setpoint_entity") or "").strip()
+
+    def bind(self, hass) -> "GoodWeEsphomeDriver":
+        self._hass, self._getter = hass, None
+        return self
+
+    def bind_getter(self, getter) -> "GoodWeEsphomeDriver":
+        self._hass, self._getter = None, getter
+        return self
+
+    def bound(self) -> bool:
+        return self._hass is not None or self._getter is not None
+
+    def _h(self):
+        hass = self._getter() if self._getter is not None else self._hass
+        if hass is None:
+            raise DriverNotBound(f"{self.brand} driver for "
+                                 f"{self.setpoint_entity!r} has no Home Assistant")
+        return hass
+
+    def configured(self) -> bool:
+        return bool(self.setpoint_entity)
+
+    def actuator_id(self) -> str:
+        return self.setpoint_entity
+
+    def soc_source(self) -> str:
+        return self.soc_entity
+
+    def batt_source(self) -> str:
+        return self.batt_entity
+
+    async def limits(self) -> tuple:
+        dev_min, dev_max, _step = await self._h().limits(self.setpoint_entity)
+        return dev_min, dev_max
+
+    async def read(self) -> tuple:
+        hass = self._h()
+        soc = await hass.number(self.soc_entity)
+        raw = await hass.number(self.batt_entity, self.batt_invert)
+        return soc, raw, {}
+
+    async def write(self, w: float) -> bool:
+        return await self._h().set_number(self.setpoint_entity, w)
+
+    async def probe(self) -> Probe:
+        dev_min, dev_max = await self.limits()
+        soc, raw, extra = await self.read()
+        return Probe(dev_min, dev_max, soc, raw, extra)
+
+
+class UnavailableDriver(Driver):
+    """What an entry naming a driver this build does not have gets. Validation
+    refuses such an entry before a Controller exists; this is only what
+    `build_roster()` (which refuses nothing) builds meanwhile: a unit that is
+    not configured, so the refusal branch names it and commands 0 W."""
+
+    name = ""
+    # The census still counts a `setpoint_entity` on such an entry: refusing
+    # too much is the safe direction for SAFETY-28.
+    writer_key = "setpoint_entity"
+    brand = ""
+
+    def __init__(self, wanted=""):
+        self.wanted = str(wanted or "")
+
+    @classmethod
+    def from_entry(cls, entry: dict) -> "UnavailableDriver":
+        return cls(entry.get("driver"))
+
+    @classmethod
+    def actuator_id_of(cls, entry: dict) -> str:
+        return str(entry.get("setpoint_entity") or "").strip()
+
+    def configured(self) -> bool:
+        return False
+
+    def actuator_id(self) -> str:
+        return ""
+
+    async def limits(self) -> tuple:
+        return None, None
+
+    async def read(self) -> tuple:
+        return None, None, {}
+
+    async def write(self, w: float) -> bool:
+        return False
+
+    async def probe(self) -> Probe:
+        return Probe()
+
+
+# name -> class. The options' `driver` value picks one. A test may add a fake;
+# `SELECTABLE` (not this) is what the options accept.
+REGISTRY = {GoodWeEsphomeDriver.name: GoodWeEsphomeDriver}
+
+
+def driver_name(entry) -> str:
+    """The `driver` an options entry selects; blank or absent is the default."""
+    raw = entry.get("driver") if isinstance(entry, dict) else None
+    return str(raw).strip() if raw not in (None, "") else DEFAULT_DRIVER
+
+
+def driver_class(entry):
+    """The class an options entry selects, `UnavailableDriver` if none."""
+    return REGISTRY.get(driver_name(entry), UnavailableDriver)
+
+
+def build_driver(entry) -> Driver:
+    """The driver for ONE options entry (or the flat options, for one inverter)."""
+    entry = entry if isinstance(entry, dict) else {}
+    return driver_class(entry).from_entry(entry)

goodwe_controller/app/inverter.py

@@ -124,9 +124,12 @@ from datetime import datetime
 from .clamp import ClampResult
 from .delivery import DeliveryConfig, DeliveryWatchdog
 from . import messages
+# SMA-2. The driver seam: how this unit's inverter is read and steered.
+from .drivers import (GOODWE_LINK_STALE_S, GoodWeEsphomeDriver,
+                      build_driver)
 # SAFETY-37. The freshness policy, and the ONE place its bounds are written.
 # freshness.py imports nothing from this package, so the direction is one-way.
-from .freshness import BOUNDS, Freshness
+from .freshness import Freshness
 from .idle import IdleReading
 # CFG-03's ONE in-code mirror of config.yaml's SAFETY-24 defaults. Imported
 # rather than retyped: the flat-keys path below needs `bms_soc_floor`'s
@@ -149,7 +152,11 @@ from .watchdog import SignWatchdog, SignWatchdogConfig
 # decided it will not. No config key: CFG-03 tracks three places a default has
 # to be kept in step and this does not earn a fourth - it is derived from a
 # constant the firmware and validate.py already share.
-LINK_STALE_S = 30.0
+#
+# SMA-2. This is the GoodWe driver's `link_stale_s` (drivers.py), kept under its
+# old name for the checks and for main.py. `is_live()` and `link_note()` now ask
+# each unit's DRIVER for its window; a driver for another make states its own.
+LINK_STALE_S = GOODWE_LINK_STALE_S
 
 # The name of the inverter a SINGLE-INVERTER config describes. ⚠️ A NAME, NOT A
 # CONFIGURATION - it exists so that a per-inverter claim has an actuator to be
@@ -195,12 +202,14 @@ class Inverter:
     def __init__(self, name, *, setpoint_entity="", soc_entity="",
                  batt_entity="", batt_invert=False, max_w=2000.0,
                  bms_soc_floor=None, capacity_kwh=None, store=None,
-                 sign_cfg=None, delivery_cfg=None):
+                 sign_cfg=None, delivery_cfg=None, driver=None):
         self.name = str(name)
-        self.setpoint_entity = str(setpoint_entity or "")
-        self.soc_entity = str(soc_entity or "")
-        self.batt_entity = str(batt_entity or "")
-        self.batt_invert = bool(batt_invert)
+        # SMA-2. HOW THIS UNIT IS READ AND STEERED. The entity keywords above
+        # are the GoodWe driver's, kept so every existing caller builds the
+        # same unit; a roster entry that selects another driver passes
+        # `driver` and the entity keywords stay empty.
+        self.driver = driver if driver is not None else GoodWeEsphomeDriver(
+            setpoint_entity, soc_entity, batt_entity, batt_invert)
         # The supervised cap on what may be written to THIS setpoint entity.
         # See the module docstring on why this is a number and not a `Tuning`.
         self.max_w = float(max_w)
@@ -256,7 +265,7 @@ class Inverter:
         # constructed here: SoC's obligation term is energy through the pack, not
         # command travel, and the constant it needs has not been measured. That
         # is TEL-08's, and freshness.py's docstring says exactly what it owes.
-        self.batt_fresh = Freshness(BOUNDS["batt"])
+        self.batt_fresh = Freshness(self.driver.batt_bound)
 
         # --- what this unit was last told, and what it last accepted ---------
         # `target` is THIS unit's share of the site target from the last
@@ -338,6 +347,50 @@ class Inverter:
     def idle_w(self) -> float:
         return self.idle.w
 
+    # -- the GoodWe driver's entities, read through for the callers that name
+    # them (the panel's range check, the status rows, the checks). A driver
+    # without such an entity answers "" - the same as "not set".
+    @property
+    def setpoint_entity(self) -> str:
+        return getattr(self.driver, "setpoint_entity", "")
+
+    @property
+    def soc_entity(self) -> str:
+        return getattr(self.driver, "soc_entity", "")
+
+    @property
+    def batt_entity(self) -> str:
+        return getattr(self.driver, "batt_entity", "")
+
+    @property
+    def batt_invert(self) -> bool:
+        return bool(getattr(self.driver, "batt_invert", False))
+
+    # Assigning one writes through to the driver, as the attributes did before
+    # SMA-2 (nothing in the repo assigns them today; this keeps it working).
+    # A driver that has no such field refuses, rather than losing the value.
+    def _set_driver_field(self, name, value):
+        if not hasattr(self.driver, name):
+            raise AttributeError(
+                f"{type(self.driver).__name__} has no {name}")
+        setattr(self.driver, name, value)
+
+    @setpoint_entity.setter
+    def setpoint_entity(self, v):
+        self._set_driver_field("setpoint_entity", str(v or ""))
+
+    @soc_entity.setter
+    def soc_entity(self, v):
+        self._set_driver_field("soc_entity", str(v or ""))
+
+    @batt_entity.setter
+    def batt_entity(self, v):
+        self._set_driver_field("batt_entity", str(v or ""))
+
+    @batt_invert.setter
+    def batt_invert(self, v):
+        self._set_driver_field("batt_invert", bool(v))
+
     # -- store ---------------------------------------------------------------
     def key(self, name: str) -> str:
         """This unit's namespaced Store key for `name`. See store.unit_key()."""
@@ -387,7 +440,7 @@ class Inverter:
         return out
 
     def has_writer(self) -> bool:
-        return bool(self.setpoint_entity)
+        return self.driver.configured()
 
     # -- liveness ------------------------------------------------------------
     def note_write(self, ok: bool, now: datetime) -> None:
@@ -398,7 +451,7 @@ class Inverter:
         elif self.rejected_since is None:
             self.rejected_since = now
 
-    def is_live(self, now: datetime, stale_s: float = LINK_STALE_S):
+    def is_live(self, now: datetime, stale_s: float | None = None):
         """`(live, why_not)` - may this controller command this unit THIS TICK?
 
         Read the module docstring before widening this - in particular on why a
@@ -411,7 +464,11 @@ class Inverter:
 
         ⚠️ ANSWERED PER TICK, NEVER CACHED, for `allocator.plan()`'s reason: a
         link goes stale between one cycle and the next.
+
+        `stale_s` is the driver's `link_stale_s` unless a caller passes one.
         """
+        if stale_s is None:
+            stale_s = self.driver.link_stale_s
         if self.rejected_since is not None:
             held = (now - self.rejected_since).total_seconds()
             if held >= float(stale_s):
@@ -451,9 +508,11 @@ class Inverter:
         held = (0.0 if self.rejected_since is None
                 else (now - self.rejected_since).total_seconds())
         if not live:
-            return messages.row_not_controllable(self.name, held, LINK_STALE_S)
+            return messages.row_not_controllable(self.name, held,
+                                                 self.driver.link_stale_s)
         if self.rejected_since is not None:
-            return messages.row_commands_refused(self.name, held, LINK_STALE_S)
+            return messages.row_commands_refused(self.name, held,
+                                                 self.driver.link_stale_s)
         return ""
 
     def __repr__(self) -> str:
@@ -556,10 +615,8 @@ def build_roster(opts: dict, store, sign_cfg=None,
         # identically, which is AC 3.
         return [Inverter(
             UNIT_A,
-            setpoint_entity=opts.get("setpoint_entity", ""),
-            soc_entity=opts.get("soc_entity", ""),
-            batt_entity=opts.get("batt_entity", ""),
-            batt_invert=bool(opts.get("batt_invert")),
+            # SMA-2. The flat keys are the GoodWe shape and only that.
+            driver=GoodWeEsphomeDriver.from_entry(opts),
             max_w=_flat_max_w(opts),
             # SAFETY-32. THE FLAT KEY, because this IS the single-inverter shape
             # - the one pack the flat options describe is the one pack whose
@@ -588,10 +645,9 @@ def build_roster(opts: dict, store, sign_cfg=None,
         entry = entry if isinstance(entry, dict) else {}
         roster.append(Inverter(
             name,
-            setpoint_entity=entry.get("setpoint_entity", ""),
-            soc_entity=entry.get("soc_entity", ""),
-            batt_entity=entry.get("batt_entity", ""),
-            batt_invert=bool(entry.get("batt_invert")),
+            # SMA-2. The entry's `driver` (default goodwe_esphome) picks the
+            # class; the entry itself says where its entities are.
+            driver=build_driver(entry),
             # ⚠️ NOT `opts["max_w"]`. See the docstring: a rating an entry did
             # not state is not the flat one, it is a rating rule 9 refuses.
             max_w=_entry_max_w(entry),

goodwe_controller/app/lateness.py

@@ -10,7 +10,8 @@ already fixed. The readings show it: they answer each power request late.
   house does not move that reading, so a repeating load cannot fool it. It
   needs the request to move FIT_MOVE_W; one reading per 10 s makes it slow.
 - `MeterFit`: how long after the site request the meter shows it. Quick at a
-  1-3 s meter, none at a slower one, and a house load that repeats can fool it.
+  1-3 s meter, slower and for long lags only at a 5 s one (SAFETY-129), and a
+  house load that repeats can fool it.
 The meter counts only when an ask was slow to start and no battery fit said
 "on time"; with hysteresis (ON_S / OFF_S) the result is `late_seen`. Then (`observe()`, each cycle):
 - a new command more than ANSWER_W from the battery reading is PENDING until
@@ -163,13 +164,22 @@ class Delay:
 
 
 # The meter's delay: fitted over this much history, from this many new
-# values (a meter of about 3.5 s or faster: a slower one gives too few to
-# fit without false delays); late over M_LATE_S (the plant's 6 s settle and one 2 s meter tick),
-# but only when the batteries' own readings have not just shown them on time.
+# values (a meter of about 3.5 s or faster; a slower one see M_SLOW_FIT_S);
+# late over M_LATE_S (the plant's 6 s settle and one 2 s meter tick), but only
+# when the batteries' own readings have not just shown them on time.
 M_FIT_S = 60.0
 M_ROWS = 16
 M_LATE_S = freshness.PLANT_SETTLE_S + 2.0                          # 8.0
 M_SHARE, M_BETTER = 0.4, 0.7
+# SAFETY-129: a meter slower than that (a 5 s P1) has fewer than M_ROWS values
+# in M_FIT_S, so it is fitted over M_SLOW_FIT_S instead, from the same M_ROWS
+# values. It calls a delay late only over M_SLOW_LATE_S and on ON_FITS fits in
+# a row. Alone it reads 35-90 s on healthy runs, a few times in a row; what
+# keeps those from a verdict is the slow-start and battery-fit check in
+# watch(), not the limit. A 10 s lag is not found: it cannot be told from a
+# slow battery (R2-1), nor does it swing at 5 s.
+M_SLOW_FIT_S = 120.0
+M_SLOW_LATE_S = 16.0
 # A meter fit counts only after an ask that took this long to start moving a
 # reading: on time they start within 1-2 s (the recorded house, at 1-5 s).
 SLOW_START_S = 3.0
@@ -183,19 +193,30 @@ class MeterFit:
     up with an earlier request, so it only counts while the batteries' own
     fit is not on time (`Answer.watch`)."""
 
-    def __init__(self):
+    def __init__(self, fit_s=M_FIT_S, late_s=M_LATE_S):
+        self.fit_s, self.late_s = fit_s, late_s
         self.cmds = deque()      # (t, request) when it changed
         self.grid = deque()      # (t, meter) new values only
 
+    def slow(self):
+        """The meter is too slow for M_ROWS values in `fit_s`: the usual gap
+        between its new values (median, so a pause does not count) is over
+        fit_s / M_ROWS."""
+        ts = [t for t, _g in self.grid]
+        if len(ts) < 5:
+            return False
+        gaps = sorted(b - a for a, b in zip(ts, ts[1:]))
+        return gaps[len(gaps) // 2] * M_ROWS > self.fit_s
+
     def add(self, now, cmd, grid):
         if cmd is not None and (not self.cmds or self.cmds[-1][1] != cmd):
             self.cmds.append((now, float(cmd)))
         if grid is not None and (not self.grid or self.grid[-1][1] != grid):
             self.grid.append((now, float(grid)))
-        while self.grid and self.grid[0][0] < now - M_FIT_S:
+        while self.grid and self.grid[0][0] < now - self.fit_s:
             self.grid.popleft()
         while (len(self.cmds) > 1
-               and self.cmds[1][0] <= now - M_FIT_S - MAX_DELAY_S):
+               and self.cmds[1][0] <= now - self.fit_s - MAX_DELAY_S):
             self.cmds.popleft()
 
     def estimate(self):
@@ -214,9 +235,9 @@ class MeterFit:
         best = min(range(len(pure)), key=lambda d: (pure[d], d))
         if g_var <= 0 or pure[best] > (1 - M_SHARE) * g_var:
             return None
-        if best <= M_LATE_S:
+        if best <= self.late_s:
             return float(best)
-        if pure[best] > M_BETTER * min(pure[:int(M_LATE_S) + 1]):
+        if pure[best] > M_BETTER * min(pure[:int(self.late_s) + 1]):
             return None
         return float(best)
 
@@ -236,6 +257,8 @@ class Answer:
         self._was = None         # (command, reading, meter) at the last call
         self.delay = Delay()
         self.meter = MeterFit()
+        self.meter_slow = MeterFit(M_SLOW_FIT_S, M_SLOW_LATE_S)   # SAFETY-129
+        self._ms_run = 0         # late slow-meter fits in a row
         self._batt_fit = None     # (when, d): the batteries' last fit
         self._b_run = 0          # late battery fits in a row
         self._start_age = None   # (when, how long the last ask took to start)
@@ -279,14 +302,18 @@ class Answer:
         delays. Late on ON_FITS battery fits from ON_S, or on a meter fit
         over M_LATE_S when an ask was slow to start (SLOW_START_S) and no
         battery fit within FIT_S said on time: a house load that repeats can
-        fool the meter fit, not these. On time again on a battery fit under
-        OFF_S, or a meter fit within M_LATE_S unless a battery fit said late.
+        fool the meter fit, not these. A meter too slow for M_ROWS values in
+        M_FIT_S (a 5 s P1) is fitted over M_SLOW_FIT_S and counts only over
+        M_SLOW_LATE_S, on ON_FITS fits in a row. On time again on a battery fit
+        under OFF_S, or a meter fit within M_LATE_S unless a battery fit said
+        late.
         """
         if self._start is None:
             self._start = now
         for name, cmd, reading in units:
             self.delay.add(now, name, cmd, reading)
         self.meter.add(now, standing_w, grid_w)
+        self.meter_slow.add(now, standing_w, grid_w)
         if self._fit_at is not None and now - self._fit_at < FIT_EVERY_S:
             return
         self._fit_at = now
@@ -295,6 +322,9 @@ class Answer:
                 and now - self._exempt_at <= FIT_S:
             b = None          # a battery that stopped following: no late fit
         m = self.meter.estimate()
+        ms = None
+        if self.meter.slow():
+            ms = self.meter_slow.estimate()
         if b is not None:
             self._batt_fit = (now, b)
             self.shown_s = b if b >= ON_S else None
@@ -304,7 +334,13 @@ class Answer:
         b_late = recent and self._batt_fit[1] >= ON_S
 
         self._b_run = self._b_run + 1 if (b or 0.0) >= ON_S else 0
-        d = b if b is not None else m
+        self._ms_run = self._ms_run + 1 if (ms or 0.0) > M_SLOW_LATE_S else 0
+        # A slow-meter fit only ever adds a late verdict: it counts as a fit
+        # when it is late on ON_FITS in a row, and never ends one, so a slow
+        # battery (R2-1) runs exactly as before.
+        if self._ms_run < ON_FITS:
+            ms = None
+        d = b if b is not None else m if m is not None else ms
         if d is None:
             if self._fit_ok is None or now - self._fit_ok > FIT_GONE_S:
                 self.late_seen = False            # nothing to tell it by
@@ -312,12 +348,13 @@ class Answer:
         self.fits += 1
         self.delay_s, self._fit_ok = d, now
         warm = now - self._start >= MAX_DELAY_S + FIT_S
-        m_late = (m or 0.0) > M_LATE_S
+        m_late = ((m or 0.0) > M_LATE_S and warm) or (
+            self._ms_run >= ON_FITS
+            and now - self._start >= MAX_DELAY_S + M_SLOW_FIT_S)
         slow = (self._start_age is not None
                 and now - self._start_age[0] <= M_FIT_S
                 and self._start_age[1] >= SLOW_START_S)
-        if self._b_run >= ON_FITS or (
-                m_late and warm and slow and not b_fresh):
+        if self._b_run >= ON_FITS or (m_late and slow and not b_fresh):
             self.late_seen = True
         elif (b is not None and b < OFF_S) or (
                 m is not None and m <= M_LATE_S and not b_late):

goodwe_controller/app/main.py

@@ -71,6 +71,7 @@ from . import messages
 from .freeze import (
     FreezeWatchdog, FreezeWatchdogConfig, looks_frozen, rail_unreachable, rail_w,
 )
+from .drivers import GoodWeEsphomeDriver
 from .hass import CORE_API, OUTAGE, HomeAssistant
 # ⚠️ `LINK_STALE_S` IS RE-EXPORTED DELIBERATELY. `live_units()` reasons about it
 # and two checks read `app.main.LINK_STALE_S` to build a stale link, so the
@@ -3774,6 +3775,18 @@ class Controller:
         elif self.inputs_bad_since is None:
             self.inputs_bad_since = datetime.now(timezone.utc)
 
+    def _driver(self, inv):
+        """`inv`'s driver, wired to this Controller's Home Assistant on first use.
+
+        SMA-2. Bound HERE rather than when the roster is built, so a unit added
+        to `self.inverters` afterwards is wired too, and the driver follows
+        whatever `self.hass` is at the time (a check may swap it).
+        """
+        drv = inv.driver
+        if not drv.bound():
+            drv.bind_getter(lambda: self.hass)
+        return drv
+
     async def _read_unit(self, inv, now_mono: float) -> None:
         """One inverter's own SoC and battery power, from ITS OWN entities.
 
@@ -3793,8 +3806,10 @@ class Controller:
         second spelling of "what is standing" would let the two disagree about
         the number they are both reasoning about.
         """
-        inv.soc = await self.hass.number(inv.soc_entity)
-        raw = await self.hass.number(inv.batt_entity, inv.batt_invert)
+        # SMA-2. The driver reads, in the order this method always did: SoC,
+        # then power. `extra` is whatever else the driver learned; nothing
+        # consumes it yet.
+        inv.soc, raw, _extra = await self._driver(inv).read()
         was_stale = inv.batt_fresh.stale
         was_releases = inv.batt_fresh.releases
         inv.batt_fresh.submit(raw, inv.standing_w(), now_mono)
@@ -4029,9 +4044,9 @@ class Controller:
                                for inv in targets))
 
     async def _write_unit(self, inv, value: float, *, force: bool = False) -> None:
-        entity = inv.setpoint_entity
-        if not entity:
+        if not inv.driver.configured():
             return
+        entity = inv.driver.actuator_id()       # for the log line below
 
         # --- SAFETY-03: THE CLAMP, AND IT IS ON THE ONLY PATH TO THE WIRE -----
         # Every write in this add-on comes through here - the cycle, the
@@ -4154,7 +4169,7 @@ class Controller:
         # tick by itself: `last_write` is updated only on an ACCEPTED write, so
         # `changed` is still true.
         try:
-            ok = await asyncio.wait_for(self.hass.set_number(entity, value),
+            ok = await asyncio.wait_for(self._driver(inv).write(value),
                                         timeout=WRITE_BUDGET_S)
         except (asyncio.TimeoutError, TimeoutError):
             ok = False
@@ -6730,9 +6745,8 @@ class Controller:
         # Concurrently, for `read_inputs()`'s reason: a slow ESPHome on one unit
         # must not delay the controller's start on every other.
         async def _limits(inv):
-            if inv.setpoint_entity:
-                inv.dev_min, inv.dev_max, _ = await self.hass.limits(
-                    inv.setpoint_entity)
+            if inv.driver.configured():
+                inv.dev_min, inv.dev_max = await self._driver(inv).limits()
         await asyncio.gather(*(_limits(inv) for inv in self.inverters))
         entity = self.o.get("setpoint_entity", "")
         # ⚠️ THIS USED TO WARN AND CARRY ON, and that was the bug CFG-01
@@ -7316,9 +7330,10 @@ class Controller:
         rows = []
         for inv in self.inverters:
             unit = None if len(self.inverters) == 1 else inv.name
-            rows.append((messages.LABEL_SOC, unit, inv.soc, inv.soc_entity))
+            rows.append((messages.LABEL_SOC, unit, inv.soc,
+                         inv.driver.soc_source()))
             rows.append((messages.LABEL_BATTERY, unit, inv.batt,
-                         inv.batt_entity))
+                         inv.driver.batt_source()))
         if not self.p1_enabled:
             # In P1 mode the check above replaces this one; leaving both in
             # would report "no entity configured" for a meter_entity that is
@@ -7360,9 +7375,8 @@ class Controller:
                             "text": messages.row_sensor(label, value, unit)})
 
         for inv in self.inverters:
-            sp = inv.setpoint_entity
             unit = None if len(self.inverters) == 1 else inv.name
-            if not sp:
+            if not inv.driver.configured():
                 out.append({"ok": False, "warn": False, "id": "control",
                             "text": messages.row_control_unset(unit)})
             elif inv.dev_max is None:
@@ -8205,10 +8219,13 @@ async def amain() -> None:
         # validator cannot do for itself; (None, None) when ESPHome has not come
         # up yet is not a refusal - run_control() re-checks once the real rating
         # is known. See validate.py for where the line is drawn and why.
+        # SMA-2. The flat options are the one-inverter shape, so it is the
+        # GoodWe driver's, asked without a Controller (none exists yet).
         _sp = str(opts.get("setpoint_entity", "") or "").strip()
+        _drv = GoodWeEsphomeDriver(setpoint_entity=_sp).bind(hass)
         _dev_min = _dev_max = None
         if _sp:
-            _dev_min, _dev_max, _ = await hass.limits(_sp)
+            _dev_min, _dev_max = await _drv.limits()
             _LOG.info("setpoint device %s reports range %s…%s W", _sp, _dev_min, _dev_max)
         _problems = validate_startup_config(opts, _dev_min, _dev_max)
         # UX-15: a new install names no inverter. When that is the ONLY

goodwe_controller/app/selfconfig.py

@@ -343,10 +343,12 @@ async def problems_and_ranges(opts: dict, hass=None, notes=None):
                      f"disagreement between the two and should be reported."]
         return found, []
     if hass is not None:
+        # SMA-2. Each unit's driver, bound to the panel's own (read-only)
+        # Home Assistant: there is no Controller here.
         async def _limits(unit):
-            if unit.setpoint_entity:
-                unit.dev_min, unit.dev_max, _ = await hass.limits(
-                    unit.setpoint_entity)
+            if unit.driver.configured():
+                unit.dev_min, unit.dev_max = await unit.driver.bind(
+                    hass).limits()
         await asyncio.gather(*(_limits(unit) for unit in units))
 
     dev_min = dev_max = None
@@ -357,8 +359,8 @@ async def problems_and_ranges(opts: dict, hass=None, notes=None):
     problems = list(validate_startup_config(opts, dev_min, dev_max,
                                             notes=notes))
     problems += validate_unit_ranges(units, skip_entity=flat)
-    unread = [unit.setpoint_entity for unit in units
-              if unit.setpoint_entity and unit.dev_min is None
+    unread = [unit.driver.actuator_id() for unit in units
+              if unit.driver.configured() and unit.dev_min is None
               and unit.dev_max is None]
     return problems, unread
 

goodwe_controller/app/solarfc.py

@@ -34,7 +34,9 @@ evening and api.forecast.solar is sometimes out for a poll. A kept reading
 stands in for its sensor (a) for up to `FRESH_S` when the sensor is missing
 or unavailable, as a sensor that merely stopped updating is already
 tolerated, and (b) from the end of the bell (`BELL_TO_H`) until local
-midnight, if it is at most `KEPT_MAX_AGE_S` old when used.
+midnight, if it is at most `KEPT_MAX_AGE_S` old when used. A TOMORROW
+reading needs no evening for (b), at any hour (OPT-38: one Forecast.Solar
+sensor stopped at 14:54 and tomorrow vanished from 17:54 to 20:00).
 Never beyond that, and never for another day: a today figure is only today's,
 a tomorrow figure is tomorrow's until midnight and then, until the bell starts
 (`BELL_FROM_H`, 06:00 local), today's (OPT-34: at most `KEPT_MORNING_MAX_AGE_S`
@@ -330,8 +332,9 @@ class Hold:
 
         The kept reading must be for `day0`, and either (a) at most `FRESH_S`
         old, or (b) used at night: from the end of the bell to local midnight,
-        at most `KEPT_MAX_AGE_S` old. `kept` is true when (b) was needed,
-        or `as_today` (OPT-34: a tomorrow figure standing in for today, after
+        at most `KEPT_MAX_AGE_S` old; a "tomorrow" reading at any hour
+        (OPT-38; `read_sum()` flags it kept only from the end of the bell).
+        `kept` is true when (b) was needed, or `as_today` (OPT-34: a tomorrow figure standing in for today, after
         midnight and until the bell starts, `KEPT_MORNING_MAX_AGE_S`)."""
         rec = self._r.get((kind, entity, day0))
         if rec is None:
@@ -347,8 +350,8 @@ class Hold:
             return (day, at, True) if ok else None
         if age > KEPT_MAX_AGE_S:
             return None
-        # A tomorrow sensor is asked with day0 = the next midnight, which is
-        # in the future all day, so the evening ends at that day's end.
+        if kind == "tomorrow":
+            return day, at, True    # OPT-38: no evening needed
         evening = _local_hour(now) >= BELL_TO_H and now < _bounds(day0)[1]
         return (day, at, True) if evening else None
 
@@ -556,6 +559,9 @@ def read_sum(today, tomorrow, now: datetime, hold: Hold | None = None):
     if why0 is not None:
         return None, why0, gaps
     tday = _sum(tdays) if tomorrow and len(tdays) == len(tomorrow) else None
-    kept = kept or (tday is not None and tkept)
+    # OPT-38: before the bell ends a kept tomorrow reading is no "kept for
+    # tonight": it must not replace today's line or the log.
+    kept = kept or (tday is not None and tkept
+                    and _local_hour(now) >= BELL_TO_H)
     # The oldest member's report is when the sum was last fully known.
     return Forecast(_sum(days), tday, min(ats), kept), None, gaps

goodwe_controller/app/validate.py

@@ -328,12 +328,14 @@ WHAT IS DELIBERATELY *NOT* REFUSED, because it is untidy rather than unsafe:
 """
 
 import logging
+import math
 import re
 from typing import NamedTuple
 
 from . import dayplan, messages, pv, solarfc
 from .arbitrage import enabled as arbitrage_on
 from .control import SLEW_RAIL_FRAC, LOOP_TICK_S, PLANT_SETTLE_S
+from . import drivers
 from .custom_formula import (
     EXPORT_FORMULA_OPTION, FormulaInvalid, implausible_all_in,
     parse_export_formula, parse_price_formula, stale_card)
@@ -457,13 +459,28 @@ def _collect_writers(node, path: str, out: list) -> None:
     however deeply nested and under whatever container.
     """
     if isinstance(node, dict):
+        # SMA-2. THE DRIVER NAMES THE ACTUATOR. An entry's `driver` (default
+        # goodwe_esphome) says which key declares its writer - `setpoint_entity`
+        # for GoodWe, `host` for a driver reached over a socket - and what that
+        # entry's `actuator_id()` is. The walk is as structural as before: it
+        # still finds an entry wherever the options keep it.
+        cls = drivers.driver_class(node)
         for k, v in node.items():
             here = f"{path}.{k}" if path else str(k)
+            # Every key in ACTUATOR_WRITE_KEYS declares a writer, as before
+            # SMA-2 and whatever the entry's driver; PLUS the entry driver's
+            # own writer key. The second is added to the first, never in place
+            # of it (census parity with 23ff9d0 is graded in test_sma2_driver).
             if k in ACTUATOR_WRITE_KEYS:
                 entity = str(v or "").strip()
                 if entity:
                     out.append((here, entity))
                 continue
+            if k == cls.writer_key:
+                actuator = cls.actuator_id_of(node)
+                if actuator:
+                    out.append((here, actuator))
+                continue
             _collect_writers(v, here, out)
     elif isinstance(node, (list, tuple)):
         for i, v in enumerate(node):
@@ -498,7 +515,7 @@ def actuator_writers(opts: dict) -> dict:
     _collect_writers(opts, "", found)
     census: dict = {}
     for where, entity in found:
-        census.setdefault(entity.lower(), []).append(where)
+        census.setdefault(drivers.census_key(entity), []).append(where)
     return census
 
 
@@ -729,11 +746,16 @@ def _roster_problems(opts: dict) -> list[str]:
         else:
             seen[name] = where
         who = f"{where} ({name})" if name else where
-        if not str(entry.get("setpoint_entity") or "").strip():
+        # SMA-2. WHETHER THE ENTRY REACHES ITS INVERTER IS THE DRIVER'S TO SAY.
+        # For goodwe_esphome that is a setpoint entity, and the sentence is the
+        # one it always was. An unknown driver is refused by (h) below.
+        _cls = drivers.driver_class(entry)
+        if _cls is not drivers.UnavailableDriver and not _cls.actuator_id_of(entry):
             problems.append(
                 f"{who} has no {S('setpoint_entity')}, so the add-on cannot "
                 "control it. Set its ESPHome number entity, or remove the "
                 "entry.")
+        problems += _driver_problems(entry, who)
         if _number(entry, "max_w", None) is None or (_number(entry, "max_w", None) or 0) <= 0:
             problems.append(
                 f"{who} has no usable {S('max_w')}: it is "
@@ -775,6 +797,56 @@ def _roster_problems(opts: dict) -> list[str]:
     return problems
 
 
+def _whole(raw, lo: int, hi: int):
+    """`raw` as an int in [lo, hi], or None. A bool or a fraction is not one."""
+    if isinstance(raw, bool):
+        return None
+    try:
+        val = float(raw)
+        # NaN and infinity: int() raises on them, so refuse them first.
+        if not math.isfinite(val) or val != int(val):
+            return None
+    except (TypeError, ValueError, OverflowError):
+        return None
+    return int(val) if lo <= val <= hi else None
+
+
+def _driver_problems(entry: dict, who: str) -> list[str]:
+    """Rule 9(h) (SMA-2): the keys that choose and address an inverter driver.
+
+    NOT IN config.yaml's SCHEMA, so the Supervisor form and its validator never
+    see them (owner decision 2026-11-07: clients see nothing new until SMA-4).
+    They can still reach this rule from callers that skip the schema (the
+    panel's save path, sim/inject, the checks), so they are graded here, and
+    named by key because they have no label.
+
+    Only `goodwe_esphome` is selectable for now. `host`, `port`, `unit_id` and
+    `maintenance` are accepted and checked for shape, and nothing reads them
+    yet: a later driver's entry must not be refused for carrying them.
+    """
+    out: list[str] = []
+    raw = entry.get("driver")
+    if raw not in (None, "") and str(raw).strip() not in drivers.SELECTABLE:
+        out.append(
+            f"{who} has `inverters.driver` {messages.shown(raw)}. Use "
+            f"{', '.join(drivers.SELECTABLE)}, or leave it empty.")
+    host = entry.get("host")
+    if host not in (None, "") and (not isinstance(host, str)
+                                   or re.search(r"\s", host.strip())):
+        out.append(f"{who} has `inverters.host` {messages.shown(host)}, which is "
+                   "not an address. Use a name or an IP address.")
+    for key, lo, hi in (("port", 1, 65535), ("unit_id", 0, 247)):
+        val = entry.get(key)
+        if val not in (None, "") and _whole(val, lo, hi) is None:
+            out.append(f"{who} has `inverters.{key}` {messages.shown(val)}. Use a "
+                       f"whole number from {lo} to {hi}.")
+    maint = entry.get("maintenance")
+    if maint is not None and not isinstance(maint, bool):
+        out.append(f"{who} has `inverters.maintenance` {messages.shown(maint)}. "
+                   "Use on or off.")
+    return out
+
+
 def validate_unit_ranges(units, *, skip_entity: str = "") -> list[str]:
     """Rule 2, per inverter, against each unit's OWN reported device range.
 
@@ -787,8 +859,8 @@ def validate_unit_ranges(units, *, skip_entity: str = "") -> list[str]:
     2's own failure mode restored by a config shape: the operator's stated
     ceiling is not the ceiling and nothing says which won.
 
-    `units` is the roster (anything with `.name`, `.setpoint_entity`, `.max_w`,
-    `.dev_min`, `.dev_max`). A unit whose device did not answer is SKIPPED, not
+    `units` is the roster (anything with `.name`, `.driver` or `.setpoint_entity`,
+    `.max_w`, `.dev_min`, `.dev_max`). A unit whose device did not answer is SKIPPED, not
     guessed at, for the reason in this module's docstring: ESPHome is often not
     up when the add-on boots and (None, None) means "not read", not "zero".
 
@@ -800,7 +872,12 @@ def validate_unit_ranges(units, *, skip_entity: str = "") -> list[str]:
     problems: list[str] = []
     skip = str(skip_entity or "").strip().lower()
     for unit in units:
-        entity = str(getattr(unit, "setpoint_entity", "") or "").strip()
+        # SMA-2. THE DRIVER'S ACTUATOR, not a GoodWe key: a unit reached some
+        # other way is graded too. A bare object with only `.setpoint_entity`
+        # (the checks pass some) is read as it always was.
+        drv = getattr(unit, "driver", None)
+        entity = str((drv.actuator_id() if drv is not None
+                      else getattr(unit, "setpoint_entity", "")) or "").strip()
         if not entity or entity.lower() == skip:
             continue
         cap = float(getattr(unit, "max_w", 0.0) or 0.0)

goodwe_controller/test_freshness.py

@@ -790,7 +790,7 @@ print(f"    BOUNDS['soc']  subscripted in CODE by: {_uses_soc or 'nothing'}")
 check("⚠️ `wired=True` MEANS SOMETHING CONSUMES IT. A policy table nobody "
       "checks is a comment; this is the row `batt` claims and the module that "
       "claims it",
-      BOUNDS["batt"].wired is True and _uses_batt == ["inverter.py"])
+      BOUNDS["batt"].wired is True and _uses_batt == ["drivers.py"])
 check("⚠️ AND `wired=False` MEANS NOTHING DOES. SoC is TEL-08's and it is "
       "STATED here rather than wired, because its obligation term is energy "
       "through the pack and the constant that needs has never been measured on "

goodwe_controller/test_opt27.py

@@ -144,8 +144,9 @@ rs(st(5.0, at(4, 9, 5)), st(12.0, at(4, 9, 5)), at(4, 9, 6), h9)
 for label, tm in (("stale", st(12.0, at(4, 9, 5))), ("unavailable", GONE)):
     fd = rs(st(3.0, at(4, 13, 55)), tm, at(4, 14, 0), h9)[0]
     check(f"14:00, today fresh, tomorrow {label} since 09:05 (over 3 h): "
-          "no tomorrow, nothing kept",
-          fd is not None and fd.tomorrow is None and not fd.kept)
+          "the 09:05 figure covers it, kept (OPT-38: at any hour)",
+          fd is not None and fd.tomorrow is not None
+          and fd.tomorrow.kwh == 12.0 and not fd.kept)
     fn = rs(st(0.1, at(4, 20, 55)), tm, at(4, 21, 0), h9)[0]
     check(f"21:00, tomorrow {label}: the 09:05 figure covers it, kept",
           fn is not None and fn.tomorrow is not None

goodwe_controller/test_opt34.py

@@ -186,8 +186,9 @@ check("a fresh figure replaces the kept one at once (04:00: 5.0 today, "
       and fy.today.kwh == 5.0 and fy.tomorrow.kwh == 9.0)
 fy2 = rs(st(5.0, at(4, 7, 55)), GONE, at(4, 8, 0), hd)[0]
 check("... and at 08:00, today fresh and tomorrow's sensor down for over 3 h: "
-      "no tomorrow (the 3rd's figure is not the 5th's)",
-      fy2 is not None and fy2.tomorrow is None and fy2.today.kwh == 5.0)
+      "tomorrow is the 03:50 figure for the 5th (9.0, not flagged kept before 20:00: OPT-38), never "
+      "the 3rd's 12.0", fy2 is not None and fy2.today.kwh == 5.0
+      and fy2.tomorrow is not None and fy2.tomorrow.kwh == 9.0 and not fy2.kept)
 
 # Restart: the saved hold keeps the figure.
 for label, now in (("01:00", at(4, 1, 0)), ("04:00", at(4, 4, 0))):

goodwe_controller/test_opt38.py

@@ -0,0 +1,159 @@
+"""Runnable check for OPT-38.  `python3 test_opt38.py`
+
+Live 0.20.1, 2026-11-06: one of three Forecast.Solar tomorrow sensors last
+reported 14:54. From 17:54 (over FRESH_S) to 20:00 (BELL_TO_H) nothing stood in
+for it, the OPT-17 sum was dropped and the Forecast tab had no tomorrow. A kept
+tomorrow reading now stands in at any hour (`Hold.stand_in()`). Today members,
+and the OPT-34 morning rules, are as before.
+
+Mutations (MUTATION.md, purged copies): see the pull request.
+"""
+
+import os
+import sys
+from datetime import datetime, timedelta, timezone
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+
+from app import homeview, solarfc                                        # noqa: E402
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+L = timezone(timedelta(hours=1))             # Brussels in winter (CET)
+
+
+def at(day, hh, mm=0):
+    return datetime(2026, 11, day, hh, mm, tzinfo=L).astimezone(timezone.utc)
+
+
+def st(kwh, when):
+    return {"state": f"{kwh}", "attributes": {"unit_of_measurement": "kWh"},
+            "last_updated": when.isoformat(), "last_reported": when.isoformat()}
+
+
+TODAY = [("sensor.t", None)]
+M = ["sensor.m1", "sensor.m2", "sensor.m3"]
+
+
+def run(hold, readings, now, today_at=None):
+    t = [("sensor.t", st(0.5, today_at or now - timedelta(minutes=5)))]
+    return solarfc.read_sum(
+        t, [(e, st(k, w)) for e, (k, w) in zip(M, readings)], now, hold)
+
+
+# The live day: m2 last reported 14:54, the others keep reporting.
+h = solarfc.Hold()
+run(h, [(4.0, at(6, 14, 0)), (5.0, at(6, 14, 54)), (6.0, at(6, 14, 0))],
+    at(6, 14, 55))
+fresh = lambda w: [(4.0, w), (5.0, at(6, 14, 54)), (6.0, w)]  # noqa: E731
+f, why, gaps = run(h, fresh(at(6, 18, 20)), at(6, 18, 30))
+check("18:30, one member last reported 14:54: tomorrow sum is there (not "
+      "flagged kept before 20:00)",
+      f is not None and f.tomorrow is not None and f.tomorrow.kwh == 15.0
+      and not f.kept and gaps == ())
+f, _w, _g = run(h, fresh(at(6, 17, 40)), at(6, 17, 50))
+check("17:50 (just under 3 h): fresh, not kept",
+      f is not None and f.tomorrow.kwh == 15.0 and not f.kept)
+f, _w, _g = run(h, fresh(at(6, 17, 55)), at(6, 18, 0))
+check("18:00 (stale since 17:54): tomorrow there",
+      f is not None and f.tomorrow is not None and not f.kept)
+f, _w, _g = run(h, fresh(at(6, 19, 50)), at(6, 19, 55))
+check("19:55 (before the bell ends): tomorrow there", f.tomorrow is not None)
+
+# Too old: over KEPT_MAX_AGE_S (12 h) when used.
+h2 = solarfc.Hold()
+run(h2, [(4.0, at(6, 5, 0)), (5.0, at(6, 5, 0)), (6.0, at(6, 5, 0))],
+    at(6, 5, 5))
+w = at(6, 16, 50)
+f, why, gaps = run(h2, [(4.0, w), (5.0, at(6, 5, 0)), (6.0, w)],
+                   at(6, 17, 10))
+check("member 12 h 10 min old: no tomorrow sum (OPT-17 rule), today stays",
+      f is not None and f.tomorrow is None
+      and [g[0] for g in gaps] == ["tomorrow"])
+f, _w, _g = run(h2, [(4.0, w), (5.0, at(6, 5, 0)), (6.0, w)],
+                at(6, 16, 50))
+check("... and 11 h 50 min old is still kept",
+      f is not None and f.tomorrow is not None and not f.kept)
+
+# Not a partial sum: a member never read has nothing to stand in.
+f, _w, gaps = run(solarfc.Hold(), [(4.0, at(6, 18, 0)),
+                                   (5.0, at(6, 14, 54)),
+                                   (6.0, at(6, 18, 0))], at(6, 18, 30))
+check("a member with no kept reading: no tomorrow sum, no partial sum",
+      f is not None and f.tomorrow is None and len(gaps) == 1)
+
+# Today members are as before: a stale today member at 18:30 is not kept.
+h3 = solarfc.Hold()
+solarfc.read_sum([("sensor.t", st(1.0, at(6, 14, 54)))],
+                 [(M[0], st(9.0, at(6, 14, 54)))], at(6, 14, 55), h3)
+f, why, _g = solarfc.read_sum([("sensor.t", st(1.0, at(6, 14, 54)))],
+                              [(M[0], st(9.0, at(6, 18, 20)))],
+                              at(6, 18, 30), h3)
+check("stale today member at 18:30: not kept, no forecast",
+      f is None and why == solarfc.STALE)
+f, _w, _g = solarfc.read_sum([("sensor.t", st(1.0, at(6, 14, 54)))],
+                             [(M[0], st(9.0, at(6, 20, 0)))],
+                             at(6, 20, 30), h3)
+check("... and at 20:30 it is kept, as before (the bell has ended)",
+      f is not None and f.kept)
+
+# After midnight: a tomorrow reading from before midnight is never tomorrow's.
+h4 = solarfc.Hold()
+run(h4, [(4.0, at(6, 14, 0)), (5.0, at(6, 14, 0)), (6.0, at(6, 14, 0))],
+    at(6, 14, 5))
+for label, now in (("00:30", at(7, 0, 30)), ("10:00", at(7, 10, 0)),
+                   ("15:00", at(7, 15, 0))):
+    f, _w, _g = solarfc.read_sum(
+        [("sensor.t", st(2.0, now - timedelta(minutes=5)))],
+        [(e, st(k, at(6, 14, 0))) for e, k in zip(M, (4.0, 5.0, 6.0))],
+        now, h4)
+    check(f"{label} on the 7th: the 6th's tomorrow figure is not the "
+          "7th's tomorrow", f is not None and f.tomorrow is None)
+
+# Before 20:00 a kept tomorrow reading is not "kept for tonight": today's
+# line stays the normal one (EN and NL) and the event log (main.py, from
+# `fc.kept`) has no "kept".
+h5 = solarfc.Hold()
+solarfc.read_sum([("t", st(3.0, at(6, 9, 0)))], [("m", st(12.0, at(6, 9, 5)))],
+                 at(6, 9, 6), h5)
+for hh in (14, 18, 19):
+    f, _w, _g = solarfc.read_sum([("t", st(2.5, at(6, hh - 1, 55)))],
+                                 [("m", st(12.0, at(6, 9, 5)))],
+                                 at(6, hh, 0), h5)
+    check(f"{hh}:00, today fresh, tomorrow kept from 09:05: tomorrow there, "
+          "not flagged kept", f is not None and f.tomorrow is not None
+          and f.tomorrow.kwh == 12.0 and f.kept is False)
+f, _w, _g = solarfc.read_sum([("t", st(2.5, at(6, 13, 55)))],
+                             [("m", st(12.0, at(6, 9, 5)))], at(6, 14, 0), h5)
+for lang in ("en", "nl"):
+    with homeview.in_lang(lang):
+        line = homeview.line_solar(f.today.kwh, f.today.kwh * 0.7,
+                                   homeview.hhmm(f.at), kept=f.kept)
+        kept_line = homeview.line_solar(1, 1, homeview.hhmm(f.at), kept=True)
+    print(f"    {lang}: {line!r}")
+    if lang == "en":
+        check("... EN says 'Solar still to come today'",
+              line.startswith("Solar still to come today"))
+    check(f"14:00 Home line ({lang}) is the normal line, not the kept one",
+          line != kept_line and "2,5" in line.replace(".", ",")
+          and "13:55" in line)
+f, _w, _g = solarfc.read_sum([("t", st(0.2, at(6, 20, 55)))],
+                             [("m", st(12.0, at(6, 9, 5)))], at(6, 21, 0), h5)
+check("21:00 as before: tomorrow kept from 09:05 is flagged kept",
+      f is not None and f.kept and f.tomorrow.kwh == 12.0)
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED:")
+    for f in fails:
+        print("  -", f)
+    sys.exit(1)
+print("all checks passed")

goodwe_controller/test_safety128.py

@@ -229,9 +229,11 @@ late5 = R.Rig(lag_s=30.0, lag_on=ON, lag_off=ON + 300.0, p1_s=5.0)
 late5.run(ON + 300.0, lambda rig, c, t: rows5.extend(
     str(r["text"]) for r in c.checks() if r.get("id") == "late"))
 print(f"    5 s meter, 30 s late: {sorted(set(rows5))}")
-check("at a 5 s meter (the batteries' fit) the row names the delay: about "
-      "20-32 s for 30 s late",
-      rows5 and set(rows5) <= _ok_texts and any("about" in x for x in rows5))
+# SAFETY-129: the 5 s meter's fit finds it first (no figure); the hold then
+# leaves the batteries' fit nothing to measure, so the row names no delay.
+check("at a 5 s meter the row says the readings are late, with no figure "
+      "(the meter alone found it)",
+      rows5 and set(rows5) == {str(messages.row_readings_late(None))})
 check("logged once for the episode, and once when it ends, not per cycle",
       n_late == 1 and n_back == 1)
 check("the row goes once a fit shows the readings on time, or 2 minutes "

goodwe_controller/test_safety129.py

@@ -0,0 +1,238 @@
+"""Runnable check for SAFETY-129: late Home Assistant readings found at a 5 s meter.
+`python3 test_safety129.py`
+
+SAFETY-128 fitted the meter's delay only at a meter of about 3.5 s or faster
+(16 new values in 60 s). The owner's P1 reports every 5 s, so there only the
+batteries' fit ran: with readings 30 s late the loop swung to -420 W export
+and was found after about 3 minutes (-1 325 W before the hold).
+`lateness.MeterFit` now also runs at a slow meter over 120 s, and calls a
+delay late only over `M_SLOW_LATE_S` on two fits in a row, with the same
+corroboration as before (an ask slow to start, no battery fit on time).
+DECISIONS/SAFETY-128-late-readings.md.
+
+The REAL Controller on TEST-23's plant (tools/safety128_rig.py).
+
+1. 5 s meter, readings 30 s late (N=2 house 1 800 W, N=1 house 1 500 W): found
+   within 100 s, and the export never goes below -200 W (red on c66aa7f: found
+   after 190 s, export to -420 W).
+2. 5 s meter, readings 10 s late: no swing, nothing to hold; and the limit that
+   goes with it: a battery that answers in 10 s (R2-1) is still not read as
+   late at a 5 s meter, which a lower slow-meter threshold would break.
+3. No false holds at a slow meter: the owner's shape at 2, 5 and 10 s, HA
+   restarts and meter pauses at 2 and 5 s, 8-10 s dead times: never
+   late, so the commands are those of the run without the gate.
+4. The slow-meter fit at its edges.
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import logging
+import os
+import sys
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+sys.path.insert(0, os.path.join(HERE, "tools"))
+logging.disable(logging.WARNING)
+
+import safety128_rig as R                                      # noqa: E402
+import app.main as app_main                                    # noqa: E402
+from app import lateness                                       # noqa: E402
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+def no_gate(fn):
+    """`fn()` with the answer gate off: the tree before SAFETY-128."""
+    real = app_main.Controller._readings_late
+    app_main.Controller._readings_late = lambda self, standing_w: None
+    try:
+        return fn()
+    finally:
+        app_main.Controller._readings_late = real
+
+
+def run_late(**kw):
+    """Run the rig; (rig, seconds the verdict stood, first second it stood)."""
+    dur = kw.pop("dur")
+    seen = []
+    r = R.Rig(**kw)
+    r.run(dur, lambda rig, c, t: seen.append(t) if c.answer.late_seen
+          else None)
+    return r, len(seen), min(seen) if seen else None
+
+
+def grid_min(rows, t0, t1):
+    return min(r[1] for r in rows if t0 <= r[0] < t1)
+
+
+AB, A = ("a", "b"), ("a",)
+ON = 300.0
+
+# --------------------------------------------------------------------------- #
+print("1. 5 s meter, readings 30 s late for 300 s")
+for units, house in ((AB, 1800.0), (A, 1500.0)):
+    n = f"N={len(units)}"
+    r, secs, first = run_late(house=house, lag_s=30.0, lag_on=ON,
+                              lag_off=ON + 300.0, units=units, p1_s=5.0,
+                              dur=ON + 420.0)
+    low = grid_min(r.rows, ON, ON + 300.0)
+    print(f"    {n}: found {None if first is None else first - ON:.0f} s "
+          f"after the start, export to {low:.0f} W")
+    check(f"{n}: found within 100 s of the lag starting",
+          first is not None and first - ON <= 100.0)
+    check(f"{n}: the export never goes below -200 W", low >= -200.0)
+    if len(units) == 2:
+        old = R.Rig(house=house, lag_s=30.0, lag_on=ON, lag_off=ON + 300.0,
+                    units=units, p1_s=5.0)
+        no_gate(lambda: old.run(ON + 300.0))
+        check(f"{n}: without the gate the export goes below -1 kW",
+              grid_min(old.rows, ON, ON + 300.0) < -1000.0)
+
+# --------------------------------------------------------------------------- #
+print("\n2. 5 s meter, readings 10 s late; a battery that answers in 10 s")
+r, secs, first = run_late(house=1800.0, lag_s=10.0, lag_on=ON,
+                          lag_off=ON + 150.0, units=AB, p1_s=5.0,
+                          dur=ON + 200.0)
+low = grid_min(r.rows, ON, ON + 150.0)
+print(f"    10 s late: export to {low:.0f} W, verdict stood {secs} s")
+check("10 s late: no swing, the export never goes below -200 W", low >= -200.0)
+
+
+def owner(t):
+    w = 400.0
+    if 600 <= t < 780:
+        w += 2200
+    if 1200 <= t < 2400 and ((t - 1200) % 105) < 45:
+        w += 2000
+    return w
+
+
+for dead in (8.0, 10.0):
+    r, secs, first = run_late(house=owner, noise=80.0, units=AB, p1_s=5.0,
+                              dead_s=dead, dur=800.0)
+    print(f"    {dead:.0f} s dead time, 5 s meter: late {secs} s")
+    check(f"{dead:.0f} s dead time at a 5 s meter is not read as late "
+          f"(R2-1 does not get worse)", secs == 0)
+
+# At a 6 s meter an 8 s dead time reads late on the slow fit alone; only the
+# slow-start and battery-fit check keeps it from a verdict (122 s late
+# without the check, from 1 084 s).
+r, secs, first = run_late(house=owner, noise=80.0, units=AB, p1_s=6.0,
+                          dead_s=8.0, dur=1500.0)
+print(f"    8 s dead time, 6 s meter: late {secs} s")
+check("an 8 s dead time at a 6 s meter is never late (the slow fit needs an "
+      "ask slow to start and no battery fit on time)", secs == 0)
+
+# A 12 s dead time is already read as late (R2-1, SAFETY-128); the slow-meter
+# fit must not make that longer. The same run with the slow fit off is the
+# tree before this card.
+_slow = lateness.MeterFit.slow
+secs = {}
+for label in ("with", "without"):  # 12 s dead time: late from about 140 s
+    lateness.MeterFit.slow = _slow if label == "with" else (lambda self: False)
+    try:
+        secs[label] = run_late(house=owner, noise=80.0, units=A, p1_s=5.0,
+                               dead_s=12.0, dur=900.0)[1]
+    finally:
+        lateness.MeterFit.slow = _slow
+print(f"    12 s dead time, N=1, 5 s meter: late {secs['with']} s, with the "
+      f"slow fit off {secs['without']} s")
+check("a 12 s dead time at a 5 s meter is late no longer than before "
+      "(R2-1 does not get worse)", secs["with"] <= secs["without"])
+
+# --------------------------------------------------------------------------- #
+print("\n3. no false holds")
+_view = R.Plant.view
+
+
+def frozen(windows, meter_only):
+    def view(self, t, lag_s, lag_on, lag_off):
+        cur = _view(self, t, lag_s, lag_on, lag_off)
+        for a, b in windows:
+            if a <= t < b:
+                for row in reversed(self.hist):
+                    if row[0] <= a + 1e-9:
+                        old = row[1:]
+                        return (old[0], old[1], cur[2], cur[3]) \
+                            if meter_only else old
+        return cur
+    return view
+
+
+for p1 in (2.0, 5.0, 10.0):
+    kw = dict(house=owner, noise=80.0, units=AB, p1_s=p1)
+    r, secs, first = run_late(dur=800.0, **kw)
+    new = r
+    old = R.Rig(**kw)
+    no_gate(lambda: old.run(800.0))
+    same = max(abs(x[2] - y[2]) for x, y in zip(new.rows, old.rows))
+    print(f"    owner shape, {p1:.0f} s meter: late {secs} s, command "
+          f"difference {same:.0f} W")
+    check(f"{p1:.0f} s meter, owner shape: never late, same commands",
+          secs == 0 and same == 0)
+for label, windows, meter_only in (
+        ("HA restart", ((650.0, 710.0),), False),
+        ("meter pause", ((640.0, 700.0),), True)):
+    for p1 in (2.0, 5.0):
+        R.Plant.view = frozen(windows, meter_only)
+        try:
+            r, secs, first = run_late(house=owner, noise=80.0, units=AB,
+                                      p1_s=p1, dur=800.0)
+        finally:
+            R.Plant.view = _view
+        check(f"{label} at a {p1:.0f} s meter: never late", secs == 0)
+
+# --------------------------------------------------------------------------- #
+print("\n4. the slow-meter fit at its edges")
+
+
+def meter_with(period, pause_after=None, pause_s=40.0, n=24):
+    f = lateness.MeterFit()
+    f.add(0.0, 100.0, 0.0)
+    for k in range(1, n):
+        gap = pause_s if pause_after is not None and k > pause_after else 0.0
+        f.add(period * k + gap, None, float(k))
+    return f
+
+
+check("a 5 s and a 10 s meter are slow",
+      meter_with(5.0).slow() and meter_with(10.0).slow())
+check("a 1 s, a 2 s and a 3 s meter are not slow",
+      not any(meter_with(p, n=60).slow() for p in (1.0, 2.0, 3.0)))
+check("a pause in a 2 s meter does not make it slow",
+      not meter_with(2.0, pause_after=10, n=40).slow())
+def fed(values):
+    """Answer.watch() with the slow fit reading `values`, one per fit."""
+    a = lateness.Answer()
+    a.meter.slow = lambda: True
+    a.meter.estimate = a.delay.estimate = lambda: None
+    it = iter(values)
+    a.meter_slow.estimate = lambda: next(it, None)
+    for k in range(len(values)):
+        a.watch(2.0 * k, [], 0.0, 0.0)
+    return a
+
+
+check("one late slow-meter fit is no fit; two in a row are one, and "
+      "one on time or in between is none",
+      fed([30.0]).fits == 0 and fed([30.0, 30.0]).fits == 1
+      and fed([30.0, None, 30.0]).fits == 0 and fed([30.0, 12.0]).fits == 0
+      and fed([3.0, 3.0]).fits == 0 and not fed([30.0, 30.0]).late_seen)
+check("the slow fit asks for a longer window and a later delay than the "
+      "fast one", lateness.M_SLOW_FIT_S > lateness.M_FIT_S
+      and lateness.M_SLOW_LATE_S > lateness.M_LATE_S)
+check("the slow fit's late limit is above the fast one's and above a "
+      "10 s lag", lateness.M_SLOW_LATE_S >= 16.0)
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED: " + ", ".join(fails))
+    sys.exit(1)
+print("all checks passed")

goodwe_controller/test_sma2_driver.py

@@ -0,0 +1,750 @@
+"""SMA-2: the inverter driver seam.  `python3 test_sma2_driver.py`
+
+Part 1 grades the GoodWe driver against a Home Assistant that logs every call:
+it must make exactly the calls the code it replaced made, in the same order.
+Part 2 grades what `Inverter` reads from its driver. Part 3 drives a REAL
+Controller with a roster of one fake-driver unit and one GoodWe unit and
+checks that it writes, reads, limits, grades, drops and publishes each by its
+own driver, with a Home Assistant that records every entity it is asked about
+so a call site that goes round the seam is seen. Part 4 is the SAFETY-28
+census and rule 9 on the new keys. Part 5 is the schema.
+
+Mutating a module this file grades? Read MUTATION.md first (purge __pycache__).
+The seam mutants this file was run against are listed at the end.
+"""
+
+import asyncio
+import io
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+from app import drivers
+from app.drivers import (DEFAULT_DRIVER, REGISTRY, SELECTABLE, Driver,
+                         DriverNotBound, GoodWeEsphomeDriver, Probe,
+                         UnavailableDriver, build_driver, census_key)
+from app.freshness import BOUNDS, Bound
+from app.inverter import LINK_STALE_S, Inverter, build_roster
+from app.main import Controller
+from app.store import DEFAULTS
+from app.validate import (actuator_writers, validate_startup_config,
+                          validate_unit_ranges)
+
+fails = []
+HERE = Path(__file__).resolve().parent
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+class MemStore:
+    def __init__(self):
+        self.data = dict(DEFAULTS)
+
+    def set(self, k, v):
+        self.data[k] = v
+
+    def get_time(self, *a):
+        return None
+
+    def set_time(self, *a):
+        pass
+
+
+class Pub:
+    """Keeps the last published values, like test_mqtt.py's."""
+    enabled = False
+
+    def __init__(self):
+        self.last = {}
+
+    def publish(self, values):
+        self.last = dict(values)
+
+    def close(self):
+        pass
+
+
+class LogHass:
+    """Home Assistant for the GoodWe unit. EVERY call is logged with its entity,
+    so a path that talks to Home Assistant about anything else is on record."""
+
+    def __init__(self, numbers=None, limits=None, accept=True):
+        self.numbers = dict(numbers or {})
+        self.limits_by = dict(limits or {})
+        self.accept = accept
+        self.log = []                       # (method, entity, extra)
+
+    async def number(self, entity, invert=False):
+        self.log.append(("number", entity, invert))
+        v = self.numbers.get(entity)
+        return None if v is None else (-v if invert else v)
+
+    async def limits(self, entity):
+        self.log.append(("limits", entity, None))
+        return self.limits_by.get(entity, (None, None, None))
+
+    async def set_number(self, entity, value):
+        self.log.append(("set_number", entity, value))
+        if self.accept:
+            self.numbers[entity.replace("number.sp", "sensor.batt")] = value
+        return self.accept
+
+    async def state(self, entity):
+        self.log.append(("state", entity, None))
+        return None
+
+    async def call(self, *a):
+        self.log.append(("call", repr(a), None))
+        return True
+
+    def entities(self):
+        """Every entity name Home Assistant was asked about (service calls
+        like notifications name none)."""
+        return {e for m, e, _x in self.log if m != "call"}
+
+
+def run(coro):
+    return asyncio.run(coro)
+
+
+def raises(exc, make):
+    """Does `run(make())` raise `exc`? `make` builds the coroutine."""
+    try:
+        run(make())
+    except exc:
+        return True
+    except Exception:
+        return False
+    return False
+
+
+# ---------------------------------------------------------------------------
+# a fake driver: a unit reached by host:port:unit, no Home Assistant at all
+# ---------------------------------------------------------------------------
+FAKE_BOUND = Bound(signal="fake", unchanged_s=11.0, move_w=100.0,
+                   release_s=5.0, wired=True, derivation="test")
+
+
+class FakeDriver(Driver):
+    name = "fake"
+    writer_key = "host"
+    brand = "Fake"
+    link_stale_s = 7.0
+    batt_bound = FAKE_BOUND
+    settle_s = 3.0
+    words = {"idle": "handed back"}
+
+    def __init__(self, host="", port=502, unit_id=1, rating=1000.0, soc=60.0):
+        self.host, self.port, self.unit_id = host, port, unit_id
+        self.rating, self.soc, self.power = rating, soc, 0.0
+        self.accept = True
+        self.calls = []                     # (what, arg)
+
+    @classmethod
+    def from_entry(cls, entry):
+        return cls(entry.get("host", ""), entry.get("port", 502),
+                   entry.get("unit_id", 1), rating=entry.get("_rating", 1000.0))
+
+    @classmethod
+    def actuator_id_of(cls, entry):
+        host = str(entry.get("host") or "").strip()
+        return (f"{host}:{entry.get('port', 502)}:{entry.get('unit_id', 1)}"
+                if host else "")
+
+    def configured(self):
+        return bool(self.host)
+
+    def actuator_id(self):
+        return f"{self.host}:{self.port}:{self.unit_id}"
+
+    def soc_source(self):
+        return "fake-soc"
+
+    def batt_source(self):
+        return "fake-power"
+
+    async def limits(self):
+        self.calls.append(("limits", None))
+        return -self.rating, self.rating
+
+    async def read(self):
+        self.calls.append(("read", None))
+        return self.soc, self.power, {"state": "ok"}
+
+    async def write(self, w):
+        self.calls.append(("write", w))
+        if self.accept:
+            self.power = w                  # the battery follows its command
+        return self.accept
+
+    async def release(self):
+        self.calls.append(("release", None))
+        return True
+
+    async def probe(self):
+        self.calls.append(("probe", None))
+        return Probe(-self.rating, self.rating, self.soc, self.power, {})
+
+
+def calls(drv, what):
+    return [a for w, a in drv.calls if w == what]
+
+
+print("SMA-2 - the inverter driver seam")
+print()
+# ==============================================================================
+print("1. the GoodWe driver makes the calls the code it replaced made")
+
+GW = dict(setpoint_entity="number.sp_a", soc_entity="sensor.soc_a",
+          batt_entity="sensor.batt_a")
+h1 = LogHass({"sensor.soc_a": 55.0, "sensor.batt_a": 700.0},
+             {"number.sp_a": (-5000.0, 4000.0, 1.0)})
+d1 = GoodWeEsphomeDriver(**GW).bind(h1)
+check("a driver built from an entry names its three entities and is configured",
+      GoodWeEsphomeDriver.from_entry(dict(GW)).setpoint_entity == "number.sp_a"
+      and d1.configured() and d1.actuator_id() == "number.sp_a"
+      and (d1.soc_source(), d1.batt_source())
+      == ("sensor.soc_a", "sensor.batt_a"))
+check("...and one with no setpoint entity is not",
+      not GoodWeEsphomeDriver().configured()
+      and not GoodWeEsphomeDriver(soc_entity="sensor.soc_a").configured())
+check("limits() is hass.limits(setpoint) and hands back (min, max)",
+      run(d1.limits()) == (-5000.0, 4000.0)
+      and h1.log == [("limits", "number.sp_a", None)])
+h1.log.clear()
+check("read() asks for the SoC first, then the power, with the invert flag, "
+      "and nothing else",
+      run(d1.read()) == (55.0, 700.0, {})
+      and h1.log == [("number", "sensor.soc_a", False),
+                     ("number", "sensor.batt_a", False)])
+h1.log.clear()
+d1i = GoodWeEsphomeDriver(batt_invert=True, **GW).bind(h1)
+check("...and a reversed power sensor is reversed by hass.number, as before",
+      run(d1i.read()) == (55.0, -700.0, {})
+      and h1.log[1] == ("number", "sensor.batt_a", True))
+h1.log.clear()
+check("write(w) is hass.set_number(setpoint, w) and says whether it took it",
+      run(d1.write(-250.0)) is True and h1.log == [("set_number", "number.sp_a",
+                                                    -250.0)])
+h1.log.clear()
+h1.accept = False
+check("...and False when Home Assistant refuses",
+      run(d1.write(10.0)) is False)
+h1.accept = True
+h1.log.clear()
+check("release() is write(0): GoodWe is handed back by its board's own watchdog",
+      run(d1.release()) is True
+      and h1.log == [("set_number", "number.sp_a", 0.0)])
+h1.log.clear()
+h1.numbers.update({"sensor.soc_a": 55.0, "sensor.batt_a": 700.0})
+pr = run(d1.probe())
+check("probe() reads the range, the SoC and the power and WRITES NOTHING",
+      pr == Probe(-5000.0, 4000.0, 55.0, 700.0, {})
+      and all(m != "set_number" for m, _e, _x in h1.log))
+check("a driver with no Home Assistant is loud, not a silent 'no answer'",
+      raises(DriverNotBound, GoodWeEsphomeDriver(**GW).limits)
+      and raises(DriverNotBound, GoodWeEsphomeDriver(**GW).read)
+      and raises(DriverNotBound, lambda: GoodWeEsphomeDriver(**GW).write(1.0))
+      and not GoodWeEsphomeDriver(**GW).bound())
+_swap = {"h": h1}
+d1g = GoodWeEsphomeDriver(**GW).bind_getter(lambda: _swap["h"])
+h1b = LogHass({"sensor.soc_a": 1.0, "sensor.batt_a": 2.0})
+_swap["h"] = h1b
+check("bind_getter() looks Home Assistant up at each use, so a Controller whose "
+      ".hass is replaced afterwards is followed",
+      d1g.bound() and run(d1g.read()) == (1.0, 2.0, {}) and h1b.log)
+check("its numbers are the ones the code it replaced used",
+      (d1.link_stale_s, d1.batt_bound, d1.brand)
+      == (LINK_STALE_S, BOUNDS["batt"], "GoodWe") and LINK_STALE_S == 30.0
+      and d1.settle_s == 18.0 and d1.words == {})
+
+# ==============================================================================
+print()
+print("2. what Inverter reads from its driver")
+
+NOW = datetime(2026, 11, 7, 12, 0, tzinfo=timezone.utc)
+fk = FakeDriver(host="10.9.9.9", unit_id=3)
+inv_f = Inverter("fake-a", driver=fk, max_w=2000.0)
+inv_g = Inverter("inverter-b", max_w=3000.0, **{
+    "setpoint_entity": "number.sp_b", "soc_entity": "sensor.soc_b",
+    "batt_entity": "sensor.batt_b"})
+check("an Inverter built from the old keywords gets the GoodWe driver, with "
+      "the same entities on the same attributes",
+      isinstance(inv_g.driver, GoodWeEsphomeDriver)
+      and (inv_g.setpoint_entity, inv_g.soc_entity, inv_g.batt_entity,
+           inv_g.batt_invert) == ("number.sp_b", "sensor.soc_b",
+                                  "sensor.batt_b", False))
+check("...and one given another driver has none of those entities",
+      (inv_f.setpoint_entity, inv_f.soc_entity, inv_f.batt_entity) == ("", "", ""))
+check("has_writer() is the driver's configured()",
+      inv_f.has_writer() and inv_g.has_writer()
+      and not Inverter("x", driver=FakeDriver(host="")).has_writer()
+      and not Inverter("y").has_writer())
+check("the freshness gate is built on the DRIVER's bound",
+      inv_f.batt_fresh.bound is FAKE_BOUND
+      and inv_g.batt_fresh.bound is BOUNDS["batt"])
+for u in (inv_f, inv_g):
+    u.note_write(False, NOW - timedelta(seconds=10))
+check("is_live() takes its window from the driver: 10 s of rejected writes "
+      "drops a 7 s unit and not a GoodWe one",
+      inv_f.is_live(NOW)[0] is False and inv_g.is_live(NOW)[0] is True)
+check("...and an explicit window still wins",
+      inv_f.is_live(NOW, stale_s=60.0)[0] is True
+      and inv_g.is_live(NOW, stale_s=5.0)[0] is False)
+inv_f5 = Inverter("fake-5", driver=FakeDriver(host="h"), max_w=2000.0)
+inv_f5.note_write(False, NOW - timedelta(seconds=5))
+check("link_note() says the driver's window: 'after 7 seconds' for the fake "
+      "unit, 'after 30 seconds' for the GoodWe one, and nothing for a healthy unit",
+      "after 7 seconds" in inv_f5.link_note(NOW)
+      and "after 30 seconds" in inv_g.link_note(NOW)
+      and "cannot be controlled" in inv_f.link_note(NOW)
+      and Inverter("z", driver=FakeDriver(host="h")).link_note(NOW) == "")
+check("an unconfigured unit gets the no-setpoint row, as before",
+      Inverter("x", driver=FakeDriver(host="")).link_note(NOW).startswith("x: ")
+      and Inverter("x", driver=FakeDriver(host="")).link_note(NOW)
+      == Inverter("x").link_note(NOW) != "")
+check("an entry that names a driver this build lacks is an unconfigured unit, "
+      "not a crash and not GoodWe",
+      isinstance(build_driver({"driver": "nope", "setpoint_entity": "number.x"}),
+                 UnavailableDriver)
+      and not build_driver({"driver": "nope"}).configured()
+      and build_roster({"inverters": [{"name": "n", "driver": "nope",
+                                       "setpoint_entity": "number.x",
+                                       "max_w": 1000}]}, None)[0].has_writer()
+      is False)
+
+# ==============================================================================
+print()
+print("3. a roster of one fake unit and one GoodWe unit, in a real Controller")
+
+REGISTRY["fake"] = FakeDriver
+OPTS = dict(meter_source="off", meter_entity="sensor.grid", max_w=5000,
+            gain=0.6, slew_w=0, deadband_w=15, target_grid_w=0, step_w=10,
+            heartbeat_s=10, stale_input_s=15, shadow_mode="off",
+            inverters=[
+                dict(name="fake-a", driver="fake", host="10.9.9.9", port=502,
+                     unit_id=3, max_w=2000, bms_soc_floor=10),
+                dict(name="inverter-b", setpoint_entity="number.sp_b",
+                     soc_entity="sensor.soc_b", batt_entity="sensor.batt_b",
+                     max_w=3000, bms_soc_floor=15)])
+B_ENT = {"number.sp_b", "sensor.soc_b", "sensor.batt_b"}
+
+
+def site(rating_f=1000.0, **over):
+    hass = LogHass({"sensor.grid": 3000.0, "sensor.soc_b": 50.0,
+                    "sensor.batt_b": 0.0},
+                   {"number.sp_b": (-3000.0, 3000.0, 1.0)})
+    pub = Pub()
+    c = Controller(dict(OPTS, **over), hass, MemStore(), pub,
+                   stop=asyncio.Event())
+    f = c.inverters[0].driver
+    f.rating = rating_f
+    c.inverters[0].dev_min, c.inverters[0].dev_max = -rating_f, rating_f
+    c.inverters[1].dev_min, c.inverters[1].dev_max = -3000.0, 3000.0
+    c.auto = True
+    return c, hass, pub, f
+
+
+async def cycles(c, n=6):
+    for _ in range(n):
+        await c.read_inputs()
+        await c.cycle()
+
+
+c3, h3, p3, f3 = site()
+check("the roster is built through the registry: a fake unit and a GoodWe unit",
+      isinstance(c3.inverters[0].driver, FakeDriver)
+      and isinstance(c3.inverters[1].driver, GoodWeEsphomeDriver)
+      and f3.actuator_id() == "10.9.9.9:502:3")
+run(cycles(c3))
+fa, fb = c3.inverters
+print(f"    after 6 cycles: fake-a wrote {calls(f3, 'write')}, GoodWe wrote "
+      f"{[x for m, e, x in h3.log if m == 'set_number']}, shares "
+      f"{c3.allocation.shares if c3.allocation else None}")
+check("READ: the unit's SoC and power come from its driver, the GoodWe unit's "
+      "from its entities",
+      calls(f3, "read") and fa.soc == 60.0 and fb.soc == 50.0
+      and ("number", "sensor.soc_b", False) in h3.log)
+check("WRITE: the fake unit's share went to ITS driver and the GoodWe unit's to "
+      "ITS entity, each non-zero, the same direction",
+      calls(f3, "write") and calls(f3, "write")[-1] > 0
+      and [x for m, e, x in h3.log if m == "set_number" and e == "number.sp_b"][-1] > 0)
+check("ALLOCATE: each unit holds a share inside its own rating, and the shares "
+      "are the site's (3000 W of grid, fake capped at 1000 W by its driver)",
+      c3.allocation is not None
+      and set(c3.allocation.shares) == {"fake-a", "inverter-b"}
+      and 0 < c3.allocation.shares["fake-a"] <= 1000.0
+      and 0 < c3.allocation.shares["inverter-b"] <= 3000.0
+      and calls(f3, "write")[-1] == c3.allocation.shares["fake-a"])
+check("GRADE: the fake unit's rating is the driver's range (1000 W), the "
+      "GoodWe unit's is its entry's (3000 W), each from its own source",
+      (fa.rating_w(), fb.rating_w()) == (1000.0, 3000.0))
+check("the fake's power follows its command, so the freshness gate (the fake's "
+      "own bound) lets the reading through",
+      fa.batt == calls(f3, "write")[-1] and fa.batt_fresh.bound is FAKE_BOUND)
+run(c3.send_heartbeats(10_000.0, 10.0))
+n_before = len(calls(f3, "write"))
+check("HEARTBEAT: it re-emits the fake unit's standing command through its driver",
+      n_before >= 2 and calls(f3, "write")[-1] == calls(f3, "write")[-2])
+run(c3.command_site_zero())
+check("STOP: command_site_zero() reaches the fake unit through its driver, as "
+      "0 W (the hand-back is SMA-10's, and release() has no caller yet)",
+      calls(f3, "write")[-1] == 0.0 and not calls(f3, "release"))
+c3.publish()
+print(f"    published unit rows: "
+      f"{sorted(k for k in p3.last if k.startswith('unit/'))}")
+check("PUBLISH: one set of rows per unit, with that unit's own numbers",
+      p3.last.get("unit/fake_a/soc") == 60.0
+      and p3.last.get("unit/inverter_b/soc") == 50.0
+      and "unit/fake_a/battery" in p3.last and "unit/inverter_b/battery" in p3.last)
+rows = c3.checks()
+check("STATUS: the fake unit's sensor rows are not 'unset' (its driver has a "
+      "source), and it has the control row",
+      not [r for r in rows if r.get("id") == "sensor" and not r["ok"]
+           and "fake" in r["text"]])
+# liveness: the fake's 7 s window, the GoodWe unit's 30 s
+c4, h4, p4, f4 = site()
+run(cycles(c4, 2))
+f4.accept = False                  # the fake's link is down; GoodWe's is fine
+_now = datetime.now(timezone.utc)
+for u in c4.inverters:
+    u.note_write(False, _now - timedelta(seconds=10))
+check("LIVENESS: after 10 s of rejected writes the fake unit is left out of the "
+      "split and the GoodWe unit stays in it, per its driver's window",
+      c4.live_units() == ("inverter-b",))
+run(cycles(c4, 4))
+check("...and the survivor carries the site: the allocation names only it",
+      c4.allocation is not None and set(c4.allocation.shares) == {"inverter-b"}
+      and c4.allocation.shares["inverter-b"] > 1000.0)
+# the tripwire: nothing in the world was asked about the fake unit
+asked = h3.entities() | h4.entities()
+print(f"    Home Assistant was asked about: {sorted(asked)}")
+check("TRIPWIRE: Home Assistant was only ever asked about the GoodWe unit's "
+      "entities and the meter - no empty name, no host - on read, write, "
+      "heartbeat, stop and limits alike",
+      asked <= B_ENT | {"sensor.grid"} and "" not in asked
+      and not any("10.9.9.9" in e or "fake" in e for e in asked))
+check("...and the GoodWe unit was never touched through the fake",
+      not [1 for w, a in f3.calls + f4.calls
+           if isinstance(a, str) and "number.sp_b" in a])
+
+# limits and the per-unit range check, through run_control()
+async def drive(c, secs):
+    try:
+        await asyncio.wait_for(c.run_control(), timeout=secs)
+    except asyncio.TimeoutError:
+        pass
+
+
+c5, h5, p5, f5 = site(rating_f=2500.0)
+f5.rating = 2500.0
+c5.inverters[0].dev_min = c5.inverters[0].dev_max = None
+c5.inverters[1].dev_min = c5.inverters[1].dev_max = None
+run(drive(c5, 3.0))
+check("LIMITS: run_control() reads the fake unit's range from its driver and "
+      "the GoodWe unit's from its entity",
+      calls(f5, "limits")
+      and (c5.inverters[0].dev_min, c5.inverters[0].dev_max) == (-2500.0, 2500.0)
+      and (c5.inverters[1].dev_min, c5.inverters[1].dev_max) == (-3000.0, 3000.0)
+      and [e for m, e, x in h5.log if m == "limits"] == ["number.sp_b"])
+c6, h6, p6, f6 = site()
+f6.rating = 1500.0                         # the entry says 2000 W
+c6.inverters[0].dev_min = c6.inverters[0].dev_max = None
+run(drive(c6, 10.0))
+check("RULE 2 PER UNIT: a fake unit rated above what its driver says it accepts "
+      "is refused by name, as a GoodWe one is",
+      c6.refused and any("fake-a" in p and "10.9.9.9:502:3" in p
+                         for p in c6.refused)
+      and not calls(f6, "write"))
+
+# the settings panel has no Controller: it binds each unit's driver itself
+from app import selfconfig                                    # noqa: E402
+hs = LogHass({}, {"number.sp_b": (-3000.0, 3000.0, 1.0)})
+_ps, _unread = run(selfconfig.problems_and_ranges(
+    dict(OPTS, inverters=[dict(OPTS["inverters"][0], _rating=1500.0),
+                          OPTS["inverters"][1]]), hs))
+check("PANEL: the settings panel reads the fake unit's range from its driver "
+      "and the GoodWe unit's from its entity, and grades the fake unit by name",
+      any("fake-a" in p and "10.9.9.9:502:3 accepts at most 1500" in p
+          for p in _ps)
+      and [e for m, e, x in hs.log if m == "limits"] == ["number.sp_b"]
+      and _unread == [])
+
+# the seam is the only way in: a fake-only roster touches Home Assistant for
+# nothing but the meter
+hz = LogHass({"sensor.grid": 800.0})
+cz = Controller(dict(OPTS, inverters=[OPTS["inverters"][0]]), hz, MemStore(),
+                Pub(), stop=asyncio.Event())
+cz.inverters[0].dev_min, cz.inverters[0].dev_max = -1000.0, 1000.0
+cz.auto = True
+run(cycles(cz, 3))
+check("A FAKE-ONLY ROSTER never asks Home Assistant for anything but the "
+      "meter, and still reads and writes (nothing on the control path is "
+      "keyed on a GoodWe entity)",
+      hz.entities() <= {"sensor.grid"} and calls(cz.inverters[0].driver, "read")
+      and calls(cz.inverters[0].driver, "write"))
+
+# ==============================================================================
+print()
+print("4. SAFETY-28's census counts actuator ids; rule 9 on the new keys")
+
+def entry(**kw):
+    return dict(name="u", max_w=3000, bms_soc_floor=10, **kw)
+
+
+check("GoodWe entries are counted as before: one entity named twice is one "
+      "actuator with two writers, whatever its case",
+      {k: len(v) for k, v in actuator_writers({"inverters": [
+          entry(setpoint_entity="number.X"),
+          entry(setpoint_entity=" number.x ")]}).items()} == {"number.x": 2}
+      and actuator_writers({"setpoint_entity": "number.a", "inverters": [
+          entry(setpoint_entity="number.a")]})["number.a"]
+      == ["setpoint_entity", "inverters[0].setpoint_entity"])
+cen = actuator_writers({"inverters": [
+    entry(driver="fake", host="gw", port=502, unit_id=1),
+    entry(driver="fake", host="gw", port=502, unit_id=2),
+    entry(driver="fake", host="gw", port=503, unit_id=1),
+    entry(driver="fake", host="GW", port=502, unit_id=1)]})
+print(f"    census {cen}")
+check("another driver's actuator is host:port:unit: two units behind one "
+      "gateway are two actuators, and one unit named twice is one",
+      set(cen) == {"gw:502:1", "gw:502:2", "gw:503:1"}
+      and cen["gw:502:1"] == ["inverters[0].host", "inverters[3].host"])
+check("a GoodWe key on a fake entry STILL declares a writer (the pre-SMA-2 "
+      "census kept, the driver's actuator added to it), and a `host` on a "
+      "GoodWe entry declares none",
+      actuator_writers({"inverters": [entry(driver="fake", host="gw",
+                                            setpoint_entity="number.x")]})
+      == {"number.x": ["inverters[0].setpoint_entity"],
+          "gw:502:1": ["inverters[0].host"]}
+      and actuator_writers({"inverters": [entry(setpoint_entity="number.x",
+                                                host="gw")]})
+      == {"number.x": ["inverters[0].setpoint_entity"]})
+check("an entry naming a driver this build lacks still counts its setpoint "
+      "entity - refusing too much is the safe direction",
+      actuator_writers({"inverters": [entry(driver="nope",
+                                            setpoint_entity="number.q")]})
+      == {"number.q": ["inverters[0].setpoint_entity"]})
+check("census_key() folds case and space",
+      census_key(" Number.X ") == "number.x")
+
+GOOD = dict(entry(setpoint_entity="number.sp_a", soc_entity="sensor.soc_a",
+                  batt_entity="sensor.batt_a"))
+BASEOPTS = {"meter_source": "off", "meter_entity": "sensor.grid"}
+base_problems = validate_startup_config(dict(BASEOPTS, inverters=[dict(GOOD)]))
+print(f"    baseline problems for a plain GoodWe roster: {base_problems}")
+
+
+def prob(**kw):
+    return validate_startup_config(dict(BASEOPTS, inverters=[dict(GOOD, **kw)]))
+
+
+check("an existing entry (no new key) is graded exactly as before",
+      prob() == base_problems)
+check("...and so is one that says driver goodwe_esphome, or carries every new "
+      "key in a valid shape",
+      prob(driver="goodwe_esphome") == base_problems
+      and prob(driver="", host="", maintenance=False) == base_problems
+      and prob(host="inv.local", port=502, unit_id=3, maintenance=True)
+      == base_problems)
+def named(problems, key):
+    return [p for p in problems if f"inverters.{key}" in p or f"({key})" in p]
+
+
+check("only goodwe_esphome can be selected for now",
+      named(prob(driver="sma_modbus"), "driver")
+      and named(prob(driver="fake"), "driver")
+      and SELECTABLE == (DEFAULT_DRIVER,))
+check("a host with a space, a bad port or unit number and a non-boolean "
+      "maintenance are each refused by key; the right shapes are not",
+      named(prob(host="a b"), "host")
+      and named(prob(port=0), "port") and named(prob(port=70000), "port")
+      and named(prob(port="x"), "port") and named(prob(port=True), "port")
+      and named(prob(unit_id=248), "unit_id")
+      and named(prob(unit_id=1.5), "unit_id")
+      and named(prob(maintenance="yes"), "maintenance")
+      and not named(prob(port=65535, unit_id=0, host="10.0.0.1"), "port"))
+check("a GoodWe entry with no setpoint entity is refused with the same sentence "
+      "as before",
+      any("has no" in p and "setpoint entity" in p.lower()
+          for p in validate_startup_config(dict(BASEOPTS, inverters=[
+              dict(GOOD, setpoint_entity="")]))))
+fu = FakeDriver(host="gw")
+check("rule 2 grades any unit by its driver's actuator id, so a unit with no "
+      "setpoint entity is no longer skipped",
+      validate_unit_ranges([Inverter("a", driver=fu, max_w=2000.0)])
+      == [] and validate_unit_ranges([_u := Inverter("a", driver=fu,
+                                                      max_w=2000.0)]) == []
+      and (setattr(_u, "dev_max", 1000.0) or True)
+      and len(validate_unit_ranges([_u])) == 1
+      and "gw:502:1" in validate_unit_ranges([_u])[0])
+
+# ==============================================================================
+print()
+print("5. the schema")
+cfg = io.open(HERE / "config.yaml", encoding="utf-8").read()
+sch = cfg[cfg.index("\nschema:"):]
+ent = sch[sch.index("\n  inverters:"):]
+ent = ent[:ent.index("\n\n")]
+decl = dict(ln.strip().lstrip("- ").split(": ", 1)
+            for ln in ent.splitlines()[1:] if ": " in ln)
+NEW_KEYS = ("driver", "host", "port", "unit_id", "maintenance")
+check("HIDDEN FROM CLIENTS (owner, 2026-11-07): config.yaml's per-entry schema "
+      "is what it was on 23ff9d0 - the five driver keys are not in it, and "
+      "setpoint_entity is still required - so the Supervisor form shows "
+      "nothing new",
+      not set(NEW_KEYS) & set(decl) and not decl["setpoint_entity"].endswith("?")
+      and set(decl) == {"name", "setpoint_entity", "max_w", "bms_soc_floor",
+                        "soc_entity", "batt_entity", "batt_invert",
+                        "capacity_kwh"})
+_tr = {lg: io.open(HERE / "translations" / f"{lg}.yaml", encoding="utf-8").read()
+       for lg in ("en", "nl")}
+from app import messages as _msgs                             # noqa: E402
+check("...and no translation, label or description was added for them",
+      not any(f"      {k}:\n" in t.split("  estop_fitted:")[0].split("  inverters:")[1]
+              for t in _tr.values() for k in NEW_KEYS)
+      and not set(NEW_KEYS) & (set(_msgs.LABELS) | set(_msgs.LABELS_NL)))
+
+# ==============================================================================
+print()
+print("6. round 2: census parity, setters, rule 9(d), NaN ports, hidden keys")
+
+import copy, itertools, math                                  # noqa: E402
+from app import uimodel                        # noqa: E402
+from app.validate import ACTUATOR_WRITE_KEYS                  # noqa: E402
+
+
+def old_census(opts):
+    """23ff9d0's census, copied: every ACTUATOR_WRITE_KEYS key, wherever it is."""
+    found = []
+
+    def walk(node, path):
+        if isinstance(node, dict):
+            for k, v in node.items():
+                here = f"{path}.{k}" if path else str(k)
+                if k in ACTUATOR_WRITE_KEYS:
+                    e = str(v or "").strip()
+                    if e:
+                        found.append((here, e))
+                    continue
+                walk(v, here)
+        elif isinstance(node, (list, tuple)):
+            for i, v in enumerate(node):
+                walk(v, f"{path}[{i}]")
+    walk(opts, "")
+    cen = {}
+    for where, e in found:
+        cen.setdefault(e.lower(), []).append(where)
+    return cen
+
+
+check("ACTUATOR_WRITE_KEYS is exactly ('setpoint_entity',): the parity below "
+      "is about that key", tuple(ACTUATOR_WRITE_KEYS) == ("setpoint_entity",))
+SP = ["number.a", "Number.A", " number.b ", "", None, 5, ["x"]]
+shapes = []
+for flat, e0, e1 in itertools.product(SP, SP, SP):
+    shapes.append({"setpoint_entity": flat, "inverters": [
+        {"name": "a", "setpoint_entity": e0}, {"name": "b", "setpoint_entity": e1}]})
+shapes += [{"x": {"y": [{"setpoint_entity": "number.z"}]}, "setpoint_entity": "n.q"},
+           {"inverters": "junk"}, {"inverters": [{"driver": "nope",
+                                                   "setpoint_entity": "n.k"}]},
+           {"inverters": [{"driver": "goodwe_esphome", "setpoint_entity": "n.k"}]},
+           {}, {"inverters": []}]
+bad = [i for i, o in enumerate(shapes) if actuator_writers(o) != old_census(o)]
+check(f"CENSUS PARITY: on {len(shapes)} flat/roster/odd option shapes the "
+      "census equals 23ff9d0's, keys, order and all", not bad)
+mixed = {"setpoint_entity": "number.a", "inverters": [
+    {"name": "a", "setpoint_entity": "number.a"},
+    {"name": "f", "driver": "fake", "host": "gw", "port": 502, "unit_id": 1,
+     "setpoint_entity": "number.ignored"},
+    {"name": "g", "setpoint_entity": "number.g"}]}
+new_c, old_c = actuator_writers(mixed), old_census(mixed)
+check("...and on a mixed config (a fake-driver entry too) the census is a "
+      "SUPERSET of the pre-SMA-2 one: every old writer is still counted, "
+      "plus the driver's own actuator",
+      all(new_c.get(k) and set(v) <= set(new_c[k]) for k, v in old_c.items())
+      and "gw:502:1" in new_c)
+
+gi = Inverter("s", setpoint_entity="number.s", soc_entity="sensor.a",
+              batt_entity="sensor.b")
+gi.setpoint_entity, gi.soc_entity, gi.batt_entity, gi.batt_invert = (
+    "number.t", "sensor.c", "sensor.d", True)
+check("SETTERS: assigning setpoint_entity/soc_entity/batt_entity/batt_invert "
+      "on an Inverter writes through to its GoodWe driver, as it did before",
+      (gi.driver.setpoint_entity, gi.driver.soc_entity, gi.driver.batt_entity,
+       gi.driver.batt_invert) == ("number.t", "sensor.c", "sensor.d", True)
+      and gi.has_writer() and gi.driver.actuator_id() == "number.t")
+
+
+def _assign():
+    inv_f.setpoint_entity = "x"
+
+
+try:
+    _assign()
+    _refused = False
+except AttributeError:
+    _refused = True
+check("...and on a driver that has no such field it refuses, not loses the value",
+      _refused)
+
+nogood = validate_startup_config(dict(BASEOPTS, inverters=[
+    {k: v for k, v in GOOD.items() if k != "setpoint_entity"}]))
+check("RULE 9(d): a GoodWe entry with no setpoint_entity is refused, by name "
+      "and by position, with the sentence it always had",
+      any(p.startswith("inverters[0] (u) has no") and "setpoint entity" in p.lower()
+          and "Set its ESPHome number entity, or remove the entry." in p
+          for p in nogood)
+      and nogood != base_problems)
+check("...also with `driver: goodwe_esphome` said out loud",
+      any("has no" in p for p in validate_startup_config(dict(
+          BASEOPTS, inverters=[{k: v for k, v in dict(GOOD, driver="goodwe_esphome").items()
+                                if k != "setpoint_entity"}]))))
+
+nan, inf = float("nan"), float("inf")
+for key in ("port", "unit_id"):
+    for bad_v, nm in ((nan, "NaN"), (inf, "inf"), (-inf, "-inf")):
+        try:
+            got = prob(**{key: bad_v})
+            ok_ = bool(named(got, key))
+        except Exception as e:                                 # noqa: BLE001
+            got, ok_ = repr(e), False
+        check(f"{key} = {nm} is refused by key and does not raise", ok_)
+
+ent = dict(name="a", setpoint_entity="number.a", max_w=3000, bms_soc_floor=10,
+           soc_entity="sensor.s", batt_entity="sensor.b", driver="goodwe_esphome",
+           host="h", port=502, unit_id=3, maintenance=False)
+opts_rt = {"inverters": [ent, dict(ent, name="b", setpoint_entity="number.b")]}
+back = uimodel.ui_to_options(uimodel.options_to_ui(copy.deepcopy(opts_rt)))
+check("an entry that carries the driver keys (from a caller that skips the "
+      "schema) comes back from a save through the page model unchanged",
+      back.get("inverters") == opts_rt["inverters"])
+
+del REGISTRY["fake"]
+
+# THE SEAM MUTANTS THIS FILE WAS RUN AGAINST (twelve, each one edit to a copy of
+# the tree; all twelve turn it red):
+#   write_unit calls hass.set_number itself           -> part 3 WRITE (and the tripwire)
+#   _read_unit calls hass.number(soc/batt entity)     -> part 3 READ
+#   run_control's limits keyed on setpoint_entity     -> part 3 LIMITS
+#   the panel's range check keyed on setpoint_entity  -> part 3 PANEL
+#   has_writer() back to bool(setpoint_entity)        -> part 2 has_writer
+#   is_live() window back to LINK_STALE_S             -> part 2 is_live
+#   link_note() window back to LINK_STALE_S           -> part 2 link_note
+#   Freshness built on BOUNDS["batt"]                 -> part 2 freshness bound
+#   _write_unit's guard keyed on setpoint_entity      -> part 3 WRITE
+#   the census back to the GoodWe key only            -> part 4 census
+#   rule 2 per unit back to .setpoint_entity          -> part 3 RULE 2 PER UNIT
+#   the status rows' source test back to the entities -> part 3 STATUS
+# NOT COVERED: the call in amain() that reads the flat setpoint entity's range
+# before any Controller exists. It runs only inside amain(), which no check
+# starts; it is two lines and the GoodWe driver's own limits(), graded in part 1.
+print()
+if fails:
+    print(f"{len(fails)} FAILED")
+    sys.exit(1)
+print("all passed")

goodwe_controller/tools/supervisor_shape_probe.py

@@ -257,6 +257,23 @@ for title, entry, expect in (
 ok, out = validate(dict(D_BASE, inverters="[]"))
 show("a JSON STRING where the list belongs is refused", not ok, out)
 
+print("\n== K. SMA-2: the driver keys are NOT in the schema (owner, 2026-11-07) ==")
+# `driver`, `host`, `port`, `unit_id` and `maintenance` are graded by validate.py
+# rule 9(h) but kept out of config.yaml so the Configuration tab shows nothing
+# new. This pins both halves: the shipped entry is the one it always was, and the
+# Supervisor (measured) silently DROPS an unknown key from an entry, so they
+# cannot arrive by the form or by YAML.
+ok, out = validate(dict(D_BASE, inverters=[D_FULL]))
+show("the pre-SMA-2 entry is ACCEPTED unchanged", ok and out["inverters"] == [D_FULL], out)
+for _k in ("driver", "host", "port", "unit_id", "maintenance"):
+    ok, out = validate(dict(D_BASE, inverters=[dict(D_FULL, **{_k: "x"})]))
+    show(f"an entry carrying `{_k}` is accepted and the key is dropped",
+         ok and out["inverters"] == [D_FULL], out)
+show("the per-entry schema has exactly the eight keys it had",
+     sorted(schema["inverters"][0]) == sorted(["name", "setpoint_entity", "max_w",
+        "bms_soc_floor", "soc_entity", "batt_entity", "batt_invert", "capacity_kwh"]),
+     json.dumps(schema["inverters"]))
+
 
 print("\n== E. the ROLLBACK direction, and the operator's UI ==")
 # The old (flat) schema, i.e. what an operator would be running after a