Merge pull request 'SAFETY-91: a held battery is drawn only after the others are full' (#165) from safety-91-held-battery-last into release/1.0

goodwe_controller/DECISIONS/SAFETY-91-held-battery-last.md

@@ -0,0 +1,74 @@
+# SAFETY-91 - a held battery is drawn only after the others are full
+
+**Status: DECIDED 2026-10-01 (owner: held battery last, SAFETY-25 option A
+kept).** Follows SAFETY-87. Tests: `test_safety91.py`.
+
+**The defect** (SAFETY-87 rig test, F1 and F2). At N>=2 a battery at or under
+its own peak reserve was held back (weight 0) only on cycles where the peak
+guard asked for power.
+- F1: B draining at 1000 W, A held at 15 %, house 5500. Once the grid was
+  near zero the loop set the target, A had its normal weight and B's drain
+  pin bound, so the loop drew A at 2500 W for the whole quarter (15.0 ->
+  10.7 %) while B stayed at 1000 W. In the next quarter A reached its
+  cut-out, and B stayed pinned at the peak floor (~2422 W), not its 2500 W.
+- F2: in a HOLD (A holding, B held), at a quarter's edge the guard saw a block
+  with no seconds left and asked for nothing for one cycle. The HOLD's
+  ceiling lapsed, A went back to 0 W and the loop's target went to B: 2500 W
+  for 1 s.
+
+**The fix** (`main.py`, N>=2 only).
+- `_held_back()`: the batteries at or under their own reserve while another
+  can still give (live, not refusing), graded as in `_peak_on_able()`. They
+  weigh 0 in every split where the site gives, not only for the peak. With
+  nobody able (all held, or the rest refusing or dropped) nobody is held back
+  and the split is as before.
+  Review: once held back, a battery stays so until its SoC reads more than 1
+  point above its reserve. A reading flipping 20/21 % otherwise swapped
+  ~1000 W between the batteries on each flip (grid +-500 W with A 10 s and
+  B 3 s lag). The guard's own permission (SAFETY-87) is unchanged.
+- `_drain_rise()`: while a battery is held back, a draining battery's pin
+  becomes a floor at its drain rate (`lo`), so it rises towards its rating
+  before the held one gives. Not for a held battery's own drain, not under a
+  site stop, not for a battery that is not live.
+- `tariff.next_quarter()`: past a block's end, before the next block's first
+  telegram, the guard grades the next quarter, projected from the held
+  reading (the averager credits it there). Not when that reading is too old
+  to be credited (`max_hold_s`).
+
+**What stays.** SAFETY-25 option A: once every able or draining battery is
+at full power, the loop still draws a held battery (`allocator._fill()`'s
+last-resort pass). With house 5500 and 2 x 2500 W, A still gives 2500 W
+after B is full. Nothing is placed on a battery at its cut-out: OPT-09's
+pack limit closes it (`hi = 0`), and the drain rise lets B take the rest.
+
+**Before -> after** (`test_safety91.py`, 2 x 2500 W, lagging battery 10 s):
+
+| case | before | after |
+|---|---|---|
+| B drain 1000, A held, house 4000 | A 2500, B 1000 | A 1500, B 2500 |
+| same, house 5500 | A 2500, B 1000 | A 2500, B 2500 (option A) |
+| cycles A asked while B not full | 612 of 900 | 0 |
+| A at its cut-out, house 5500: B | 1572 W mean | 2500 W, A 0 W |
+| HOLD, late telegram at the edge | A 0, B 2470 for 1 cycle | A 2050, B 0 |
+| idle, A held, house 2000 | A 1005, B 1005 | A 0, B 2020 |
+| N=1, every scenario | | byte-identical |
+
+**Why N>=2 only.** At N=1 there is no other battery to give first. The edge
+roll would also fix the N=1 one-cycle dip at a quarter's edge (SAFETY-86
+rig, "A dropped to 0 W for 1 s"), but this ticket keeps N=1 byte-identical;
+extending it is a one-line change for a later ticket.
+
+**Re-pinned.** `test_modes.py` `n2` and `n2_off`: from t=1600 battery A is
+at its 20 % reserve, so B carries the house and A gives last (100 of 540
+cycles differ). With SAFETY-87's held rule both are a5dddad's again, and the
+test grades that.
+Two fixtures changed with it: `test_safety87.py` part 10's "without the
+raise" run also turns off `_drain_rise()` (else B rises anyway), and
+`test_peakguard.py` 10g and `test_watchdog.py`'s two-inverter rig give
+their second battery a cut-out, as rule 8 requires (an unknown cut-out is held, so A alone would carry the cycle).
+
+**Not changed.** The peak guard's own rules (SAFETY-87). A drain with
+nobody held stays a pin (SAFETY-78). An Off battery beside a held one
+(SAFETY-87 Open). The first cycle after start, before any quarter is
+measured, can still put the loop's first step on a held battery in a HOLD.
+SAFETY-88, -89, -90, -92, -93, -94.

goodwe_controller/DOCS.md

@@ -1640,7 +1640,7 @@ help - see "Why the tuning is what it is".
 | `maintenance_enabled` | false | Enable the monthly cycle |
 | `maintenance_interval_days` | 28 | Minimum gap between cycles |
 | `maintenance_start_hour` | 10 | Hour of day a due cycle begins |
