SAFETY-137: the shutdown and Stop 0 W is the last write every unit hears

The 0 W of shutdown and of the Stop button now waits (bounded, 2 s) for a
write already on its way to the same unit, for every driver, GoodWe included.
Once shutdown has begun, cycle, heartbeat and safety-stop writes are dropped
right before the write is created. One shared wait (_await_inflight) serves
the 0 W and SMA-10's _release_last.

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

goodwe_controller/CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ### Not in a release yet
 
+- When the add-on stops, a command that was already on its way can no longer reach an inverter after the 0 W. The 0 W is now the last thing every inverter gets, also when Stop is pressed ([[SAFETY-137]]).
+- When the add-on stops, an inverter that can steer itself is handed back instead of held at 0 W. Safety stops still hold 0 W, and a GoodWe site is not affected ([[SMA-10]]).
+- 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]]).

goodwe_controller/DECISIONS/SAFETY-137-last-write-is-zero.md

@@ -0,0 +1,21 @@
+# SAFETY-137 - the shutdown 0 W is the last write every unit hears
+
+SMA-10 made a hand-back release wait for a write already on its way
+(`_release_last`, `Inverter.inflight`). A GoodWe unit skipped that, and rig 5b
+saw a cycle write (-2 823 W) land after the shutdown 0 W and stay until relaunch.
+
+Now the 0 W of shutdown and of the Stop button is a *last* write
+(`_write_unit(last=True)`), for every driver:
+
+- It first waits, at most `SETTLE_BUDGET_S` (2 s), for the writes already on
+  its way to that unit (`_await_inflight`, shared with `_release_last`). A hung
+  write is not waited out and not cancelled; the 0 W goes anyway.
+- Once shutdown has begun (`stopping`), every other write (cycle, heartbeat,
+  safety stop) is dropped at the same place, right before the write is created.
+  That is the gate that keeps the control cycle quiet; the cycle itself needs
+  no cancelling.
+- Worst case for a GoodWe unit is 2 s + `WRITE_BUDGET_S` (4 s) = 6 s, the same
+  as `HAND_BACK_WORST_S`; the units are handled together.
+
+Not touched: the heartbeat's own timing, safety-stop paths, the maintenance
+abort (still a plain 0 W).

goodwe_controller/DECISIONS/SMA-10-two-idle-states.md

