Merge pull request 'SAFETY-96: a battery the loop cannot move does not freeze it' (#176) from safety-96-stale-unit-site-total into release/1.0

goodwe_controller/DECISIONS/SAFETY-96-dropped-battery.md

@@ -0,0 +1,59 @@
+# SAFETY-96 - a battery the loop cannot move does not freeze it
+
+**Status: implemented 2026-10-06, for review.** Tests: `test_safety96.py`.
+
+**The defect** (SAFETY-93 rig test). At N=2 the saturation freeze compares
+the site target with the site sum of the readings. After A's link went
+stale, A was out of the split but its reading (-2000 W, still charging) was
+in the sum: "saturation freeze (800 W vs -1903 W)", B held at 1200 W for
+~236 cycles until the circulation trip. After a charge timeout or Off at
+6000 W the loop froze ~14 s at 800 W while A came out of its charge. At a
+drop, B also took A's whole share in one step.
+
+**The rule** (`main.py`, `_held_out_w()`, `_reseed_drop()`, N>=2 only).
+A battery is held out of the freeze test, counted at its command instead of
+its reading, while:
+
+- it is out of the split (writes refused for `LINK_STALE_S`);
+- its last write was refused (its board may have gone to 0 W on its own
+  watchdog);
+- it settles after a step the loop did not make: a charge end (SAFETY-93:
+  HOLD, timeout, Off, refusal, stale link) or a rejoin (then every battery,
+  as the split is made again). Settling ends when its reading is within the
+  deadband of its command, or after `PLANT_SETTLE_DOWN_S` (23 s, the
+  measured plant settle).
+
+Its real power is not lost and not counted twice: the meter sees it, and
+the loop follows the meter. Only the freeze stops reading it as the loop's.
+The freeze still fires on the other batteries' own lag and on unplaced
+watts.
+
+At a drop the battery's share leaves the loop's start point, so the others
+take over one slew step a cycle. At a rejoin it enters at its reading.
+
+**Not a drop.** A missing reading or a stale register stops the site
+(`inputs_ok`), as before. A refused battery (SAFETY-73) stays in the sum:
+it is asked ~0 W and reads ~0 W. A watchdog-stopped battery (SAFETY-02/30)
+is pinned at 0 W and stays in the sum. New: no refusal is learned on a
+battery whose writes are refused; its board going to 0 W is no refusal. Its
+count is kept, not reset, so a link that refuses 1 write in 3 does not hide
+a battery that really refuses (review).
+
+**Before -> after** (2 x 2500 W, 10 s lag, 2 s meter; freeze cycles not
+about the remaining battery's own lag):
+
+| case | before | after |
+|---|---|---|
+| A drops while giving, A holds 1200 W | 128, B steps 1210 W | 0, B stays |
+| A drops while charging, A holds -2000 W | 78 | 0 (circulation trip stays: 4 kW goes round) |
+| A drops while charging, board to 0 W | 5 | 0 |
+| A drops then comes back, A holds 1200 W | 46, grid +1.1 kW | 0, grid within 250 W |
+| charge timeout at 6000 W | 7, B held at 400 W | 0, B climbs |
+| A switched Off at 6000 W | 6, B held at 800 W | 0, B climbs |
+| N=1, N=2 with no drop | | byte-identical |
+
+**Not changed.** At a rejoin the allocator splits again at once (A 0 ->
+1205, B 2410 -> 1205 W); the site total does not step. A battery that keeps
+charging on its own after its link drops still trips the circulation check
+(it is real circulation). A trade end (`_reseed_run_edge`) still starts
+from the full site sum.

goodwe_controller/DOCS.md

@@ -1700,6 +1700,8 @@ batteries the other one gives first. The hold still ends on its clock
 (SAFETY-86).
 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).
 
 #### 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-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}).
+        self._live_was = None
+        self._loop_start = {}
+        self._settle = {}
         # Set by amain() once the transport is built, so the status page can show
         # what only the transport knows (homewizard_local's unchanged_s). Stays
         # None when P1 is off, or under a transport that has no such counter.
@@ -1393,6 +1399,12 @@ class Controller:
                     self._unit_refusal_end(
                         inv, "the battery takes power the other way")
                 continue
+            if inv.rejected_since is not None:
+                # SAFETY-96: its writes are refused, so its board may have
+                # gone to 0 W on its own; that is no refusal by the battery.
+                # The count is kept, as across a gap, so a link that refuses
+                # now and then does not hide a battery that does refuse.
+                continue
             sign = self._gave_nothing(inv)
             if sign:
                 self._unit_refusal_start(
@@ -2456,9 +2468,69 @@ class Controller:
                 continue
             now_w = (float(res.setpoint_w or 0.0) if res.owns_setpoint
                      else max(float(inv.target), 0.0))
+            if not inv.is_live(datetime.now(timezone.utc))[0]:
+                now_w = 0.0     # SAFETY-96: out of the split, it is sent 0 W
             self.target += now_w - float(inv.target)
+            # SAFETY-96: it settles from its charge to that command.
+            self._loop_start[inv.name] = now_w
+            self._settle[inv.name] = self._now_mono
         self.i_w = self.target
 
+    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
+        slew step a cycle instead of taking its share at once. One that comes
+        back enters it at its reading, and every battery settles while the
+        split is made again (`_held_out_w`).
+        DECISIONS/SAFETY-96-dropped-battery.md.
+        """
+        was, self._live_was = self._live_was, frozenset(live)
+        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:
+                continue        # SAFETY-93's edge: `_reseed_charge_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:
+                now_w = float(inv.batt)
+                # The split is made again, so every battery's command steps.
+                self._settle.update(
+                    {u: self._now_mono for u in live})
+            else:
+                continue
+            self.target += now_w - float(inv.target)
+            self._loop_start[inv.name] = now_w
+        self.i_w = self.target
+
+    def _held_out_w(self, live) -> float:
+        """SAFETY-96, N>=2: what to add to the site sum so that a battery
+        the loop cannot move right now counts at its command, not at its
+        reading. That is a battery out of the split, one whose writes are
+        refused, or one settling after a step the loop did not make (a charge
+        end, a rejoin) until its reading is within the deadband of its
+        command or PLANT_SETTLE_DOWN_S has passed. Its real power still
+        reaches the loop through the meter, so it is neither lost nor counted
+        twice; only the saturation freeze stops seeing it as the loop's.
+        """
+        if len(self.inverters) < 2:
+            return 0.0
+        out = 0.0
+        for inv in self.inverters:
+            cmd = self._loop_start.get(inv.name, float(inv.target))
+            since = self._settle.get(inv.name)
+            age = (None if since is None or self._now_mono is None
+                   else self._now_mono - since)
+            if since is not None and (
+                    age is None or age >= _ctl.PLANT_SETTLE_DOWN_S
+                    or (age > 0 and abs(float(inv.batt) - cmd)
+                        < self.tuning.deadband_w)):
+                del self._settle[inv.name]
+            if (inv.name not in live or inv.rejected_since is not None
+                    or inv.name in self._settle):
+                out += cmd - float(inv.batt)
+        return out
+
     def _mode_fault_log(self, resolution) -> None:
         """OPT-11: say once, per battery, that its mode was dropped as a fault.
 
@@ -4730,6 +4802,8 @@ class Controller:
             mode_trade = self._mode_trade(claims, now)
         self._mode_trade_was = mode_trade
         self._charge_track(claims, maint_results)
+        _live96 = set(self.live_units())
+        self._loop_start = {}
         if (self.auto or self.shadow) and self.inputs_ok:
             # ⚠️ prev is the ARBITER's last output, not the loop's own last
             # wish. If something outranked the loop, that is what the hardware
@@ -4756,6 +4830,9 @@ class Controller:
             # is what seeds it, and dropping it there restarts the ~70 s cycle.
             if self.loop_w is not None and self.target != self.loop_w:
                 self._refusal_end("another writer")
+            # SAFETY-96: a battery that dropped out of the split leaves the
+            # loop's start point (SAFETY-93's own if its charge ended too).
+            self._reseed_drop(_live96)
             # SAFETY-93: a charge that ended leaves the loop's start point.
             self._reseed_charge_end(maint_results)
             probe = self._refusal_probe()
@@ -4775,6 +4852,9 @@ class Controller:
             # level: the two halves of §6.2's identity break in opposite
             # directions. See `site_batt_w` and test_alloc.py part 8.
             site_batt = self.site_batt_w
+            # SAFETY-96: a battery the loop cannot move counts at its command,
+            # so its own power does not freeze the loop (the meter has it).
+            site_batt += self._held_out_w(_live96)
             # TEL-10. ⚠️ ONE NAME FOR THE COMMAND THE LOOP WAS MEASURED
             # AGAINST. `self.target` is reassigned further down this method, so
             # a `decision.i_w - self.target` written at publish time would
@@ -4883,6 +4963,7 @@ class Controller:
             # Stopped or blind: no accumulation may survive the outage.
             self.i_w = 0.0
             self.loop_w = None
+            self._live_was = frozenset(_live96)
             # TEL-10, and it is the same zero the line above writes rather than
             # a second opinion about it. ⚠️ `not-running` rather than a stale
             # `tracking`: a reason row that keeps repeating the last word the

goodwe_controller/test_safety78.py

@@ -187,7 +187,7 @@ check("⭐ a draining battery that is dropped (stale link) adds no floor: B "
       "covers the 1500 W house, not the 2000 W drain rate, and the wait is "
       "logged once",
       "dropped inverter-a" in _l_rows[-1][7]
-      and cmd(_l_rows[-1])["b"] == 1500.0
+      and cmd(_l_rows[-1])["b"] == 1510.0
       and sum("cannot give power now" in e for e in _l.events) == 1)
 
 # ============================================================================ #

goodwe_controller/test_safety96.py

@@ -0,0 +1,573 @@
+"""Runnable check for SAFETY-96: a battery the loop cannot move does not
+freeze the loop. `python3 test_safety96.py`
+
+At N=2, after A's link went stale the site sum still counted A's power, so
+`compute()` saw the site "not following" and froze B (SAFETY-93 rig test:
+B held at 1200 W for ~236 cycles). After a charge timeout or Off at 6000 W
+it froze ~14 s while A came out of its charge. Now a battery out of the
+split, with refused writes, or settling after a step the loop did not make
+counts at its command; its real power reaches the loop through the meter.
+Drives the REAL Controller on a lagging battery (10 s, 2 s meter).
+DECISIONS/SAFETY-96-dropped-battery.md.
+
+1. A's link drops while it discharges: A holds its last command (stuck), or
+   its board goes to 0 W 25 s after the first refused write.
+2. A's link drops while it charges (maintenance).
+3. B's link drops; both drop.
+4. A drops, then comes back.
+5. A's charge times out, or A is switched Off, at 6000 W.
+6. N=1 is untouched; the rule's parts one by one.
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import hashlib
+import logging
+import math
+import os
+import sys
+from datetime import datetime, timedelta, timezone
+
+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
+import safety84_identity as s84                                # noqa: E402
+from app import packmode                                       # noqa: E402
+from app.main import Controller                                # noqa: E402
+from app.maintenance import CHARGE_TIMEOUT_MIN, IDLE, MaintResult  # noqa: E402
+from app.store import unit_key                                 # noqa: E402
+from named_absence import first, present                       # noqa: E402
+
+fails = []
+SLEW = 400.0
+FROZEN = ("saturated-freeze", "freeze-released")
+OPT = dict(s84.OPT, peak_cap_w=3500)
+PLAIN = dict(s79.P1, peak_cap_w=3500)
+
+
+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, "_held_out_w", None)
+real_drop = getattr(Controller, "_reseed_drop", None)
+
+
+def without(fn):
+    """Run `fn` with SAFETY-96 taken out (the base behaviour)."""
+    if fixed is None:
+        return fn()
+    Controller._held_out_w = lambda self, live: 0.0
+    Controller._reseed_drop = lambda self, live: setattr(
+        self, "_live_was", frozenset(live))
+    try:
+        return fn()
+    finally:
+        Controller._held_out_w, Controller._reseed_drop = fixed, real_drop
+
+
+def run(n=2, *, house=lambda t: 2400.0, dur=600.0, store=None, opt=None,
+        soc=None, reject=None, back=None, wd=None, before=None, lag=10.0,
+        blind=None, flaky=None, dead=None):
+    """Rows of one run. Writes to the units in `reject` (name -> from s) are
+    refused from that time until `back` s; the add-on sees the refusal at
+    once (a heartbeat write). A refused unit holds its last command, or with
+    `wd` its board goes to 0 W `wd` s after the first refused write. B's
+    reading is missing in the `blind` (from, to) window. `flaky(u, t)`:
+    writes to u refused at t, each cycle's verdict seen (a heartbeat).
+    `dead` (u, from s): u gives nothing from then on, whatever it is told.
+    Each row: (t, plant commands, add-on commands, loop reason, site target,
+    grid, circulation tripped)."""
+    reject = reject or {}
+    real_p, real_s, real_tick = rig._Hass.power, rig._Hass.set_number, rig.TICK
+
+    def refused(u, t):
+        if flaky is not None:
+            return flaky(u, t)
+        return u in reject and t >= reject[u] and (back is None or t < back)
+
+    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 wd is not None and refused(u, now) and now - reject[u] >= wd:
+                c = 0.0
+            if dead is not None and u == dead[0] and now >= dead[1]:
+                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
+
+    rows = []
+
+    def hook(c, t):
+        for inv in c.inverters:
+            if flaky is not None:
+                inv.note_write(not flaky(inv.name[-1], t),
+                               rig.BASE + timedelta(seconds=t))
+            elif refused(inv.name[-1], t) and inv.rejected_since is None:
+                inv.rejected_since = rig.BASE + timedelta(seconds=t)
+        if before is not None:
+            before(c, t)
+        if blind is not None and not getattr(c.hass, "_s96", False):
+            read = c.hass.number
+
+            async def number(e, invert=False):
+                if e == "sensor.batt_b" and blind[0] <= c.hass.t < blind[1]:
+                    return None
+                return await read(e, invert)
+            c.hass.number, c.hass._s96 = number, True
+        (s84.feed(True) if opt is not None else s79._feed)(c, t)
+
+    def on_cycle(c, t):
+        rows.append((t, tuple(sorted(c.hass.cmd.items())),
+                     tuple(inv.target for inv in c.inverters), c.loop_reason,
+                     c.target, c.hass.grid(), c.circ_wd.tripped,
+                     tuple(inv.batt for inv in c.inverters)))
+    rig._Hass.power, rig._Hass.set_number, rig.TICK = power, set_number, 2.0
+    try:
+        _r, ctl = rig.run(n, extra=dict(opt if opt is not None else PLAIN),
+                          dur=dur, load=house, soc=soc or (lambda u, t: 60.0),
+                          cut=lambda u, t: False, before=hook,
+                          on_cycle=on_cycle, store=store or rig._Store())
+    finally:
+        rig._Hass.power, rig._Hass.set_number, rig.TICK = real_p, real_s, \
+            real_tick
+    return rows, ctl
+
+
+def freeze(rows, lo, hi=1e9):
+    return sum(1 for r in rows if lo <= r[0] < hi and r[3] in FROZEN)
+
+
+def phantom(rows, k, lo, hi=1e9):
+    """Freeze cycles while battery `k` (the one taking over) reads within
+    saturation_w (500 W) of what the loop last asked of it plus anything
+    unplaced (last site target less the other's command): the freeze is
+    then about the other battery, not about `k`'s own lag."""
+    return sum(1 for a, b in zip(rows, rows[1:]) if lo <= b[0] < hi
+               and b[3] in FROZEN
+               and abs(a[4] - a[2][1 - k] - b[7][k]) <= 500.0)
+
+
+def ramp(rows, k, lo, hi=1e9):
+    """The largest change of unit `k`'s command (0 = A) between cycles."""
+    return max(abs(b[2][k] - a[2][k]) for a, b in zip(rows, rows[1:])
+               if lo < b[0] <= hi)
+
+
+def site_step(rows, lo, hi=1e9):
+    return max(abs(b[4] - a[4]) for a, b in zip(rows, rows[1:])
+               if lo < b[0] <= hi)
+
+
+def grid_err(rows, lo, hi=1e9):
+    """Mean |grid - aim| over [lo, hi) (aim -10 W)."""
+    xs = [abs(r[5] + 10.0) for r in rows if lo <= r[0] < hi]
+    return sum(xs) / len(xs)
+
+
+def trip_at(rows):
+    """When the circulation guard tripped (its own 0 W step is no ramp)."""
+    return next((r[0] for r in rows if r[6]), 1e9)
+
+
+def tripped(rows):
+    return rows[-1][6]
+
+
+def at(rows, t, k):
+    return first((r[2][k] for r in rows if r[0] == t), f"row {t}")
+
+
+TABLE = []
+LAST = [None]               # the controller of the last run with the fix
+
+
+def case(name, fn, lo, k, hi=1e9):
+    """Run `fn` with and without SAFETY-96, print and record the numbers:
+    freeze cycles (and those not about `k`'s own lag), `k`'s largest step,
+    the mean grid error, a circulation trip."""
+    (now, LAST[0]), old = fn(), without(fn)[0]
+    row = (name, freeze(now, lo, hi), freeze(old, lo, hi),
+           phantom(now, k, lo, hi), phantom(old, k, lo, hi),
+           ramp(now, k, lo, min(hi, trip_at(now) - 1)),
+           ramp(old, k, lo, min(hi, trip_at(old) - 1)),
+           grid_err(now, lo, hi), grid_err(old, lo, hi),
+           tripped(now), tripped(old))
+    TABLE.append(row)
+    print(f"    {name}: freeze {row[1]} (base {row[2]}), not about its own lag "
+          f"{row[3]} (base {row[4]}); step {row[5]:.0f} (base {row[6]:.0f}) "
+          f"W/cycle; grid error {row[7]:.0f} (base {row[8]:.0f}) W; "
+          f"circulation trip {row[9]} (base {row[10]})")
+    return now, old
+
+
+STUCK, BOARD = ("stuck", None), ("board 0 W", 25.0)
+
+# ============================================================================ #
+print("1. A's link drops while it discharges (house 2400 W, each 1200 W)")
+for v, w in (STUCK, BOARD):
+    now, old = case(f"A drop discharging, {v}",
+                    lambda w=w: run(reject={"a": 300.0}, wd=w), 300.0, 1)
+    check(f"⭐ {v}: no freeze about A, B moves at most one slew step a "
+          f"cycle, no circulation trip", phantom(now, 1, 300) == 0
+          and ramp(now, 1, 300) <= SLEW and not tripped(now))
+    check(f"non-vacuous ({v}): without the fix the loop freezes about A, or "
+          f"B jumps more than a slew step at the drop",
+          phantom(old, 1, 300) >= 10 or ramp(old, 1, 300) > SLEW)
+    if w is None:
+        check("stuck: B stays at its share, since A still gives its 1200 W "
+              "(no freeze at all, grid within 15 W on average)",
+              freeze(now, 300) == 0 and grid_err(now, 300) < 15.0
+              and freeze(old, 300) >= 100)
+    else:
+        check("board 0 W: B ends up carrying the whole house (2410 W)",
+              abs(now[-1][2][1] - 2410.0) <= 20.0)
+
+# ============================================================================ #
+print("\n2. A's link drops while it charges (maintenance, house 1200 W)")
+
+
+def charging(**kw):
+    return run(house=lambda t: 1200.0, dur=1500.0,
+               store=s79.store_charging(2), opt=dict(
+                   OPT, maintenance_hold_min=600),
+               soc=lambda u, t: 90.0 if u == "a" else 60.0, **kw)
+
+
+for v, w in (STUCK, BOARD):
+    now, old = case(f"A drop charging, {v}",
+                    lambda w=w: charging(reject={"a": 1300.0}, wd=w),
+                    1300.0, 1)
+    trip = trip_at(now)
+    check(f"⭐ {v}: no freeze about A, and B climbs at most one slew step a "
+          f"cycle and never charges (until any circulation trip)",
+          phantom(now, 1, 1300, trip) == 0
+          and ramp(now, 1, 1300, trip - 1) <= SLEW
+          and min(r[2][1] for r in now if r[0] >= 1300) >= 0.0)
+    check(f"non-vacuous ({v}): without the fix the loop freezes about A",
+          phantom(old, 1, 1300) >= 3)
+    if w is None:
+        check("stuck (A charges 2000 W on its own): B reaches its 2500 W "
+              "before the circulation trip, which still trips on the real "
+              "4000 W going round (as on base)",
+              max(r[2][1] for r in now if r[0] < trip) == 2500.0
+              and tripped(now) and tripped(old))
+    else:
+        check("board 0 W: B carries the house (grid within 50 W from 100 s "
+              "after the drop), no circulation trip",
+              grid_err(now, 1400) < 50.0 and not tripped(now))
+
+# ============================================================================ #
+print("\n3. B's link drops; both drop")
+for v, w in (STUCK, BOARD):
+    now, old = case(f"B drop discharging, {v}",
+                    lambda w=w: run(reject={"b": 300.0}, wd=w), 300.0, 0)
+    check(f"⭐ B drops ({v}): no freeze about B, A moves at most one slew "
+          f"step a cycle, no circulation trip", phantom(now, 0, 300) == 0
+          and ramp(now, 0, 300) <= SLEW and not tripped(now))
+for v, w in (STUCK, BOARD):
+    now, old = case(f"both drop, {v}",
+                    lambda w=w: run(reject={"a": 300.0, "b": 300.0}, wd=w),
+                    300.0, 1)
+    check(f"⭐ both drop ({v}): no freeze; both are commanded 0 W once "
+          f"nothing can be placed", freeze(now, 300) == 0
+          and now[-1][2] == (0.0, 0.0) and not tripped(now))
+
+# ============================================================================ #
+print("\n4. A drops, then comes back")
+
+
+def rejoined(rows):
+    return next((r[0] for r in rows if r[0] > 420 and r[2][0] != 0.0), 1e9)
+
+
+for v, w in (STUCK, BOARD):
+    now, old = case(f"A drop and rejoin, {v}",
+                    lambda w=w: run(reject={"a": 300.0}, back=420.0, wd=w,
+                                    dur=700.0), 300.0, 1)
+    back = rejoined(now)
+    check(f"rejoin ({v}): A is back in the split within 4 s of its link "
+          f"returning (no refusal was learned while its writes were refused)",
+          back <= 424.0 and not any("REFUSES" in e for e in LAST[0].events))
+    print(f"      A back in the split at {back:.0f} s; A "
+          f"{[r[2][0] for r in now if back - 2 <= r[0] <= back + 6]}, B "
+          f"{[r[2][1] for r in now if back - 2 <= r[0] <= back + 6]}; grid "
+          f"{[round(r[5]) for r in now if back - 2 <= r[0] <= back + 6]}; "
+          f"base grid {[round(r[5]) for r in old if back - 2 <= r[0] <= back + 6]}")
+    a0 = first((r for r in now if r[0] == back - 2), "row before the rejoin")
+    a1 = first((r for r in now if r[0] == back), "the rejoin row")
+    check(f"⭐ rejoin ({v}): no freeze after it; the site target takes in only "
+          f"A's own reading, then moves at most one slew step a cycle; the "
+          f"grid stays within 250 W of its aim; no circulation trip",
+          present(a0) and present(a1) and freeze(now, back) == 0
+          and abs(a1[4] - a0[4] - a0[7][0]) <= SLEW
+          and site_step(now, back) <= SLEW
+          and max((abs(r[5] + 10.0) for r in now if r[0] >= back),
+                  default=1e9) < 250.0
+          and not tripped(now))
+    if w is None:
+        check("non-vacuous (stuck): without the fix the grid swings over "
+              "1000 W after the rejoin", max((abs(r[5] + 10.0) for r in old
+                                              if r[0] >= back), default=0.0)
+              >= 1000.0)
+
+
+
+# Review: a battery that does refuse, behind a link that refuses one write
+# in three, is still found out (the count is kept over a refused write).
+_fl_rows, _fl = run(dur=700.0, dead=("a", 300.0),
+                    flaky=lambda u, t: u == "a" and t >= 300.0
+                    and int(t / 2) % 3 == 0)
+_fl_at = next((r[0] for r in _fl_rows if r[2][0] == 0.0 and r[0] > 300.0),
+              None)
+_fl_end = first((r[2] for r in _fl_rows[-1:]), "the last row")
+print(f"    A gives nothing from 300 s, 1 write in 3 refused: A's share is "
+      f"taken off at {_fl_at} s; A {_fl.inverters[0].refused}, last "
+      f"commands {_fl_end}")
+check("⭐ a battery that refuses behind a link refusing 1 write in 3: its "
+      "refusal is still learned and B carries the house",
+      _fl.inverters[0].refused == 1 and _fl_at is not None
+      and _fl_at <= 400.0 and present(_fl_end)
+      and abs(_fl_end[1] - 2410.0) <= 20.0)
+
+# Review: a fast battery whose board goes to 0 W 1 s after the first refused
+# write reads ~0 W long before its link is stale (30 s): no refusal either.
+_ref5 = []
+_e5_rows, _e5 = run(reject={"a": 300.0}, back=420.0, wd=1.0, lag=1.0,
+                    dur=700.0,
+                    before=lambda c, t: _ref5.append(c.inverters[0].refused))
+check("fast battery, board 0 W 1 s after the first refused write: no "
+      "refusal is learned, A is back in the split within 4 s of its link "
+      "returning",
+      not any(_ref5) and rejoined(_e5_rows) <= 424.0)
+
+_seen = []
+
+
+def blind_seen(c, t):
+    """Record the last cycle: whether the loop ran, is A live, what settles."""
+    if t > 340.0:
+        _seen.append((t - 2.0, c.loop_reason, "inverter-a" in c.live_units(),
+                      dict(c._settle)))
+
+
+run(reject={"a": 300.0}, back=420.0, blind=(340.0, 470.0), dur=490.0,
+    before=blind_seen)
+_back = first((x for x in _seen if x[1] != "not-running" and x[0] > 400),
+              "the restart")
+print(f"    B's reading missing 340-470 s, A back at 420 s: control holds from "
+      f"{first((x[0] for x in _seen if x[1] == 'not-running'), '?')} s, "
+      f"restarts at {_back[0]:.0f} s, settling then: {_back[3]}")
+check("A drops, then comes back while control holds (a reading missing): no "
+      "rejoin at the restart, which starts from what is on the wire",
+      any(x[1] == "not-running" and x[2] for x in _seen) and _back[3] == {})
+
+
+# ============================================================================ #
+print("\n5. A's charge times out, or A is switched Off, at a 6000 W house")
+
+
+def timed_out(end):
+    """A's charge passes CHARGE_TIMEOUT_MIN at `end` s; A then stays free
+    (its cycle started today, so no restart) and B is not due."""
+    s = s79.store_charging(2)
+    s.data[unit_key("inverter-a", "phase_started")] = (
+        rig.BASE + timedelta(seconds=end - CHARGE_TIMEOUT_MIN * 60 - 1)
+    ).isoformat()
+    s.data[unit_key("inverter-a", "last_start_attempt")] = rig.BASE.isoformat()
+    s.data[unit_key("inverter-b", "last_completed")] = rig.BASE.isoformat()
+    return s
+
+
+def b_done():
+    s = s79.store_charging(2)
+    s.data[unit_key("inverter-b", "last_completed")] = rig.BASE.isoformat()
+    return s
+
+
+def switch_off(at_t):
+    def before(c, t):
+        if t >= at_t:
+            c.store.data[unit_key("inverter-a", packmode.USE_KEY)] = False
+    return before
+
+
+def high(t):
+    return 6000.0 if t >= 1760.0 else 1200.0
+
+
+ENDS = {
+    "timeout": lambda: run(house=high, dur=1798.0, store=timed_out(1780.0),
+                           opt=dict(OPT, maintenance_hold_min=600),
+                           soc=lambda u, t: 90.0 if u == "a" else 60.0),
+    "Off": lambda: run(house=high, dur=1798.0, store=b_done(),
+                       before=switch_off(1780.0),
+                       opt=dict(OPT, maintenance_hold_min=600,
+                                battery_modes_enabled=True),
+                       soc=lambda u, t: 90.0 if u == "a" else 60.0),
+}
+for k, fn in ENDS.items():
+    now, old = case(f"{k} at 6000 W", fn, 1780.0, 1)
+    print(f"      B {[r[2][1] for r in now if 1778 <= r[0] <= 1796]}, base "
+          f"{[r[2][1] for r in old if 1778 <= r[0] <= 1796]}")
+    check(f"⭐ {k}: no freeze about A while it comes out of its charge; B "
+          f"climbs one slew step a cycle and never charges",
+          phantom(now, 1, 1780) == 0 and ramp(now, 1, 1780) <= SLEW
+          and min(r[2][1] for r in now if r[0] >= 1780) >= 0.0)
+    check(f"non-vacuous ({k}): without the fix the loop freezes about A and "
+          f"B ends lower", phantom(old, 1, 1780) >= 3
+          and at(now, 1796.0, 1) > at(old, 1796.0, 1))
+
+# ============================================================================ #
+print("\n6. N=1 is untouched; the rule's parts")
+
+
+def digest(rows):
+    return hashlib.sha256(repr(rows).encode()).hexdigest()
+
+
+def wave(t):
+    return 2400.0 + 1500 * math.sin(t / 40)
+
+
+_n1 = [lambda: run(1, reject={"a": 300.0}, back=420.0, wd=25.0),
+       lambda: run(1, house=high, dur=1798.0, store=timed_out(1780.0),
+                   opt=dict(OPT, maintenance_hold_min=600)),
+       lambda: run(1, house=wave)]
+check("⭐ N=1 (a refused link that comes back, a charge timeout, a moving "
+      "house): every row byte-identical with and without SAFETY-96",
+      [digest(f()[0]) for f in _n1] == [digest(without(f)[0]) for f in _n1])
+check("N=2 healthy (a moving house, no drop): every row byte-identical with "
+      "and without SAFETY-96",
+      digest(run(house=wave)[0]) == digest(without(lambda: run(
+          house=wave))[0]))
+
+if fixed is None:
+    print(f"\n{len(fails)} FAILED (no _held_out_w): " + ", ".join(fails))
+    sys.exit(1)
+
+_c = run(dur=4.0)[1]
+AB = ("inverter-a", "inverter-b")
+
+
+def held(batts, targets, live=AB, rejected=(), settle=None, now=100.0,
+         start=None):
+    for inv, b, t in zip(_c.inverters, batts, targets):
+        inv.batt, inv.target = b, t
+        inv.rejected_since = rig.BASE if inv.name in rejected else None
+    _c._settle = dict(settle or {})
+    _c._loop_start = dict(start or {})
+    _c._now_mono = now
+    return _c._held_out_w(set(live))
+
+
+check("a healthy pair: nothing held out, however far a reading lags",
+      held((-2000.0, 400.0), (0.0, 400.0)) == 0.0)
+check("a dropped battery counts at its command (0 W), not its reading",
+      held((-2000.0, 800.0), (0.0, 800.0), live=("inverter-b",)) == 2000.0)
+check("a battery whose writes are refused counts at its command",
+      held((900.0, 800.0), (1200.0, 800.0), rejected=("inverter-a",))
+      == 300.0)
+check("a settling battery counts at its command until its reading is within "
+      "the deadband (15 W) of it, then it is the loop's again",
+      held((-20.0, 800.0), (0.0, 800.0), settle={"inverter-a": 95.0})
+      == 20.0 and "inverter-a" in _c._settle
+      and held((-10.0, 800.0), (0.0, 800.0), settle={"inverter-a": 95.0})
+      == 0.0 and "inverter-a" not in _c._settle)
+check("...but not on the cycle it starts",
+      held((-10.0, 800.0), (0.0, 800.0), settle={"inverter-a": 100.0})
+      == 10.0 and "inverter-a" in _c._settle)
+check("...or until PLANT_SETTLE_DOWN_S (23 s) has passed, whichever is first",
+      held((-2000.0, 800.0), (0.0, 800.0), settle={"inverter-a": 77.1})
+      == 2000.0 and held((-2000.0, 800.0), (0.0, 800.0),
+                         settle={"inverter-a": 77.0}) == 0.0)
+check("a reseeded start (SAFETY-93's charge end) is the command it counts at",
+      held((-2000.0, 800.0), (-2000.0, 800.0), settle={"inverter-a": 99.0},
+           start={"inverter-a": 0.0}) == 2000.0)
+
+
+def dropped(was, live, target, shares, batts=(0.0, 0.0), start=None):
+    _c._live_was, _c.target, _c.i_w = (None if not was else frozenset(was),
+                                        target, 123.0)
+    _c._loop_start, _c._settle, _c._now_mono = dict(start or {}), {}, 50.0
+    for inv, b in zip(_c.inverters, batts):
+        inv.target, inv.batt = shares[inv.name[-1]], b
+    _c._reseed_drop(set(live))
+    return _c.target, _c.i_w
+
+
+check("a drop: the dropped battery's share leaves the start point, the "
+      "integrator with it", dropped(AB, ("inverter-b",), 2400.0,
+                                    {"a": 1200.0, "b": 1200.0})
+      == (1200.0, 1200.0) and _c._live_was == frozenset({"inverter-b"}))
+_c._charge_ended = frozenset({"inverter-a"})
+check("...but a battery whose charge ended too is left to SAFETY-93's reseed",
+      dropped(AB, ("inverter-b",), -2000.0, {"a": -2000.0, "b": 0.0})
+      == (-2000.0, -2000.0))
+_c._charge_ended = frozenset()
+_free = MaintResult(IDLE, False, None, False, [])
+
+
+def charge_end(rejected_s):
+    """SAFETY-93's reseed for A (giving 1385 W under the loop, its charge
+    suspended), A's writes refused for `rejected_s` s: (target, start)."""
+    _c.target, _c.i_w, _c._charge_ended = 2770.0, 0.0, frozenset({AB[0]})
+    _c._loop_start, _c._settle = {}, {}
+    for inv in _c.inverters:
+        inv.target, inv.rejected_since = 1385.0, None
+    _c.inverters[0].rejected_since = datetime.now(timezone.utc) - timedelta(
+        seconds=rejected_s)
+    _c._reseed_charge_end([_free, _free])
+    _c.inverters[0].rejected_since = None
+    return _c.target, _c._loop_start[AB[0]]
+
+
+check("a charge that ends as its battery drops: the battery's start is 0 W "
+      "(it is sent 0 W), not its last give; still live, its give stays",
+      charge_end(31.0) == (1385.0, 0.0) and charge_end(5.0) == (2770.0, 1385.0))
+check("a rejoin: the battery enters the start point at its reading, and "
+      "both settle while the split is made again",
+      dropped(("inverter-b",), AB, 1200.0, {"a": 0.0, "b": 1200.0},
+              batts=(-700.0, 1200.0)) == (500.0, 500.0)
+      and _c._settle == {"inverter-a": 50.0, "inverter-b": 50.0})
+check("no change: nothing moves (the integrator either)",
+      dropped(AB, AB, 1200.0, {"a": 600.0, "b": 600.0}) == (1200.0, 123.0)
+      and dropped((), ("inverter-b",), 1200.0, {"a": 0.0, "b": 1200.0})
+      == (1200.0, 123.0))
+_n1c = run(1, dur=4.0)[1]
+_n1c._live_was, _n1c.target, _n1c.i_w = frozenset({"inverter-a"}), 900.0, 5.0
+_n1c._reseed_drop(set())
+check("N=1: nothing moves and nothing is held out",
+      (_n1c.target, _n1c.i_w) == (900.0, 5.0)
+      and _n1c._held_out_w(set()) == 0.0)
+
+print("\n    case | freeze head/base | not about its own lag head/base | "
+      "step W head/base | grid error W head/base | circulation trip head/base")
+for r in TABLE:
+    print(f"    {r[0]} | {r[1]}/{r[2]} | {r[3]}/{r[4]} | {r[5]:.0f}/{r[6]:.0f} |"
+          f" {r[7]:.0f}/{r[8]:.0f} | {r[9]}/{r[10]}")
+print()
+if fails:
+    print(f"{len(fails)} FAILED: " + ", ".join(fails))
+    sys.exit(1)
+print("all SAFETY-96 checks passed")