-| `maintenance_discharge_w` | 2000 | Drain rate (exports the surplus). The peak guard may ask for more (SAFETY-83; at N≥2 while another battery is at its peak reserve, SAFETY-87). Above `max_w` it is **clamped** to `max_w` — the drain phase SETs the setpoint and the arbiter bounds it, saying so on the status page (`limited by supervised(max_w)`). Warned about, not refused: the cycle still completes, just slower |
+| `maintenance_discharge_w` | 2000 | Drain rate (exports the surplus). The peak guard may ask for more (SAFETY-83; at N≥2 while another battery is at its peak reserve, SAFETY-87), and at N≥2 so may the house while another battery is at its peak reserve (SAFETY-91). Above `max_w` it is **clamped** to `max_w` — the drain phase SETs the setpoint and the arbiter bounds it, saying so on the status page (`limited by supervised(max_w)`). Warned about, not refused: the cycle still completes, just slower |
 | `maintenance_charge_w` | 2000 | Charge ceiling, capped again by peak headroom. Must stay **at or below `max_w`** or the cheap-window shaping (`hi = -maintenance_charge_w`) contradicts the supervised limit (`lo = -max_w`), the arbiter resolves it to **0 W**, and maintenance silently never charges in a cheap window — **the add-on refuses to start** |
 | `maintenance_soc_floor` | 11 | Drain target — stay just above the inverter's own floor. ⚠️ **This is not `peak_soc_floor`.** It is the depth the monthly recalibration deliberately goes to, once, on purpose; `peak_soc_floor` is the depth a money-saving discharge may never reach on any day. The right values differ (11 vs 20 as shipped), which is why they are two keys |
 | `maintenance_soc_target` | 99 | Charge target |
@@ -1667,6 +1667,9 @@ others only add power when the house needs more; none charges while it drains
 (SAFETY-78). Below the drain rate the rest is exported, as with one battery.
 If the house would still go over the peak limit, the battery being emptied
 gives more than its drain rate, with one battery too (SAFETY-83).
+A battery at its peak reserve gives last: the others give their full power
+first, and the battery being emptied rises above its drain rate before it
+(SAFETY-91).
 In a cheap window the battery being filled charges and the others only add
 charge; none gives while it charges. The charge is cut to what keeps the
 15-minute peak at the limit minus `peak_margin_w` (SAFETY-84).
@@ -1809,7 +1812,9 @@ exactly like the SoC drift the monthly maintenance cycle exists to correct.
   It will discharge straight past this number and keep going until the pack's own
   BMS cut-out stops it. What you are promised by `peak_soc_floor: 20` is *"the
   capacity-peak guard will not spend the pack for money below 20 %"*, and
-  **nothing more**. See below for what that costs on a real evening.
+  **nothing more**. See below for what that costs on a real evening. With two
+  or more batteries, one under it is used last: only once the others give
+  their full power (SAFETY-91).
 - **When it binds you are told**, once per episode, with notification
   `goodwe_peak_soc_floor` and a red status-page row reading *Battery at its
   peak reserve* (`PEAK_GUARD_FLOOR_LIMITED` in the log): the quarter **will**

goodwe_controller/app/arbiter.py

