SAFETY-95 review: test the stale-link and contradiction legs; refusal +1 only

F1: test_safety95 checks that a dropped draining battery is this rule's
(stale link, N=3, 4000 W: grid import 676 W vs base 786 W); kills R2.
F2: _charge_track tested directly with a contradicting drain (SAFETY-91's
pin as a floor against a closed pack); kills R3. Today a drain contradicts
only through a refusal; the clause stays, as a contradiction sends 0 W.
F3: a drain ends on a refusal only when the battery will not give
(refused > 0, as _held_back()); one that will not take still drains.
Tested both signs; kills R8.

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

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/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,11 @@ 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).
+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,18 @@ 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}).
+        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 +1405,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(
@@ -2428,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(
-                    [c for c in claims if c.actuator == u]).contradiction:
-                now.add(inv.name)
-        self._charge_ended |= self._charging_was - now
-        self._charging_was = frozenset(now)
+            if res.phase in now and u in live and not pre_resolve(
+                    [c for c in claims if c.actuator == u]).contradiction \
+                    and not (res.phase == DRAIN and (inv.refused or 0) > 0):
+                now[res.phase].add(inv.name)
+        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
@@ -2456,9 +2480,116 @@ 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_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
+        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 | self._drain_ended:
+                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:
+                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 +4861,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,8 +4889,14 @@ 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)
+            # 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())
@@ -4775,6 +4914,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
@@ -4784,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,
@@ -4883,6 +5027,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_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",
-      _now["n2_maint"] == SAFETY78_N2_MAINT != A5DDDAD["n2_maint"])
+      "SAFETY-85, its end handed back by SAFETY-95",
+      _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_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_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")

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")