Merge pull request 'SAFETY-95: a maintenance drain that ends hands back smoothly' (#177) from safety-95-drain-end-reseed into release/1.0

goodwe_controller/DECISIONS/SAFETY-95-drain-end.md

@@ -0,0 +1,69 @@
+# SAFETY-95 - a maintenance drain that ends hands back smoothly
+
+**Status: implemented 2026-10-06, for review.** Tests: `test_safety95.py`.
+
+**The defect** (SAFETY-93 review, O1). At N=2, when A's drain ended (A
+switched Off, say) with the house at 1200 W, B went 0 -> +1600 W in one
+cycle and exported. It is the mirror of SAFETY-93 on the drain end.
+
+**Cause.** During a drain the site target carries A's drain floor
+(SAFETY-78, `drain`, 2000 W) above what the house needs. At the end A's
+command drops (pin, Off, abort, stale link, refusal), but the loop slews
+from last cycle's target (2000 -> 1600) and only B is left to place it.
+Under a solar surplus B gave 1600 W into the export.
+
+**The rule** (`main.py`, `_reseed_drain_end()`, `_drain_grid_w()`, N>=2
+only). On the cycle a drain ends:
+
+1. The start point is what the others stood at plus A's new command (its
+   pin, else 0 W; a dropped battery is sent 0 W), as SAFETY-93 does.
+2. If the house needs more, the start is raised to the house's own point
+   (live batteries' readings + grid - target, as `_reseed_run_edge()`),
+   but never above last cycle's start. So the others cover the house at
+   once and never take over the drain's export.
+3. While A settles (SAFETY-96's window: within the deadband, at most
+   `PLANT_SETTLE_DOWN_S`), the loop's error is the house's point minus its
+   start, not the meter. The meter still shows A's tail and the others'
+   climb; chasing it sent B down and then up again.
+4. Then the loop moves one slew step a cycle.
+
+A drain ends when its battery leaves DRAIN, is not live, its claims
+contradict, or it will not give (SAFETY-73's limit outranks the pin
+without a contradiction, so the refusal is read from the battery; +1 only,
+as `_held_back()`). Today a contradiction on a drain only arises through
+a refusal; the clause stays because a contradiction sends A 0 W, which is
+an end (tested directly). The end is kept
+until the loop runs (a missed poll). A dropped draining battery is this
+rule's, not `_reseed_drop()`'s. Order in one cycle: drop/rejoin
+(SAFETY-96), charge end (SAFETY-93), drain end (this) last; each moves the
+start by its own battery, and the drain end's raise reads the start the
+others left.
+
+**Why not SAFETY-93's rule alone** (step 1, then slew). Measured on the
+same plant: at 1200 W the grid imports 849 W and B goes to -465 W
+(charging) while the loop chases A's tail; at N=3 and 4000 W it imports
+784 W where base imports 42 W, as the others give up A's share of the house.
+The house point keeps the grid at its aim; the cap at last cycle's start
+keeps it from exporting.
+
+**Before -> after** (`test_safety95.py`, 2500 W batteries, 10 s lag, 2 s
+meter; base = the same tree with the change taken out):
+
+| case | before | after |
+|---|---|---|
+| drain done / Off / timeout, N=2, 1200 W | B 0 -> 1600, +390 W commanded export, grid then +559 W | B 0 -> 1210, grid within 40 W of its aim |
+| the same, N=3, 1200 W | B, C 0 -> 800, grid +695 W | 0 -> 605, grid +39 W |
+| the same, solar surplus | B gives 1600 W into the surplus | B never gives; site moves 400 W a cycle |
+| refusal, N=3, 1200 W | B, C step 1020 W, +2000 W commanded export | step 20 W, +70 W |
+| N=3, 4000 W | others take A's share at once | the same (grid +1 W, base +42 W) |
+| abort (site zeroed by the operator) | climbs from 0, grid +669 W | climbs from 0, grid +16 W |
+| stale link (board watchdog 25 s) | | as base, but N=3 4000 W: grid +676 W (base +786) |
+| N=1, every drain end | | byte-identical |
+
+`tools/opt11_identity.py` n2_maint moved on purpose (A's drain refused at
+1655 s; B stays at the house's 1420 W instead of 2500 W). Re-pinned in
+`test_modes.py`. The other 45 digests are byte-identical.
+
+**Not changed.** The peak guard's floor still binds (arbiter, after the
+loop). Held batteries still give last; the idle clamp and the freeze test
+are as they were. A charge end (SAFETY-93) and a trade end are unchanged.

goodwe_controller/DOCS.md

@@ -1702,6 +1702,9 @@ When a charge ends, the other batteries take over one slew step a cycle,
 and none starts charging from the grid (SAFETY-93).
 When a battery's link drops, the others take over one slew step a cycle,
 and control does not stall on the dropped battery's power (SAFETY-96).
+When a drain ends (done, Off, timeout, abort, lost link, refused), the
+other batteries cover the house at once but do not export its drain; then
+they move one slew step a cycle (SAFETY-95).
 
 #### Capacity-tariff exposure of the grid-following loop
 

goodwe_controller/app/main.py

@@ -1025,6 +1025,12 @@ class Controller:
         # cycle, and those whose charge ended since the loop last ran.
         self._charging_was = frozenset()
         self._charge_ended = frozenset()
+        # SAFETY-95: the same for a maintenance drain.
+        self._draining_was = frozenset()
+        self._drain_ended = frozenset()
+        # SAFETY-95: batteries settling after their drain ended; meanwhile
+        # the loop's error counts every battery at its command.
+        self._drain_tail = set()
         # SAFETY-96: last cycle's commandable batteries, this cycle's start
         # point per battery where a reseed moved it, and batteries settling
         # after a step the loop did not make ({name: since, monotonic}).
@@ -2440,18 +2446,24 @@ class Controller:
 
     def _charge_track(self, claims, maint_results) -> None:
         """SAFETY-93, N>=2: a charge ends when its battery leaves CHARGE or can
-        no longer be placed (stale link, refusal). Kept until the loop runs."""
+        no longer be placed (stale link, refusal). Kept until the loop runs.
+        SAFETY-95: a drain likewise; its refusal (SAFETY-73) outranks the pin
+        without a contradiction, so it is read from the battery (+1, will not
+        give; a battery that will not take still drains)."""
         if len(self.inverters) < 2:
             return
         live = set(self.live_units())
-        now = set()
+        now = {CHARGE: set(), DRAIN: set()}
         for inv, res in zip(self.inverters, maint_results):
             u = self.actuator_of("maintenance", inv)
-            if res.phase == CHARGE and u in live and not pre_resolve(
+            if res.phase in now and u in live and not pre_resolve(
-                    [c for c in claims if c.actuator == u]).contradiction:
+                    [c for c in claims if c.actuator == u]).contradiction \
-                now.add(inv.name)
+                    and not (res.phase == DRAIN and (inv.refused or 0) > 0):
-        self._charge_ended |= self._charging_was - now
+                now[res.phase].add(inv.name)
-        self._charging_was = frozenset(now)
+        self._charge_ended |= self._charging_was - now[CHARGE]
+        self._charging_was = frozenset(now[CHARGE])
+        self._drain_ended |= self._draining_was - now[DRAIN]
+        self._draining_was = frozenset(now[DRAIN])
 
     def _reseed_charge_end(self, maint_results) -> None:
         """SAFETY-93, N>=2: when a battery's maintenance charge ends, the loop
@@ -2476,6 +2488,53 @@ class Controller:
             self._settle[inv.name] = self._now_mono
         self.i_w = self.target
 
+    def _reseed_drain_end(self, maint_results, live) -> None:
+        """SAFETY-95, N>=2: when a battery's maintenance drain ends, its
+        drain leaves the loop's start point, as a charge does (SAFETY-93): the
+        start is what the others stood at plus its new command (its pin, else
+        0 W). Where the house needs more, the start is the house's own point
+        (the live batteries' readings + grid - target), never above last
+        cycle's start. The loop then moves one slew step a cycle. Else the
+        drain's floor stays in the target and lands on the others: B 0 ->
+        +1600 W in one cycle, exporting. DECISIONS/SAFETY-95-drain-end.md.
+        """
+        ended, self._drain_ended = self._drain_ended, frozenset()
+        if len(self.inverters) < 2 or not ended:
+            return
+        hi = self.target
+        for inv, res in zip(self.inverters, maint_results):
+            if inv.name not in ended:
+                continue
+            now_w = (float(res.setpoint_w or 0.0)
+                     if res.owns_setpoint and res.phase != DRAIN
+                     else min(float(inv.target), 0.0))
+            if inv.name not in live:
+                now_w = 0.0     # SAFETY-96: out of the split, it is sent 0 W
+            else:
+                self._drain_tail.add(inv.name)
+            self.target += now_w - float(inv.target)
+            self._loop_start[inv.name] = now_w
+            self._settle[inv.name] = self._now_mono
+        lo, hi = sorted((self.target, hi))
+        # A dropped battery's power is the meter's, like the house's.
+        here = float(self.grid) - self.tuning.target_grid_w + sum(
+            float(inv.batt) for inv in self.inverters if inv.name in live)
+        self.target = self.i_w = min(hi, max(lo, here))
+
+    def _drain_grid_w(self, live) -> float:
+        """SAFETY-95: the grid the loop regulates. While a battery settles
+        from its ended drain (`_held_out_w`: until within the deadband, at
+        most PLANT_SETTLE_DOWN_S), the meter still shows the drain's tail and
+        the others' climb, so the loop would chase them (B down, then up
+        again). Meanwhile the batteries count at the site command: the error
+        is the house's own point minus the loop's start."""
+        grid = float(self.grid)
+        self._drain_tail &= set(self._settle) & set(live)
+        if not self._drain_tail:
+            return grid
+        return grid - float(self.target) + sum(
+            float(inv.batt) for inv in self.inverters if inv.name in live)
+
     def _reseed_drop(self, live) -> None:
         """SAFETY-96, N>=2: a battery that drops out of the split (stale
         link) leaves the loop's start point, so the others take over at one
@@ -2488,8 +2547,8 @@ class Controller:
         if len(self.inverters) < 2 or was is None or was == live:
             return
         for inv in self.inverters:
-            if inv.name in self._charge_ended:
+            if inv.name in self._charge_ended | self._drain_ended:
-                continue        # SAFETY-93's edge: `_reseed_charge_end()`
+                continue        # SAFETY-93/95's edge: `_reseed_*_end()`
             if inv.name in was and inv.name not in live:
                 now_w = 0.0
             elif inv.name in live and inv.name not in was:
@@ -4835,6 +4894,9 @@ class Controller:
             self._reseed_drop(_live96)
             # SAFETY-93: a charge that ended leaves the loop's start point.
             self._reseed_charge_end(maint_results)
+            # SAFETY-95: then a drain that ended, last (it reads the start
+            # the others left).
+            self._reseed_drain_end(maint_results, _live96)
             probe = self._refusal_probe()
             if len(self.inverters) > 1:
                 probe = any(_probing.values())
@@ -4864,7 +4926,9 @@ class Controller:
             loop_prev = self.target
             decision = compute(
                 prev_w=loop_prev,
-                grid_w=self.grid,
+                # SAFETY-95: a drain's tail is not the loop's to chase.
+                grid_w=(self._drain_grid_w(_live96) if self._drain_tail
+                        else self.grid),
                 actual_w=site_batt,
                 tuning=self.tuning,
                 sat_count=self.sat_count,

goodwe_controller/test_modes.py

@@ -129,6 +129,11 @@ finally:
 # cd4d9cb5b7e29ff4ab4fe65dab5106683d8a599f09323fbdee4018ad9b28fcd1.
 SAFETY78_N2_MAINT = \
     "1afcd46b1b987868b110d2b0f4a7555848c9aa4728a836d4a2ce2a2107268497"
+# SAFETY-95 changes it again ON PURPOSE: A's drain ends at t=1655 (its
+# refusal), and B now stays at the house's 1420 W instead of jumping to
+# 2500 W (1100 W export) and slewing back. Rows differ only at 1655-2105 s.
+SAFETY95_N2_MAINT = \
+    "949e7c093dcf065257417de3f2f908e4d57311790abc9fba0a09aa4b5d64c0d2"
 # SAFETY-91 changes n2 and n2_off ON PURPOSE: from t=1600 battery A is at its
 # 20 % reserve, so B carries the house first and A gives last. With
 # `_held_back()` reverted to SAFETY-87's (held only for the peak) both are
@@ -165,8 +170,8 @@ check("...n2 and n2_off are the SAFETY-91 digest (A at its reserve gives "
       and _n2_s87 == SAFETY94_N2_S87)
 check("...and n2_maint is the SAFETY-78 digest: the N=2 drain that a5dddad "
       "refused on 392 of 540 cycles now runs (test_safety78.py), bounded by "
-      "SAFETY-85",
+      "SAFETY-85, its end handed back by SAFETY-95",
-      _now["n2_maint"] == SAFETY78_N2_MAINT != A5DDDAD["n2_maint"])
+      _now["n2_maint"] == SAFETY95_N2_MAINT != A5DDDAD["n2_maint"])
 check("...and no cycle in those runs filed a `mode` claim, and only "
       "n2_maint's (SAFETY-85: maintenance at N=2) handed `compute()` a bound",
       _claims and not any(c.source == "mode" for cl in _claims for c in cl)

goodwe_controller/test_safety95.py

@@ -0,0 +1,578 @@
+"""Runnable check for SAFETY-95: a maintenance drain that ends hands back
+smoothly at N>=2. `python3 test_safety95.py`
+
+At N=2, when battery A's drain ended (A switched Off, say) with the house at
+1200 W, the drain's floor stayed in the loop's site target and landed on B:
+B went 0 -> +1600 W in one cycle and exported. Now the loop starts from what
+the others stood at plus A's new command, raised to the house's own point
+when the house needs more (never above last cycle's start), and counts every
+battery at its command while A settles. Drives the REAL Controller on a
+lagging battery (10 s, 2 s meter, as on the rig), at N=2 and N=3.
+DECISIONS/SAFETY-95-drain-end.md.
+
+1. Every drain end (complete -> charge, Off, timeout, operator abort, stale
+   link, refusal) at houses of 1200 W and 4000 W and under a solar surplus.
+2. The peak guard still binds across a drain end.
+3. A drain end in the same cycle as another battery's drop; a missed poll.
+4. N=1 is untouched; the rule at its boundaries.
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import asyncio
+import functools
+import hashlib
+import logging
+import math
+import os
+import sys
+import tempfile
+from datetime import timedelta
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+sys.path.insert(0, os.path.join(HERE, "tools"))
+logging.disable(logging.WARNING)
+
+import opt11_identity as rig                                   # noqa: E402
+import safety79_identity as s79                                # noqa: E402
+from app import main as app_main, packmode                     # noqa: E402
+from app.arbiter import Claim, P_MAINTENANCE, P_SAFETY         # noqa: E402
+from app.main import Controller                                # noqa: E402
+from app.maintenance import (CHARGE, DRAIN, DRAIN_TIMEOUT_MIN,  # noqa: E402
+                             HOLD, IDLE, MaintResult)
+from app.store import unit_key                                 # noqa: E402
+
+fails = []
+SLEW = 400.0            # the rig's slew_w
+TOL = SLEW / 4          # the loop's own overshoot on a lagging battery
+AIM = -10.0             # the rig's target_grid_w
+T = 300.0               # the drain ends here
+M = {"maintenance_enabled": True, "maintenance_start_hour": 11}
+A = "inverter-a"
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+# None on a tree without the fix, where this file prints the base numbers
+# and runs red.
+fixed = getattr(Controller, "_reseed_drain_end", None)
+real_grid = getattr(Controller, "_drain_grid_w", None)
+real_drop = Controller._reseed_drop
+
+
+def _base_drop(self, live):
+    """SAFETY-96's drop reseed as it was: a draining battery is not
+    skipped."""
+    keep, self._drain_ended = self._drain_ended, frozenset()
+    try:
+        real_drop(self, live)
+    finally:
+        self._drain_ended = keep
+
+
+def without(fn):
+    """Run `fn` with SAFETY-95 taken out (the base behaviour)."""
+    if fixed is None:
+        return fn()
+    Controller._reseed_drain_end = lambda self, r, live: setattr(
+        self, "_drain_ended", frozenset())
+    Controller._drain_grid_w = lambda self, live: float(self.grid)
+    Controller._reseed_drop = _base_drop
+    try:
+        return fn()
+    finally:
+        Controller._reseed_drain_end = fixed
+        Controller._drain_grid_w = real_grid
+        Controller._reseed_drop = real_drop
+
+
+def options(n, extra):
+    o = rig._options(n, extra)
+    if n > 1:
+        o["inverters"] = [
+            {"name": f"inverter-{u}", "setpoint_entity": f"number.sp_{u}",
+             "soc_entity": f"sensor.soc_{u}", "batt_entity": f"sensor.batt_{u}",
+             "max_w": 2500, "bms_soc_floor": 10} for u in "abc"[:n]]
+    return o
+
+
+def run(n, house, *, store, extra=None, soc=None, reject=None, dead=None,
+        before=None, dur=T + 120.0, lag=10.0, p1=False, back=None):
+    """Rows of one run at N=`n` on plantsim's battery (800 W/s ramp, `lag`
+    s first-order lag) at a 2 s meter tick. Writes to the units in `reject`
+    (name -> from s) are refused, until `back` s; `dead` (u, from s[, to
+    s]): u gives nothing from then on (until `to`). Each row: (t, the
+    add-on's commands, site target, grid, phases, loop reason, peak floor).
+    opt11_identity's cycle, with a third battery."""
+    units = tuple("abc"[:n])
+    soc = soc or (lambda u, t: 50.0)
+
+    def refused(u, t):
+        return bool(reject) and u in reject and t >= reject[u] and (
+            back is None or t < back)
+
+    class Hass(rig._Hass):
+        def power(self, u):
+            st = self.__dict__.setdefault("_plant", {})
+            a, tgt, tl = st.get(u, (0.0, 0.0, self.t))
+            for k in range(int(round((self.t - tl) / 0.1))):
+                now = tl + (k + 1) * 0.1
+                c = self.cmd[u]
+                if dead is not None and u == dead[0] and dead[1] <= now < (
+                        dead[2] if len(dead) > 2 else math.inf):
+                    c = 0.0
+                d = c - tgt
+                tgt += math.copysign(min(abs(d), 80.0), d)
+                a += (tgt - a) * (1 - math.exp(-0.1 / lag))
+            st[u] = (a, tgt, self.t)
+            return float(round(a))
+
+        async def set_number(self, e, v):
+            if refused(e[-1], self.t):
+                return False
+            self.cmd[e[-1]] = float(v)
+            return True
+
+    opts = options(n, dict(M, **(s79.P1 if p1 else {}), **(extra or {})))
+    db = os.path.join(tempfile.mkdtemp(prefix="s95-"), "tariff.db")
+
+    async def go():
+        hass = Hass(units, soc, lambda u, t: False,
+                    lambda t: house(t) if callable(house) else house)
+        c = Controller(opts, hass, store, rig._Pub(),
+                       stop=asyncio.Event(), peak_db_path=db)
+        for i in c.inverters:
+            i.dev_min, i.dev_max = -5000.0, 5000.0
+        c.auto = True
+        real_dt, clock = app_main.datetime, [0.0]
+
+        class _DT(real_dt):
+            @classmethod
+            def now(cls, tz=None):
+                return rig.BASE + timedelta(seconds=clock[0])
+        app_main.datetime = _DT
+        rows = []
+        try:
+            t = 0.0
+            while t < dur:
+                hass.t = clock[0] = t
+                for inv in c.inverters:
+                    u = inv.name[-1]
+                    if refused(u, t) and inv.rejected_since is None:
+                        inv.rejected_since = rig.BASE + timedelta(seconds=t)
+                if before is not None:
+                    r = before(c, t)
+                    if asyncio.iscoroutine(r):
+                        await r
+                if p1:
+                    s79._feed(c, t)
+                await c.read_inputs(now_mono=t)
+                await c.cycle()
+                g = c.peak_guard
+                rows.append((t, tuple(i.target for i in c.inverters),
+                             c.target,
+                             hass.grid(), tuple(m.phase for m in c.maints),
+                             c.loop_reason,
+                             None if g is None or not g.engaged
+                             else g.floor_w))
+                t += 2.0
+        finally:
+            app_main.datetime = real_dt
+        return rows, c
+    return asyncio.run(go())
+
+
+def store(n, **a):
+    """A due for its monthly cycle (it drains from the first cycle); the
+    others finished theirs, so none starts here."""
+    s = rig._Store()
+    if n == 1:
+        for k, v in a.items():
+            s.data[k] = v
+        return s
+    for u in "bc"[:n - 1]:
+        s.data[unit_key(f"inverter-{u}", "last_completed")] = \
+            rig.BASE.isoformat()
+    for k, v in a.items():
+        s.data[unit_key(A, k)] = v
+    return s
+
+
+def timed_store(n):
+    """A's drain passes DRAIN_TIMEOUT_MIN at T; its last start was at 0 s,
+    so the abort's cooldown keeps it from starting again."""
+    return store(n, phase=DRAIN, phase_started=(
+        rig.BASE + timedelta(seconds=T - DRAIN_TIMEOUT_MIN * 60 - 1)
+    ).isoformat(), last_start_attempt=rig.BASE.isoformat())
+
+
+def off(c, t):
+    if t >= T:
+        c.store.data[unit_key(A, packmode.USE_KEY)] = False
+
+
+async def abort(c, t):
+    if t == T:
+        await c.handle_action("maint_abort")
+
+
+ENDS = {
+    "complete": lambda n, h, **k: run(
+        n, h, store=store(n), soc=lambda u, t:
+        5.0 if (u == "a" and t >= T) else 50.0, **k),
+    "Off": lambda n, h, **k: run(n, h, store=store(n), before=off,
+                                 extra={"battery_modes_enabled": True}, **k),
+    "timeout": lambda n, h, **k: run(n, h, store=timed_store(n), **k),
+    "abort": lambda n, h, **k: run(n, h, store=store(n), before=abort, **k),
+    # A's link goes stale; its board's watchdog sends it to 0 W after 25 s.
+    "stale link": lambda n, h, **k: run(n, h, store=store(n),
+                                        reject={"a": T}, dead=("a", T + 25.0),
+                                        dur=T + 200.0, **k),
+    # A gives nothing from T; SAFETY-73 learns it ~55 s later.
+    "refusal": lambda n, h, **k: run(n, h, store=store(n), dead=("a", T),
+                                     dur=T + 300.0, **k),
+}
+HOUSES = (1200.0, 4000.0, -1500.0)
+
+
+def end_at(rows):
+    """The first cycle A is not draining, or its command left the drain;
+    None if its drain never ended."""
+    return next((r[0] for r in rows if r[0] >= T and (
+        r[4][0] != DRAIN or r[1][0] < 2000.0)), None)
+
+
+def measure(rows, n, house):
+    """From the end on: the site target's largest step after the end
+    cycle (the slew's subject); the commanded export in the 20 s after it (the site command over
+    what the house needs, or over what the others stood at); the largest
+    grid import; the export beyond what the site exported before the end;
+    the largest grid step; the others' largest command step."""
+    te = end_at(rows)
+    if te is None:      # no end: every measure is NaN, so every check FAILs
+        return dict(end=math.nan, site_step=math.nan, cmd_export=math.nan,
+                    imp=math.nan, exp=math.nan, gstep=math.nan,
+                    ostep=math.nan, rows=rows)
+    pre = next((r for r in reversed(rows) if r[0] < te), rows[0])
+    seg = [r for r in rows if r[0] >= te - 2.0]
+    aft = [r for r in rows if te <= r[0] <= te + 120.0]
+    others0 = sum(pre[1][1:])
+    site = [r[2] for r in seg]
+    return dict(
+        end=te,
+        site_step=max(abs(b - a) for a, b in zip(site[1:], site[2:])),
+        cmd_export=max(sum(r[1]) - max(house - AIM, others0) for r in aft
+                       if r[0] <= te + 20.0),
+        imp=max(r[3] for r in aft),
+        exp=min(min(pre[3], AIM) - r[3] for r in aft),
+        gstep=max(abs(b[3] - a[3]) for a, b in zip(seg, seg[1:])),
+        ostep=max(abs(b[1][i] - a[1][i]) for a, b in zip(seg, seg[1:])
+                  for i in range(1, n)),
+        rows=rows)
+
+
+TABLE = []
+
+
+def case(kind, n, house):
+    now = measure(ENDS[kind](n, house)[0], n, house)
+    old = measure(without(lambda: ENDS[kind](n, house))[0], n, house)
+    TABLE.append((kind, n, house, now, old))
+    return now, old
+
+
+# ============================================================================ #
+print("1. every drain end, N=2 and N=3, houses 1200 W, 4000 W and -1500 W")
+for kind in ENDS:
+    for n in (2, 3):
+        for house in HOUSES:
+            now, old = case(kind, n, house)
+            tag = f"{kind}, N={n}, house {house:.0f} W"
+            # Abort zeroes the site (operator); a stale link's start is
+            # SAFETY-96's drop. Those are graded against base.
+            own = kind not in ("abort", "stale link")
+            check(f"{'⭐' if own else '  '} {tag}: no battery gives into an "
+                  f"export after the end (commanded export "
+                  f"{now['cmd_export']:.0f} W, base {old['cmd_export']:.0f}) "
+                  f"and the site target moves at most one slew step a cycle "
+                  f"after it ({now['site_step']:.0f} W, base "
+                  f"{old['site_step']:.0f})",
+                  now["cmd_export"] <= (TOL if own else max(
+                      TOL, old["cmd_export"] + TOL))
+                  and now["site_step"] <= SLEW)
+            check(f"   {tag}: the grid imports no more than base (+50 W) "
+                  f"({now['imp']:.0f} W, base {old['imp']:.0f}) and exports "
+                  f"no more than it did before the end, or base (+50 W) "
+                  f"({now['exp']:.0f} W more, base {old['exp']:.0f})",
+                  now["imp"] <= old["imp"] + 50.0
+                  and now["exp"] <= max(0.0, old["exp"]) + 50.0)
+
+_by = {(k, n, h): (a, b) for k, n, h, a, b in TABLE}
+for kind in ("complete", "Off", "timeout", "refusal"):
+    _now, _old = _by[(kind, 2, 1200.0)]
+    check(f"non-vacuous ({kind}, N=2, 1200 W): without the fix B steps "
+          f"more than a slew step into an export",
+          _old["cmd_export"] >= 300.0 and _old["ostep"] > SLEW
+          and _now["ostep"] < _old["ostep"])
+    _now, _old = _by[(kind, 2, -1500.0)]
+    check(f"non-vacuous ({kind}, solar surplus): without the fix B gives "
+          f"into the surplus", _old["cmd_export"] >= 1000.0
+          and _now["cmd_export"] <= TOL)
+_now, _old = _by[("complete", 2, 1200.0)]
+check("⭐ the rig case (A's drain completes at 1200 W): B covers the house "
+      "at once (0 -> 1210 W, not 1600), and the grid stays within 50 W of "
+      "its aim while A settles (base: +559 W import, B down to 230 W)",
+      [r[1][1] for r in _now["rows"] if r[0] == _now["end"]] == [1210.0]
+      and _now["imp"] <= 50.0 and _old["imp"] > 400.0)
+_now, _old = _by[("complete", 3, 4000.0)]
+check("N=3 at 4000 W: the others take A's share at once, as before (the "
+      "site does not drop), so the grid does not import",
+      _now["imp"] <= 50.0 and abs(_now["imp"] - _old["imp"]) <= 50.0)
+_now, _old = _by[("stale link", 3, 4000.0)]
+check(f"⭐ stale link, N=3, 4000 W: a dropped draining battery is this "
+      f"rule's, not the link-drop reseed's, so the grid imports at least "
+      f"50 W less than base ({_now['imp']:.0f} W, base {_old['imp']:.0f})",
+      _now["imp"] <= _old["imp"] - 50.0)
+
+# ============================================================================ #
+print("\n2. the peak guard still binds across a drain end")
+_g = ENDS["Off"](2, 4000.0, p1=True, dur=T + 120.0)[0]
+_gs = [r for r in _g if r[0] >= T - 20 and r[6] is not None]
+print(f"    guard engaged on {len(_gs)} cycles around the end; site "
+      f"{[sum(r[1]) for r in _g if T - 4 <= r[0] <= T + 8]}")
+check("⭐ house 4000 W over a 2500 W limit, A switched Off mid-drain: on "
+      "every cycle the guard is engaged, the site gives at least its floor "
+      "(or every battery is at its rating)",
+      _gs and all(sum(r[1]) >= min(r[6], 2500.0 * 1) - 1.0 for r in _gs))
+
+# ============================================================================ #
+print("\n3. a drain end with another edge in the same cycle; a missed poll")
+
+
+def c_drops(c, t):
+    """C's writes are refused from T - 30 s, so it drops at T, the cycle A
+    is switched Off; its board goes to 0 W at once (T - 30 s)."""
+    off(c, t)
+
+
+_e3 = run(3, 4000.0, store=store(3), before=c_drops, reject={"c": T - 30.0},
+          dead=("c", T - 30.0), extra={"battery_modes_enabled": True})[0]
+_e3m = measure(_e3, 3, 4000.0)
+print(f"    A Off as C drops, house 4000 W: B "
+      f"{[r[1][1] for r in _e3 if T - 4 <= r[0] <= T + 10]}")
+check("⭐ A switched Off on the cycle C drops (N=3, 4000 W): B takes over "
+      "with no export and the site moves one slew step a cycle after",
+      _e3m["cmd_export"] <= TOL and _e3m["site_step"] <= SLEW
+      and _e3m["end"] == T)
+
+
+def c_back(c, t):
+    """C's link (stale since T - 60 s, its board holding 0 W) answers again
+    at T, the cycle A is switched Off."""
+    off(c, t)
+    if t == T:
+        c.inverters[2].note_write(True, rig.BASE + timedelta(seconds=t))
+
+
+_r3 = run(3, 1200.0, store=store(3), before=c_back, reject={"c": T - 60.0},
+          back=T, dead=("c", T - 60.0, T), extra={"battery_modes_enabled": True})[0]
+_r3m = measure(_r3, 3, 1200.0)
+print(f"    A Off as C comes back, house 1200 W: B, C "
+      f"{[r[1][1:] for r in _r3 if T - 4 <= r[0] <= T + 6]}")
+check("⭐ A switched Off on the cycle C comes back (N=3, 1200 W): no "
+      "export, one slew step a cycle after",
+      _r3m["cmd_export"] <= TOL and _r3m["site_step"] <= SLEW
+      and _r3m["end"] == T)
+
+
+def missed_poll(at_t):
+    """Battery B's power reading is missing on the cycle at `at_t` s."""
+    def before(c, t):
+        off(c, t)
+        if not getattr(c.hass, "_s95", False):
+            read = c.hass.number
+
+            async def number(e, invert=False):
+                if e == "sensor.batt_b" and c.hass.t == at_t:
+                    return None
+                return await read(e, invert)
+            c.hass.number, c.hass._s95 = number, True
+    return before
+
+
+_mp = run(2, 1200.0, store=store(2), before=missed_poll(T),
+          extra={"battery_modes_enabled": True})[0]
+_mpo = without(lambda: run(2, 1200.0, store=store(2), before=missed_poll(T),
+                           extra={"battery_modes_enabled": True}))[0]
+print(f"    a missed poll on the end cycle: B "
+      f"{[r[1][1] for r in _mp if T - 2 <= r[0] <= T + 8]}, without the fix "
+      f"{[r[1][1] for r in _mpo if T - 2 <= r[0] <= T + 8]}")
+check("⭐ a missed poll on the cycle A is switched Off: the end still "
+      "reaches the loop on the next cycle (B covers the house, no export)",
+      max(sum(r[1]) for r in _mp if r[0] > T) <= 1200.0 - AIM + TOL
+      and max(sum(r[1]) for r in _mpo if r[0] > T) >= 1500.0)
+
+# ============================================================================ #
+print("\n4. N=1 is untouched; the rule at its boundaries")
+
+
+def digest(rows):
+    return hashlib.sha256(repr(rows).encode()).hexdigest()
+
+
+_n1 = [functools.partial(ENDS[k], 1, h) for k in ENDS
+       for h in (1200.0, -1500.0)]
+check("⭐ N=1, every drain end (complete, Off, timeout, abort, stale link, "
+      "refusal), at 1200 W and under a surplus: every row byte-identical "
+      "with and without SAFETY-95",
+      [digest(f()[0]) for f in _n1] == [digest(without(f)[0]) for f in _n1])
+_wave = functools.partial(run, 2, lambda t: 2400.0 + 1500 * math.sin(t / 40),
+                          store=store(2, last_completed=rig.BASE.isoformat()),
+                          dur=600.0)
+check("N=2 healthy (a moving house, no maintenance): every row "
+      "byte-identical with and without SAFETY-95",
+      digest(_wave()[0]) == digest(without(_wave)[0]))
+
+if fixed is None:
+    print(f"\n{len(fails)} FAILED (no _reseed_drain_end): " + ", ".join(fails))
+    sys.exit(1)
+
+_c = run(2, 1200.0, store=store(2), dur=4.0)[1]
+_pin = MaintResult(IDLE, True, 0.0, False, [])
+_free = MaintResult(CHARGE, False, None, True, [])
+_b = MaintResult(IDLE, False, None, False, [])
+AB = {"inverter-a", "inverter-b"}
+
+
+def reseeded(target, shares, batts, grid, res=_pin, live=AB, ended=(A,)):
+    """`_reseed_drain_end` on a 2-battery controller: (target, i_w,
+    A's start)."""
+    _c.target, _c.i_w, _c._drain_ended = target, 123.0, frozenset(ended)
+    _c._loop_start, _c._settle, _c._drain_tail = {}, {}, set()
+    _c._now_mono, _c.grid = 50.0, grid
+    for inv in _c.inverters:
+        inv.target, inv.batt = shares[inv.name[-1]], batts[inv.name[-1]]
+    _c._reseed_drain_end([res, _b], set(live))
+    return _c.target, round(_c.i_w, 6), _c._loop_start.get(A)
+
+
+# House 1200 W: A drains 2000, B 0, grid -800.
+check("the house needs more than the others stand at: the start is the "
+      "house's own point (1210 W), the integrator with it",
+      reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
+               -800.0) == (1210.0, 1210.0, 0.0)
+      and _c._settle == {A: 50.0} and _c._drain_tail == {A}
+      and _c._drain_ended == frozenset())
+check("...and while A settles the loop's error is the house's point minus "
+      "its start (0 here: it holds), not the meter's -790 W",
+      _c._drain_grid_w(AB) - AIM == 0.0)
+check("a solar surplus (house -1500 W): the start is what the others stood "
+      "at plus A's new command (0 W), and the loop moves down from there",
+      reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
+               -3500.0)[0] == 0.0)
+check("the house needs more than stood placed (4000 W, B at 2000): never "
+      "above last cycle's start",
+      reseeded(4000.0, {"a": 2000.0, "b": 2000.0},
+               {"a": 2000.0, "b": 2000.0}, 0.0)[0] == 4000.0)
+check("a free battery (drain complete -> charge): its drain leaves the "
+      "start; a pin's value is its new command",
+      reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
+               -3500.0, res=_free)[2] == 0.0
+      and reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
+                   -3500.0, res=MaintResult(HOLD, True, 0.0, False, []))[2]
+      == 0.0)
+check("a dropped battery's power is the meter's: a board stuck at 2000 W "
+      "keeps B where it stood, and there is no tail to count",
+      reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
+               -800.0, live={"inverter-b"})[0] == 0.0
+      and _c._drain_tail == set())
+_chg = MaintResult(CHARGE, True, -1500.0, False, [])
+check("a drain that ends into a pinned charge: the pin is A's new command "
+      "while A is live, but 0 W once it is dropped (it is sent 0 W)",
+      reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
+               -3500.0, res=_chg)[2] == -1500.0
+      and reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
+                   -3500.0, res=_chg, live={"inverter-b"}) == (0.0, 0.0, 0.0))
+check("no drain ended: nothing moves (the integrator either)",
+      reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
+               -800.0, ended=()) == (2000.0, 123.0, None))
+_c._drain_tail, _c._settle, _c.grid = {A}, {}, -800.0
+check("the tail ends with A's settling (SAFETY-96's window)",
+      _c._drain_grid_w(AB) == -800.0 and _c._drain_tail == set())
+_UA, _UB = (_c.actuator_of("maintenance", i) for i in _c.inverters)
+_drain = MaintResult(DRAIN, True, 2000.0, False, [])
+
+
+def drain_ends(extra, refused=0):
+    """`_charge_track` with A draining last cycle, its drain SET and the
+    claims `extra` on A: which drains it ends."""
+    _c._draining_was, _c._drain_ended = frozenset({A}), frozenset()
+    _c._charging_was, _c._charge_ended = frozenset(), frozenset()
+    _c.live_units = lambda: (_UA, _UB)
+    _c.inverters[0].refused = refused
+    try:
+        _c._charge_track([Claim.set("maintenance", P_MAINTENANCE, 2000.0,
+                                    actuator=_UA)] + extra, [_drain, _b])
+    finally:
+        del _c.live_units
+        _c.inverters[0].refused = 0
+    return set(_c._drain_ended)
+
+
+_rating = [Claim.limit("device", P_SAFETY, -2500.0, 2500.0, actuator=_UA)]
+check("a drain that still stands does not end (its rating binds, no "
+      "contradiction, no refusal)", drain_ends(_rating) == set())
+check("a drain whose limits contradict ends: A is sent 0 W (SAFETY-91's "
+      "pin as a floor against a closed pack)",
+      drain_ends(_rating + [
+          Claim.limit("maintenance", P_MAINTENANCE, lo=2000.0, actuator=_UA),
+          Claim.limit("pack", P_SAFETY, hi=0.0, actuator=_UA)]) == {A})
+check("a refusal ends a drain only if A will not give (+1); one that will "
+      "not take (-1) leaves the drain standing",
+      drain_ends(_rating, refused=1) == {A}
+      and drain_ends(_rating, refused=-1) == set())
+
+_n1c = run(1, 1200.0, store=store(1), dur=4.0)[1]
+_n1c.target, _n1c.i_w, _n1c._drain_ended = 2000.0, 5.0, frozenset({A})
+_n1c._reseed_drain_end([_pin], {A})
+check("N=1: nothing moves", (_n1c.target, _n1c.i_w) == (2000.0, 5.0))
+
+_seen = []
+
+
+def _spy(self, maint_results, live):
+    _seen.append(sorted(self._drain_ended))
+    return fixed(self, maint_results, live)
+
+
+Controller._reseed_drain_end = _spy
+try:
+    run(2, 1200.0, store=store(2), soc=lambda u, t:
+        5.0 if (u == "a" and T <= t < T + 200) else 99.5 if u == "a" and
+        t >= T + 200 else 50.0, dur=T + 400.0,
+        extra={"maintenance_hold_min": 2})
+finally:
+    Controller._reseed_drain_end = fixed
+check("it fires once, at the drain end: not at the charge end, the HOLD "
+      "end, or on any other cycle", [x for x in _seen if x] == [[A]]
+      and len(_seen) > 300)
+
+print("\n    drain end | N | house W | site step W head/base | commanded "
+      "export W head/base | others' step W head/base | grid import W "
+      "head/base | grid step W head/base | export beyond the end's W head/base")
+for k, n, h, a, b in TABLE:
+    print(f"    {k} | {n} | {h:.0f} | {a['site_step']:.0f}/"
+          f"{b['site_step']:.0f} | {a['cmd_export']:.0f}/{b['cmd_export']:.0f}"
+          f" | {a['ostep']:.0f}/{b['ostep']:.0f} | {a['imp']:.0f}/"
+          f"{b['imp']:.0f} | {a['gstep']:.0f}/{b['gstep']:.0f} | "
+          f"{a['exp']:.0f}/{b['exp']:.0f}")
+print()
+if fails:
+    print(f"{len(fails)} FAILED: " + ", ".join(fails))
+    sys.exit(1)
+print("all SAFETY-95 checks passed")