@@ -737,9 +737,9 @@ def resolve(claims: list, *, units=None, weights=None, live=None,
     gate, held = _unit_modes(out, site_claims, bounds, live)
     if callable(weights):
         weights = weights(out.target_w)
-    # SAFETY-87: units under their peak reserve weigh 0 while the site gives,
-    # so the others carry the peak floor. An Off battery gives the peak's
-    # shortfall, so it goes before them when no other unit is left.
+    # SAFETY-87/91: units under their peak reserve weigh 0 while the site
+    # gives, so the others carry the target first. An Off battery gives the
+    # peak's shortfall, so it goes before them when no other unit is left.
     reserve = set(reserve) if out.target_w > 0.0 else set()
     if held and reserve and not set(bounds) - held - reserve:
         held = set()

goodwe_controller/app/main.py

@@ -91,7 +91,7 @@ from .price import required_spread as price_required_spread
 from .price import uncertainty_of as price_uncertainty_of
 from .price_history import PriceHistory
 from .static_tariff import resolve_tariff, static_status
-from .tariff import PeakStore, PeakTracker
+from .tariff import PeakStore, PeakTracker, next_quarter
 from .store import migrate_maint_keys, migrate_unit_keys
 from .validate import (FIRMWARE_INGEST_VETO_S, ConfigRefused, refuse,
                        validate_startup_config, validate_unit_ranges)
@@ -992,6 +992,8 @@ class Controller:
         # SAFETY-78: names of draining packs that cannot give power now, so
         # the "waits" line is logged once, not every cycle.
         self._drain_waits = set()
+        # SAFETY-91: last cycle's held-back batteries (`_held_back()`).
+        self._held_was = frozenset()
         # 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.
@@ -2567,6 +2569,58 @@ class Controller:
             return frozenset()
         return frozenset(g.held)
 
+    def _held_back(self) -> frozenset:
+        """SAFETY-91, N>=2: the batteries at or under their own peak reserve
+        while another can still give (live, not refusing). They weigh 0 in
+        every split where the site gives, so they give last (SAFETY-25
+        option A: still drawn once the others are full). One held back
+        stays so until its SoC reads more than 1 point above its reserve, so
+        a reading flipping across it does not swap the split each cycle."""
+        if len(self.inverters) < 2:
+            return frozenset()
+        floor = float(self.o.get(
+            "peak_soc_floor", peakguard.PEAK_OPTION_DEFAULTS["peak_soc_floor"]))
+        ok, held = peakguard.reserve_split(self.packs, floor)
+        held = set(held) | {
+            p.name for p in ok if p.name in self._held_was
+            and float(p.soc) <= peakguard.pack_floor_pct(p, floor) + 1.0}
+        live = set(self.live_units())
+        if not held or not any(
+                p.name not in held and inv.name in live
+                and not (inv.refused or 0) > 0
+                for p, inv in zip(self.packs, self.inverters)):
+            self._held_was = frozenset()
+        else:
+            self._held_was = frozenset(held)
+        return self._held_was
+
+    def _drain_rise(self, claims, maint_results) -> None:
+        """SAFETY-91, N>=2: while a battery is held back (`_held_back()`), a
+        draining battery's pin becomes a floor at its drain rate, so it rises
+        towards its rating before the held one gives. A site stop still
+        decides alone. DECISIONS/SAFETY-91-held-battery-last.md."""
+        held = self._held_back()
+        if not held or any(c.kind == SET and c.actuator == ACTUATOR_SITE
+                           and c.priority > P_PEAK for c in claims):
+            return
+        live = set(self.live_units())
+        for inv, res in zip(self.inverters, maint_results):
+            u = self.actuator_of("maintenance", inv)
+            if res.phase != DRAIN or not res.owns_setpoint \
+                    or inv.name in held or u not in live:
+                continue
+            b = pre_resolve([c for c in claims if c.actuator == u])
+            p = (None if b.contradiction or b.pinned_by != "maintenance"
+                 else allocator.pin_w(b))
+            if p is None or not p > 0.0:
+                continue
+            for i, c in enumerate(claims):
+                if c.source == "maintenance" and c.kind == SET \
+                        and c.actuator == u:
+                    claims[i] = Claim.limit("maintenance", P_MAINTENANCE,
+                                            lo=p, reason=c.reason,
+                                            actuator=u)
+
     def _peak_over_drain(self, claims, maint_results) -> None:
         """SAFETY-83: at N=1 a peak floor above the drain rate outranks it.
 
@@ -4082,6 +4136,11 @@ class Controller:
         # gives for SAFETY-18 on a transport that cannot see `unchanged_s`.
         self._peak_proj = (None if not self.p1_enabled
                            else self.peak.project(self.p1.averager, now))
+        # SAFETY-91: at N>=2 the quarter's edge is graded on the next quarter,
+        # so a HOLD's ceiling does not lapse for a cycle onto a held battery.
+        if self._peak_proj is not None and len(self.inverters) > 1:
+            self._peak_proj = next_quarter(self._peak_proj,
+                                           self.p1.averager.max_hold_s)
         self.peak_guard = None if not self.p1_enabled else peakguard.decide(
             self._peak_proj,
             # ⚠️ THE SITE SUM. The guard sizes a floor from what the battery is
@@ -4482,6 +4541,8 @@ class Controller:
             claims.append(Claim.limit("hold", P_MAINTENANCE, hi=hold_hi,
                                       reason="maintenance hold",
                                       actuator=self.actuator_of("hold")))
+        # --- SAFETY-91: a drain rises before a held battery gives ------------
+        self._drain_rise(claims, maint_results)
         # --- SAFETY-84: a cheap-window charge is a ceiling over the site ----
         charge = self._charge_ceiling(claims)
         if charge is not None:
@@ -4680,7 +4741,7 @@ class Controller:
         while True:
             resolution = resolve(claims, units=self.units, live=_live,
                                  weights=self.pack_weights, tick=self.alloc_tick,
-                                 reserve=self._peak_reserve())
+                                 reserve=self._held_back())
             # SAFETY-73: a probe pin stands only while the resolved site target
             # asks at least PROBE_W its way. Below that a sibling would carry
             # the opposite direction (circulation), so the pin goes back to the

goodwe_controller/app/tariff.py

@@ -822,6 +822,22 @@ class PeakTracker:
             held_w=averager.held_net_w)
 
 
+def next_quarter(p: Projection, max_hold_s: float) -> Projection:
+    """SAFETY-91: `p` past its block's end, before the next block's first
+    telegram: the next quarter, projected from the held reading (the averager
+    credits it there). `p` itself otherwise, or when that reading is too old
+    to be credited. DECISIONS/SAFETY-91-held-battery-last.md."""
+    if p.block_start is None or p.elapsed_s < QUARTER_S \
+            or p.elapsed_s * (1.0 - p.covered_frac) > float(max_hold_s):
+        return p
+    held = max(p.held_w, 0.0)
+    return Projection(projected_w=held, elapsed_w=held,
+                      elapsed_s=p.elapsed_s - QUARTER_S, covered_frac=1.0,
+                      complete=True,
+                      block_start=p.block_start + timedelta(seconds=QUARTER_S),
+                      held_w=p.held_w)
+
+
 @dataclass(frozen=True)
 class _ClosedBlock:
     """A block reconstructed from a checkpoint. Shaped like `p1.QuarterBlock` so

goodwe_controller/test_modes.py

@@ -129,6 +129,19 @@ finally:
 # cd4d9cb5b7e29ff4ab4fe65dab5106683d8a599f09323fbdee4018ad9b28fcd1.
 SAFETY78_N2_MAINT = \
     "1afcd46b1b987868b110d2b0f4a7555848c9aa4728a836d4a2ce2a2107268497"
+# 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
+# a5dddad's again, which the check below grades.
+SAFETY91_N2 = \
+    "9535c08f556d6d50bec8fb0872445173480c29b10df1a7c437ef73d6ddb2814d"
+_real_hb = app_main.Controller._held_back
+app_main.Controller._held_back = lambda self: self._peak_reserve()
+try:
+    _n2_s87 = rig.digest(rig.run(**rig.SCENARIOS["n2"])[0])
+finally:
+    app_main.Controller._held_back = _real_hb
+_S91 = ("n2", "n2_off", "n2_maint")
 for k in A5DDDAD:
     print(f"    {k:7} {'same' if _now[k] == A5DDDAD[k] else 'DIFFERS'} "
           f"{_now[k][:16]}")
@@ -136,10 +149,14 @@ check("⭐ at N=1 and N=2, with the option absent and with it false, every "
       "command, reason and allocation record over 540 cycles each is "
       "byte-identical to a5dddad (house steps, a solar surplus, a battery "
       "at its cut-out level, at N=2 a refusal with probes, both batteries "
-      "due for maintenance; n2_maint apart, below), and so is the arbiter "
-      "on 6000 claim lists",
-      {k: v for k, v in _now.items() if k != "n2_maint"}
-      == {k: v for k, v in A5DDDAD.items() if k != "n2_maint"})
+      "due for maintenance; n2, n2_off and n2_maint apart, below), and so "
+      "is the arbiter on 6000 claim lists",
+      {k: v for k, v in _now.items() if k not in _S91}
+      == {k: v for k, v in A5DDDAD.items() if k not in _S91})
+check("...n2 and n2_off are the SAFETY-91 digest (A at its reserve gives "
+      "last), and SAFETY-87's held rule gives a5dddad's back",
+      _now["n2"] == _now["n2_off"] == SAFETY91_N2 != A5DDDAD["n2"]
+      and _n2_s87 == A5DDDAD["n2"])
 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",

