Deploy 0.16.0 to master (#179)

SAFETY-103 swing 'by about 800 W for about 5 s'; SAFETY-99 'late, by less
than 30 s'; MAINT-10 row: one-at-a-time is not new; SAFETY-96: drop the
test-only 800 W; SAFETY-90 and MAINT-10 rows trimmed to two sentences.

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

goodwe_controller/CHANGELOG.md

@@ -1,5 +1,49 @@
 # Changelog
 
+## 0.16.0
+
+**A minor bump: changed behaviour, no option added or removed, and one default
+and its meaning changed.** There are still 83 options. `maintenance_start_hour`
+used to be read as UTC and its default was 10. It is now a local hour (in Home
+Assistant's time zone) and its default is 19. The other changes below apply
+with one inverter or two, unless a line says "Two inverters".
+
+**Before you update (MAINT-10).** A saved `maintenance_start_hour` is kept, and
+it now means local time. If you set it in UTC to get an evening start (for
+example 17 for 19:00 in summer), set it back to the local hour (19). A saved
+10 now starts at 10:00, not 12:00 in summer. The new default 19 only reaches
+installs that have no hour saved.
+
+Contents of this release, in merge order since 0.15.0 (`3ed3302`):
+
+| | |
+|---|---|
+| `7ef7347` | After a gap in the P1 meter, the peak guard counts the missed time as high use and keeps the 15-minute period near the limit; before, it was blind until the next period. In the test, a 5 min meter silence at one inverter ended the period at 3.3 kW, against 4.8 kW on 0.15.0 ([[SAFETY-90]]) |
+| `1bf6b7f` | The maintenance start hour is now a local hour, and its default is 19 (was 10, read as UTC). The log says which time zone it used; with two inverters, both use that hour and still cycle one at a time ([[MAINT-10]]) |
+| `b2f5573` | Two inverters: when one battery's link drops, the other takes over in small steps, and control no longer stalls. The battery that comes back rejoins within 2 s, without a false "battery refuses" ([[SAFETY-96]]) |
+| `340c7ee` | Two inverters: when a drain ends (done, switched Off, timeout, lost link), the other battery covers the house right away, without jumping to 1.6 kW, and then moves in small steps. It no longer gives into a solar surplus ([[SAFETY-95]]) |
+
+### Known issues
+
+- A false "Battery is not keeping up" can still show when the peak guard pauses a maintenance charge ([[SAFETY-92]]).
+- A battery capped at a low limit with a perfectly flat reading can still trip the stale check ([[SAFETY-74]]).
+- A battery whose reading at rest is beyond about ±170 W is still not caught when it refuses, as in 0.14.0 ([[SAFETY-97]]).
+- The peak guard can still back off near the end of a 15-minute period when a meter reading is late, by less than 30 s ([[SAFETY-99]]).
+- A charge to 100 % can time out at 98 %. The add-on then counts the cycle as failed and drains the battery again the next day ([[MAINT-13]]).
+- Two inverters: the maintenance charge can switch on and off while the house is above the peak aim ([[SAFETY-88]]).
+- Two inverters: the "circulating" figure counts a battery's reading at rest ([[SAFETY-98]]).
+- Two inverters: after a drain ends, the other battery can swing by about 800 W for about 5 s, because some readings are old ([[SAFETY-103]]).
+- Two inverters: a drain that starts during a big step in house use can send the other battery to full power, and about 1 kW goes to the grid for about 10 s ([[SAFETY-104]]).
+- Two inverters: when a battery comes back during a maintenance charge, the grid can step by several kW at once ([[SAFETY-101]]).
+- One inverter: after a link drop, the battery's own safety stop to 0 W can be taken as a refusal, and the site stays at 0 W for up to 15 min ([[SAFETY-102]]).
+
+### If you are crossing from a version older than 0.10.0
+
+`## 0.10.0` below can refuse an old config: `slew_w` above its limit, and the
+withdrawn `price_provider` ids `engie_dynamic`, `totalenergies_dynamic` and
+`eneco_eddy`. Read it before you update. Run `tools/predeploy_check.py` with the
+bare options (`jq '.data.options'`) first.
+
 ## 0.15.0
 
 **A minor bump: changed behaviour, and no option changes.** No option was

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/config.yaml

@@ -1,5 +1,5 @@
 name: GoodWe RS485 Controller
-version: "0.15.0"
+version: "0.16.0"
 slug: goodwe_controller
 description: >-
   Drives a GoodWe ES/BP battery inverter over RS485 by emulating its smart

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