Merge release/1.0 (16029d7) into MAINT-20-2

Nine conflicts in app/main.py, all in comments around new code: every code
line is release/1.0's, every comment follows the part 2 rule.

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

goodwe_controller/app/cheaproll.py

@@ -0,0 +1,201 @@
+"""UX-77: where the cheap run rolls on. Display only; control is unchanged.
+
+A cheap run is sized for the batteries charging together, at their full
+rating, and starts at the first of the cheapest hours. The peak limit leaves
+less power than that, so a battery is still short when the run ends, and
+control plans the next run then (`Controller._price_runs()`). On a flat night
+that goes on until the batteries are at their level or the cheap hours end.
+
+`roll_end()` asks what control asks at each run end, of the same functions
+(`modetrade.energy()`, `priceruns.plan()`), until nothing is short or the next
+run does not follow on. `for_today()` reads the batteries at the level they
+will have by then: the charge power is the peak limit's room (peak aim minus
+the house, as `dayplan` counts it), shared as control shares it: by room
+left to full (`packshare.weights()`), read again as the levels move, so a
+battery that fills drops out and its share goes to the others. Where that
+room is not known (no house profile or no aim), the rating is used.
+DECISIONS/UX-54-forecast-tab.md has the record.
+"""
+
+import dataclasses
+import math
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+from . import arbitrage, dayplan, modetrade, packshare, priceruns
+from .dayplan import EFF
+from .price import required_spread, uncertainty_of
+
+SLOT_S = 900                    # a price slot
+STEP = timedelta(minutes=5)     # how finely the level is followed
+LIMIT = 24                      # rolls at most (a day of one-hour runs)
+
+
+def roll_end(run, points, packs_at, *, required, floor_pct, ceiling_pct,
+             targets=None, limit=LIMIT):
+    """Where the cheap run `run` ends once control has rolled it on.
+
+    `packs_at(end)` gives the `modetrade.Pack`s as they stand when a run
+    ends at `end`. At each end the charge still short is asked of
+    `modetrade.energy()` and the next run of `priceruns.plan()`, as control
+    asks them. It stops when nothing is short or the next run does not follow
+    on, and is `run.end` when control would not roll it."""
+    end_t = run.end
+    for _ in range(limit):
+        e = modetrade.energy(packs_at(end_t), floor_pct, ceiling_pct,
+                             targets=targets)
+        if e[0] <= 1e-6:
+            break
+        new = priceruns.plan(points, end_t, required_eur_kwh=required,
+                             charge_kwh=e[0], charge_kw=e[1], sell_kwh=0.0,
+                             sell_kw=0.0)
+        nxt = new.cheap
+        if nxt is None or not new.charge_useful() or nxt.start > end_t \
+                or nxt.end <= end_t:
+            break
+        end_t = nxt.end
+    return end_t
+
+
+@dataclass(frozen=True)
+class Roll:
+    """`end`: where control stops charging, `stops`: name -> when that battery
+    reaches its level (rounded up to a price slot), None while it still
+    charges at `end`. A battery not in `stops` does not charge in the run."""
+    end: datetime
+    stops: dict
+
+    def end_for(self, name, base_end):
+        """The end of `name`'s block: its stop, no later than `end`."""
+        if name not in self.stops:
+            return base_end
+        stop = self.stops[name]
+        return self.end if stop is None else min(self.end, stop)
+
+    def last(self, base_end):
+        """The last end of any battery's block (the chart's cheap bar)."""
+        return max([base_end] + [self.end_for(n, base_end)
+                                 for n in self.stops])
+
+
+def _ceil_slot(t):
+    return datetime.fromtimestamp(
+        math.ceil(t.timestamp() / SLOT_S) * SLOT_S, tz=t.tzinfo
+        or timezone.utc)
+
+
+class _Charge:
+    """The batteries' levels followed from `t0` while the chargers charge at
+    the room the peak limit leaves: `min(sum of ratings, aim - house)`, shared
+    by room left to full as control does (`_shares()`), each stopping at `modetrade.charge_to()`."""
+
+    def __init__(self, packs, t0, ceiling_pct, targets, aim_w, house):
+        self.packs = {p.name: p for p in packs}
+        self.t = t0
+        self.ceil = ceiling_pct
+        self.targets = targets
+        self.aim = aim_w
+        self.house = house
+        self.stops = {}
+
+    def at(self, end):
+        """The packs when a run ends at `end`."""
+        self._to(end)
+        return list(self.packs.values())
+
+    def _room_w(self, hour, total):
+        if self.aim is None or self.house is None:
+            return total
+        house_w, base_w = self.house
+        return min(total, max(0.0, float(self.aim)
+                              - max(house_w(hour), base_w, 0.0)))
+
+    def _to(self, end):
+        while self.t < end:
+            nxt = min(end, self.t + STEP)
+            for s, e, hour in dayplan._hour_slices(self.t, nxt):
+                self._charge(s, e, hour)
+            self.t = nxt
+
+    def _shares(self, ch, w):
+        """Watts per charger: `w` shared by `packshare.weights()` over all
+        packs (room left to full, as control passes it), no charger above its
+        rating, the surplus going to the others. By rating where the room left
+        is not known (no capacity)."""
+        wt = packshare.weights(list(self.packs.values()), -1.0)
+        if wt is None:
+            wt = {p.name: abs(p.rating_w) for p in ch}
+        wt = {p.name: wt.get(p.name, 0.0) for p in ch}
+        cap = {p.name: abs(p.rating_w) for p in ch}
+        out, open_, left = {}, set(wt), float(w)
+        while open_:
+            tot = sum(wt[n] for n in open_)
+            if tot <= 0.0:
+                for n in open_:
+                    out[n] = 0.0
+                break
+            over = [n for n in open_ if left * wt[n] / tot > cap[n] + 1e-9]
+            if not over:
+                for n in open_:
+                    out[n] = left * wt[n] / tot
+                break
+            for n in over:
+                out[n] = cap[n]
+                left -= cap[n]
+                open_.discard(n)
+        return out
+
+    def _charge(self, s, e, hour):
+        ch = modetrade.chargers(list(self.packs.values()), self.ceil,
+                                targets=self.targets)
+        total = sum(abs(p.rating_w) for p in ch)
+        if not ch or total <= 0.0:
+            return
+        w = self._room_w(hour, total)
+        hrs = (e - s).total_seconds() / 3600.0
+        share = self._shares(ch, w)
+        for p in ch:
+            self.stops.setdefault(p.name, None)
+            if not p.capacity_kwh:
+                continue
+            to = modetrade.charge_to(p, self.ceil, self.targets)
+            kwh = share[p.name] * hrs / 1000.0 * EFF
+            room = (to - p.soc) / 100.0 * float(p.capacity_kwh)
+            if kwh <= 0.0:
+                continue
+            if kwh >= room - 1e-9:
+                self.stops[p.name] = _ceil_slot(
+                    s + (e - s) * (room / kwh))
+                kwh = room
+            self.packs[p.name] = dataclasses.replace(
+                p, soc=p.soc + kwh / float(p.capacity_kwh) * 100.0)
+
+
+def for_today(c, rp, now):
+    """The `Roll` of the cheap run `rp.cheap` now under way or to come today,
+    or None when control does not roll it on (no run left, charging not
+    useful, the trade gate shut, nothing short at its end). Display only:
+    nothing here is fed back to control."""
+    if rp is None or not rp.on or rp.cheap is None or rp.cheap.end <= now \
+            or not rp.charge_useful() \
+            or getattr(c, "mode_gate", None) is not None:
+        return None
+    plan = c.fresh_tariff_plan(now)
+    if plan is None:
+        return None
+    floor, ceil = arbitrage.soc_bounds(c.o)
+    hp = getattr(c, "house_profile", None)
+    sim = _Charge(c._mode_packs(), max(now, rp.cheap.start), ceil,
+                  c.plan_targets(), c.peak_aim_w(),
+                  None if hp is None else hp.estimator())
+    required = required_spread(
+        arbitrage.threshold_eur_kwh(c.o),
+        uncertainty_of(plan.read.product)
+        | arbitrage.weather_uncertainty(True, c.solar_fc,
+                                        bool(c.weather_held(now))))
+    end = roll_end(rp.cheap, plan.read.points, sim.at, required=required,
+                   floor_pct=floor, ceiling_pct=ceil,
+                   targets=c.plan_targets())
+    if end <= rp.cheap.end:
+        return None
+    return Roll(end, dict(sim.stops))

goodwe_controller/app/control.py

@@ -298,6 +298,9 @@ PROBE_S = 34.0
 # command: every battery that may move this way is at its mode's limit.
 MODE_BOUND = "mode-bound"
 
+# SAFETY-128. The loop's reason while it waits for the readings to answer.
+WAITING = "waiting for readings"
+
 
 @dataclass(frozen=True)
 class Tuning:
@@ -486,6 +489,8 @@ def compute(
     bound: tuple | None = None,
     gave_nothing: bool | None = None,
     idle_w: float = 0.0,
+    hold: int = 0,
+    hold_w: float | None = None,
 ) -> Decision:
     """One control cycle. A cycle is one meter update (~5 s on a HomeWizard P1).
 
@@ -506,6 +511,9 @@ def compute(
     `idle_w`   SAFETY-97: what `actual_w` reads while the batteries give
                nothing (app/idle.py). The refusal's band and its end are
                centred on it. 0 is the law as it was.
+    `hold`     SAFETY-128: +1 / -1, the batteries have not answered the last
+               command that way (app/lateness.py), so the loop may not go
+               past `hold_w` (None: `prev_w`) that way. 0 is the law as it was.
     """
     reason = "tracking"
     # OPT-11: a bound that leaves out 0 W would FORCE a direction, which no