@@ -0,0 +1,62 @@
+# SMA-10 - two idle states: hold 0 W, or hand the inverter back
+
+**Status: for review.** Tests: `test_sma10_two_idle_states.py`. Source: SMA-9
+report sections 1.4, 3.2; owner decision 2026-10-20 (Q1, "as proposed").
+
+A unit is idle in one of two ways, and the driver says which:
+`Driver.hands_back`. **False (GoodWe):** the add-on holds 0 W, `release()` is
+`write(0)` and the Controller never calls it, so Stop keeps today's 0 W write and
+today's 0 W heartbeat. **True:** `release()` gives the unit back to whatever
+steers it by itself (SMA energy management), once, and then nothing is written
+to it until control starts again.
+
+**Why two.** When the add-on goes away (Stop, shutdown, SIGTERM, restart, task
+death, a config refusal) nobody is left to hold a number, so the unit goes back
+to its own controller. When the add-on itself decides to stop (sign trip, site
+sign, circulating power, a refused or non-finite write, missing readings,
+another writer, an inverter fault) it has just said it cannot trust that
+battery, and handing it to something else would let that something do what it
+likes with it. Those hold 0 W through `write(0)` for every unit, whatever its
+driver. Shadow mode on a unit that hands back uses `release()`: it must leave
+the unit alone, and 0 W would be a command.
+
+**Where.** `Controller.command_site_release()` sits beside `command_site_zero()`
+(unchanged). Stop and `shutdown()` call it; the two maintenance-abort paths keep
+zero (the add-on runs on, the sentence says 0 W for a moment). With no
+hand-back unit it is `command_site_zero()`. `_write_unit()` also refuses to write
+a hand-back unit while the add-on is idle (stopping, stopped, or shadow mode): it
+releases once and returns, so the heartbeat and the cycle's own "stopped" 0 W
+cannot reach it. A safety stop runs while control runs, so it falls through to
+the write. Start re-arms: `released` drops and `last_write` clears.
+
+**Bounds.** `RELEASE_BUDGET_S` equals `WRITE_BUDGET_S` (4 s): units are released
+together, so one slow unit costs the kill window no more than a hung GoodWe
+write does. A release at the bound is abandoned, not awaited. A release that
+raises, hangs or returns false leaves that unit un-handed (the next idle write
+tries again) and does not touch the others. Exit 78 before the Controller
+exists has no driver to release and keeps today's path.
+
+**Words.** Every sentence that said "set to 0 W" for the add-on going away or
+for shadow mode is built from the roster in `app/handback.py` (EN and NL): a
+roster with no hand-back unit gets None and keeps its old literal, so a GoodWe
+site sees no changed byte. A hand-back unit's words (`Driver.words`: `back`,
+`back_nl`) come from its driver; one without them gets "its own control" /
+"de eigen sturing" (the article as in the generic clause; PR #239 review
+comment 2495, not the brief). MQTT `status` is unchanged; the Home battery
+row of a handed-back unit says "Back under SMA control" and carries
+`handed_back`.
+
+**Not done here (SMA-4, SMA-5, SMA-11):** any SMA/FoxESS driver, the read-back
+after `release()`, per-driver timing, DOCS.md's safety model and the
+"control stalled" sentence, the site-level "Held at 0 W" tiles in shadow mode on
+a mixed roster, and the MQTT per-unit setpoint of a handed-back unit (it still
+publishes 0).
+
+**After a failed release** (returns False, raises or hangs past the bound) the
+unit gets one `write(0)` through `Driver.release_fallback()` (default
+`write(0)`; a driver where 0 W is not always safe overrides it), then nothing:
+a repeated 0 W would keep resetting the inverter's own fallback timer, so after
+that the unit is left to its own fallback. The release is retried once per
+heartbeat; one error line and one action-log event per failure streak, one line
+on recovery. Worst case at shutdown is `RELEASE_BUDGET_S` + `FALLBACK_BUDGET_S`
+(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,320 @@
+"""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 = ""
+    #: SMA-10. Does `release()` give the inverter BACK to something that steers
+    #: it by itself (SMA energy management)? False: `release()` is `write(0)`,
+    #: the add-on goes on holding 0 W, and the Controller never calls it.
+    hands_back = False
+    #: per-driver sentences. A driver that hands back says whose control the
+    #: unit returns to: `{"back": "SMA control", "back_nl": "de sturing van SMA"}`
+    #: ("goes back to SMA control"). Missing words fall back to a plain phrase.
+    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.
+
+        SMA-10. Only a driver with `hands_back = True` has this called by the
+        Controller, and only when the add-on goes away (Stop, shutdown, restart,
+        task death) or sits in shadow mode. After it the Controller writes
+        nothing more to the unit until control starts again. A safety stop is
+        never a release: that is `write(0)`."""
+        return await self.write(0.0)
+
+    async def release_fallback(self) -> bool:
+        """SMA-10 round 2. What to send, ONCE, when `release()` has failed
+        (False, raised, or hung past the bound), so the unit does not carry its
+        last command while nothing else steers it. Default: `write(0)`. It is
+        not repeated: a repeat would keep resetting the inverter's own fallback
+        timer. A driver for which 0 W is not always a safe hold overrides this."""
+        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)` and `hands_back` is False: the board's own watchdog
+    is what hands back, so after Stop the unit keeps its 0 W write and heartbeat.
+    """
+
+    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/handback.py

@@ -0,0 +1,386 @@
+"""SMA-10: the two idle states, in words.
+
+A unit is either HELD at 0 W (GoodWe: the add-on goes on writing 0 W) or
+HANDED BACK (its driver has `hands_back = True`: `release()` gives it back to
+whatever steers it by itself, and the add-on then writes nothing to it). So every
+sentence that said "every inverter is set to 0 W" for the add-on going away or
+for shadow mode has to be built from the roster. DECISIONS/SMA-10-two-idle-
+states.md.
+
+THE RULE FOR EVERY BUILDER HERE: a roster with no hand-back unit gets None, and
+the caller then uses the sentence it always used, untouched. So a GoodWe-only
+site cannot see a changed byte; test_sma10_two_idle_states.py renders each one
+and compares it with the literal from before this card.
+
+The words for a hand-back unit come from its driver (`Driver.words`), not from
+here: `back` is "SMA control" in "goes back to SMA control", `back_nl` is the
+Dutch noun phrase after "terug naar". A driver without them gets a plain,
+brand-free phrase.
+
+Safety stops and refusals are not in this file on purpose: they hold 0 W for
+every unit, and their sentences say so.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+_BACK_EN = "its own control"
+_BACK_NL = "de eigen sturing"
+
+
+@dataclass(frozen=True)
+class Unit:
+    name: str
+    hands_back: bool = False
+    back_en: str = _BACK_EN
+    back_nl: str = _BACK_NL
+
+
+def unit_of(inv) -> Unit:
+    """One roster entry's view: its name, and (for a hand-back driver) the
+    words of whose control it returns to."""
+    drv = inv.driver
+    back = bool(getattr(drv, "hands_back", False))
+    words = getattr(drv, "words", None) or {}
+    return Unit(str(inv.name), back,
+                str(words.get("back") or _BACK_EN),
+                str(words.get("back_nl") or _BACK_NL))
+
+
+def units_of(inverters) -> tuple:
+    return tuple(unit_of(i) for i in inverters)
+
+
+# The roster of the Controller that is running, for the callers that have no
+# Controller in hand (the page, the banner). One process runs one Controller.
+_CURRENT: list = [()]
+
+
+def set_roster(units) -> None:
+    _CURRENT[0] = tuple(units)
+
+
+def current() -> tuple:
+    return _CURRENT[0]
+
+
+def active(units) -> bool:
+    """Is any unit handed back? False is every GoodWe-only roster."""
+    return any(u.hands_back for u in units)
+
+
+# --------------------------------------------------------------------------- #
+# The grammar. English and Dutch, a few verb forms each.
+# --------------------------------------------------------------------------- #
+def _join(names, lang) -> str:
+    names = list(names)
+    word = " en " if lang == "nl" else " and "
+    if len(names) <= 1:
+        return "".join(names)
+    return ", ".join(names[:-1]) + word + names[-1]
+
+
+def _forms(lang):
+    """kind -> (zero template, back template); `{n}` the names, `{b}` the words.
+    `one`/`many` pick the verb by how many units the clause is about."""
+    if lang == "nl":
+        return {
+            "set": (("{n} gaat naar 0 W", "{n} gaan naar 0 W"),
+                    ("{n} gaat terug naar {b}", "{n} gaan terug naar {b}")),
+            "at": (("{n} op 0 W",) * 2, ("{n} onder {b}",) * 2),
+            "is": (("{n} staat op 0 W", "{n} staan op 0 W"),
+                   ("{n} staat onder {b}", "{n} staan onder {b}")),
+            "stays": (("{n} blijft op 0 W", "{n} blijven op 0 W"),
+                      ("{n} blijft onder {b}", "{n} blijven onder {b}")),
+        }
+    return {
+        "set": (("{n} is set to 0 W", "{n} are set to 0 W"),
+                ("{n} goes back to {b}", "{n} go back to {b}")),
+        "at": (("{n} at 0 W",) * 2, ("{n} under {b}",) * 2),
+        "is": (("{n} is at 0 W", "{n} are at 0 W"),
+               ("{n} is under {b}", "{n} are under {b}")),
+        "stays": (("{n} stays at 0 W", "{n} stay at 0 W"),
+                  ("{n} stays under {b}", "{n} stay under {b}")),
+    }
+
+
+def _groups(units, lang):
+    """`(zero names, [(names, back words)])`: the held units, and the handed-back
+    ones grouped by whose control they return to, in roster order."""
+    zeros = [u.name for u in units if not u.hands_back]
+    groups: dict = {}
+    for u in units:
+        if u.hands_back:
+            groups.setdefault(u.back_nl if lang == "nl" else u.back_en,
+                              []).append(u.name)
+    return zeros, [(names, back) for back, names in groups.items()]
+
+
+# A clause may not eat the sentence it sits in: STYLE.md allows 25 words, and
+# the longest tail after a clause is about 11.
+CLAUSE_MAX_WORDS = 14
+
+
+def _count_txt(n, lang) -> str:
+    if lang == "nl":
+        return f"{n} omvormer" if n == 1 else f"{n} omvormers"
+    return f"{n} inverter" if n == 1 else f"{n} inverters"
+
+
+def _clause(units, lang, kind, sep, generic, counted) -> str:
+    forms = _forms(lang)[kind]
+    zeros, groups = _groups(units, lang)
+    if generic:                 # one group, and no make named
+        names = [n for g, _b in groups for n in g]
+        own = ("de eigen sturing" if lang == "nl" else
+               "their own control" if len(names) > 1 else "its own control")
+        groups = [(names, own)] if names else []
+    subject = ("de omvormers", "de omvormer") if lang == "nl" else (
+        "the inverters", "the inverter")
+    parts = []
+    if zeros:
+        parts.append(forms[0][1 if len(zeros) > 1 else 0].format(
+            n=(str(len(zeros)) if groups else _count_txt(len(zeros), lang))
+            if counted else _join(zeros, lang)))
+    for names, back in groups:
+        if not zeros and len(groups) == 1:
+            names_txt = subject[0 if len(names) > 1 else 1]
+        elif counted:
+            # the bare number after a counted first clause: "1 inverter is set
+            # to 0 W and 2 go back to their own control"
+            names_txt = (str(len(names)) if zeros
+                         else _count_txt(len(names), lang))
+        else:
+            names_txt = _join(names, lang)
+        parts.append(forms[1][1 if len(names) > 1 else 0].format(
+            n=names_txt, b=back))
+    if sep:
+        return sep.join(parts)
+    word = " en " if lang == "nl" else " and "
+    return (", ".join(parts[:-1]) + word + parts[-1]) if len(parts) > 1 \
+        else "".join(parts)
+
+
+def clause(units, lang, kind, sep=None) -> str:
+    """One clause about the whole roster, `kind` one of set/at/is/stays.
+
+    Held units first, then the handed-back ones, joined by "and" (or `sep`).
+    When EVERY unit is handed back to the same control the subject is "the
+    inverters", not their names, as the card's table has it.
+
+    A long roster would break the 25-word rule, so a clause over
+    CLAUSE_MAX_WORDS is said again without naming the make (one "own control"
+    for all), and then by counting units instead of naming them."""
+    text = _clause(units, lang, kind, sep, False, False)
+    if sep or _words(text) <= CLAUSE_MAX_WORDS:
+        return text
+    text = _clause(units, lang, kind, sep, True, False)
+    if _words(text) <= CLAUSE_MAX_WORDS:
+        return text
+    return _clause(units, lang, kind, sep, True, True)
+
+
+def _cap(text) -> str:
+    return text[:1].upper() + text[1:]
+
+
+def _bi(en, nl):
+    from .messages import Bi
+    return Bi(en, nl)
+
+
+def _words(text) -> int:
+    return len([w for w in re.findall(r"\S+", text)
+                if re.search(r"[A-Za-z0-9]", w)])
+
+
+# --------------------------------------------------------------------------- #
+# The sentences. Each returns None for a roster with no hand-back unit.
+# --------------------------------------------------------------------------- #
+def stop_dialog(units, lang):
+    """Stop control?  (was "Every inverter is set to 0 W until you start
+    control again.")"""
+    if not active(units):
+        return None
+    c = clause(units, lang, "set")
+    return (f"{_cap(c)} tot u de regeling opnieuw start." if lang == "nl"
+            else f"{_cap(c)} until you start control again.")
+
+
+def restart_dialog(units, lang):
+    """Restart now? / Save and restart? (live)"""
+    if not active(units):
+        return None
+    c = clause(units, lang, "set")
+    return (f"{_cap(c)} terwijl de add-on herstart, daarna gelden de nieuwe "
+            f"instellingen." if lang == "nl" else
+            f"{_cap(c)} while the add-on restarts, and then the new settings "
+            f"apply.")
+
+
+def restart_dialog_shadow(units, lang):
+    """Save and restart? when shadow mode keeps the units idle afterwards."""
+    if not active(units):
+        return None
+    c = clause(units, lang, "set")
+    return (f"{_cap(c)} terwijl de add-on herstart, en de schaduwmodus houdt "
+            f"het zo." if lang == "nl" else
+            f"{_cap(c)} while the add-on restarts, and shadow mode keeps it "
+            f"that way.")
+
+
+def setup_dialog_shadow(units, lang):
+    """Save and restart? in setup, shadow mode on."""
+    if not active(units):
+        return None
+    c = clause(units, lang, "at")
+    return (f"De add-on herstart met deze instellingen, en de schaduwmodus "
+            f"houdt {c}." if lang == "nl" else
+            f"The add-on restarts with these settings, and shadow mode keeps "
+            f"{c}.")
+
+
+def control_stopped(units, lang):
+    """The page's control line when stopped."""
+    if not active(units):
+        return None
+    c = clause(units, lang, "is")
+    return (f"De regeling is gestopt. {_cap(c)}." if lang == "nl"
+            else f"Control is stopped. {_cap(c)}.")
+
+
+def control_shadow(units, lang):
+    """The page's control line in shadow mode."""
+    if not active(units):
+        return None
+    c = clause(units, lang, "stays")
+    return (f"Schaduwmodus: {c}." if lang == "nl" else f"Shadow mode: {c}.")
+
+
+def attn_shadow_body(units, lang):
+    """The attention card's body in shadow mode."""
+    if not active(units):
+        return None
+    c = clause(units, lang, "stays")
+    return (f"De add-on beslist alles, maar {c}." if lang == "nl" else
+            f"The add-on decides everything, but {c}.")
+
+
+def shadow_state(units, lang, many=True):
+    """The tail of a notification: "Shadow mode is on, so <this>." For the
+    three output-clamp sentences that said every inverter stays at 0 W."""
+    if not active(units):
+        return None
+    return clause(units, lang, "stays")
+
+
+def banner_stopped(units):
+    """Stopped. 12 words at most (STYLE.md)."""
+    if not active(units):
+        return None
+    out = {}
+    for lang, stopped in (("en", "Stopped"), ("nl", "Gestopt")):
+        text = f"{stopped} — {clause(units, lang, 'at', sep=', ')}"
+        if _words(text) > 12:
+            text = _banner_counts(units, lang, stopped)
+        out[lang] = text
+    return _bi(out["en"], out["nl"])
+
+
+def _banner_counts(units, lang, stopped) -> str:
+    """The long-roster form: how many, not which."""
+    zeros, groups = _groups(units, lang)
+    if len(groups) == 1:
+        back = groups[0][1]
+        n_back = len(groups[0][0])
+        if lang == "nl":
+            held = f"{len(zeros)} op 0 W, " if zeros else ""
+            return f"{stopped} — {held}{n_back} onder {back}"
+        held = f"{len(zeros)} at 0 W, " if zeros else ""
+        return f"{stopped} — {held}{n_back} under {back}"
+    return (f"{stopped} — omvormers op 0 W of teruggegeven" if lang == "nl"
+            else f"{stopped} — inverters at 0 W or handed back")
+
+
+def banner_shadow(units):
+    """Shadow mode, nothing standing. The GoodWe sentence adds "Inverters stay
+    at 0 W." which is false for a unit that hands back, and a banner is 12
+    words, so the unit-by-unit state is left to the control line."""
+    if not active(units):
+        return None
+    return _bi("Shadow mode: the battery is not controlled.",
+               "Schaduwmodus: de batterij wordt niet geregeld.")
+
+
+def hint_shadow_on(units):
+    """"Shadow mode on, inverters stay at 0 W" for the Last event line."""
+    if not active(units):
+        return None
+    return "Shadow mode on, " + clause(units, "en", "at", sep=", ")
+
+
+def unit_state(unit, lang):
+    """One handed-back unit's own state, in place of "Held at 0 W"."""
+    if lang == "nl":
+        return f"Terug onder {unit.back_nl}"
+    return f"Back under {unit.back_en}"
+
+
+def log_shutdown(units):
+    """The INFO line at shutdown (was "shutting down - commanding 0 W on every
+    inverter")."""
+    if not active(units):
+        return None
+    zeros, groups = _groups(units, "en")
+    held = f"commanding 0 W on {_join(zeros, 'en')}" if zeros else ""
+    back = " and ".join(f"handing back {_join(n, 'en')} to {b}"
+                        for n, b in groups)
+    return "shutting down - " + " and ".join(x for x in (held, back) if x)
+
+
+def task_died_state(units) -> str:
+    """The part of the CRITICAL line after "so" when the control task dies.
+    For a roster with no hand-back unit it is the sentence it always was."""
+    if not active(units):
+        return "every inverter is set to 0 W"
+    zeros, groups = _groups(units, "en")
+    backs = [n for names, _b in groups for n in names]
+    parts = []
+    if zeros:
+        parts.append(f"{_join(zeros, 'en')} "
+                     f"{'are' if len(zeros) > 1 else 'is'} set to 0 W")
+    parts.append(f"{_join(backs, 'en')} "
+                 f"{'are' if len(backs) > 1 else 'is'} handed back")
+    return " and ".join(parts)
+
+
+# --------------------------------------------------------------------------- #
+# The page words: the ui_text sentences that move with the roster, by language.
+# --------------------------------------------------------------------------- #
+def overlay(units) -> dict:
+    """`{lang: nested dict}` of ui_text paths to replace; {} when no unit is
+    handed back, so the page is the file as it always was."""
+    if not active(units):
+        return {}
+    out = {}
+    for lang in ("en", "nl"):
+        out[lang] = {
+            "home": {"dialog": {"stop": {
+                "description": stop_dialog(units, lang)}}},
+            "settings": {
+                "dialog": {"description": restart_dialog(units, lang)},
+                "setup": {
+                    "dialog": {"description":
+                               restart_dialog_shadow(units, lang)},
+                    "dialog_live": {"description":
+                                    restart_dialog(units, lang)},
+                    "dialog_setup": {"description":
+                                     setup_dialog_shadow(units, lang)}}},
+            "home_words": {
+                "attn_shadow_body": attn_shadow_body(units, lang),
+                "control": {"shadow": control_shadow(units, lang),
+                            "stopped": control_stopped(units, lang)}},
+        }
+    return out

goodwe_controller/app/homeview.py

@@ -19,7 +19,7 @@ section at the bottom and is held to STYLE.md by test_messages.py.
 import contextlib
 from datetime import datetime, timedelta, timezone
 
-from . import arbitrage, cheaproll, messages, peakguard, priceruns, uitext
+from . import arbitrage, cheaproll, handback, messages, peakguard, priceruns, uitext
 from .dayplan import EFF
 from .maintenance import CHARGE as M_CHARGE, DRAIN as M_DRAIN, HOLD as M_HOLD
 from .maintenance import IDLE as M_IDLE, next_slot_start
@@ -936,6 +936,13 @@ def battery_rows(c, shadow, running, held=False):
                              else other, at=at,
                              **plan_words(c, inv.name, why)),
         })
+        if getattr(inv, "released", False):
+            # SMA-10. A unit handed back says so, in its driver's words, not
+            # "Held at 0 W": nothing is written to it. The key exists only for
+            # such a unit, so a GoodWe-only page is unchanged.
+            rows[-1]["now"] = handback.unit_state(handback.unit_of(inv),
+                                                  _LANG[0])
+            rows[-1]["handed_back"] = True
         if modes_on:
             # OPT-11 P1: the battery's switch and mode, for the Batteries
             # screen. Only with battery modes on, so the page is unchanged
@@ -1211,6 +1218,9 @@ _WORDS: dict = {}
 
 def _hw() -> dict:
     lang = _LANG[0]
+    if _WORDS.get("") != uitext.generation():     # SMA-10: the roster's words moved
+        _WORDS.clear()
+        _WORDS[""] = uitext.generation()
     if lang not in _WORDS:
         _WORDS[lang] = uitext.words(lang)
     return _WORDS[lang]

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
@@ -266,6 +275,21 @@ class Inverter:
         self.target = 0.0
         self.last_write = None
         self.last_write_at = None
+        # SMA-10. True from the moment this unit is handed back (`release()`)
+        # until control starts again. While it is True nothing is written to the
+        # unit - not the heartbeat, not the cycle - so the unit's own
+        # controller keeps what it was given. Only a driver with
+        # `hands_back = True` ever sets it.
+        self.released = False
+        # SMA-10. Writes to a hand-back driver that are on their way (see
+        # Controller._write_unit), so the release can wait for them and be LAST.
+        self.inflight: set = set()
+        # SMA-10 round 2. A failed hand-back: how many tries in a row failed
+        # (the log is written at the first and on recovery only), when the next
+        # try is due (monotonic s), and whether the one fallback 0 W was sent.
+        self.release_fails = 0
+        self.release_retry_at = 0.0
+        self.release_zero_sent = False
         self.last_write_ok = None
         # ⚠️ TWO TIMESTAMPS, NOT ONE FLAG. `last_write_ok` is the last write's
         # verdict and it flips on a single transient 400; liveness needs a RUN
@@ -338,6 +362,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 +455,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 +466,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 +479,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 +523,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 +630,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 +660,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/main.py

@@ -67,10 +67,12 @@ from . import freshness
 from . import lateness
 from . import dayview, homeview
 from .daycurve import SolarToday
+from . import handback
 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
@@ -88,6 +90,7 @@ from . import peakguard
 from . import packshare
 from . import pv as pv_read
 from . import shadow
+from . import uitext
 from .p1 import P1Ingest, build_source, is_enabled
 from .price import Absence, Uncertainty, brussels, price_option, provider_configured
 from .price import required_spread as price_required_spread
@@ -137,6 +140,26 @@ __all__ = ["UNIT_A", "Controller", "amain", "main"]
 # the re-issue happens by itself) than hold the control loop for ten seconds.
 WRITE_BUDGET_S = 4.0
 
+# SMA-10. The same bound for handing a unit back (`driver.release()`). It is the
+# time the shutdown path already allows one 0 W write, and the units are
+# released CONCURRENTLY, so one slow or hung unit costs the wind-down no more
+# than a hung GoodWe write does today: the Supervisor's kill window is not lost
+# to it. A release still running at the bound is abandoned, not awaited.
+RELEASE_BUDGET_S = WRITE_BUDGET_S
+
+# SMA-10 round 2. The one fallback write after a failed release (default 0 W,
+# `Driver.release_fallback()`). Worst case for a unit whose release hangs:
+# RELEASE_BUDGET_S + FALLBACK_BUDGET_S = 6 s, against the Supervisor's 10 s.
+FALLBACK_BUDGET_S = 2.0
+HAND_BACK_WORST_S = RELEASE_BUDGET_S + FALLBACK_BUDGET_S
+
+# SAFETY-137. How long a LAST write (the Stop 0 W, the shutdown 0 W, a release)
+# waits for a write already on its way to the same unit. Half the release
+# budget, as `_release_last` always waited. For a GoodWe unit that is
+# SETTLE_BUDGET_S + WRITE_BUDGET_S = 6 s at worst, which is HAND_BACK_WORST_S:
+# the wind-down is no slower than SMA-10's worst case.
+SETTLE_BUDGET_S = RELEASE_BUDGET_S / 2
+
 # SAFETY-34: a Core-only restart wipes persistent_notification and does NOT
 # restart this add-on (measured on TEST-23), so a latched alarm is re-asserted
 # this often. Why: goodwe docs/DECISIONS.md, SAFETY-34.
@@ -560,6 +583,10 @@ class Controller:
         self.store_migration = migrate_unit_keys(
             store, _roster_names[0], _roster_names)
         self.inverters = build_roster(opts, store, sign_cfg, delivery_cfg)
+        # SMA-10. Which units hand back decides what the page says about Stop,
+        # restart and shadow mode. A GoodWe-only roster adds nothing.
+        handback.set_roster(handback.units_of(self.inverters))
+        uitext.set_idle(handback.overlay(handback.current()))
         # ⚠️ CFG-04. SAY WHICH SHAPE IS IN FORCE, BECAUSE THE OTHER ONE'S KEYS
         # ARE STILL SITTING IN THE UI. A declared `inverters:` roster reads none
         # of `soc_entity`, `batt_entity`, `batt_invert` or `max_w`, and an
@@ -3774,6 +3801,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 +3832,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)
@@ -3986,7 +4027,7 @@ class Controller:
         return rec.authorises(inv.name, value, self.alloc_tick)
 
     async def write_setpoint(self, value: float, *, force: bool = False,
-                             unit=None) -> None:
+                             unit=None, last: bool = False) -> None:
         """Put `value` on ONE inverter, or - with `unit=None` - on all of them.
 
         ⚠️ `unit=None` IS FOR A NUMBER THAT IS NOT A SHARE. The stop button,
@@ -4025,13 +4066,127 @@ class Controller:
             targets = [unit]
         if not targets:
             return
-        await asyncio.gather(*(self._write_unit(inv, value, force=force)
+        await asyncio.gather(*(self._write_unit(inv, value, force=force,
+                                                last=last)
                                for inv in targets))
 
-    async def _write_unit(self, inv, value: float, *, force: bool = False) -> None:
-        entity = inv.setpoint_entity
-        if not entity:
+    async def _await_inflight(self, inv, budget_s: float) -> None:
+        """Wait, at most `budget_s`, for the writes already on their way to
+        `inv` (SAFETY-137: every driver, GoodWe too). A write still running at
+        the bound is left alone, not cancelled: its own WRITE_BUDGET_S ends it."""
+        pending = [t for t in inv.inflight if not t.done()]
+        if pending:
+            await asyncio.wait(pending, timeout=budget_s)
+
+    async def _release_last(self, inv) -> bool:
+        """`release()`, after a write already on its way to `inv` has finished
+        (at most half the budget is spent waiting for it), so the release is the
+        LAST thing the unit hears and a late write cannot undo it."""
+        await self._await_inflight(inv, SETTLE_BUDGET_S)
+        return await self._driver(inv).release()
+
+    async def _hand_back(self, inv, *, force: bool = False) -> bool:
+        """SMA-10. Give ONE unit back: exactly one `release()`, then nothing.
+
+        True when the unit is handed back (now, or already). Never raises: a
+        driver that raises or hangs is logged and the unit is tried again on the
+        next idle write, and it cannot stop a sibling being released or zeroed.
+
+        ⚠️ `released` IS SET BEFORE THE FIRST AWAIT. A second caller (the
+        heartbeat while the Stop button is mid-release) then sees it and does
+        nothing, so a unit gets one `release()` and not two. If the release
+        did not take, the flag goes back down and the unit stays un-handed
+        until a later idle write tries again.
+
+        ⚠️ IT IS NOT A WRITE: no `note_write`, no `last_write`, so a hand-back
+        cannot be read as a link that answered. `last_write` is cleared so the
+        first write after control starts again is always sent.
+        """
+        if inv.released:
+            return True
+        if not inv.driver.configured():
+            return False
+        # After a failure the next try waits one heartbeat, so a hung release
+        # costs a cycle its 4 s once per heartbeat, not on every cycle. The
+        # Stop button and shutdown (`force`) always try.
+        now_m = time.monotonic()
+        if not force and now_m < inv.release_retry_at:
+            return False
+        inv.released = True
+        why = ""
+        try:
+            ok = bool(await _bounded(self._release_last(inv),
+                                     RELEASE_BUDGET_S))
+        except (asyncio.TimeoutError, TimeoutError):
+            ok, why = False, f"took over {RELEASE_BUDGET_S:.0f} s"
+        except Exception as err:  # noqa: BLE001 - one unit never stops the rest
+            ok, why = False, f"{type(err).__name__}: {err}"
+        if ok:
+            inv.last_write = None
+            if inv.release_fails:
+                _LOG.info("Handing back %s worked again after %d failed "
+                          "tries.", inv.name, inv.release_fails)
+            inv.release_fails = 0
+            inv.release_retry_at = 0.0
+            inv.release_zero_sent = False
+            self.log_event(f"{inv.name}: handed back",
+                           hint=messages.hint("handed_back", unit=inv.name))
+            return True
+        inv.released = False
+        inv.release_fails += 1
+        inv.release_retry_at = now_m + float(self.o.get("heartbeat_s", 10))
+        _LOG.debug("Handing back %s failed (try %d): %s", inv.name,
+                   inv.release_fails, why)
+        if inv.release_fails == 1:          # one line per failure streak
+            _LOG.error("Handing back %s failed, so it is sent 0 W once and "
+                       "handing back is tried again.", inv.name)
+            self.log_event(f"{inv.name}: hand-back failed",
+                           hint=messages.hint("hand_back_failed", unit=inv.name))
+        if not inv.release_zero_sent:
+            # ⚠️ ONCE. The unit must not keep carrying its last command, but a
+            # 0 W re-sent on every try would reset the inverter's own fallback
+            # timer for ever. The driver says what the one write is.
+            inv.release_zero_sent = True
+            try:
+                if await _bounded(self._driver(inv).release_fallback(),
+                                  FALLBACK_BUDGET_S):
+                    inv.last_write = 0.0
+            except Exception as err:  # noqa: BLE001
+                _LOG.debug("The fallback write to %s failed: %r", inv.name, err)
+        return False
+
+    async def _write_unit(self, inv, value: float, *, force: bool = False,
+                          last: bool = False) -> None:
+        if not inv.driver.configured():
             return
+        # --- SMA-10: A UNIT THAT HANDS BACK IS NOT WRITTEN WHILE THE ADD-ON IS
+        # IDLE. Idle is: shutting down, control stopped, or shadow mode. The
+        # unit is released once and then nothing is written to it - not the
+        # heartbeat, not the cycle's "stopped" 0 W - so its own controller keeps
+        # it. This is the passive half; `command_site_release()` is the active
+        # one, for the moment the add-on goes away.
+        #
+        # ⚠️ A SAFETY STOP IS NOT IDLE. It happens while control runs (`auto`,
+        # not shadow), so it falls through to the 0 W write below: the add-on's
+        # own decision to hold 0 W is never turned into a hand-back.
+        #
+        # ⚠️ AND CONTROL STARTING AGAIN RE-ARMS THE UNIT: the flag drops and
+        # `last_write` is cleared, so the first write goes out whatever it is.
+        # `stopping` never re-arms: a shutdown in progress stays handed back.
+        if inv.driver.hands_back:
+            if self.stopping or self.shadow or not self.auto:
+                await self._hand_back(inv)
+                return
+            if inv.released or inv.release_fails or inv.release_zero_sent:
+                # Control has the unit again: forget the hand-back, and any
+                # failed streak with it, so the next Stop starts clean (its own
+                # one 0 W, its own error line and event).
+                inv.released = False
+                inv.last_write = None
+                inv.release_fails = 0
+                inv.release_retry_at = 0.0
+                inv.release_zero_sent = False
+        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
@@ -4144,6 +4299,21 @@ class Controller:
                        value, inv.name, shadow.WIRE_W)
             value = shadow.WIRE_W
 
+        # --- SAFETY-137: THE 0 W THAT ENDS THINGS IS THE LAST WRITE THE UNIT
+        # HEARS. `last` is the Stop 0 W and the shutdown 0 W (and nothing else).
+        # It first waits, bounded, for a write already on its way to this unit,
+        # so that write cannot land after it. And once shutdown has begun every
+        # other write (cycle, heartbeat, a safety stop) is dropped here: the
+        # check sits right before the write is created with no await between,
+        # so a write that starts after it cannot exist, and one that started
+        # before is in `inflight` for the wait above. The idle gate for a
+        # hand-back unit is higher up and does the same for those.
+        if last:
+            await self._await_inflight(inv, SETTLE_BUDGET_S)
+        elif self.stopping:
+            _LOG.debug("Shutting down, so %+.0f W to %s is not sent.",
+                       value, inv.name)
+            return
         changed = (inv.last_write is None or value != inv.last_write)
         if not (changed or force):
             return
@@ -4154,8 +4324,13 @@ 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),
-                                        timeout=WRITE_BUDGET_S)
+            _w = self._driver(inv).write(value)
+            # SMA-10 / SAFETY-137: kept for every driver, so a release or a last
+            # 0 W can wait for it (`_await_inflight`).
+            _w = asyncio.ensure_future(_w)
+            inv.inflight.add(_w)
+            _w.add_done_callback(inv.inflight.discard)
+            ok = await asyncio.wait_for(_w, timeout=WRITE_BUDGET_S)
         except (asyncio.TimeoutError, TimeoutError):
             ok = False
             _LOG.error("Sending %+.0f W to %s (%s) took over %.0f s and was "
@@ -6730,9 +6905,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
@@ -6772,9 +6946,10 @@ class Controller:
             self.log_event("REFUSED: " + problems[0],
                            hint=messages.hint("refused"))
             # Signal, then return WITHOUT entering the loop. amain() wakes
-            # on this, runs the normal wind-down (which commands 0 W) and
-            # re-raises ConfigRefused afterwards, so the exit code is 78 and
-            # the inverter is left at zero rather than latched.
+            # on this, runs the normal wind-down (which commands 0 W, or hands
+            # back a unit whose driver does: SMA-10) and re-raises
+            # ConfigRefused afterwards, so the exit code is 78 and the
+            # inverter is left at zero rather than latched.
             self.stop_event.set()
             return
 
@@ -7038,11 +7213,43 @@ class Controller:
     async def shutdown(self) -> None:
         """Deterministic wind-down. Do not skip this."""
         self.stopping = True
-        _LOG.info("shutting down - commanding 0 W on every inverter")
-        await self.command_site_zero()
+        _LOG.info(handback.log_shutdown(handback.current())
+                  or "shutting down - commanding 0 W on every inverter")
+        await self.command_site_release()
         self.mqtt.close()
 
-    async def command_site_zero(self) -> None:
+    async def command_site_release(self) -> None:
+        """SMA-10. The add-on is going away (Stop, shutdown, restart, task
+        death, a config refusal): every unit that hands back is given back, and
+        every other unit is held at 0 W exactly as `command_site_zero()` does.
+
+        ⚠️ THE SAFETY STOPS AND REFUSALS DO NOT COME HERE. A sign trip, a
+        refused write, a missing reading, another writer: those are the add-on's
+        own decisions and hold 0 W through `write(0)` for EVERY unit, handed-back
+        drivers included. `command_site_zero()` is that, and it is unchanged. A
+        unit the add-on has just declared it cannot trust must not be handed to
+        something else that will then do what it likes with the battery.
+
+        ⚠️ A ROSTER WITH NO HAND-BACK UNIT IS `command_site_zero()`, line for
+        line: GoodWe's release is its 0 W write, so Stop keeps today's write
+        and today's heartbeat.
+
+        ⚠️ ONE UNIT'S FAILURE IS ITS OWN. `_hand_back()` never raises and is
+        bounded by RELEASE_BUDGET_S; the units run concurrently, so the whole
+        call takes no longer than its slowest bound.
+        """
+        if not any(inv.driver.hands_back for inv in self.inverters):
+            await self.command_site_zero(last=True)
+            return
+        self.target = 0.0
+        for inv in self.inverters:
+            inv.target = 0.0
+        await asyncio.gather(*(
+            self._hand_back(inv, force=True) if inv.driver.hands_back
+            else self._write_unit(inv, 0.0, force=True, last=True)
+            for inv in self.inverters))
+
+    async def command_site_zero(self, *, last: bool = False) -> None:
         """Every inverter to 0 W, and every inverter's TARGET to 0 W with it.
 
         ⚠️ THE SECOND HALF IS NOT BOOKKEEPING, AND IT IS A HAZARD DEV-02
@@ -7064,7 +7271,7 @@ class Controller:
         self.target = 0.0
         for inv in self.inverters:
             inv.target = 0.0
-        await self.write_setpoint(0.0, force=True)
+        await self.write_setpoint(0.0, force=True, last=last)
 
     # -- UI ------------------------------------------------------------------
     def checks(self, when: datetime | None = None) -> list:
@@ -7316,9 +7523,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 +7568,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:
@@ -7997,7 +8204,9 @@ class Controller:
                 # Zeroing `self.target` alone would leave each unit's last share
                 # standing for its next heartbeat to re-emit, which an operator
                 # would read as the stop button not working.
-                await self.command_site_zero()
+                # SMA-10: the Stop button is the add-on going quiet, so a unit
+                # that hands back is given back; every other unit is zeroed.
+                await self.command_site_release()
             return {"auto": self.auto}
         if what in ("maint_start", "maint_abort") and unit is not None:
             # MAINT-11: the operator named the battery. A request with no unit
@@ -8041,6 +8250,21 @@ MAINT_CHARGING_W = 50.0
 EXIT_CONTROL_DIED = 70
 
 
+async def _bounded(awaitable, budget_s: float):
+    """SMA-10. `await awaitable`, but give up after `budget_s` and do NOT wait
+    for it to wind down. `asyncio.wait_for` waits for the cancelled task to
+    finish, so a driver that ignores its cancellation would hold the whole
+    shutdown; this abandons it. Raises `asyncio.TimeoutError` at the bound."""
+    task = asyncio.ensure_future(awaitable)
+    done, _ = await asyncio.wait({task}, timeout=budget_s)
+    if not done:
+        task.cancel()
+        # a result or error that arrives later has nobody to read it
+        task.add_done_callback(lambda t: t.cancelled() or t.exception())
+        raise asyncio.TimeoutError
+    return task.result()
+
+
 class ControlTaskDied(RuntimeError):
     """The control task ended while nothing had asked it to stop."""
 
@@ -8067,10 +8291,12 @@ def control_task_done(task: asyncio.Task, *, controller, stop: asyncio.Event,
     if stop.is_set() or controller.stopping:
         return          # the normal wind-down, or run_control()'s refusal path
     exc = None if task.cancelled() else task.exception()
-    _LOG.critical("Control stopped unexpectedly (%s), so every inverter is "
-                  "set to 0 W and the add-on exits with code %d.",
+    _LOG.critical("Control stopped unexpectedly (%s), so %s and the add-on "
+                  "exits with code %d.",
                   "cancelled" if task.cancelled() else
-                  (repr(exc) if exc else "returned"), EXIT_CONTROL_DIED,
+                  (repr(exc) if exc else "returned"),
+                  handback.task_died_state(handback.current()),
+                  EXIT_CONTROL_DIED,
                   exc_info=exc)
     died.append(exc)
     stop.set()
@@ -8205,10 +8431,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/messages.py

@@ -12,6 +12,8 @@ import re
 from dataclasses import dataclass
 from datetime import datetime
 
+from . import handback
+
 # Severity (STYLE.md section 3). An ALARM may use up to three sentences so
 # that no action a person must take is lost; everything else gets one.
 ALARM, WARNING, INFO, RESOLVED = "alarm", "warning", "info", "resolved"
@@ -228,6 +230,19 @@ class Clamp:
     limit: str          # a key of LIMIT_WORDS
 
 
+def _shadow_so(held: str) -> str:
+    """"Shadow mode is on, so <held>." SMA-10: with a unit handed back, the
+    roster's own words say where each unit stays; otherwise `held`, as ever."""
+    moved = handback.shadow_state(handback.current(), "en")
+    return f"Shadow mode is on, so {moved or held}."
+
+
+def _shadow_keeps(held: str) -> str:
+    moved = handback.clause(handback.current(), "en", "at") \
+        if handback.active(handback.current()) else None
+    return moved or held
+
+
 def output_clamp(clamps, *, shadow: bool = False) -> Notice:
     """SAFETY-03: a computed command above the limit. A software fault.
 
@@ -238,7 +253,7 @@ def output_clamp(clamps, *, shadow: bool = False) -> Notice:
     fault = "This is a software fault, so please report it to your installer."
     if not clamps:
         cut = ("would have been cut back" if shadow else "was cut back")
-        state = ("Shadow mode is on, so every inverter stays at 0 W." if shadow
+        state = ((_shadow_so("every inverter stays at 0 W")) if shadow
                  else "Control is still running.")
         return Notice(ID_OUTPUT_CLAMP, ALARM, T_OUTPUT_CLAMP,
                       f"The add-on asked for more power than {RATING} allows "
@@ -250,11 +265,11 @@ def output_clamp(clamps, *, shadow: bool = False) -> Notice:
     if others:
         verb = "was" if len(others) == 1 else "were"
         state = (f"{names(others)} would have been cut too, and shadow mode "
-                 f"keeps every inverter at 0 W." if shadow else
+                 f"keeps {_shadow_keeps('every inverter at 0 W')}." if shadow else
                  f"{names(others)} {verb} cut too, and control is still "
                  f"running.")
     else:
-        state = ("Shadow mode is on, so the inverter stays at 0 W." if shadow
+        state = (_shadow_so("the inverter stays at 0 W") if shadow
                  else "Control is still running.")
     return Notice(ID_OUTPUT_CLAMP, ALARM, T_OUTPUT_CLAMP,
                   f"{who} add-on asked for {watts(c.asked_w)}, above "
@@ -576,6 +591,9 @@ def banner_running() -> str:
 
 
 def banner_stopped() -> str:
+    moved = handback.banner_stopped(handback.current())
+    if moved is not None:       # SMA-10: a unit is handed back, not held
+        return moved
     return Bi("Stopped — the inverters are set to 0 W",
               "Gestopt — de omvormers staan op 0 W")
 
@@ -601,6 +619,9 @@ def banner_shadow(latched: bool) -> str:
     mode. Each stop has its own red row below, which names it and says how it
     clears.
     """
+    moved = None if latched else handback.banner_shadow(handback.current())
+    if moved is not None:       # SMA-10: a unit is handed back, not held
+        return moved
     if latched:
         return Bi("Shadow mode: battery not controlled. A safety stop stays "
                   "on, see below.",
@@ -1367,6 +1388,8 @@ HINTS = {
     "setup_only": "In setup, nothing is sent to any inverter",
     "roster": "Inverter list read: {n} inverters",
     "maint_history": "Maintenance history did not move over cleanly",
+    "handed_back": "{unit}: handed back, the add-on writes nothing to it",
+    "hand_back_failed": "{unit}: could not be handed back, set to 0 W once, trying again",
     "stop_restored": "{unit}: an earlier safety stop is back, it stays at 0 W",
     "readings_late": "Home Assistant readings arrive late, control holds",
     "readings_back": "Home Assistant readings are on time again",
@@ -1470,6 +1493,10 @@ def hint(key: str, **kw) -> str:
         kw["w"] = watts(kw["w"])
     if "sw" in kw:
         kw["sw"] = signed_watts(kw["sw"])
+    if key == "shadow_on":      # SMA-10: a unit handed back is not at 0 W
+        moved = handback.hint_shadow_on(handback.current())
+        if moved is not None:
+            return moved
     return HINTS[key].format(**kw)
 
 

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/uitext.py

@@ -13,6 +13,11 @@ import os
 DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ui_text")
 LANGS = ("en", "nl")
 _CACHE: dict = {}
+# SMA-10. The sentences that move with the roster (handback.overlay): empty for
+# every GoodWe-only roster, and then `raw()` is the file as it always was.
+_IDLE: dict = {}
+_GEN = [0]
+_MERGED: dict = {}
 
 
 def lang_of(value) -> str:
@@ -21,14 +26,36 @@ def lang_of(value) -> str:
     return head if head in LANGS else "en"
 
 
-def raw(lang: str) -> dict:
-    """One language file as written, without the English fallback."""
+def _file(lang: str) -> dict:
     if lang not in _CACHE:
         with io.open(os.path.join(DIR, f"{lang}.json"), encoding="utf-8") as f:
             _CACHE[lang] = json.load(f)
     return _CACHE[lang]
 
 
+def set_idle(overlay: dict) -> None:
+    """SMA-10: the roster-built sentences, `{lang: nested dict}` (or {})."""
+    _IDLE.clear()
+    _IDLE.update(overlay or {})
+    _MERGED.clear()
+    _GEN[0] += 1
+
+
+def generation() -> int:
+    """Changes whenever `set_idle()` does, so a cache of words can tell."""
+    return _GEN[0]
+
+
+def raw(lang: str) -> dict:
+    """One language file as written, without the English fallback (and with the
+    roster's own sentences laid over it when a unit is handed back)."""
+    if lang not in _IDLE:
+        return _file(lang)
+    if lang not in _MERGED:
+        _MERGED[lang] = _merge(_file(lang), _IDLE[lang])
+    return _MERGED[lang]
+
+
 def _merge(base, over):
     if not (isinstance(base, dict) and isinstance(over, dict)):
         return copy.deepcopy(over)

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/app/web.py

@@ -442,7 +442,8 @@ def build_app(controller, selfcfg=None) -> web.Application:
         """Restart the add-on, at the operator's explicit request.
 
         ⚠️ THIS IS THE ONE ROUTE ON THIS SURFACE THAT COSTS ANYTHING. The
-        wind-down commands 0 W on every inverter (`Controller.shutdown()`), so
+        wind-down commands 0 W on every inverter, or hands a unit back when its
+        driver does (`Controller.shutdown()`, SMA-10), so
         the request must carry the intent token; a reload, a retry or a stray
         POST cannot spend a control interruption. It restarts THIS ADD-ON and
         nothing else: no setpoint is written from here, and a latched failsafe

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_messages.py

@@ -793,6 +793,7 @@ from datetime import datetime as _dt, timedelta as _td, timezone as _tz  # noqa:
 
 from app import clamp as _clamp, freshness as _fresh, peakguard as _pg   # noqa: E402
 from app import store as _store                                         # noqa: E402
+from app import handback                                                # noqa: E402
 from app.circulation import CirculationConfig, CirculationWatchdog      # noqa: E402
 from app.control import Tuning, compute                                 # noqa: E402
 from app.delivery import (DeliveryConfig, DeliveryWatchdog, UnitDelivery,  # noqa: E402
@@ -1058,6 +1059,12 @@ TEXT = {
     "main.py:messages.reason_code(inv.batt_fresh.reason)": _coded(_stale_texts),
     "main.py:rep.explain()": _migration_texts,
     "store.py:rep.explain()": _migration_texts,
+    # SMA-10: the control-task CRITICAL line's state, for a GoodWe-only roster
+    # (the sentence it always was) and a mixed one.
+    "main.py:handback.task_died_state(handback.current())": lambda: [
+        handback.task_died_state(()),
+        handback.task_died_state((handback.Unit("inverter-a"),
+                                  handback.Unit("inverter-b", True)))],
     "main.py:plan.words": _tariff_texts,
     # OPT-11 P2: why an MQTT command was ignored.
     "main.py:why": lambda: list(messages.PACK_REFUSED.values()),

goodwe_controller/test_safety137_last_write.py

@@ -0,0 +1,332 @@
+"""SAFETY-137: the 0 W of shutdown and of the Stop button is the last write every
+unit hears.  `python3 test_safety137_last_write.py`
+
+A cycle write already on its way (0.3-0.6 s, or hung) must not land after the
+0 W, for a GoodWe unit and for a fake unit that does not hand back, at N=1 and
+N=2, at shutdown and at Stop. The control cycle (and the heartbeat) issue no
+new write once shutdown has begun. The whole wind-down stays inside
+HAND_BACK_WORST_S (6 s) when a write hangs.
+
+Mutating a module this file grades? Read MUTATION.md first (purge __pycache__).
+The mutants this file was run against are listed at the end.
+"""
+
+import asyncio
+import sys
+import time
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE))
+
+from app import main as M                                    # noqa: E402
+from app.drivers import Driver, Probe, REGISTRY             # noqa: E402
+from app.main import Controller                             # noqa: E402
+from app.store import DEFAULTS                              # noqa: E402
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+class PlainFake(Driver):
+    """A unit that does not hand back (so, like GoodWe, no release()). A write
+    of anything but 0 W takes `delay` s to land; every write is logged when it
+    LANDS (`landed`) and when it STARTS (`started`)."""
+    name = "plainfake"
+    writer_key = "host"
+    brand = "Fake"
+    hands_back = False
+    delay = 0.0
+
+    def __init__(self, host="", port=502, unit_id=1, rating=2000.0):
+        self.host, self.port, self.unit_id = host, port, unit_id
+        self.rating, self.power = rating, 0.0
+        self.landed, self.started = [], []
+
+    @classmethod
+    def from_entry(cls, entry):
+        return cls(entry.get("host", ""), entry.get("port", 502),
+                   entry.get("unit_id", 1))
+
+    @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):
+        return -self.rating, self.rating
+
+    async def read(self):
+        return 60.0, self.power, {}
+
+    async def write(self, w):
+        self.started.append((time.monotonic(), float(w)))
+        if w != 0 and self.delay:
+            await asyncio.sleep(self.delay)
+        self.landed.append((time.monotonic(), float(w)))
+        self.power = w
+        return True
+
+    async def probe(self):
+        return Probe(-self.rating, self.rating, 60.0, self.power, {})
+
+
+REGISTRY["plainfake"] = PlainFake
+
+
+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:
+    enabled = False
+
+    def publish(self, values):
+        pass
+
+    def close(self):
+        pass
+
+
+class SlowHass:
+    """Home Assistant for the GoodWe units. A set_number of anything but 0 W
+    takes `delay` s to land; the log gets the write when it lands."""
+
+    def __init__(self):
+        self.delay = 0.0
+        self.landed, self.started = [], []        # (t, entity, value)
+        self.numbers = {"sensor.grid": 3000.0, "sensor.soc_a": 50.0,
+                        "sensor.batt_a": 0.0, "sensor.soc_b": 50.0,
+                        "sensor.batt_b": 0.0}
+
+    async def number(self, entity, invert=False):
+        return self.numbers.get(entity)
+
+    async def limits(self, entity):
+        return (-3000.0, 3000.0, 1.0)
+
+    async def set_number(self, entity, value):
+        self.started.append((time.monotonic(), entity, value))
+        if value != 0 and self.delay:
+            await asyncio.sleep(self.delay)
+        self.landed.append((time.monotonic(), entity, value))
+        return True
+
+    async def state(self, entity):
+        return None
+
+    async def call(self, *a):
+        return True
+
+
+def gw(name, suffix):
+    return dict(name=name, setpoint_entity=f"number.sp_{suffix}",
+                soc_entity=f"sensor.soc_{suffix}",
+                batt_entity=f"sensor.batt_{suffix}", max_w=3000,
+                bms_soc_floor=10)
+
+
+def fk(name):
+    return dict(name=name, driver="plainfake", host="10.9.9.9", port=502,
+                unit_id=3, max_w=2000, bms_soc_floor=10)
+
+
+ROSTERS = {
+    "N=1 GoodWe": [gw("inverter-a", "a")],
+    "N=1 fake": [fk("inverter-a")],
+    "N=2 GoodWe+GoodWe": [gw("inverter-a", "a"), gw("inverter-b", "b")],
+    "N=2 GoodWe+fake": [gw("inverter-a", "a"), fk("inverter-b")],
+}
+
+
+def site(inverters, delay):
+    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=inverters)
+    hass = SlowHass()
+    hass.delay = delay
+    c = Controller(opts, hass, MemStore(), Pub(), stop=asyncio.Event())
+    for inv in c.inverters:
+        inv.dev_min, inv.dev_max = -3000.0, 3000.0
+        if isinstance(inv.driver, PlainFake):
+            inv.driver.delay = delay
+    c.auto = True
+    return c, hass
+
+
+def per_unit(c, hass, kind="landed"):
+    """name -> [(t, value)] in the order the writes LANDED (or started)."""
+    out = {}
+    for inv in c.inverters:
+        d = inv.driver
+        if isinstance(d, PlainFake):
+            out[inv.name] = list(getattr(d, kind))
+        else:
+            ent = d.setpoint_entity
+            out[inv.name] = [(t, v) for t, e, v in getattr(hass, kind)
+                             if e == ent]
+    return out
+
+
+async def cycles(c, n=2):
+    for _ in range(n):
+        await c.read_inputs()
+        await c.cycle()
+
+
+async def in_flight(c):
+    """A cycle write on its way to every unit (what rig 5b had): the real
+    cycle, whose writes are authorised and take `delay` s to land."""
+    tasks = [asyncio.ensure_future(cycles(c, 1))]
+    await asyncio.sleep(0.1)
+    assert all(not t.done() for t in tasks)
+    return tasks
+
+
+async def finish(tasks):
+    for t in tasks:
+        t.cancel()
+    for t in tasks:
+        try:
+            await t
+        except BaseException:           # noqa: BLE001 - cancelled or timed out
+            pass
+
+
+def last_is_zero(c, hass):
+    got = per_unit(c, hass)
+    return all(w and w[-1][1] == 0.0 for w in got.values()), got
+
+
+def scenario(label, inverters, how, delay):
+    """how: 'shutdown' or 'stop'. Returns (seconds, ok, detail)."""
+    c, hass = site(inverters, delay)
+
+    async def go():
+        tasks = await in_flight(c)
+        t0 = time.monotonic()
+        if how == "shutdown":
+            await c.shutdown()
+        else:
+            await c.handle_action("auto_toggle")        # Stop
+        dt = time.monotonic() - t0
+        # let a write that is still on its way land, to see if it overtakes
+        await asyncio.sleep(0.8 if delay < 5 else 0.0)
+        await finish(tasks)
+        return dt
+
+    dt = asyncio.run(go())
+    ok, got = last_is_zero(c, hass)
+    n_sent = {n: len(w) for n, w in got.items()}
+    check(f"{label}: the cycle write was on its way at {how} (test sees it)",
+          all(len(w) >= 2 for w in per_unit(c, hass, "started").values()))
+    print(f"    {label}: {how}, delay {delay} s: {dt:.2f} s, "
+          f"landed per unit {n_sent}, last "
+          f"{ {n: (w[-1][1] if w else None) for n, w in got.items()} }")
+    return dt, ok, got
+
+
+print("SAFETY-137 - the 0 W is the last write")
+
+print("1. a slow write on its way (0.3 s, 0.6 s) when shutdown / Stop starts")
+for label, roster in ROSTERS.items():
+    for how in ("shutdown", "stop"):
+        for delay in (0.3, 0.6):
+            dt, ok, got = scenario(label, roster, how, delay)
+            check(f"{label}, {how}, {delay} s in flight: the last write per "
+                  f"unit is 0 W", ok)
+            check(f"{label}, {how}, {delay} s in flight: the slow write "
+                  f"landed first, then the 0 W (not dropped, not overtaken)",
+                  all(len(w) >= 2 and w[-2][1] != 0.0
+                      for w in got.values()))
+            check(f"{label}, {how}: took about the write, not the budget "
+                  f"({dt:.2f} s)", dt < delay + 1.0)
+
+print("2. a hung write on its way: the 0 W still reaches every unit, in time")
+for label, roster in ROSTERS.items():
+    for how in ("shutdown", "stop"):
+        dt, ok, got = scenario(label, roster, how, 3600.0)
+        check(f"{label}, {how}, hung write: 0 W landed on every unit and is "
+              f"its last write", ok)
+        check(f"{label}, {how}, hung write: inside HAND_BACK_WORST_S "
+              f"({dt:.2f} s <= {M.HAND_BACK_WORST_S:.0f} s)",
+              dt <= M.HAND_BACK_WORST_S)
+        check(f"{label}, {how}, hung write: the 0 W goes after the settle "
+              f"wait ({M.SETTLE_BUDGET_S:.0f} s), not after the write's own "
+              f"budget ({dt:.2f} s)", dt <= M.SETTLE_BUDGET_S + 0.5)
+
+print("3. no new write once shutdown has begun (cycle, heartbeat)")
+for label, roster in ROSTERS.items():
+    c, hass = site(roster, 0.4)
+
+    async def go3():
+        tasks = await in_flight(c)
+        s = asyncio.ensure_future(c.shutdown())
+        await asyncio.sleep(0)                      # shutdown has begun
+        begun = time.monotonic()
+        await cycles(c, 3)                          # would write: grid 3000 W
+        await c.send_heartbeats(10 ** 9, 10)        # every unit is due
+        await asyncio.wait_for(s, 10)
+        await asyncio.sleep(0.8)
+        await finish(tasks)
+        return begun
+
+    begun = asyncio.run(go3())
+    starts = per_unit(c, hass, "started")
+    late = {n: [v for t, v in w if t >= begun and v != 0.0]
+            for n, w in starts.items()}
+    check(f"{label}: no non-zero write started after shutdown began "
+          f"({late})", all(not v for v in late.values()))
+    ok, got = last_is_zero(c, hass)
+    check(f"{label}: and the last write per unit is 0 W", ok)
+    # sanity: the same cycles DO write before shutdown
+    c2, hass2 = site(roster, 0.0)
+    asyncio.run(cycles(c2, 3))
+    before = per_unit(c2, hass2, "started")
+    check(f"{label}: control, same cycles before shutdown, write non-zero "
+          f"(the test can see a write)",
+          all(any(v != 0.0 for _t, v in w) for w in before.values()))
+
+print()
+if fails:
+    print(f"FAILED {len(fails)}:")
+    for f in fails:
+        print("  -", f)
+    sys.exit(1)
+print("all ok")
+
+# MUTANTS this file was run against (each must turn it red):
+#  (a) _await_inflight does nothing (no wait for in-flight writes), GoodWe too
+#  (b) the `elif self.stopping: return` gate removed (cycle not stopped)
+#  (c) `last=True` not passed by command_site_release (Stop/shutdown 0 W is a
+#      plain write)
+#  (d) the wait unbounded (no timeout): red on the settle-bound check

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