goodwe_controller/test_peakguard.py

@@ -3132,6 +3132,10 @@ check("...and between them the nineteen appends account for exactly the "
 # through this file's reading of main.py. It grades three of the eight appends.
 # The static pin above is what covers the other five.
 _b2, _restore_b2 = second_inverter(c_src)
+# SAFETY-91: B states a cut-out, as every roster entry must (rule 8). Without
+# one it is held back at its reserve and gives last, so A alone would carry
+# this cycle's 500 W and the broadcast check below could not see a split.
+_b2.bms_soc_floor = c_src.inverters[0].bms_soc_floor
 _n2_writes_before = len(c_src.hass.per_entity)
 try:
     _, _cl2 = cycle_claims(lambda: run_cycle(c_src))

goodwe_controller/test_safety87.py

@@ -559,11 +559,14 @@ def _drain_run():
 
 DR = _drain_run()
 _real_dor = Controller._drain_over_reserve
+_real_rise = Controller._drain_rise     # SAFETY-91 would raise B as well
 Controller._drain_over_reserve = lambda self, claims, results: None
+Controller._drain_rise = lambda self, claims, results: None
 try:
     DR_OLD = _drain_run()
 finally:
     Controller._drain_over_reserve = _real_dor
+    Controller._drain_rise = _real_rise
 
 
 def _a_for_peak(rows):

goodwe_controller/test_safety91.py

@@ -0,0 +1,452 @@
+"""Runnable check for SAFETY-91: a held battery is drawn only after the others
+are full. `python3 test_safety91.py`
+
+At N>=2 a battery at or under its peak reserve was held back only while the
+peak guard asked for power. The self-consumption loop drew it at its full
+rating while a draining battery stayed at its drain rate (SAFETY-87 rig test
+F1), and at a quarter's edge in a HOLD the held battery gave 2500 W for one
+cycle (F2). Now a held battery weighs 0 whenever the site gives, a draining
+battery rises above its drain rate before it, and the edge is graded on the
+next quarter. SAFETY-25 option A stands: once the others are full the loop
+may still draw it. Drives the REAL Controller on a lagging battery (10 s, as
+on the rig). DECISIONS/SAFETY-91-held-battery-last.md.
+
+1. B draining, A held, house over B's rating: B rises to full before A gives.
+2. A held reaching its cut-out: nothing is placed on it, B gives its rating.
+3. HOLD across a quarter edge: the held battery gives nothing at the edge.
+4. Idle, A held: B carries the house first; both held or B refusing: as before;
+   a SoC reading flipping across the reserve does not swap the split.
+5. The rules at their boundaries (claim lists, N=3, a site stop).
+6. next_quarter() at its boundaries.
+7. N=1 is untouched.
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import logging
+import math
+import os
+import sys
+import time
+import types
+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 safety84_identity as s84                                # noqa: E402
+from app import peakguard                                      # noqa: E402
+from app.arbiter import (Claim, P_MAINTENANCE, P_SAFETY,        # noqa: E402
+                         resolve)
+from app.main import Controller                                # noqa: E402
+from app.maintenance import DRAIN, HOLD, IDLE, MaintResult     # noqa: E402
+from app.p1 import make_sample, split_signed                   # noqa: E402
+from app.store import unit_key                                 # noqa: E402
+from app.tariff import QUARTER_S, Projection, next_quarter     # noqa: E402
+
+fails = []
+A, B = "inverter-a", "inverter-b"
+FULL = 2500.0
+CUT = 10.0              # bms_soc_floor of both test batteries
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+def lagging(fn, lag=10.0, tick=1.0, lags=None):
+    """`fn` on plantsim's battery (800 W/s ramp, `lag` s first-order lag, or
+    `lags[u]`), a 1 s meter tick, and a BMS that stops discharge at its
+    cut-out."""
+    real_p, real_t = rig._Hass.power, rig.TICK
+
+    def power(self, u):
+        st = self.__dict__.setdefault("_plant", {})
+        a, tgt, tl = st.get(u, (0.0, 0.0, self.t))
+        for _ in range(int(round((self.t - tl) / 0.1))):
+            d = self.cmd[u] - tgt
+            tgt += math.copysign(min(abs(d), 80.0), d)
+            a += (tgt - a) * (1 - math.exp(-0.1 / (lags or {}).get(u, lag)))
+        st[u] = (a, tgt, self.t)
+        return 0.0 if self.soc(u, self.t) <= CUT and a > 0 else float(round(a))
+    rig._Hass.power, rig.TICK = power, tick
+    try:
+        return fn()
+    finally:
+        rig._Hass.power, rig.TICK = real_p, real_t
+
+
+def store(phase_of, n=2):
+    """`phase_of`: {name: phase}; the others have just done their cycle."""
+    s = rig._Store()
+    for u in (A, B)[:n]:
+        key = (lambda k: k) if n == 1 else (lambda k: unit_key(u, k))
+        if u in phase_of:
+            s.data[key("phase")] = phase_of[u]
+            s.data[key("phase_started")] = rig.BASE.isoformat()
+        else:
+            s.data[key("last_completed")] = rig.BASE.isoformat()
+    return s
+
+
+def run(house, phase_of, socs, dur, late=0.0, kwh=15.36, n=2, read=None,
+        lags=None):
+    """Rows and controller. Levels follow each battery's measured power; the
+    P1 telegram read at each cycle is `late` s old. `read(u, t, level)` is
+    the SoC the controller reads (default the level)."""
+    soc, last, levels = dict(socs), [0.0], {}
+    grid = []
+
+    def before(c, t):
+        grid.append((t, c.hass.grid()))
+        for u in soc:
+            soc[u] = max(CUT, soc[u] - c.hass.power(u) * (t - last[0])
+                         / 3600.0 / (kwh * 10.0))
+        last[0], levels[t] = t, dict(soc)
+        imp, exp = split_signed(c.hass.grid())
+        c.p1.submit(make_sample("ha_signed", imp, exp, phases=1,
+                                ingest_ts=rig.BASE + timedelta(seconds=t - late),
+                                ingest_mono=time.monotonic()))
+    rows, c = lagging(lambda: s84.run(
+        n, load=house if callable(house) else (lambda t: house), cheap=False,
+        store=store(phase_of, n), before=before, dur=dur,
+        soc=(lambda u, t: soc[u]) if read is None
+        else (lambda u, t: read(u, t, soc[u])),
+        extra={"maintenance_discharge_w": 1000,
+               "maintenance_hold_min": 600}), lags=lags)
+    c.levels, c.grid = levels, grid
+    return rows, c
+
+
+def without_fix(fn):
+    """release/1.0 8c438ab: held only for the peak, a drain stays pinned, the
+    quarter's edge is not rolled over."""
+    real = (Controller._held_back, Controller._drain_rise)
+    import app.main as m
+    real_nq = m.next_quarter
+    Controller._held_back = lambda self: self._peak_reserve()
+    Controller._drain_rise = lambda self, claims, results: None
+    m.next_quarter = lambda p, max_hold_s: p
+    try:
+        return fn()
+    finally:
+        Controller._held_back, Controller._drain_rise = real
+        m.next_quarter = real_nq
+
+
+def cmd(r):
+    return dict(r[2])
+
+
+def seg(rows, lo, hi):
+    """Mean A and B commands over [lo, hi) s."""
+    rs = [r for r in rows if lo <= r[0] < hi]
+    return (round(sum(cmd(r)["a"] for r in rs) / len(rs)),
+            round(sum(cmd(r)["b"] for r in rs) / len(rs)))
+
+
+def a_before_b(rows, t0=0.0):
+    """Cycles where held A is asked to give while B is not at full power."""
+    return sum(1 for r in rows if r[0] >= t0
+               and cmd(r)["a"] > 0.0 and cmd(r)["b"] < FULL)
+
+
+def unmet(rows, lo, hi):
+    return sum(1 for r in rows if lo <= r[0] < hi and r[7] and "UNMET" in r[7])
+
+
+# ============================================================================ #
+print("1. B draining at 1000 W, A held at 15 %, house over B's rating")
+STEPS = (800.0, 4000.0, 5500.0)
+
+
+def steps(t):
+    return STEPS[min(int(t // 300), 2)]
+
+
+DR = run(steps, {B: DRAIN}, {"a": 15.0, "b": 60.0}, 900.0)
+DR_OLD = without_fix(lambda: run(steps, {B: DRAIN}, {"a": 15.0, "b": 60.0},
+                                 900.0))
+for tag, (rows, c) in (("after", DR), ("before", DR_OLD)):
+    print(f"    {tag}: house 800 A/B {seg(rows, 60, 300)}, 4000 "
+          f"{seg(rows, 360, 600)}, 5500 {seg(rows, 660, 900)}; A asked while "
+          f"B not full: {a_before_b(rows)} cycles; UNMET "
+          f"{unmet(rows, 0, 900)}")
+check("⭐ A (held) is never asked to give while B has room: every cycle that "
+      "puts power on A has B at its full 2500 W",
+      a_before_b(DR[0]) == 0 and DR[1].maints[1].phase == DRAIN)
+check("...house 4000: B rises from its 1000 W drain rate to 2500 W and A "
+      "gives only the 1500 W rest (before: A 2500, B 1000)",
+      seg(DR[0], 360, 600)[1] >= 2480 and seg(DR[0], 360, 600)[0] <= 1520
+      and seg(DR_OLD[0], 360, 600)[0] >= 2480
+      and seg(DR_OLD[0], 360, 600)[1] <= 1020)
+check("...house 800: B still drains at its 1000 W rate (the rest is "
+      "exported, as before) and A gives nothing",
+      all(cmd(r)["b"] >= 1000.0 and cmd(r)["a"] == 0.0
+          for r in DR[0] if 60 <= r[0] < 300))
+check("...house 5500 (over both ratings): option A stands, A is drawn as a "
+      "last resort once B is full, and nothing is left UNMET",
+      seg(DR[0], 660, 900) == (2500, 2500) and unmet(DR[0], 0, 900) == 0)
+check("non-vacuous: before the fix A was asked while B had room",
+      a_before_b(DR_OLD[0]) >= 100)
+
+# ============================================================================ #
+print("\n2. the held battery reaches its cut-out (house 5500)")
+CO = run(5500.0, {B: DRAIN}, {"a": 11.0, "b": 60.0}, 1200.0)
+CO_OLD = without_fix(lambda: run(5500.0, {B: DRAIN}, {"a": 11.0, "b": 60.0},
+                                 1200.0))
+
+
+def after_cut(rows, c):
+    """Rows once A reads its cut-out (its level follows its power)."""
+    return [r for r in rows if c.levels[r[0]]["a"] <= CUT]
+
+
+_ac, _ac_old = after_cut(*CO), after_cut(*CO_OLD)
+print(f"    after A's cut-out: B {seg(_ac, 0, 9e9)[1]} W over {len(_ac)} "
+      f"cycles (before: {seg(_ac_old, 0, 9e9)[1]} W); A soc "
+      f"{CO[1].inverters[0].soc:.2f} %")
+check("⭐ at its cut-out nothing is placed on A (0 W on every cycle), B "
+      "gives its full 2500 W and nothing is left UNMET on the empty A",
+      CO[1].inverters[0].soc <= CUT and len(_ac) >= 100
+      and all(cmd(r)["a"] == 0.0 and cmd(r)["b"] == FULL for r in _ac)
+      and unmet(CO[0], 600, 1200) == 0)
+check("non-vacuous: before the fix B stayed pinned under its rating on most "
+      "of those cycles (the rig saw ~2422 W)",
+      len(_ac_old) >= 100
+      and sum(1 for r in _ac_old if cmd(r)["b"] < FULL) > len(_ac_old) / 2)
+
+# ============================================================================ #
+print("\n3. HOLD across a quarter edge (A full holding, B held at 15 %)")
+EDGE = run(5500.0, {A: HOLD}, {"a": 99.0, "b": 15.0}, 1000.0, late=0.5)
+EDGE_OLD = without_fix(lambda: run(5500.0, {A: HOLD}, {"a": 99.0, "b": 15.0},
+                                   1000.0, late=0.5))
+
+
+def at_edge(rows):
+    return [(r[0], cmd(r)["a"], cmd(r)["b"]) for r in rows
+            if QUARTER_S - 1 <= r[0] <= QUARTER_S + 1]
+
+
+print(f"    edge after {at_edge(EDGE[0])}, before {at_edge(EDGE_OLD[0])}")
+check("⭐ the telegram at the edge comes late (0.5 s): B gives nothing on any "
+      "cycle (the first cycle, before any quarter is measured, aside), A "
+      "keeps giving and stays in its HOLD",
+      all(cmd(r)["b"] == 0.0 for r in EDGE[0] if r[0] > 0)
+      and all(a > 2000.0 for _t, a, _b in at_edge(EDGE[0]))
+      and EDGE[1].maints[0].phase == HOLD)
+check("non-vacuous: before the fix B gave over 2000 W at the edge (F2)",
+      any(b > 2000.0 for _t, _a, b in at_edge(EDGE_OLD[0])))
+
+# ============================================================================ #
+print("\n4. idle, house 2000 W")
+ID = run(2000.0, {}, {"a": 15.0, "b": 60.0}, 300.0)
+ID_OLD = without_fix(lambda: run(2000.0, {}, {"a": 15.0, "b": 60.0}, 300.0))
+print(f"    A held: after {seg(ID[0], 120, 300)}, before "
+      f"{seg(ID_OLD[0], 120, 300)}")
+check("⭐ A held, B able: B carries the house alone; A is asked only on the "
+      "cycles where the lagging B is already at its full command",
+      a_before_b(ID[0]) == 0 and seg(ID[0], 120, 300)[0] == 0
+      and seg(ID[0], 120, 300)[1] >= 1950)
+check("non-vacuous: before the fix the weights gave A a share",
+      seg(ID_OLD[0], 120, 300)[0] > 50)
+BOTH = run(2000.0, {}, {"a": 15.0, "b": 14.0}, 300.0)
+BOTH_OLD = without_fix(lambda: run(2000.0, {}, {"a": 15.0, "b": 14.0}, 300.0))
+check("...both held: nothing to give first, the split is as before",
+      [r[:3] for r in BOTH[0]] == [r[:3] for r in BOTH_OLD[0]])
+
+
+def held_back(socs, refused=(0, 0), live=(A, B)):
+    c = ID[1]
+    for inv, s, r in zip(c.inverters, socs, refused):
+        inv.soc, inv.refused = s, r
+    c.live_units = lambda: live
+    try:
+        return c._held_back()
+    finally:
+        del c.live_units
+        for inv in c.inverters:
+            inv.refused = 0
+
+
+check("...held back only while another can give: A held beside B able; "
+      "nothing with B refusing to give, B dropped, or both under",
+      held_back((15.0, 60.0)) == {A} and held_back((15.0, 60.0), (0, 1))
+      == frozenset() and held_back((15.0, 60.0), live=(A,)) == frozenset()
+      and held_back((15.0, 14.0)) == frozenset()
+      and held_back((15.0, 60.0), (0, -1)) == {A})
+
+# A's SoC READING flips 20/21 % across its 20 % reserve every 10 s, and B
+# follows its command faster (3 s) than A (10 s), as two different inverters.
+def _flip(u, t, level):
+    return (20.0 + int(t // 10) % 2) if u == "a" else level
+
+
+def _a_jumps(rows, t0):
+    rs = [r for r in rows if r[0] >= t0]
+    return sum(1 for p, q in zip(rs, rs[1:])
+               if abs(cmd(p)["a"] - cmd(q)["a"]) > 300.0)
+
+
+def _grid_rms(c, t0):
+    g = [w + 10.0 for t, w in c.grid if t >= t0]
+    return math.sqrt(sum(w * w for w in g) / len(g))
+
+
+_lags = {"a": 10.0, "b": 3.0}
+FL = run(2000.0, {}, {"a": 20.0, "b": 60.0}, 600.0, read=_flip, lags=_lags)
+_real_was = Controller.__dict__["_held_back"]
+
+
+def _no_hyst(self):
+    self._held_was = frozenset()
+    return _real_was(self)
+
+
+Controller._held_back = _no_hyst
+try:
+    FL_OLD = run(2000.0, {}, {"a": 20.0, "b": 60.0}, 600.0, read=_flip,
+                 lags=_lags)
+finally:
+    Controller._held_back = _real_was
+print(f"    A's reading flipping 20/21 %: A jumps {_a_jumps(FL[0], 120)}, "
+      f"grid rms {_grid_rms(FL[1], 120):.0f} W (no hysteresis: "
+      f"{_a_jumps(FL_OLD[0], 120)}, {_grid_rms(FL_OLD[1], 120):.0f} W)")
+check("⭐ a SoC reading flipping across the reserve does not swap the split: "
+      "A stays held back (0 W) and the grid stays near its target",
+      _a_jumps(FL[0], 120) == 0 and seg(FL[0], 120, 600)[0] == 0
+      and _grid_rms(FL[1], 120) < 30.0)
+check("non-vacuous: without the 1-point hysteresis the split swaps on "
+      "every flip and the grid swings",
+      _a_jumps(FL_OLD[0], 120) >= 20 and _grid_rms(FL_OLD[1], 120) > 100.0)
+REL = run(2000.0, {}, {"a": 20.0, "b": 60.0}, 300.0,
+          read=lambda u, t, lv: (20.0 if t < 100 else 21.0 if t < 200
+                                 else 22.0) if u == "a" else lv)
+check("...held back at 20 %, still at 21 %, released at 22 % (more than 1 "
+      "point above its reserve)",
+      seg(REL[0], 40, 100)[0] == 0 and seg(REL[0], 140, 200)[0] == 0
+      and seg(REL[0], 240, 300)[0] > 500)
+ID[1]._held_was = frozenset({A, B})
+check("...a battery kept held back by the hysteresis is not one that can "
+      "give: A at 20 % and B kept at 21 % hold nobody back",
+      held_back((20.0, 21.0)) == frozenset())
+
+# ============================================================================ #
+print("\n5. the rules at their boundaries")
+
+
+def stub(names, held, live=None):
+    """A Controller stand-in for `_drain_rise()`, on SIMULATED units."""
+    s = types.SimpleNamespace(
+        inverters=[types.SimpleNamespace(name=n) for n in names],
+        actuator_of=lambda src, inv=None: inv.name,
+        live_units=lambda: tuple(live or names),
+        _held_back=lambda: frozenset(held))
+    return s
+
+
+def rise(names, held, target, drains, extra=(), live=None):
+    """`_drain_rise()` then the split: (claims kinds, shares)."""
+    cl = [Claim.limit("supervised", P_SAFETY, -FULL, FULL, actuator=n)
+          for n in names]
+    cl += [Claim.set("maintenance", P_MAINTENANCE, w, DRAIN, actuator=n)
+           for n, w in drains.items()]
+    cl += list(extra)
+    res = [MaintResult(DRAIN, True, drains[n], False, []) if n in drains
+           else MaintResult(IDLE, False, None, False, []) for n in names]
+    Controller._drain_rise(stub(names, held, live), cl, res)
+    r = resolve(cl + [Claim.set("loop", 10, target)], units=tuple(names),
+                live=set(live or names), reserve=held)
+    kinds = {c.actuator: c.kind for c in cl if c.source == "maintenance"}
+    return kinds, {u: round(v) for u, v in r.shares.items()}
+
+
+_n3 = rise(("x", "y", "z"), {"z"}, 1500.0, {"x": 1000.0})
+print(f"    N=3, x draining 1000, y able, z held, 1500 W: {_n3[1]}")
+check("⭐ N=3: the draining battery keeps its drain rate as a floor (x >= "
+      "1000 W) while sharing with the able one; the held one gives nothing",
+      _n3[0] == {"x": "limit"} and _n3[1]["x"] >= 1000 and _n3[1]["z"] == 0
+      and sum(_n3[1].values()) == 1500)
+_stop = rise((A, B), {A}, 3000.0, {B: 1000.0},
+             (Claim.set("safety", P_SAFETY, 0.0, "stopped"),))
+check("⭐ a site stop decides alone: the drain stays a pin, nothing rises",
+      _stop[0] == {B: "set"} and _stop[1] == {A: 0, B: 0})
+_hd = rise((A, B), {A}, 5000.0, {A: 1000.0})
+check("...a held battery's own drain stays a pin (it is not raised for the "
+      "loop): A 1000 W, B 2500 W",
+      _hd[0] == {A: "set"} and _hd[1] == {A: 1000, B: 2500})
+_nh = rise((A, B), set(), 3000.0, {B: 1000.0})
+check("...with nothing held the drain stays a pin, as before (SAFETY-78)",
+      _nh[0] == {B: "set"} and _nh[1] == {A: 2000, B: 1000})
+_dead = rise((A, B), {A}, 3000.0, {B: 1000.0}, live=(A,))
+check("...a draining battery that is not live is not touched",
+      _dead[0] == {B: "set"})
+
+# ============================================================================ #
+print("\n6. next_quarter()")
+T0 = rig.BASE
+
+
+def proj(elapsed, covered=1.0, held=3000.0, start=T0):
+    return Projection(projected_w=2000.0, elapsed_w=1900.0, elapsed_s=elapsed,
+                      covered_frac=covered, complete=covered >= 1.0,
+                      block_start=start, held_w=held)
+
+
+_in = proj(899.0)
+_past = next_quarter(proj(900.4, covered=899.9 / 900.4), 30.0)
+check("⭐ inside its block a projection is unchanged (the same object)",
+      next_quarter(_in, 30.0) is _in)
+check("⭐ past its end with a fresh held reading: the next quarter, "
+      "projected from that reading",
+      _past.block_start == T0 + timedelta(seconds=QUARTER_S)
+      and abs(_past.elapsed_s - 0.4) < 1e-9 and _past.projected_w == 3000.0
+      and _past.covered_frac == 1.0
+      and peakguard.blindness(_past, 30.0, False) is None)
+_old = proj(960.0, covered=880.0 / 960.0)
+_edge29 = proj(901.0, covered=871.5 / 901.0)     # 29.5 s uncovered
+_edge31 = proj(901.0, covered=870.5 / 901.0)     # 30.5 s uncovered
+check("...the held reading is credited up to max_hold_s and no further "
+      "(29.5 s uncovered: rolled; 30.5 s: unchanged)",
+      next_quarter(_edge29, 30.0) is not _edge29
+      and next_quarter(_edge31, 30.0) is _edge31)
+check("...a held reading too old to be credited, no block, or an export: "
+      "unchanged, or projected at 0 W offtake",
+      next_quarter(_old, 30.0) is _old
+      and next_quarter(proj(950.0, start=None), 30.0).block_start is None
+      and next_quarter(proj(900.5, held=-800.0), 30.0).projected_w == 0.0)
+
+# ============================================================================ #
+print("\n7. N=1")
+for _name, _args in (("HOLD, late telegram at the edge",
+                      (5500.0, {A: HOLD}, {"a": 99.0}, 1000.0, 0.5)),
+                     ("drain under the reserve, house 4000",
+                      (4000.0, {A: DRAIN}, {"a": 15.0}, 600.0, 0.5))):
+    _new = run(*_args, n=1)
+    _old = without_fix(lambda: run(*_args, n=1))
+    print(f"    {_name}: A at the edge "
+          f"{[cmd(r)['a'] for r in _new[0] if 899 <= r[0] <= 901]}")
+    check(f"⭐ N=1, {_name}: every row (claims, commands, the guard) is "
+          "identical with and without the fix",
+          _new[1].maints[0].phase == _args[1][A] and len(_new[0]) > 500
+          and _new[0] == _old[0])
+_one = types.SimpleNamespace(inverters=[object()])
+_cl = [Claim.set("maintenance", P_MAINTENANCE, 1000.0, DRAIN)]
+check("⭐ at one inverter nothing is held back and no pin moves (the N=1 "
+      "identity digests in test_safety78/79/83/84/86/87 stay pinned)",
+      Controller._held_back(_one) == frozenset()
+      and Controller._drain_rise(types.SimpleNamespace(
+          inverters=[object()], _held_back=lambda: frozenset()), _cl,
+          [MaintResult(DRAIN, True, 1000.0, False, [])]) is None
+      and _cl[0].kind == "set")
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED: " + ", ".join(fails))
+    sys.exit(1)
+print("all SAFETY-91 checks passed")

goodwe_controller/test_watchdog.py

@@ -1631,7 +1631,10 @@ def _sign_site(hass, store=None, **over):
                    _P(), stop=asyncio.Event())
     b = Inverter("inverter-b", setpoint_entity="number.sp_b",
                  soc_entity="sensor.soc_b", batt_entity="sensor.batt_b",
-                 max_w=B_RATING_HYPOTHESIS, store=c.store)
+                 max_w=B_RATING_HYPOTHESIS, store=c.store,
+                 # SAFETY-91: a cut-out, as rule 8 requires; without one B is
+                 # held back at its reserve and gets no share to reverse.
+                 bms_soc_floor=c.inverters[0].bms_soc_floor)
     b.dev_min, b.dev_max = -B_RATING_HYPOTHESIS, B_RATING_HYPOTHESIS
     c.inverters.append(b)
     c.auto = True