@@ -622,6 +630,11 @@ def compute(
             i_w = min(moved, i_w)
         else:
             i_w = max(moved, i_w)
+    # SAFETY-128: no further than `hold_w` until the readings answer the ask.
+    held_at = float(prev_w if hold_w is None else hold_w)
+    if hold and hold * (i_w - held_at) > 0:
+        i_w = held_at
+        reason = WAITING
 
     # OPT-11: the integrator does not wind past what the modes let the
     # batteries place. Only a side tighter than the rail is left in `bound`

goodwe_controller/app/dayplan.py

@@ -22,7 +22,10 @@ OPT-28 for the second):
   all of them cannot hold (`Plan.uncovered_kwh`) is not bought: another
   battery (Cheap hours, Solar first) or the grid carries it.
 * Cheap hours only ignores the plan and fills to the ceiling, as in P3 and
-  the design (section 1). Skipping the sun's share cost money in the replay
+  the design (section 1). OPT-36: what it, and a Solar first battery after
+  the night's use, will give in the expensive hours is taken off `short`
+  before the Leftover solar batteries share it (`_cheap_give_kwh()`,
+  `_solar_first_give_kwh()`). Skipping the sun's share cost money in the replay
   (PR #156 review: 29.72 against 29.41 EUR on real 2026 prices).
 * A Leftover solar battery charges from the grid up to the level it needs
   for its part of `short`, less its share of the sun before the run:
@@ -45,7 +48,7 @@ import statistics
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 
-from .packmode import LEFTOVER_SOLAR, SOLAR_FIRST
+from .packmode import CHEAP_HOURS, LEFTOVER_SOLAR, SOLAR_FIRST
 from .tariff import BILLING_TZ
 
 EFF = 0.95                  # charge (and discharge) efficiency
@@ -170,6 +173,14 @@ class Plan:
     short_kwh: float = 0.0
     cover_kwh: dict = field(default_factory=dict)
     uncovered_kwh: float = 0.0
+    # OPT-36: what Cheap hours batteries will give in those hours (kWh on the
+    # house's side), and, per Leftover solar battery, the grid kWh it buys
+    # less because of it.
+    cheap_kwh: float = 0.0
+    saved_kwh: dict = field(default_factory=dict)
+    # ...and the same for Solar first batteries (owner, 2026-11-02): their
+    # level when the expensive hours start, after the night's use.
+    solar_first_kwh: float = 0.0
 
 
 def _hour_slices(t0, t1):
@@ -219,12 +230,100 @@ def _room(p, to_pct):
         p.capacity_kwh)
 
 
+def _cheap_give_kwh(runs, now, house, packs, *, floor_pct, ceiling_pct,
+                    aim_w, until=None, slots=None, spread=None):
+    """OPT-36: what the Cheap hours batteries in `packs` give in the
+    expensive hours, in kWh on the house's side: (the level at the start of
+    those hours - the floor) x capacity x `EFF`.
+
+    That level is the one its cheap charging reaches: the ceiling, or less
+    when the time left cannot get there. That time is the cheap run from now
+    (or the whole run ahead of it) and, given the firm price `slots`, each
+    later slot before the expensive hours (`until`, a give run) still worth
+    charging in (its price plus the spread control asks, `spread`, at most
+    the give run's dearest slot, `priceruns.plan()`'s test) and no dearer
+    than every slot from the start of those hours on: control plans the
+    charge left in the cheapest window ahead, so a slot with a cheaper one
+    after it may never be used (round 4, a three-price night). The run is
+    sized at the batteries' ratings (`priceruns.need_hours()`), not the peak
+    limit, so a run too short to fill them is planned again in that time;
+    counting the run alone let A's level fall in the night after A had
+    bought. In that time the batteries together charge at their ratings,
+    held to what the peak aim leaves above the house's use that hour
+    (`aim_w`, None: no limit), shared by rating; x `EFF` of it is stored.
+    No cheap run worth charging in: the level now.
+    A battery that refuses a direction counts 0."""
+    ps = [p for p in packs if p.mode == CHEAP_HOURS and p.refused == 0]
+    if not ps:
+        return 0.0
+    house_w, base_w = house
+    gain = {p.name: 0.0 for p in ps}
+    rate = {p.name: abs(float(p.rating_w)) for p in ps}
+    total = sum(rate.values())
+    run = runs.cheap
+    spans = [(max(now, run.start), run.end)]
+    top = max((eur for s, _e, eur in slots or ()
+               if until is not None and until.start <= s < until.end),
+              default=None)
+    low = min((eur for s, _e, eur in slots or ()
+               if until is not None and s >= until.start), default=None)
+    if top is not None and spread is not None:
+        spans += [(max(s, now, run.end), min(e, until.start))
+                  for s, e, eur in slots
+                  if eur + float(spread) <= top + 1e-9
+                  and (low is None or eur <= low + 1e-9)
+                  and e > run.end and s < until.start]
+    if runs.charge_useful() and total > 0.0:
+        for t0, t1 in spans:
+            for s, e, h in _hour_slices(t0, t1):
+                w = total
+                if aim_w is not None:
+                    w = min(w, max(0.0, float(aim_w)
+                                   - max(house_w(h), base_w, 0.0)))
+                hrs = (e - s).total_seconds() / 3600.0
+                for p in ps:
+                    gain[p.name] += w * rate[p.name] / total * hrs \
+                        / 1000.0 * EFF
+    out = 0.0
+    for p in ps:
+        cap = float(p.capacity_kwh)
+        lvl = max(float(p.soc), min(float(ceiling_pct),
+                                    float(p.soc) + gain[p.name] / cap * 100.0))
+        out += max(0.0, lvl - float(floor_pct)) / 100.0 * cap * EFF
+    return out
+
+
+def _solar_first_give_kwh(house, now, start, packs, *, floor_pct):
+    """OPT-36 (owner, 2026-11-02): what the Solar first batteries in `packs`
+    will still give when the expensive hours start at `start`, in kWh on the
+    house's side. They give at any hour, so the house's use from now to
+    `start` is taken off first, shared by the energy each holds above the
+    floor, and what is left x `EFF` counts. Each hour is read at no less than
+    the night base: the profile is net of the sun the house used, and read
+    net it let a dark day's morning drain a battery that the plan still
+    counted (round 4, real-price replays). A battery
+    that runs down to the floor before then counts 0; one that refuses a
+    direction counts 0."""
+    ps = [p for p in packs if p.mode == SOLAR_FIRST and p.refused == 0]
+    if not ps:
+        return 0.0
+    house_w, base_w = house
+    use = sum(max(0.0, house_w(h), base_w) * (e - s).total_seconds() / 3.6e6
+              for s, e, h in _hour_slices(now, start))
+    have = {p.name: max(0.0, float(p.soc) - float(floor_pct)) / 100.0
+            * float(p.capacity_kwh) for p in ps}
+    took, _ = _spread(use / EFF, have, have)
+    return sum(have[n] - took[n] for n in have) * EFF
+
+
 def plan(forecast, house, runs, now, packs, *, trust_pct, floor_pct,
-         ceiling_pct):
+         ceiling_pct, aim_w=None, slots=None, spread=None):
     """The day plan, or None when there is none to make: no forecast, no
     house profile yet, no price plan, or no expensive run after the cheap
     run. `house` is `HouseProfile.estimator()`'s answer. `packs` are
-    `modetrade.Pack`s."""
+    `modetrade.Pack`s. `aim_w`, `slots` (`priceruns.firm_slots()`) and
+    `spread` (the price gap control asks, EUR/kWh) size a Cheap hours
+    battery's charge (`_cheap_give_kwh()`)."""
     if forecast is None or house is None or runs is None or not runs.on \
             or runs.cheap is None:
         return None
@@ -273,11 +372,23 @@ def plan(forecast, house, runs, now, packs, *, trust_pct, floor_pct,
                          - trust * pv)
     caps = {p.name: float(p.capacity_kwh) for p in price}
     span = max(0.0, float(ceiling_pct) - float(floor_pct))
-    held, left = _spread(short / EFF,
-                         {n: span / 100.0 * c for n, c in caps.items()},
-                         caps)
+    # OPT-36: the Cheap hours and Solar first batteries give first;
+    # Leftover solar holds the rest. A Leftover solar battery's own stored energy counts through its
+    # level: it buys from where it is to the level it needs.
+    cheap = _cheap_give_kwh(runs, now, house, usable, floor_pct=floor_pct,
+                            ceiling_pct=ceiling_pct, aim_w=aim_w,
+                            until=until, slots=slots, spread=spread) \
+        if price else 0.0
+    cheap = min(cheap, short)
+    sfirst = _solar_first_give_kwh(house, now, until.start, usable,
+                                   floor_pct=floor_pct) if price else 0.0
+    sfirst = min(sfirst, short - cheap)
+    others = cheap + sfirst
+    room_of = {n: span / 100.0 * c for n, c in caps.items()}
+    held, left = _spread((short - others) / EFF, room_of, caps)
+    held0 = _spread(short / EFF, room_of, caps)[0] if others > 0.0 else held
 
-    targets, need, share, cover = {}, {}, {}, {}
+    targets, need, share, cover, saved = {}, {}, {}, {}, {}
     for p in price:
         sh = excess * rooms[p.name] / total_room if total_room > 0 else 0.0
         sh_pct = sh / caps[p.name] * 100.0
@@ -285,6 +396,9 @@ def plan(forecast, house, runs, now, packs, *, trust_pct, floor_pct,
         level = max(0.0, top - sh_pct)
         targets[p.name] = round(level, 2)
         need[p.name] = round(_room(p, level), 3)
+        top0 = float(floor_pct) + held0[p.name] / caps[p.name] * 100.0
+        saved[p.name] = round(
+            _room(p, max(0.0, top0 - sh_pct)) - need[p.name], 3)
         share[p.name] = round(sh, 3)
         cover[p.name] = round(held[p.name] * EFF, 3)
     if sun < sf_room:
@@ -295,7 +409,8 @@ def plan(forecast, house, runs, now, packs, *, trust_pct, floor_pct,
         verdict = SUNNY
     return Plan(verdict, round(sun, 3), round(fc_sum, 3), until.start,
                 targets, need, share, round(short, 3), cover,
-                round(left * EFF if price else 0.0, 3))
+                round(left * EFF if price else 0.0, 3), round(cheap, 3),
+                saved, round(sfirst, 3))
 
 
 def hold(old: dict | None, new: dict, capacity: dict) -> dict:

goodwe_controller/app/dayview.py

@@ -19,7 +19,8 @@ empty chart. DECISIONS/UX-54-forecast-tab.md has the record.
 import dataclasses
 from datetime import datetime, timedelta
 
-from . import arbitrage, dayplan, homeview as hv, modetrade, priceruns
+from . import arbitrage, cheaproll, dayplan, homeview as hv, modetrade
+from . import priceruns
 from . import solarfc
 from .daycurve import Batt, curve
 from .packmode import SOLAR_FIRST
@@ -210,25 +211,18 @@ def _levels(c, est, fc, trust, now, end, runs, floor, ceil, notes_b, shadow,
             arbitrage.threshold_eur_kwh(c.o),
             uncertainty_of(plan.read.product)
             | arbitrage.weather_uncertainty(True, c.solar_fc))
-        end_t = cheap.end
-        for _ in range(24):
-            ext = dataclasses.replace(rp, cheap=priceruns.Run(
+
+        def packs_at(end_t):
+            """The packs at the curve's level when a run ends at `end_t`."""
+            at = dataclasses.replace(rp, cheap=priceruns.Run(
                 cheap.start, end_t, cheap.mean_eur_kwh))
-            rt, lv = settle(ext, lv0)
-            at_end = [dataclasses.replace(packs[n], soc=v)
-                      for n, v in ((n, _level_at(lv.get(n, ()), end_t))
-                                   for n in lv) if v is not None]
-            e = modetrade.energy(at_end, floor, ceil, targets=targets)
-            if e[0] <= 1e-6:
-                break
-            new = priceruns.plan(
-                plan.read.points, end_t, required_eur_kwh=required,
-                charge_kwh=e[0], charge_kw=e[1], sell_kwh=0.0, sell_kw=0.0)
-            nxt = new.cheap
-            if nxt is None or not new.charge_useful() or nxt.start > end_t \
-                    or nxt.end <= end_t:
-                break
-            end_t = nxt.end
+            _rt, lv = settle(at, lv0)
+            return [dataclasses.replace(packs[n], soc=v)
+                    for n, v in ((n, _level_at(lv.get(n, ()), end_t))
+                                 for n in lv) if v is not None]
+        end_t = cheaproll.roll_end(
+            cheap, plan.read.points, packs_at, required=required,
+            floor_pct=floor, ceiling_pct=ceil, targets=targets)
         ext = dataclasses.replace(rp, cheap=priceruns.Run(
             cheap.start, end_t, cheap.mean_eur_kwh))
     rt, lv = settle(ext, lv0)

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, messages, peakguard, 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
@@ -218,14 +218,20 @@ def price_windows(c, plan, verdict, now=None):
     """`[(kind, start, end)]`: where the add-on itself buys ("cheap") and
     gives ("dear"). With battery modes on that is the kept `priceruns.RunPlan`
     (`c.mode_runs`): the cheap run when charging in it pays and the trade gate
-    allows it (UX-53), and every give run. With them off it is the arbitrage
-    verdict's legs the add-on will take, the ones `trade_runs()` draws
-    (`trade_legs()`, given `now`). Nothing is classified here."""
+    allows it (UX-53), and every give run. UX-77: the cheap run is drawn to
+    where control stops charging as it rolls on (`cheaproll`). With them off
+    it is the arbitrage verdict's legs the add-on will take, the ones
+    `trade_runs()` draws (`trade_legs()`, given `now`). Nothing is classified
+    here."""
     if getattr(c, "modes_on", False):
         rp = getattr(c, "mode_runs", None)
         if rp is None or not rp.on or not _price_mode_in_use(c):
             return []
-        out = [("cheap", rp.cheap.start, rp.cheap.end)] \
+        end = rp.cheap.end
+        roll = None if now is None else cheaproll.for_today(c, rp, now)
+        if roll is not None:
+            end = roll.last(end)            # UX-77: to where control stops
+        out = [("cheap", rp.cheap.start, end)] \
             if rp.charge_useful() and getattr(c, "mode_gate", None) is None \
             else []
         return out + [("dear", g.start, g.end) for g in rp.give]
@@ -422,6 +428,7 @@ def mode_runs(c, day0, day1, now, shadow, ahead=False, rp=None):
     `ahead=True`) a charge from the floor and a give from the ceiling. The
     same filter, the same `MIN_KWH`. `rp`: a run plan to read instead of
     `c.mode_runs`. Home never passes either."""
+    today = rp is None and not ahead
     rp = rp if rp is not None else getattr(c, "mode_runs", None)
     if getattr(c, "mode_tick", None) is None:
         return [], None             # no cycle yet: nothing is known
@@ -431,6 +438,9 @@ def mode_runs(c, day0, day1, now, shadow, ahead=False, rp=None):
         return [], "not_worth"
     floor, ceiling = arbitrage.soc_bounds(c.o)
     gate = getattr(c, "mode_gate", None)
+    # UX-77: today's grid block runs to where control stops charging (the run
+    # rolls on when it ends), each battery's to its own stop.
+    roll = cheaproll.for_today(c, rp, now) if today else None
     out = []
     for i, inv in enumerate(c.inverters):
         pm = c.pack_mode(inv)
@@ -444,9 +454,16 @@ def mode_runs(c, day0, day1, now, shadow, ahead=False, rp=None):
             return got if got is not None else (
                 floor if kind == "grid" else ceiling)
 
+        grids = []                  # (end, kWh) of the grid blocks drawn
+
         def room(kind, hrs, start):
             kwh = inv.max_w / 1000.0 * hrs
             lvl = start_pct(kind, start) if ahead else soc
+            if kind != "grid" and not ahead and cap and lvl is not None:
+                # OPT-36: a give block reads the level after the grid block
+                # before it, not the level now.
+                lvl = min(float(ceiling), lvl + sum(
+                    k for e, k in grids if e <= start) / cap * 100.0)
             if cap and lvl is not None:
                 kwh = min(kwh, max(0.0, (ceiling - lvl) if kind == "grid"
                                    else (lvl - floor)) / 100.0 * cap)
@@ -458,7 +475,10 @@ def mode_runs(c, day0, day1, now, shadow, ahead=False, rp=None):
         if gate is None and rp.charge_useful() and (
                 need > 0.0 if need is not None
                 else pm.mode in GRID_CHARGE_MODES):
-            blocks.append(("grid", rp.cheap))
+            cheap = rp.cheap if roll is None else priceruns.Run(
+                rp.cheap.start, roll.end_for(inv.name, rp.cheap.end),
+                rp.cheap.mean_eur_kwh)
+            blocks.append(("grid", cheap))
         sell = rp.sell if pm.sell and pm.can_sell else None
         for g in rp.give:
             if sell is not None and g.start <= sell.start < g.end:
@@ -474,6 +494,8 @@ def mode_runs(c, day0, day1, now, shadow, ahead=False, rp=None):
                 kwh = min(kwh, need)
             if kwh < MIN_KWH:
                 continue
+            if kind == "grid":
+                grids.append((run.end, kwh))
             price = run.mean_eur_kwh
             if kind == "grid":
                 cov = plan_cover(c, inv.name)
@@ -522,17 +544,16 @@ def plan_cover(c, name):
     grid: the level it charges to, and the expensive-hour use of the house
     that charge covers and the sun will not. Read from the level control
     holds (`plan_targets()`), so Home says what control does. None when the
-    plan buys nothing for it."""
+    plan buys nothing for it. OPT-36: the kWh is the plan's own cover figure
+    (`Plan.cover_kwh`), not one rebuilt from the level."""
     dp = getattr(c, "day_plan", None)
     targets = c.plan_targets() if hasattr(c, "plan_targets") else None
     inv = next((i for i in c.inverters if i.name == name), None)
     if dp is None or not targets or name not in targets or inv is None \
-            or not inv.capacity_kwh or targets[name] <= 0.0:
+            or not inv.capacity_kwh or targets[name] <= 0.0 \
+            or name not in dp.cover_kwh:
         return None
-    floor, _ceil = arbitrage.soc_bounds(c.o)
-    cap = float(inv.capacity_kwh)
-    top = targets[name] + dp.share_kwh.get(name, 0.0) / cap * 100.0
-    return targets[name], max(0.0, top - floor) / 100.0 * cap * EFF
+    return targets[name], dp.cover_kwh[name]
 
 
 def _latched_below(c) -> bool:
@@ -915,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
@@ -931,7 +959,8 @@ def battery_rows(c, shadow, running, held=False):
 
 def day_key(c, name):
     """P4: the day plan's word for one Leftover solar or Cheap hours
-    battery: `sun_fills`, `tops_up`, `no_forecast`, or None."""
+    battery: `sun_fills`, `others_cover`, `tops_up`, `no_forecast`, or
+    None."""
     st = plan_state(c)
     if st == "no_forecast":
         mode = c.pack_mode(next(i for i in c.inverters if i.name == name))
@@ -941,7 +970,14 @@ def day_key(c, name):
     need = plan_need(c, name)
     if need is None:
         return None
-    return "tops_up" if need >= MIN_KWH else "sun_fills"
+    if need >= MIN_KWH:
+        return "tops_up"
+    # OPT-36: no grid charge because other batteries (Cheap hours only,
+    # Solar first) cover the hours, not the sun.
+    dp = getattr(c, "day_plan", None)
+    if dp is not None and dp.saved_kwh.get(name, 0.0) >= MIN_KWH:
+        return "others_cover"
+    return "sun_fills"
 
 
 def battery_why(*, modes_on, mode, off, reading, phase, shadow, running,
@@ -1182,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]
@@ -1473,7 +1512,8 @@ def line_solar(fc_kwh, count_kwh, at, kept=False) -> str:
 
 def line_uncovered(kwh) -> str:
     """OPT-28: the Leftover solar batteries cannot hold the expensive hours'
-    shortfall. The rest comes from another battery or the grid."""
+    shortfall. The rest comes from the grid: OPT-36 counts the other
+    batteries first."""
     return _t("solar.uncovered", kwh=_kwh(kwh))
 
 

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

@@ -0,0 +1,410 @@
+"""SAFETY-128: a reading that does not answer the last command is not current.
+
+When Home Assistant delivers readings late but still changing, every gate that
+judges arrival (SAFETY-01) or an unchanged value (SAFETY-18, SAFETY-37) passes
+them, and the loop corrects again, each cycle, an error its last command has
+already fixed. The readings show it: they answer each power request late.
+
+`Answer.watch()` (each 1 s read) fits two delays every FIT_EVERY_S:
+- `Delay`: how long after its request each battery's reading shows it. The
+  house does not move that reading, so a repeating load cannot fool it. It
+  needs the request to move FIT_MOVE_W; one reading per 10 s makes it slow.
+- `MeterFit`: how long after the site request the meter shows it. Quick at a
+  1-3 s meter, slower and for long lags only at a 5 s one (SAFETY-129), and a
+  house load that repeats can fool it.
+The meter counts only when an ask was slow to start and no battery fit said
+"on time"; with hysteresis (ON_S / OFF_S) the result is `late_seen`. Then (`observe()`, each cycle):
+- a new command more than ANSWER_W from the battery reading is PENDING until
+  that reading comes within ANSWER_W of it, or half way;
+- while pending the loop may not lead the battery reading by more than
+  ANSWER_W that way (`hold`, `hold_w`; not while `exempt`: a battery that
+  gives nothing under its ask or refuses is SAFETY-52/94's case) and the peak
+  guard may not push past what stands (`peak_cap`). Pending for LATE_S with
+  no reading moving towards it is LATE: control holds, then 0 W after
+  `stale_input_s`, as for a missing input.
+
+Not `late_seen`, nothing here acts. DECISIONS/SAFETY-128-late-readings.md.
+"""
+
+import bisect
+from collections import deque
+
+from . import freshness
+from .control import PROBE_ANSWER_W
+
+# Within this of the command the reading has answered it (a probe's answer).
+ANSWER_W = PROBE_ANSWER_W                                         # 100
+# The slowest healthy battery answer: SAFETY-75's 6 s dead time, the 6 s plant
+# settle, one 10 s publish and one 1 s read.
+LATE_S = (2 * freshness.PLANT_SETTLE_S + freshness.PUBLISH_PERIOD_S
+          + freshness.READ_PERIOD_S)                              # 23.0
+
+
+# Each battery reading's delay is fitted over this much history, up to this.
+FIT_S = 60.0
+MAX_DELAY_S = 90
+FIT_EVERY_S = 2.0
+# A fit needs this many new readings (all batteries) and this much movement
+# of the reading and the request.
+FIT_ROWS = 5
+FIT_MOVE_W = 3 * ANSWER_W                                         # 300
+# Late from ON_S, on time again under OFF_S (hysteresis).
+ON_S = 9.5
+OFF_S = 6.5
+# ...after this many late fits in a row.
+ON_FITS = 2
+# With no fit for this long the evidence is dropped (a steady site shows none).
+FIT_GONE_S = 120.0
+# A fit counts only if the commands explain this share of the reading's swing.
+FIT_SHARE = 0.5
+# A late delay must explain this share.
+FIT_SHARE_LATE = 0.75
+# ...and a late delay only if it fits this much better than any fresh one.
+FIT_BETTER = 0.5
+# A shorter delay within this factor of the best fit is taken instead.
+FIT_NEAR = 2.0
+# ...or within this of it (W^2): two fits this good are both exact.
+FIT_FLOOR_W2 = (ANSWER_W / 2) ** 2                               # 2 500
+
+
+def per_second(cmds, first, last):
+    """The request standing at each whole second from `first` to `last`."""
+    ts = [t for t, _c in cmds]
+    return [cmds[max(0, bisect.bisect_right(ts, k) - 1)][1]
+            for k in range(first, last + 1)]
+
+
+def pooled(sets, d):
+    """Pooled variance over [(rows, c, first, sign)] of value + sign *
+    c[t - d], each set about its own mean."""
+    tot, n = 0.0, 0
+    for rows, c, first, sign in sets:
+        e = [v + sign * c[int(t) - d - first] for t, v in rows]
+        m = sum(e) / len(e)
+        tot += sum((x - m) ** 2 for x in e)
+        n += len(e)
+    return tot / n
+
+
+class UnitFit:
+    """One battery's power requests and readings, as they were seen."""
+
+    def __init__(self):
+        self.cmds = deque()      # (t, request) when it changed
+        self.reads = deque()     # (t, reading) when it changed
+
+    def add(self, now, cmd, reading):
+        if cmd is not None and (not self.cmds or self.cmds[-1][1] != cmd):
+            self.cmds.append((now, float(cmd)))
+        if reading is not None and (not self.reads
+                                    or self.reads[-1][1] != reading):
+            self.reads.append((now, float(reading)))
+        while self.reads and self.reads[0][0] < now - FIT_S:
+            self.reads.popleft()
+        while (len(self.cmds) > 1
+               and self.cmds[1][0] <= now - FIT_S - MAX_DELAY_S):
+            self.cmds.popleft()
+
+    def series(self):
+        """(rows, request each second, first second) for the fit, or None:
+        too few new readings, or a reading or a request that has not moved
+        FIT_MOVE_W. The request is averaged over the 2 s before each second:
+        the battery's own lag and the read."""
+        rows = [r for r in self.reads
+                if self.cmds and r[0] >= self.cmds[0][0] + MAX_DELAY_S]
+        vals = [v for _t, v in rows]
+        if len(rows) < 3 or max(vals) - min(vals) < FIT_MOVE_W:
+            return None
+        cs = [c for _t, c in self.cmds]
+        if max(cs) - min(cs) < FIT_MOVE_W:
+            return None
+        first, last = int(rows[0][0]) - MAX_DELAY_S - 2, int(rows[-1][0])
+        c = per_second(self.cmds, first, last)
+        avg = [c[0], c[0]] + [(c[k - 2] + c[k - 1] + c[k]) / 3
+                              for k in range(2, len(c))]
+        return rows, avg, first
+
+
+class Delay:
+    """How long after a power request the batteries' readings show it.
+
+    A reading follows its battery's own request and the house does not move
+    it, so `reading(t) - request(t - d)` is flattest at the right `d`. Each new
+    reading counts once, at the second it was seen, so the 10 s publish adds no
+    delay. One `d` for every battery (Home Assistant delays them all). None
+    with too few readings, or when the requests explain too little of them (a
+    battery that does not follow); not late unless it fits clearly better than
+    any fresh delay, and then the shortest delay that fits about as well."""
+
+    def __init__(self):
+        self.units = {}
+
+    def add(self, now, unit, cmd, reading):
+        self.units.setdefault(unit, UnitFit()).add(now, cmd, reading)
+
+    def estimate(self):
+        ser = [x for x in (u.series() for u in self.units.values()) if x]
+        if sum(len(rows) for rows, _c, _f in ser) < FIT_ROWS:
+            return None
+        sets = [(rows, c, first, -1.0) for rows, c, first in ser]
+        r_var = pooled([(rows, [0.0] * len(c), first, -1.0)
+                        for rows, c, first in ser], 0)
+        pure = [pooled(sets, d) for d in range(0, MAX_DELAY_S + 1)]
+        # The shortest that fits about as well: a swing's period fits again.
+        best = next(d for d, v in enumerate(pure)
+                    if v <= FIT_NEAR * min(pure) + FIT_FLOOR_W2)
+        if r_var <= 0 or pure[best] > (1 - FIT_SHARE) * r_var:
+            return None
+        if best < ON_S:
+            return float(best)
+        if (pure[best] > (1 - FIT_SHARE_LATE) * r_var
+                or pure[best] > FIT_BETTER * min(pure[:int(OFF_S) + 1])):
+            return None
+        return float(best)
+
+
+# The meter's delay: fitted over this much history, from this many new
+# values (a meter of about 3.5 s or faster; a slower one see M_SLOW_FIT_S);
+# late over M_LATE_S (the plant's 6 s settle and one 2 s meter tick), but only
+# when the batteries' own readings have not just shown them on time.
+M_FIT_S = 60.0
+M_ROWS = 16
+M_LATE_S = freshness.PLANT_SETTLE_S + 2.0                          # 8.0
+M_SHARE, M_BETTER = 0.4, 0.7
+# SAFETY-129: a meter slower than that (a 5 s P1) has fewer than M_ROWS values
+# in M_FIT_S, so it is fitted over M_SLOW_FIT_S instead, from the same M_ROWS
+# values. It calls a delay late only over M_SLOW_LATE_S and on ON_FITS fits in
+# a row. Alone it reads 35-90 s on healthy runs, a few times in a row; what
+# keeps those from a verdict is the slow-start and battery-fit check in
+# watch(), not the limit. A 10 s lag is not found: it cannot be told from a
+# slow battery (R2-1), nor does it swing at 5 s.
+M_SLOW_FIT_S = 120.0
+M_SLOW_LATE_S = 16.0
+# A meter fit counts only after an ask that took this long to start moving a
+# reading: on time they start within 1-2 s (the recorded house, at 1-5 s).
+SLOW_START_S = 3.0
+
+
+class MeterFit:
+    """How long after a site request the meter shows it (round 1's fit).
+
+    `meter(t) + request(t - d)` is the house alone at the right `d`. Quick,
+    as the meter reads every 1-5 s, but a house load that repeats can line
+    up with an earlier request, so it only counts while the batteries' own
+    fit is not on time (`Answer.watch`)."""
+
+    def __init__(self, fit_s=M_FIT_S, late_s=M_LATE_S):
+        self.fit_s, self.late_s = fit_s, late_s
+        self.cmds = deque()      # (t, request) when it changed
+        self.grid = deque()      # (t, meter) new values only
+
+    def slow(self):
+        """The meter is too slow for M_ROWS values in `fit_s`: the usual gap
+        between its new values (median, so a pause does not count) is over
+        fit_s / M_ROWS."""
+        ts = [t for t, _g in self.grid]
+        if len(ts) < 5:
+            return False
+        gaps = sorted(b - a for a, b in zip(ts, ts[1:]))
+        return gaps[len(gaps) // 2] * M_ROWS > self.fit_s
+
+    def add(self, now, cmd, grid):
+        if cmd is not None and (not self.cmds or self.cmds[-1][1] != cmd):
+            self.cmds.append((now, float(cmd)))
+        if grid is not None and (not self.grid or self.grid[-1][1] != grid):
+            self.grid.append((now, float(grid)))
+        while self.grid and self.grid[0][0] < now - self.fit_s:
+            self.grid.popleft()
+        while (len(self.cmds) > 1
+               and self.cmds[1][0] <= now - self.fit_s - MAX_DELAY_S):
+            self.cmds.popleft()
+
+    def estimate(self):
+        if len(self.cmds) < 2:
+            return None
+        rows = [(t, g) for t, g in self.grid
+                if t >= self.cmds[0][0] + MAX_DELAY_S]
+        cs = [c for _t, c in self.cmds]
+        if len(rows) < M_ROWS or max(cs) - min(cs) < ANSWER_W:
+            return None
+        first, last = int(rows[0][0]) - MAX_DELAY_S, int(rows[-1][0])
+        c = per_second(self.cmds, first, last)
+        g_var = pooled([(rows, [0.0] * len(c), first, 1.0)], 0)
+        pure = [pooled([(rows, c, first, 1.0)], d)
+                for d in range(0, MAX_DELAY_S + 1)]
+        best = min(range(len(pure)), key=lambda d: (pure[d], d))
+        if g_var <= 0 or pure[best] > (1 - M_SHARE) * g_var:
+            return None
+        if best <= self.late_s:
+            return float(best)
+        if pure[best] > M_BETTER * min(pure[:int(self.late_s) + 1]):
+            return None
+        return float(best)
+
+
+class Answer:
+    """The answer gate. `observe()` once per cycle, before the loop."""
+
+    def __init__(self):
+        self.ask = None          # the pending command, W (site, + = give)
+        self.from_w = None       # the battery reading when it was asked
+        self.from_grid = None    # the meter when it was asked
+        self.since = None        # when it was asked
+        self.started = False     # a reading has moved towards it
+        self.late_seen = False   # the batteries' readings run late
+        self.late_at = None      # when the late verdict started
+        self.exempt = False      # a battery that cannot follow explains it
+        self._was = None         # (command, reading, meter) at the last call
+        self.delay = Delay()
+        self.meter = MeterFit()
+        self.meter_slow = MeterFit(M_SLOW_FIT_S, M_SLOW_LATE_S)   # SAFETY-129
+        self._ms_run = 0         # late slow-meter fits in a row
+        self._batt_fit = None     # (when, d): the batteries' last fit
+        self._b_run = 0          # late battery fits in a row
+        self._start_age = None   # (when, how long the last ask took to start)
+        self._exempt_at = None   # when a battery last gave nothing under its ask
+        self._ask_exempt = False
+        self._start = None
+        self.delay_s = None      # the last fitted delay (either fit), s
+        self.shown_s = None      # the batteries' last late delay, for the page
+        self.fits = 0            # fits made (instrumentation)
+        self._fit_at = self._fit_ok = None
+
+    @property
+    def active(self) -> bool:
+        return self.ask is not None and self.late_seen
+
+    @property
+    def hold(self) -> int:
+        """+1 / -1: the loop may not push further that way. 0: no hold."""
+        if not self.active or self.exempt:
+            return 0
+        return 1 if self.ask > self.from_w else -1
+
+    @property
+    def hold_w(self):
+        """How far the loop may go while held: ANSWER_W past the reading."""
+        if not self.hold:
+            return None
+        return self._was[1] + self.hold * ANSWER_W
+
+    def peak_cap(self, standing_w):
+        """The most the peak guard may ask, or None."""
+        return max(float(standing_w), 0.0) if self.active else None
+
+    def late_s(self, now):
+        """Seconds the late verdict has stood, or None."""
+        return None if self.late_at is None else max(0.0, now - self.late_at)
+
+    def watch(self, now, units, standing_w=None, grid_w=None):
+        """Each read (1 s): `units` = [(name, request on its wire, its battery
+        reading)], the site request and the meter. Every FIT_EVERY_S, fit both
+        delays. Late on ON_FITS battery fits from ON_S, or on a meter fit
+        over M_LATE_S when an ask was slow to start (SLOW_START_S) and no
+        battery fit within FIT_S said on time: a house load that repeats can
+        fool the meter fit, not these. A meter too slow for M_ROWS values in
+        M_FIT_S (a 5 s P1) is fitted over M_SLOW_FIT_S and counts only over
+        M_SLOW_LATE_S, on ON_FITS fits in a row. On time again on a battery fit
+        under OFF_S, or a meter fit within M_LATE_S unless a battery fit said
+        late.
+        """
+        if self._start is None:
+            self._start = now
+        for name, cmd, reading in units:
+            self.delay.add(now, name, cmd, reading)
+        self.meter.add(now, standing_w, grid_w)
+        self.meter_slow.add(now, standing_w, grid_w)
+        if self._fit_at is not None and now - self._fit_at < FIT_EVERY_S:
+            return
+        self._fit_at = now
+        b = self.delay.estimate()
+        if (b or 0.0) >= ON_S and self._exempt_at is not None \
+                and now - self._exempt_at <= FIT_S:
+            b = None          # a battery that stopped following: no late fit
+        m = self.meter.estimate()
+        ms = None
+        if self.meter.slow():
+            ms = self.meter_slow.estimate()
+        if b is not None:
+            self._batt_fit = (now, b)
+            self.shown_s = b if b >= ON_S else None
+        recent = (self._batt_fit is not None
+                  and now - self._batt_fit[0] <= FIT_S)
+        b_fresh = recent and self._batt_fit[1] < ON_S
+        b_late = recent and self._batt_fit[1] >= ON_S
+
+        self._b_run = self._b_run + 1 if (b or 0.0) >= ON_S else 0
+        self._ms_run = self._ms_run + 1 if (ms or 0.0) > M_SLOW_LATE_S else 0
+        # A slow-meter fit only ever adds a late verdict: it counts as a fit
+        # when it is late on ON_FITS in a row, and never ends one, so a slow
+        # battery (R2-1) runs exactly as before.
+        if self._ms_run < ON_FITS:
+            ms = None
+        d = b if b is not None else m if m is not None else ms
+        if d is None:
+            if self._fit_ok is None or now - self._fit_ok > FIT_GONE_S:
+                self.late_seen = False            # nothing to tell it by
+            return
+        self.fits += 1
+        self.delay_s, self._fit_ok = d, now
+        warm = now - self._start >= MAX_DELAY_S + FIT_S
+        m_late = ((m or 0.0) > M_LATE_S and warm) or (
+            self._ms_run >= ON_FITS
+            and now - self._start >= MAX_DELAY_S + M_SLOW_FIT_S)
+        slow = (self._start_age is not None
+                and now - self._start_age[0] <= M_FIT_S
+                and self._start_age[1] >= SLOW_START_S)
+        if self._b_run >= ON_FITS or (m_late and slow and not b_fresh):
+            self.late_seen = True
+        elif (b is not None and b < OFF_S) or (
+                m is not None and m <= M_LATE_S and not b_late):
+            self.late_seen = False
+
+    def observe(self, now, standing_w, batt_w, *, grid_w=None, exempt=False,
+                idle_w=0.0):
+        """`standing_w` the site command on the wire, `batt_w` the site
+        battery reading (None: unknown, nothing changes), `grid_w` the meter
+        (+ = import), `idle_w` what the batteries read while giving nothing.
+        `exempt`: a battery gives nothing under its ask, or refuses."""
+        if batt_w is None or standing_w is None:
+            return
+        cmd, r = float(standing_w), float(batt_w) - float(idle_w)
+        g = None if grid_w is None else float(grid_w)
+        self.exempt = bool(exempt)
+        if self.exempt:
+            self._exempt_at = now
+        was, self._was = self._was, (cmd, r, g)
+        if self.ask is None:
+            if was is None or cmd == was[0] or abs(cmd - was[1]) < ANSWER_W:
+                return
+            self.ask, self.from_w, self.from_grid = cmd, was[1], was[2]
+            self.since, self.started = now, False
+            self._ask_exempt = False
+        self.ask = cmd
+        # A battery that gives nothing under its ask explains a slow start.
+        self._ask_exempt = self._ask_exempt or self.exempt
+        age = now - self.since
+        gap = self.ask - self.from_w
+        way = 1.0 if gap > 0 else -1.0
+        batt_to = (r - self.from_w) * way
+        grid_to = (-1e9 if g is None or self.from_grid is None
+                   else (self.from_grid - g) * way)
+        if not self.started and max(batt_to, grid_to) >= ANSWER_W:
+            self.started, self.late_at = True, None
+            self._start_age = (now, 0.0 if self._ask_exempt else age)
+        elif (not self.started and age >= SLOW_START_S
+              and not self._ask_exempt):
+            self._start_age = (now, age)
+        if (self.ask - r) * way < ANSWER_W or batt_to >= abs(gap) / 2:
+            self._clear()                         # answered
+        elif abs(gap) < ANSWER_W:
+            self._clear()                         # withdrawn
+        elif exempt:
+            self.late_at = None
+        elif (self.late_at is None and self.late_seen and not self.started
+              and age >= LATE_S):
+            self.late_at = now
+
+    def _clear(self):
+        self.ask = self.from_w = self.from_grid = self.since = None
+        self.started, self.late_at = False, None

goodwe_controller/app/main.py

@@ -59,12 +59,15 @@ from .clamp import (
     shortfall_note, write_clamp_note,
 )
 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()` and two checks
 # read `app.main.LINK_STALE_S`.
@@ -81,6 +84,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
@@ -119,6 +123,21 @@ __all__ = ["UNIT_A", "Controller", "amain", "main"]
 # on an ACCEPTED write). DECISIONS/MAIN-orchestration.md.
 WRITE_BUDGET_S = 4.0
 
+# SMA-10: the bound for handing a unit back (`driver.release()`); units are
+# released concurrently and one still running at the bound is abandoned.
+# DECISIONS/SMA-10-two-idle-states.md.
+RELEASE_BUDGET_S = WRITE_BUDGET_S
+
+# SMA-10: the one fallback write after a failed release (default 0 W). Worst
+# case RELEASE_BUDGET_S + FALLBACK_BUDGET_S = 6 s, inside 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 waits for a write already on its way to the
+# same unit; half the release budget, so a GoodWe unit's worst case is
+# HAND_BACK_WORST_S. DECISIONS/SAFETY-137-last-write-is-zero.md.
+SETTLE_BUDGET_S = RELEASE_BUDGET_S / 2
+
 # SAFETY-34: a Core-only restart wipes persistent_notification without
 # restarting this add-on, so a latched alarm is re-asserted this often.
 REANNOUNCE_S = 300.0
@@ -405,6 +424,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 (DECISIONS/SMA-10-two-idle-states.md).
+        handback.set_roster(handback.units_of(self.inverters))
+        uitext.set_idle(handback.overlay(handback.current()))
         # CFG-04: say which shape is in force. A declared `inverters:` roster
         # ignores `soc_entity`, `batt_entity`, `batt_invert` and `max_w`, which
         # the UI still shows; validate.py rule 9(e) refuses only the dangerous
@@ -925,6 +948,8 @@ class Controller:
         self.cycle_at = None
         self._arb_action = None
         self.inputs_bad_since = None
+        self.answer = lateness.Answer()           # SAFETY-128
+        self._late_logged = self._late_zero = False
         # SAFETY-03: what the loop's OWN bounds did last cycle, `(engaged,
         # note, over_rating)` (SAFETY-40 splits the last out). Site-level: the
         # integrator that wound is the one SITE integrator; a clamp bit a value
@@ -1708,7 +1733,9 @@ class Controller:
             else self.house_profile.estimator(),
             priceruns.keep_started(self.mode_runs, runs, now), now, packs,
             trust_pct=self.solar_trust_pct, floor_pct=floor_pct,
-            ceiling_pct=ceiling_pct)
+            ceiling_pct=ceiling_pct, aim_w=self.peak_aim_w(),
+            slots=priceruns.firm_slots(plan.read.points, now),
+            spread=required)
         self.day_plan = dp
         if dp is None:
             self._plan_targets = None
@@ -2350,6 +2377,36 @@ class Controller:
         return (decision if self._peak_needed
                 else dataclasses.replace(decision, within_aim=True))
 
+    def _readings_late(self, standing_w):
+        """SAFETY-128: update the answer gate; seconds the late verdict has
+        stood, or None. Readings found late are logged once per episode.
+        DECISIONS/SAFETY-128-late-readings.md."""
+        a = self.answer
+        if self._now_mono is None:
+            return None
+        a.observe(self._now_mono, standing_w, self.site_batt_w,
+                  grid_w=self.grid, exempt=bool(self.batt_refused) or any(
+                      i.refused or (
+                          i.batt is not None
+                          and abs(i.standing_w()) >= lateness.ANSWER_W
+                          and abs(i.batt - i.idle_w) < lateness.ANSWER_W)
+                      for i in self.inverters),
+                  idle_w=self._site_idle_w())
+        if a.late_seen != self._late_logged:
+            self._late_logged = a.late_seen
+            if a.late_seen:
+                self.log_event("readings late - holding",
+                               hint=messages.hint("readings_late"))
+                _LOG.warning("Home Assistant readings arrive late, so control "
+                             "holds the batteries: READINGS_LATE.")
+                _LOG.debug("READINGS_LATE: the readings show a power request "
+                           "%.0f s after it is sent.", a.delay_s)
+            else:
+                self.log_event("readings on time", hint=messages.hint(
+                    "readings_back"))
+                _LOG.info("Home Assistant readings are on time again.")
+        return a.late_s(self._now_mono)
+
     def _peak_reserve(self) -> frozenset:
         """SAFETY-87: the batteries that give nothing for the peak this
         cycle: under their reserve while the others carry a peak floor."""
@@ -2867,6 +2924,10 @@ class Controller:
                 else:
                     _LOG.debug("Meter readings are back.")
 
+        self.answer.watch(now_mono, [(inv.name, inv.standing_w(), inv.batt)
+                                     for inv in self.inverters],  # SAFETY-128
+                          sum(inv.standing_w() for inv in self.inverters),
+                          self.grid)
         if self.inputs_ok:
             if self.inputs_bad_since is not None:
                 self.log_event("inputs recovered",
@@ -2875,14 +2936,26 @@ 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.
         SAFETY-37: the POWER reading goes through the freshness gate and the
         SoC does not, deliberately: `batt` feeds SAFETY-30's per-unit sign
         check and the delivery check, which a frozen register would fool.
         The standing command is asked of the unit, not recomputed."""
-        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 SoC, then power, as this method always did.
+        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)
@@ -2993,11 +3066,11 @@ 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. `unit=None`
         is for a number that is not a share (stop button, `shutdown()`,
         `maint_abort()`: 0 W means the same to every inverter). Writes are
-        concurrent, each under its own budget (DEV-02)."""
+        concurrent, each under its own budget (DEV-02). `last`: SAFETY-137."""
         if unit is None:
             targets = [i for i in self.inverters]
             if len(targets) > 1 and is_commandable(value) and float(value) != 0.0:
@@ -3021,13 +3094,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 is on the ONLY path to the wire (cycle,
         # heartbeat, stop button, shutdown(); no optimizer, maintenance or
@@ -3083,6 +3270,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
@@ -3091,8 +3293,13 @@ class Controller:
         # abandoned write is re-issued next tick (`last_write` moves only on an
         # ACCEPTED write).
         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 "
@@ -3403,6 +3610,8 @@ class Controller:
                     "away a partly delivering unit's output.")
 
         stale_after = float(self.o.get("stale_input_s", 15))
+        late_s = self._readings_late(site_standing)          # SAFETY-128
+        self._late_zero = late_s is not None and late_s >= stale_after
         # ⚠️ `any`, NOT the sole unit's. The hold exists so a single missed
         # poll does not become a fault; a LATCHED inverter must still reach the
         # claim list below, because its `failsafe` SET is what parks it.
@@ -3420,6 +3629,13 @@ class Controller:
                 self.why_words = messages.why_waiting(bad_for)
                 self._why_waiting_since = now - timedelta(seconds=bad_for)
                 return
+        if late_s is not None and late_s < stale_after and self.inputs_ok \
+                and (self.auto or self.shadow) \
+                and not any(i.sign_wd.tripped for i in self.inverters) \
+                and not (n_units > 1 and self.site_sign_wd.tripped):
+            self.reason = f"readings late {late_s:.0f}s"     # SAFETY-128: hold
+            self.why_words = messages.why_late()
+            return
 
         claims = []
 
@@ -3491,6 +3707,9 @@ class Controller:
                                hint=messages.hint("inputs_missing"))
             claims.append(Claim.set("safety", P_SAFETY, 0.0, "inputs missing",
                                     actuator=self.actuator_of("safety")))
+        elif late_s is not None:
+            claims.append(Claim.set("safety", P_SAFETY, 0.0, "readings late",
+                                    actuator=self.actuator_of("safety")))
 
         # SAFETY-07: the quarter-peak projection guard  ONE `lo` limit at
         # P_PEAK, and peakguard.py carries the whole argument.
@@ -3550,6 +3769,8 @@ class Controller:
             self.peak_guard = self._peak_on_able(self.peak_guard,
                                                  maint_results)
             self.peak_guard = self._peak_landing(self.peak_guard)
+            self.peak_guard = peakguard.cap_floor(          # SAFETY-128
+                self.peak_guard, self.answer.peak_cap(site_standing))
         # SITE, and peakguard.claim() says so on the object it builds.
         # (DECISIONS/MAIN-orchestration-archive.md)
         peak_claim = (peakguard.claim(self.peak_guard)
@@ -3844,6 +4065,7 @@ class Controller:
                               if len(self.inverters) == 1 else None),
                 # SAFETY-97: "gives nothing" is centred on the idle reading.
                 idle_w=self._site_idle_w(),
+                hold=self.answer.hold, hold_w=self.answer.hold_w,  # SAFETY-128
                 # OPT-11: None unless battery modes are on. SAFETY-85: or while
                 # maintenance holds a battery at N>=2, so the loop does not
                 # wind past what the others can give and jump when it ends.
@@ -4591,9 +4813,8 @@ class Controller:
         # ⚠️ EACH INVERTER'S RATING FROM ITS OWN ENTITY, WHICH IS AC 4.
         # (DECISIONS/MAIN-orchestration-archive.md)
         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 exists
@@ -4614,9 +4835,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
 
@@ -4738,18 +4960,50 @@ 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 CREATED.
         (DECISIONS/MAIN-orchestration-archive.md)"""
         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:
@@ -4939,9 +5193,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
@@ -4965,9 +5220,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:
@@ -4986,6 +5240,10 @@ class Controller:
                               (inv.refusal_note(), False)):
                 if note:
                     out.append({"ok": False, "warn": not red, "text": note})
+        if self.answer.late_seen:                             # SAFETY-128
+            out.append({"ok": False, "warn": True, "id": "late",
+                        "text": messages.row_readings_late(
+                            self.answer.shown_s, self._late_zero)})
 
         # SAFETY-03. (DECISIONS/MAIN-orchestration-archive.md)
         _cw = self.clamp_wd
@@ -5426,9 +5684,9 @@ class Controller:
                 self._delivery_announced = set()
             if not self.auto:
                 # ⚠️ THE WHOLE SITE, TARGETS INCLUDED - see
-                # command_site_zero().
+                # command_site_zero(). SMA-10: Stop gives hand-back units back.
                 # (DECISIONS/MAIN-orchestration-archive.md)
-                await self.command_site_zero()
+                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
@@ -5463,6 +5721,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."""
 
@@ -5485,10 +5758,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()
@@ -5611,10 +5886,12 @@ async def amain() -> None:
         # configuration validate.py rejects - specs §1.5: a config the system
         # cannot honour is refused at startup, not survived at runtime.
         # (DECISIONS/MAIN-orchestration-archive.md)
+        # SMA-2: the flat options are the GoodWe driver's (no Controller 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 refusal

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.",
@@ -644,6 +665,31 @@ def row_controller_stalled(age_s, veto_s) -> str:
               f"stuurmodule zet de omvormer na {seconden(veto_s)} op 0 W.")
 
 
+def row_readings_late(delay_s, zero=False) -> str:
+    """SAFETY-128: readings show the power request late; `delay_s` the
+    batteries' fitted delay (their own answer included), or None when only
+    the meter has shown it; past the hold, 0 W."""
+    if delay_s is None:
+        if zero:
+            return Bi("Home Assistant readings show requests late, so every "
+                      "inverter is at 0 W.",
+                      "Metingen uit Home Assistant tonen vermogensvraag te "
+                      "laat, omvormers op 0 W.")
+        return Bi("Home Assistant readings show requests late, so control "
+                  "holds the batteries.",
+                  "Metingen uit Home Assistant tonen vermogensvraag te laat, "
+                  "dus de batterijen wachten.")
+    if zero:
+        return Bi(f"Home Assistant readings show requests {seconds(delay_s)} "
+                  f"late, so every inverter is at 0 W.",
+                  f"Metingen uit Home Assistant tonen vermogensvraag "
+                  f"{seconden(delay_s)} te laat, omvormers op 0 W.")
+    return Bi(f"Home Assistant readings show requests about {seconds(delay_s)} "
+              f"late, so control holds the batteries.",
+              f"Metingen uit Home Assistant tonen vermogensvraag zo'n "
+              f"{seconden(delay_s)} te laat, dus de batterijen wachten.")
+
+
 def row_meter_waiting(limit_s) -> str:
     """UX-10: no reading yet since the start, and still inside the limit."""
     return Bi(f"Meter: waiting for the first reading, limit "
@@ -1270,6 +1316,7 @@ WHY_WINNER = {
     "circulation": "Safety stop, inverters worked against each other",
     "stopped": "Control is stopped",
     "inputs missing": "A reading is missing",
+    "readings late": "Home Assistant readings arrive late",
     "no claim was made": "Nothing asked for power",
     # UX-10: the claim reasons main.why_key() can return for a safety or
     # failsafe winner. test_messages.py part 7 finds every one in the code.
@@ -1320,6 +1367,11 @@ def why_missing(what, limit_s) -> str:
     return f"No usable {what} reading for over {seconds(limit_s)}: 0 W"
 
 
+def why_late() -> str:
+    """SAFETY-128: readings arrive late, so control holds the batteries."""
+    return "Home Assistant readings arrive late, control holds"
+
+
 def why_waiting(missing_s) -> str:
     """A reading went missing a moment ago: control holds and waits."""
     return f"A reading is missing for {seconds(missing_s)}, control waits"
@@ -1336,7 +1388,11 @@ 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",
     "battery_stuck": "{unit}: battery power reading stopped changing",
     "battery_back": "{unit}: battery power reading is used again",
     "battery_refuses": "The battery refuses to give power, the add-on asks 0 W",
@@ -1437,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)
 
 
@@ -1488,6 +1548,10 @@ def status_samples() -> list:
         ("row_controller_stalled", RED, row_controller_stalled(45, 30)),
         ("row_meter_stale", RED, row_meter_stale(40, 10)),
         ("row_meter_waiting", AMBER, row_meter_waiting(30)),
+        ("row_readings_late", AMBER, row_readings_late(30)),
+        ("row_readings_late", AMBER, row_readings_late(30, True)),
+        ("row_readings_late", AMBER, row_readings_late(None)),
+        ("row_readings_late", AMBER, row_readings_late(None, True)),
         ("row_meter", GREEN, row_meter(-350, 2, 3, 340)),
         ("row_meter_frozen", RED, row_meter_frozen()),
         *(("row_freeze_off", AMBER, row_freeze_off(k)) for k in FREEZE_OFF),

goodwe_controller/app/peakguard.py

@@ -446,6 +446,19 @@ def shaping_suspended(decision, peak_forecast_w, cap_w) -> bool:
     return decision is not None and decision.engaged
 
 
+def cap_floor(decision, cap_w):
+    """SAFETY-128: the floor no higher than `cap_w` (the command standing while
+    the readings have not answered it), or `decision` unchanged when None.
+    DECISIONS/SAFETY-128-late-readings.md."""
+    if (cap_w is None or decision is None or decision.floor_w is None
+            or decision.floor_w <= cap_w):
+        return decision
+    return replace(
+        decision, floor_w=float(cap_w), saturated=False,
+        reason=decision.reason + f", held at {float(cap_w):+.0f} W until the "
+        "readings answer the last power request")
+
+
 def claim(decision: PeakDecision) -> Claim | None:
     """The decision as one arbiter claim (`lo` only, `hi` stays `+inf`), or None."""
     if not decision.engaged or decision.floor_w is None:

goodwe_controller/app/selfconfig.py

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

goodwe_controller/app/solarfc.py

@@ -34,11 +34,14 @@ evening and api.forecast.solar is sometimes out for a poll. A kept reading
 stands in for its sensor (a) for up to `FRESH_S` when the sensor is missing
 or unavailable, as a sensor that merely stopped updating is already
 tolerated, and (b) from the end of the bell (`BELL_TO_H`) until local
-midnight, if it is at most `KEPT_MAX_AGE_S` old when used.
+midnight, if it is at most `KEPT_MAX_AGE_S` old when used. A TOMORROW
+reading needs no evening for (b), at any hour (OPT-38: one Forecast.Solar
+sensor stopped at 14:54 and tomorrow vanished from 17:54 to 20:00).
 Never beyond that, and never for another day: a today figure is only today's,
 a tomorrow figure is tomorrow's until midnight and then, until the bell starts
 (`BELL_FROM_H`, 06:00 local), today's (OPT-34: at most `KEPT_MORNING_MAX_AGE_S`
-old, and it must have been usable at midnight).
+old, it must have been usable at midnight, and no fresh today reading may
+have replaced it since).
 DECISIONS/OPT-27-last-fresh-forecast-for-the-night.md,
 DECISIONS/OPT-34-the-night-charge-keeps-its-level.md.
 
@@ -329,8 +332,9 @@ class Hold:
 
         The kept reading must be for `day0`, and either (a) at most `FRESH_S`
         old, or (b) used at night: from the end of the bell to local midnight,
-        at most `KEPT_MAX_AGE_S` old. `kept` is true when (b) was needed,
-        or `as_today` (OPT-34: a tomorrow figure standing in for today, after
+        at most `KEPT_MAX_AGE_S` old; a "tomorrow" reading at any hour
+        (OPT-38; `read_sum()` flags it kept only from the end of the bell).
+        `kept` is true when (b) was needed, or `as_today` (OPT-34: a tomorrow figure standing in for today, after
         midnight and until the bell starts, `KEPT_MORNING_MAX_AGE_S`)."""
         rec = self._r.get((kind, entity, day0))
         if rec is None:
@@ -346,11 +350,18 @@ class Hold:
             return (day, at, True) if ok else None
         if age > KEPT_MAX_AGE_S:
             return None
-        # A tomorrow sensor is asked with day0 = the next midnight, which is
-        # in the future all day, so the evening ends at that day's end.
+        if kind == "tomorrow":
+            return day, at, True    # OPT-38: no evening needed
         evening = _local_hour(now) >= BELL_TO_H and now < _bounds(day0)[1]
         return (day, at, True) if evening else None
 
+    def replaced(self, entities, day0) -> bool:
+        """OPT-34 F-1: every today sensor has given a fresh reading for
+        `day0` (so after midnight, newer than any evening figure). The
+        evening's tomorrow figure is then no stand-in for today any more,
+        also once those readings are stale."""
+        return all(("today", e, day0) in self._r for e in entities)
+
     def dump(self) -> dict:
         return {"v": 1, "r": [
             [kind, ent, d.day0.isoformat(), d.kwh, at.isoformat(),
@@ -492,7 +503,8 @@ def read_sum(today, tomorrow, now: datetime, hold: Hold | None = None):
     where `Hold.stand_in()` allows it (the forecast then has `kept` set when
     that took a reading older than `FRESH_S`). After midnight, if the today
     side is still short, the tomorrow figures read before midnight are
-    today's, until the bell starts (`stand_in()`). Without a `hold` this is
+    today's, until the bell starts (`stand_in()`), unless fresh today
+    readings have replaced them (`Hold.replaced()`). Without a `hold` this is
     exactly as before."""
     if not today:
         return None, NOT_SET, ()
@@ -537,7 +549,8 @@ def read_sum(today, tomorrow, now: datetime, hold: Hold | None = None):
         # day; all of them are needed, as for any sum.
         subs = [hold.stand_in("tomorrow", ent, day0, now, as_today=True)
                 for ent, _st in tomorrow]
-        if all(r is not None for r in subs):
+        if all(r is not None for r in subs) and not hold.replaced(
+                [ent for ent, _st in today], day0):
             days = [r[0] for r in subs]
             ats = [r[1] for r in subs]
             gaps = [g for g in gaps if g[0] != "today"]
@@ -546,6 +559,9 @@ def read_sum(today, tomorrow, now: datetime, hold: Hold | None = None):
     if why0 is not None:
         return None, why0, gaps
     tday = _sum(tdays) if tomorrow and len(tdays) == len(tomorrow) else None
-    kept = kept or (tday is not None and tkept)
+    # OPT-38: before the bell ends a kept tomorrow reading is no "kept for
+    # tonight": it must not replace today's line or the log.
+    kept = kept or (tday is not None and tkept
+                    and _local_hour(now) >= BELL_TO_H)
     # The oldest member's report is when the sum was last fully known.
     return Forecast(_sum(days), tday, min(ats), kept), None, gaps

goodwe_controller/app/ui_text/en.json

@@ -357,10 +357,11 @@
       "no_prices": "No prices, so it follows the house.",
       "no_spread": "Prices differ too little, so it follows the house.",
       "sun_fills": "Enough sun is forecast, so it does not charge from the grid.",
+      "others_cover": "Other batteries cover the expensive hours. It does not charge from the grid.",
       "tops_up": "It charges from the grid to {pct}. That covers {kwh} of expensive hours the sun will not.",
       "no_forecast": "No solar forecast, so it charges from solar only."
     },
-    "solar": {"line": "Solar still to come today: {kwh}, we count on {count}. Updated {at}.", "uncovered": "The expensive hours need more than Leftover solar can hold. The rest, {kwh}, comes from another battery or the grid.", "kept": "Solar forecast read {at}, kept for tonight.", "no_forecast": "No solar forecast now, so Leftover solar charges from solar only.", "latched": "No solar forecast now. The charge under way goes on to its planned level.", "learning": "The add-on is learning your house's use. The plan starts after one full day."},
+    "solar": {"line": "Solar still to come today: {kwh}, we count on {count}. Updated {at}.", "uncovered": "The expensive hours need more than Leftover solar can hold. The rest, {kwh}, comes from the grid.", "kept": "Solar forecast read {at}, kept for tonight.", "no_forecast": "No solar forecast now, so Leftover solar charges from solar only.", "latched": "No solar forecast now. The charge under way goes on to its planned level.", "learning": "The add-on is learning your house's use. The plan starts after one full day."},
     "today": {
       "not_planned": "Not planned yet.",
       "no_runs": "No cheap or expensive hours left today.",
@@ -476,6 +477,7 @@
       "stopped": "Control is stopped. Every inverter is at 0 W."
     },
     "attn": {
+      "late": "Readings arrive late",
       "home_assistant": "Home Assistant is not answering",
       "controller": "Control is not running",
       "meter": "Meter reading",

goodwe_controller/app/ui_text/nl.json

@@ -297,10 +297,11 @@
       "no_prices": "Geen prijzen, dus ze volgt het huis.",
       "no_spread": "De prijzen verschillen te weinig, dus ze volgt het huis.",
       "sun_fills": "Er is genoeg zon voorspeld, dus ze laadt niet van het net.",
+      "others_cover": "Andere batterijen dekken de dure uren. Ze laadt niet van het net.",
       "tops_up": "Ze laadt van het net tot {pct}. Dat dekt {kwh} dat de zon in dure uren niet levert.",
       "no_forecast": "Geen zonneprognose, dus ze laadt alleen met zonne-energie."
     },
-    "solar": {"line": "Nog verwachte zon vandaag: {kwh}, we rekenen op {count}. Bijgewerkt om {at}.", "uncovered": "De dure uren vragen meer dan Overschot zonne-energie kan opslaan. De rest, {kwh}, komt van een andere batterij of het net.", "kept": "Zonneprognose van {at}, bewaard voor vannacht.", "no_forecast": "Nu geen zonneprognose, dus Overschot zonne-energie laadt alleen met zon.", "latched": "Nu geen zonneprognose. De lopende lading gaat door tot het geplande niveau.", "learning": "De add-on leert eerst het verbruik van uw huis. Het plan start na één volle dag."},
+    "solar": {"line": "Nog verwachte zon vandaag: {kwh}, we rekenen op {count}. Bijgewerkt om {at}.", "uncovered": "De dure uren vragen meer dan Overschot zonne-energie kan opslaan. De rest, {kwh}, komt van het net.", "kept": "Zonneprognose van {at}, bewaard voor vannacht.", "no_forecast": "Nu geen zonneprognose, dus Overschot zonne-energie laadt alleen met zon.", "latched": "Nu geen zonneprognose. De lopende lading gaat door tot het geplande niveau.", "learning": "De add-on leert eerst het verbruik van uw huis. Het plan start na één volle dag."},
     "today": {
       "not_planned": "Nog niet gepland.",
       "no_runs": "Vandaag geen goedkope of dure uren meer.",
@@ -397,6 +398,7 @@
     "peak_not_covered": "Geen piekgrens: deze meter meet geen kwartieren.",
     "control": {"shadow": "Schaduwmodus: elke omvormer blijft op 0 W.", "running": "De regeling is actief.", "stopped": "De regeling is gestopt. Elke omvormer staat op 0 W."},
     "attn": {
+      "late": "Metingen komen te laat",
       "home_assistant": "Home Assistant antwoordt niet",
       "controller": "Regeling is niet actief",
       "meter": "Meting van de meter",

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