Merge pull request 'SAFETY-86: peak guard outranks the maintenance hold' (#161) from safety-86-peak-over-hold into release/1.0

goodwe_controller/DECISIONS/SAFETY-83-peak-over-drain.md

@@ -30,6 +30,5 @@ everything. Where the floor is at or below the drain rate nothing changes.
 **Not changed.**
 - Between the drain rate and the peak aim, the house still imports what the
   drain leaves (N=1 behaviour kept by the identity). At N=2 the loop covers it.
-- A maintenance HOLD (0 W, 120 min, battery full) still outranks the peak
-  guard at N=1. Discharging there spends the charge the BMS balances on; it is
-  an owner question, not this fix.
+- A maintenance HOLD (0 W, 120 min, battery full) still outranked the peak
+  guard at N=1. Fixed by [[SAFETY-86]].

goodwe_controller/DECISIONS/SAFETY-84-charge-at-two.md

@@ -41,6 +41,6 @@ was fine.
   and the battery gives until the quarter is safe. The same sizing would fit
   there; that is an owner question, not this fix.
 - Outside a cheap window (`hi = 0`, solar only) nothing is filed, at any N.
-- A HOLD (SAFETY-86) and the "Battery is not keeping up" alarm (SAFETY-85)
-  are untouched. In the identity runs that alarm's log line comes as often at
-  N=2 as at N=1 (2 in 480 cycles).
+- A HOLD (later fixed by [[SAFETY-86]]) and the "Battery is not keeping up"
+  alarm (SAFETY-85) are untouched. In the identity runs that alarm's log line
+  comes as often at N=2 as at N=1 (2 in 480 cycles).

goodwe_controller/DECISIONS/SAFETY-86-peak-over-hold.md

@@ -0,0 +1,47 @@
+# SAFETY-86 - the peak guard outranks the maintenance HOLD
+
+**Status: DECIDED 2026-09-30 (owner).** Follows SAFETY-83's "Not changed"
+list. Tests: `test_safety86.py`.
+
+**The defect.** The HOLD (battery resting full after the maintenance charge)
+is a SET of 0 W at `P_MAINTENANCE` (60), above the peak floor (`P_PEAK`, 50),
+so the floor did not bind. At N=1 a 6500 W house banked 6500 W quarters and
+'Load too high' fired (SAFETY-83 rig test, check 4). At N=2 only the other
+battery gave: 4500 W against a 3950 W aim (SAFETY-84 rig test, check 6).
+
+**The fix** (`main.py`, `_peak_over_hold()`).
+- N=1: during a HOLD a peak floor above 0 W is refiled at `P_MAINTENANCE`, as
+  for a drain (SAFETY-83). The battery gives the floor; the Why reads "raised
+  for the peak limit".
+- N>=2: the holding battery's pin is raised to the floor minus what the other
+  live batteries may place (their hard `hi`, or their pin). The others give
+  first, the holding one only the shortfall. Why this split and not the
+  allocator's shares: the hold is maintenance and outranks the others' use
+  (owner order), and the owner rule for an Off battery is the same, "give way
+  only for the shortfall". A battery Off by its mode (P_MODE, below
+  maintenance) therefore still gives before the holding one.
+- N>=2, while the holding battery gives: a site `hold` ceiling at the floor
+  (P_MAINTENANCE), so the loop does not ask for watts the hold keeps back.
+  Without it the moving pin re-logged the ALLOCATOR "could not be placed"
+  row each cycle and filled the action log (review, PR #161).
+- Below the aim nothing changes: the floor is 0 W, the hold is 0 W, the rows
+  are identical to before (test_safety86.py part 2).
+- A site stop (a SET above `P_PEAK`) decides alone: nothing is raised at N>=2,
+  and the refiled floor at N=1 stays under `P_SAFETY`. If another battery is
+  pinned to charge (a SAFETY-73 probe) the pin is not raised, so the site is
+  never split both ways.
+- The hold resumes by itself: the rule is evaluated each cycle, so when the
+  guard lets go the command is 0 W again.
+
+**The level dips below the target: the hold finishes anyway.** Owner
+decision 2026-10-01: the HOLD ends on its clock (`maintenance_hold_min`) even
+when the peak guard pulls the level down; no top-up. A top-up would re-enter
+CHARGE (up to 36 h, `CHARGE_TIMEOUT_MIN`) and keep the site's one maintenance
+slot meanwhile.
+
+**Not changed.**
+- The peak guard still grades the emptiest battery (SAFETY-32): at N=2 with
+  B under `peak_soc_floor` the floor is limited and A holds even when it is
+  full (SAFETY-87).
+- "Battery is not keeping up" during maintenance (SAFETY-85), the N=2 charge
+  flapping (SAFETY-88) and the N=1 charge headroom sizing are untouched.

goodwe_controller/DOCS.md

@@ -1669,6 +1669,10 @@ gives more than its drain rate, with one battery too (SAFETY-83).
 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).
+While a full battery rests after its charge (the hold), it gives just enough
+to keep the 15-minute peak at that aim if the house needs it; with two
+batteries the other one gives first. The hold still ends on its clock
+(SAFETY-86).
 
 #### Capacity-tariff exposure of the grid-following loop
 

goodwe_controller/app/arbiter.py

@@ -79,7 +79,7 @@ P_PEAK = 50           # capacity-tariff peak shaving - costs real money
 # The order in effect: safety limits and the SAFETY-73 refusal > the peak
 # guard > maintenance > modes > trade > loop. A peak floor above a drain still
 # binds: the `drain` site floor at N>=2 (SAFETY-78), refiled at P_MAINTENANCE
-# at N=1 (SAFETY-83). Not a maintenance HOLD at N=1 (DECISIONS/SAFETY-83).
+# at N=1 (SAFETY-83); over a HOLD, `_peak_over_hold()` (SAFETY-86).
 # A cheap-window charge is suspended while the peak guard holds and, at N>=2,
 # sized to the quarter's headroom (SAFETY-84).
 P_MAINT_SHAPE = 40    # maintenance shaping the loop (charge-only)
@@ -202,6 +202,10 @@ CLAIM_SCOPE = {
     # sibling would have to give against it. Never filed at N=1.
     # DECISIONS/SAFETY-84-charge-at-two.md.
     "charge": SCOPE_SITE,
+    # main.py: `Claim.limit("hold", P_MAINTENANCE, hi=...)`, SAFETY-86. While a
+    # holding pack gives the peak shortfall at N>=2, the site target stays at
+    # the peak floor, what the pin and the others place. Never filed at N=1.
+    "hold": SCOPE_SITE,
 
     # --- UNIT: about one pack, one inverter, one device -------------------
     # main.py: `Claim.limit("device", P_SAFETY, self.dev_min, self.dev_max,

goodwe_controller/app/main.py

@@ -74,7 +74,7 @@ from .hass import HomeAssistant
 from .inverter import (LINK_STALE_S, UNIT_A, build_roster,
                        roster_entries, roster_names)
 from .maintenance import (
-    DRAIN, IDLE, POWER_MOVING, MaintConfig, Maintenance, MaintResult,
+    DRAIN, HOLD, IDLE, POWER_MOVING, MaintConfig, Maintenance, MaintResult,
     arbitrate as maint_arbitrate,
 )
 from .mqtt import (MODE_CONFIGS_KEY, MODE_LABELS, USE_OFF, USE_ON,
@@ -2533,6 +2533,53 @@ class Controller:
             if c.source == peakguard.SOURCE and c.lo > rate:
                 claims[i] = dataclasses.replace(c, priority=P_MAINTENANCE)
 
+    def _peak_over_hold(self, claims, maint_results):
+        """SAFETY-86: a peak floor outranks a maintenance HOLD (0 W).
+
+        At N=1 the floor is refiled at P_MAINTENANCE, as for a drain. At N>=2
+        the holding pack gives only what the others cannot: its pin is raised
+        to the floor minus what they may place, and the floor is returned as
+        the site `hold` ceiling (else None). A site stop still decides alone.
+        DECISIONS/SAFETY-86-peak-over-hold.md.
+        """
+        hold = {self.actuator_of("maintenance", inv)
+                for inv, res in zip(self.inverters, maint_results)
+                if res.phase == HOLD and res.owns_setpoint}
+        peak = [i for i, c in enumerate(claims)
+                if c.source == peakguard.SOURCE and c.lo > 0.0]
+        if not hold or not peak:
+            return
+        if len(self.inverters) == 1:
+            for i in peak:
+                claims[i] = dataclasses.replace(claims[i],
+                                                priority=P_MAINTENANCE)
+            return
+        if any(c.kind == SET and c.actuator == ACTUATOR_SITE
+               and c.priority > P_PEAK for c in claims):
+            return
+        live = set(self.live_units())
+        others = 0.0
+        for inv in self.inverters:
+            u = self.actuator_of("maintenance", inv)
+            if u in hold or u not in live:
+                continue
+            b = pre_resolve([c for c in claims if c.actuator == u])
+            if b.contradiction:
+                continue
+            p = allocator.pin_w(b)
+            if p is not None and p < 0.0:
+                return          # a pack pinned to charge: never split both ways
+            others += allocator.hi_w(b) if p is None else p
+        floor = max(claims[i].lo for i in peak)
+        give = floor - others
+        if give <= 0.0:
+            return
+        for i, c in enumerate(claims):
+            if c.source == "maintenance" and c.kind == SET \
+                    and c.actuator in hold:
+                claims[i] = dataclasses.replace(c, value=give)
+        return floor
+
     def _charge_headroom(self, floor: float, claims) -> float:
         """SAFETY-84: at N>=2 a cheap-window charge pulls no more than the
         quarter's headroom under the peak aim (`charge_headroom_w()`), so it
@@ -3905,7 +3952,8 @@ class Controller:
         # business, not this block's: a maintenance DRAIN or HOLD owns the
         # setpoint at P_MAINTENANCE, and a maintenance CHARGE is only ever a
         # P_MAINT_SHAPE limit (40), bound by this floor. A floor above the drain
-        # rate still wins (SAFETY-78 at N>=2, `_peak_over_drain()` at N=1).
+        # rate still wins (SAFETY-78 at N>=2, `_peak_over_drain()` at N=1), and
+        # so does a floor over a HOLD (`_peak_over_hold()`, SAFETY-86).
         #
         # ⚠️ AND IT IS NOT RUN AT ALL WITHOUT P1. `meter_source: off` is the
         # legacy single-entity path: there is no telegram stream, so
@@ -4307,6 +4355,13 @@ class Controller:
                                       reason="maintenance drain",
                                       actuator=self.actuator_of("drain")))
         self._peak_over_drain(claims, maint_results)
+        # --- SAFETY-86: while a HOLD gives, the site stays at the peak floor --
+        # So the loop never asks for watts the hold keeps back.
+        hold_hi = self._peak_over_hold(claims, maint_results)
+        if hold_hi is not None:
+            claims.append(Claim.limit("hold", P_MAINTENANCE, hi=hold_hi,
+                                      reason="maintenance hold",
+                                      actuator=self.actuator_of("hold")))
         # --- SAFETY-84: a cheap-window charge is a ceiling over the site ----
         charge = self._charge_ceiling(claims)
         if charge is not None:

goodwe_controller/app/messages.py

@@ -836,6 +836,7 @@ WHY_LIMIT = {
     "reserve": "the battery reserve",
     "maintenance": "battery maintenance",
     "charge": "battery maintenance",
+    "hold": "battery maintenance",
     "refused": "a battery that refuses power",
 }
 

goodwe_controller/test_arbiter.py

@@ -149,7 +149,7 @@ check("...and `source` is still the first positional argument, which is what "
       Claim.limit("device", P_SAFETY, -1, 1).source == "device")
 
 print()
-print("ARB-03 2. the split: fifteen sources, each classified where it is BUILT")
+print("ARB-03 2. the split: sixteen sources, each classified where it is BUILT")
 # The classification is graded against what each claim is ABOUT, restated here
 # so that the table in arbiter.py and the expectation in a test are not the same
 # sentence read twice. test_peakguard.py part 10f pins this set against main.py's
@@ -179,6 +179,9 @@ SITE_SOURCES = {
     "charge":    "the site target may not go above a cheap-window maintenance "
                  "charge, or a sibling would give against it. Filed at N>1 "
                  "only",
+    # SAFETY-86.
+    "hold":      "the site target stays at the peak floor while a holding pack "
+                 "gives the shortfall. Filed at N>1 only",
 }
 UNIT_SOURCES = {
     "device":     "one inverter's own reported min/max rating",
@@ -210,9 +213,9 @@ for _src, _why in sorted(SITE_SOURCES.items()):
     print(f"    site  {_src:<11} {_why}")
 for _src, _why in sorted(UNIT_SOURCES.items()):
     print(f"    unit  {_src:<11} {_why}")
-check("all FIFTEEN claim sources are classified, and only those fifteen",
+check("all SIXTEEN claim sources are classified, and only those sixteen",
       set(CLAIM_SCOPE) == set(SITE_SOURCES) | set(UNIT_SOURCES))
-check("the eight that are about the GRID CONNECTION are site claims",
+check("the nine that are about the GRID CONNECTION are site claims",
       {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_SITE} == set(SITE_SOURCES))
 check("the seven that are about a PACK, a DEVICE or one write are per-inverter",
       {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_UNIT} == set(UNIT_SOURCES))
@@ -234,7 +237,7 @@ check("...and with none configured at all, likewise",
       all(actuator_for(s, "inverter-a", ()) == ACTUATOR_SITE for s in CLAIM_SCOPE))
 check("with TWO the identity breaks and the split becomes real: the seven "
       "per-inverter sources route to the unit this controller drives, the "
-      "eight site sources stay on the grid connection",
+      "nine site sources stay on the grid connection",
       {s: actuator_for(s, "inverter-a", TWO) for s in CLAIM_SCOPE}
       == {**{s: "inverter-a" for s in UNIT_SOURCES},
           **{s: ACTUATOR_SITE for s in SITE_SOURCES}})
@@ -515,7 +518,7 @@ check("...and at N=1, where every source legitimately routes to the site, the "
           Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0)]) == [])
 check("an unclassified ninth source is a violation wherever it is filed",
       scope_violations([Claim.limit("reserve", P_LOOP, hi=0.0)]) != [])
-check("and the eight, routed by actuator_for() at a two-inverter site, are "
+check("and the sixteen, routed by actuator_for() at a two-inverter site, are "
       "clean - which is what says the router and the checker agree",
       scope_violations(
           [Claim.limit(s, P_SAFETY, -1, 1,

goodwe_controller/test_peak_margin.py

@@ -173,14 +173,15 @@ check("...and decide() sizes the floor to land the quarter 50 W under it "
 # ============================================================================ #
 print("\n3. the bill and the reports keep the real cap")
 _c = _runs["n1_off", None][1]
-_pk = _c.peak.peak().peak_w
+_t = rig.rig.BASE       # the runs' clock, not the wall clock (a new month)
+_pk = _c.peak.peak(_t).peak_w
 print(f"    month peak {_pk:.0f} W, tracker cap {_c.peak.cap_w:.0f} W, "
-      f"warning {_c.peak.warning_frac():.3f}, headroom "
-      f"{_c.peak.headroom_w():.0f} W")
+      f"warning {_c.peak.warning_frac(_t):.3f}, headroom "
+      f"{_c.peak.headroom_w(_t):.0f} W")
 check("test_billing_uses_cap: the month-peak tracker (bill, headroom, 80 % "
       "warning) is built on the real cap, not cap - margin",
-      _c.peak.cap_w == CAP and _c.peak.warning_frac() == _pk / CAP
-      and _c.peak.headroom_w() == CAP - _pk)
+      _c.peak.cap_w == CAP and _c.peak.warning_frac(_t) == _pk / CAP
+      and _c.peak.headroom_w(_t) == CAP - _pk)
 _eng = [g for r in _runs["n1_off", None][0] for g in (r[-1][1],)
         if g and g[0] and g[1]]
 print(f"    {len(_eng)} engaged cycles; one: {_eng[0][4] if _eng else None}")

goodwe_controller/test_peakguard.py

@@ -2691,9 +2691,12 @@ check("...and so does the persistent notification, which outlives the page and "
 # site `hi` at P_MAINT_SHAPE, filed only at N>=2 while a pack's cheap-window
 # maintenance charge makes it charge, equal to that pack's own `hi`. Never at
 # N=1. DECISIONS/SAFETY-84-charge-at-two.md.
+# SAFETY-86 added `hold`: a site `hi` at the peak floor while a holding pack
+# gives at N>=2. Not a reserve either. DECISIONS/SAFETY-86-peak-over-hold.md.
 CLAIM_SOURCES = {"device", "supervised", "failsafe", "safety", "peak",
                  "arbitrage", "maintenance", "loop", "circulation",
-                 "sign_site", "pack", "refused", "mode", "drain", "charge"}
+                 "sign_site", "pack", "refused", "mode", "drain", "charge",
+                 "hold"}
 # The two appends that file a claim BUILT ELSEWHERE in the method. Named here so
 # that a THIRD such indirection reads as an unrecognised append rather than as a
 # silent gap in the scan.
@@ -2735,7 +2738,7 @@ check("every `claims.append(...)` in main.py's cycle is one this scan can READ "
       "- a claim filed through a shape it cannot parse has to fail here rather "
       "than be counted as nothing, which is how a scan like this goes quietly "
       "vacuous", len(_appends) >= 10 and None not in _filed)
-check("...and the sources that cycle can file are EXACTLY these FIFTEEN. THIS "
+check("...and the sources that cycle can file are EXACTLY these SIXTEEN. THIS "
       "is what makes part 10 a tripwire: replay_loop() hand-builds the claim "
       "list, so it cannot see a claim main.py adds - and a loop-side SoC "
       "reserve is a new source. Filing one takes this line red on the line it "
@@ -2875,7 +2878,7 @@ print("    site: " + ", ".join(sorted(s for s, v in CLAIM_SCOPE.items()
       + "  |  per-inverter: "
       + ", ".join(sorted(s for s, v in CLAIM_SCOPE.items() if v == SCOPE_UNIT)))
 check("...and the split is the one ARB-03 argued, with SAFETY-31's and "
-      "SAFETY-30's, OPT-09's, SAFETY-78's and SAFETY-84's additions: the eight about the GRID CONNECTION "
+      "SAFETY-30's, OPT-09's, SAFETY-78's, SAFETY-84's and SAFETY-86's additions: the nine about the GRID CONNECTION "
       "are site claims and the five about a pack, a device or one write are "
       "per-inverter. `circulation` is a site claim because its quantity is "
       "defined over the roster and is identically zero for any one unit; "
@@ -2885,7 +2888,7 @@ check("...and the split is the one ARB-03 argued, with SAFETY-31's and "
       "against that unit's OWN measured power",
       {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_SITE}
       == {"loop", "peak", "arbitrage", "safety", "circulation", "sign_site",
-          "drain", "charge"}
+          "drain", "charge", "hold"}
       and {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_UNIT}
       == {"device", "supervised", "failsafe", "maintenance", "pack",
           "refused", "mode"})
@@ -2990,7 +2993,8 @@ check("⚠️ AND THIS IS THE BREAKING HALF. Give the same Controller a second "
                   "loop": ACTUATOR_SITE, "peak": ACTUATOR_SITE,
                   "arbitrage": ACTUATOR_SITE, "safety": ACTUATOR_SITE,
                   "circulation": ACTUATOR_SITE, "sign_site": ACTUATOR_SITE,
-                  "drain": ACTUATOR_SITE, "charge": ACTUATOR_SITE})
+                  "drain": ACTUATOR_SITE, "charge": ACTUATOR_SITE,
+                  "hold": ACTUATOR_SITE})
 check("...and the roster it was restored to is the shipped one, so nothing "
       "below inherits a two-inverter Controller",
       c_src.units == (app_main.UNIT_A,) and len(c_src.inverters) == 1)
@@ -3069,13 +3073,13 @@ _unit_sources_unrouted = [
     if CLAIM_SCOPE.get(r[0]) == SCOPE_UNIT and "," not in r[1]]
 print(f"    {len(_direct)} direct appends, each read at its own line: "
       + ", ".join(f"main.py:{ln} {s} -> {a}" for s, a, ln in _routing))
-check("all SIXTEEN of `Controller.cycle()`'s DIRECT appends route through "
+check("all SEVENTEEN of `Controller.cycle()`'s DIRECT appends route through "
       "`self.actuator_of(<string literal>)`, so this scan can read the routing "
       "of EVERY one of them. An append that computed its actuator, or omitted "
       "the argument, or hand-wrote a string, fails HERE rather than being "
       "silently skipped - which is how a scan like this goes quietly vacuous, "
       "and 10f is in this file because that already happened once",
-      len(_direct) == 16
+      len(_direct) == 17
       and all(a is not None and a.startswith("self.actuator_of('")
               for _s, a, _ln in _routing))
 check("⚠️ AND AT EVERY ONE OF THE ELEVEN THE LITERAL IT ROUTES THROUGH IS THE "
@@ -3109,8 +3113,8 @@ check("...and the only appends this pin does NOT reach are the two that file a "
       "arbitrage.claim() by test_arbitrage.py",
       sorted(_indirect) == sorted(CLAIM_BUILDERS)
       and len(_direct) + len(_indirect) == len(_appends))
-check("...and between them the eighteen appends account for exactly the "
-      "fifteen "
+check("...and between them the nineteen appends account for exactly the "
+      "sixteen "
       "sources 10f pins, so the routing pin and the source set are one scan "
       "and neither can drift from the other",
       {s for s, _a, _ln in _routing} | set(CLAIM_BUILDERS.values())
@@ -3171,7 +3175,7 @@ check("...and naming those three is not an accident of this fixture but the "
       {c.source for c in _cl2}
       == CLAIM_SOURCES - {"failsafe", "safety", "maintenance", "peak",
                           "arbitrage", "circulation", "sign_site",
-                          "refused", "mode", "drain", "charge"})
+                          "refused", "mode", "drain", "charge", "hold"})
 # ⚠️ AND NOW THE HALF THAT WAS EVIDENCE-SHAPED AND OBSERVED NOTHING. The three
 # checks above grade WHICH ACTUATOR each claim lands on. They say nothing about
 # WHOSE NUMBERS the bound is built from - and those are two different defects

goodwe_controller/test_safety83.py

@@ -295,11 +295,10 @@ def store_hold():
 
 
 _h_rows, _h, _hb = run(6000.0, 2500, 50, 2000, dur=600.0, store=store_hold())
-check("a maintenance HOLD at N=1 is not changed (DECISIONS/SAFETY-83, not "
-      "changed): the battery stays at 0 W with the house over the limit",
+check("a maintenance HOLD at N=1 now gives way to the guard too (SAFETY-86, "
+      "test_safety86.py): the battery gives with the house over the limit",
       _h.maints[0].phase == "hold" and _h.peak_guard.engaged
-      and _h.peak_guard.floor_w > 0.0
-      and all(cmd(r) == 0.0 for r in _h_rows))
+      and cmd(_h_rows[-1]) > 0.0 and "limited by peak" in _h_rows[-1][6])
 
 print()
 if fails:

goodwe_controller/test_safety86.py

@@ -0,0 +1,421 @@
+"""Runnable check for SAFETY-86: the peak guard outranks the maintenance HOLD.
+`python3 test_safety86.py`
+
+During the HOLD the battery rests full at 0 W (P_MAINTENANCE, above the peak
+floor), so the house imported everything and quarters went over the peak
+limit. Now the holding battery gives what keeps the quarter at the aim: at N=1
+the peak floor is refiled at P_MAINTENANCE, at N>=2 the hold's pin is raised
+to what the other batteries cannot give. Drives the REAL Controller
+(tools/safety84_identity.py's harness). DECISIONS/SAFETY-86-peak-over-hold.md.
+
+1. N=1, house over the aim: every quarter lands on the aim, the hold goes on,
+   no 'Load too high'; without the fix the quarter goes over and it fires.
+2. Under the aim nothing changes, at N=1 and N=2 (rows identical).
+3. N=2, house over the aim: the other battery gives first, the holding one
+   only the shortfall; with battery modes on and B Off too.
+4. A month peak banked above the limit: the aim is that peak - margin.
+5. A safety stop still parks everything, and the hold gives again after.
+6. The level dips below the target: the hold still ends on its clock.
+7. The rule at its boundaries, on claim lists.
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import logging
+import os
+import sys
+import types
+
+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 control as _ctl                                # noqa: E402
+from app import packmode                                       # noqa: E402
+from app.arbiter import (Claim, P_MAINTENANCE, P_PEAK, P_SAFETY,  # noqa
+                         resolve)
+from app.main import Controller                                # noqa: E402
+from app.maintenance import (CHARGE, DRAIN, HOLD, IDLE,          # noqa: E402
+                             MaintResult)
+from app.store import unit_key                                 # noqa: E402
+
+fails = []
+A, B = "inverter-a", "inverter-b"
+AIM = 3450.0            # peak_cap_w 3500 - peak_margin_w 50
+TOL = 10.0              # W over the aim a banked quarter may show (SAFETY-81)
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+def store_hold(n, b_off=False):
+    """A store with battery A in its HOLD from the first cycle."""
+    s = rig._Store()
+    for k in ("phase", "phase_started"):
+        s.data[k if n == 1 else unit_key(A, k)] = (
+            "hold" if k == "phase" else rig.BASE.isoformat())
+    if b_off:
+        s.data[unit_key(B, packmode.USE_KEY)] = False
+    return s
+
+
+def full(u, t):
+    """A rests full; B is half full."""
+    return 99.0 if u == "a" else 60.0
+
+
+def run(house, n, *, hold_min=600, b_off=False, extra=None, **kw):
+    kw.setdefault("soc", full)
+    kw.setdefault("dur", 2700.0)
+    return s84.run(n, load=house if callable(house) else lambda t: house,
+                   cheap=False, store=store_hold(n, b_off),
+                   extra=dict({"maintenance_hold_min": hold_min},
+                              **(extra or {})), **kw)
+
+
+def without_fix(fn):
+    real = Controller._peak_over_hold
+    Controller._peak_over_hold = lambda self, claims, results: None
+    try:
+        return fn()
+    finally:
+        Controller._peak_over_hold = real
+
+
+def cmd(r):
+    return dict(r[2])
+
+
+def banked(c):
+    return [b.offtake_avg_w for b in c.p1.blocks]
+
+
+def on_aim(c, aim=AIM):
+    b = banked(c)
+    return len(b) >= 2 and all(aim - TOL <= v <= aim + TOL for v in b)
+
+
+def alarmed(c):
+    """Did SAFETY-81's 'Load too high' alarm fire on this run?"""
+    return any("PEAK_GUARD_SATURATED" in e for e in c.events)
+
+
+def refused(rows):
+    return sum(1 for r in rows if r[7] and "REFUSED" in r[7])
+
+
+def peak_prio(r):
+    return [c[1] for c in r[-1][0] if c[0] == "peak"]
+
+
+def hold_pin(r):
+    return [c[3] for c in r[-1][0] if c[0] == "maintenance" and c[2] == "set"]
+
+
+# ============================================================================ #
+print("1. N=1: a house over the aim during the HOLD")
+N1 = {h: run(h, 1) for h in (5000.0, 6500.0, 7000.0)}
+N1_OLD = {h: without_fix(lambda: run(h, 1)) for h in N1}
+for h in N1:
+    print(f"    house {h:.0f}: banked {[round(v) for v in banked(N1[h][1])]} W"
+          f" (before the fix {[round(v) for v in banked(N1_OLD[h][1])]} W), "
+          f"last {N1[h][0][-1][6]}")
+check("⭐ every quarter lands on the aim (3450 W) for a 5000, 6500 and "
+      "7000 W house, and 'Load too high' never fires",
+      all(on_aim(c) and not alarmed(c) for _r, c in N1.values()))
+check("...the battery gives, raised for the peak limit, and stays in its "
+      "HOLD: the phase is hold and the claim still says so",
+      all(c.maints[0].phase == HOLD and cmd(rows[-1])["a"] > 1000.0
+          and "limited by peak" in rows[-1][6]
+          and c.why_words.endswith("raised for the peak limit")
+          for rows, c in N1.values()))
+check("non-vacuous: without the fix the battery sits at 0 W, the quarter "
+      "banks the whole house and 'Load too high' fires",
+      all(all(cmd(r)["a"] == 0.0 for r in rows) and min(banked(c)) >= h - 1
+          and alarmed(c) for h, (rows, c) in N1_OLD.items()))
+
+
+def hob(t):
+    """A 6000 W house with a 5 s spike to 11000 W every 150 s."""
+    return 11000.0 if t % 150 < 5 else 6000.0
+
+
+_sp_rows, _sp = run(hob, 1)
+check("...and 5 s spikes that saturate the guard for a cycle do not fire it "
+      "(SAFETY-81's 10 s hold)",
+      any(r[-1][1] and r[-1][1][1] >= 5000.0 for r in _sp_rows)
+      and not alarmed(_sp))
+
+# ============================================================================ #
+print("\n2. under the aim the HOLD is as before")
+BELOW = {(n, h): (run(h, n, dur=1800.0),
+                  without_fix(lambda: run(h, n, dur=1800.0)))
+         for n, h in ((1, 1200.0), (1, 3000.0), (1, -1500.0),
+                      (2, 1500.0), (2, 3000.0), (2, 5000.0))}
+check("⭐ N=1 at a 1200 W and 3000 W house and a 1500 W surplus, N=2 at "
+      "1500, 3000 and 5000 W (B covers it): battery A is at 0 W on every "
+      "cycle and every row equals the run without the fix",
+      all(all(cmd(r)["a"] == 0.0 for r in new[0]) and new[0] == old[0]
+          for new, old in BELOW.values()))
+check("...and the peak claim stays at P_PEAK on every cycle",
+      all(p == P_PEAK for new, _o in BELOW.values() for r in new[0]
+          for p in peak_prio(r)))
+
+# ============================================================================ #
+print("\n3. N=2: the other battery gives first, the holding one the rest")
+N2 = {h: run(h, 2) for h in (6500.0, 7000.0)}
+N2_OLD = {h: without_fix(lambda: run(h, 2)) for h in N2}
+for h, (rows, c) in N2.items():
+    print(f"    house {h:.0f}: banked {[round(v) for v in banked(c)]} W "
+          f"(before the fix {[round(v) for v in banked(N2_OLD[h][1])]} W), "
+          f"last {rows[-1][2]}")
+check("⭐ house 6500 and 7000 W, B at most 2500 W: every quarter lands on "
+      "the aim, nothing is refused or split both ways, no 'Load too high'",
+      all(on_aim(c) and refused(rows) == 0 and not alarmed(c)
+          and all(cmd(r)["a"] * cmd(r)["b"] >= 0.0 for r in rows)
+          for rows, c in N2.values()))
+_gives = [r for rows, _c in N2.values() for r in rows if cmd(r)["a"] > 0.0]
+check("⭐ A gives only the shortfall: on every cycle A gives, B is at its "
+      "2500 W and A's pin is the guard's floor minus 2500 W",
+      len(_gives) > 500
+      and all(cmd(r)["b"] == 2500.0 and hold_pin(r)
+              and abs(hold_pin(r)[0] - (r[-1][1][1] - 2500.0)) < 1e-6
+              for r in _gives))
+check("...A stays in its HOLD, and its last command is the shortfall "
+      "(6500 - 3450 - 2500 = 550 W, give or take the meter's 1 W)",
+      all(c.maints[0].phase == HOLD for _r, c in N2.values())
+      and 540.0 <= cmd(N2[6500.0][0][-1])["a"] <= 560.0)
+check("non-vacuous: without the fix A sits at 0 W, the quarter goes over "
+      "the limit and 'Load too high' fires",
+      all(all(cmd(r)["a"] == 0.0 for r in rows) and max(banked(c)) > 3900.0
+          and alarmed(c) for rows, c in N2_OLD.values()))
+
+_off_rows, _off = run(6500.0, 2, b_off=True,
+                      extra={"battery_modes_enabled": True})
+print(f"    B Off, house 6500: banked {[round(v) for v in banked(_off)]} W, "
+      f"last {_off_rows[-1][2]}")
+check("⭐ battery modes on, B Off: the Off battery still gives before the "
+      "holding one (maintenance outranks a mode), the quarter lands on the "
+      "aim, nothing refused",
+      on_aim(_off) and cmd(_off_rows[-1])["b"] == 2500.0
+      and 540.0 <= cmd(_off_rows[-1])["a"] <= 560.0
+      and refused(_off_rows) == 0 and not alarmed(_off))
+
+
+
+def noisy(t):
+    """A 6500 W house that moves up to 150 W from one second to the next."""
+    return 6500.0 + ((int(t) * 7919) % 301) - 150.0
+
+
+def no_ceiling(fn):
+    """Run with the pin raised but no `hold` site ceiling (review check)."""
+    real = Controller._peak_over_hold
+
+    def pin_only(self, claims, results):
+        real(self, claims, results)
+    Controller._peak_over_hold = pin_only
+    try:
+        return fn()
+    finally:
+        Controller._peak_over_hold = real
+
+
+def rows_logged(c, word):
+    return sum(1 for e in c.events if word in e)
+
+
+_nz_rows, _nz = run(noisy, 2)
+_nz_old = no_ceiling(lambda: run(noisy, 2))[1]
+print(f"    noisy house: ALLOCATOR rows {rows_logged(_nz, 'ALLOCATOR')}, "
+      f"clamp rows {rows_logged(_nz, 'OUTPUT_CLAMP')} (without the site "
+      f"ceiling {rows_logged(_nz_old, 'ALLOCATOR')} and "
+      f"{rows_logged(_nz_old, 'OUTPUT_CLAMP')})")
+check("⭐ while A gives, the site target is the guard's floor (the `hold` "
+      "ceiling), so a noisy house does not fill the action log: no ALLOCATOR "
+      "row and at most one clamp row",
+      on_aim(_nz) and all(abs(r[1] - r[-1][1][1]) < 1e-6 for r in _nz_rows
+                          if cmd(r)["a"] > 0.0)
+      and rows_logged(_nz, "ALLOCATOR") == 0
+      and rows_logged(_nz, "OUTPUT_CLAMP") <= 1)
+check("non-vacuous: without the ceiling the loop asks 5000 W and the "
+      "ALLOCATOR and clamp rows fill the 40-row log",
+      rows_logged(_nz_old, "ALLOCATOR") >= 10 and len(_nz_old.events) == 40)
+
+# ============================================================================ #
+print("\n4. a month peak banked above the limit")
+BANK = {n: (run(6500.0, n, month_peak=4000.0),
+            without_fix(lambda: run(6500.0, n, month_peak=4000.0)))
+        for n in (1, 2)}
+for n, (new, old) in BANK.items():
+    print(f"    N={n} house 6500, month peak 4000: banked "
+          f"{[round(v) for v in banked(new[1])]} W (before the fix "
+          f"{[round(v) for v in banked(old[1])]} W)")
+check("⭐ under a month peak of 4000 W the aim is 3950 W (peak - margin), "
+      "at N=1 and N=2, and no 'Load too high'",
+      all(on_aim(new[1], 3950.0) and not alarmed(new[1])
+          for new, _old in BANK.values()))
+check("non-vacuous: before the fix N=1 banks the whole house and N=2 banks "
+      "4000 W, over that aim",
+      min(banked(BANK[1][1][1])) >= 6499.0
+      and min(banked(BANK[2][1][1])) >= 3999.0)
+
+# ============================================================================ #
+print("\n5. a safety stop still wins")
+
+
+def blind_at(t0, t1):
+    """A's power reading is missing from `t0` to `t1`: a site safety SET."""
+    def before(c, t):
+        s84.feed(False)(c, t)
+        if not getattr(c.hass, "_blind", False):
+            real = c.hass.number
+
+            async def number(e, invert=False):
+                if e == "sensor.batt_a" and t0 <= c.hass.t < t1:
+                    return None
+                return await real(e, invert)
+            c.hass.number, c.hass._blind = number, True
+    return before
+
+
+STOP = {n: run(6500.0, n, dur=900.0, before=blind_at(300.0, 600.0))
+        for n in (1, 2)}
+for n, (rows, _c) in STOP.items():
+    print(f"    N={n} reading lost 300-600 s: at {rows[100][0]:.0f} s "
+          f"{rows[100][2]}, {rows[100][6]}; after {rows[-1][2]}")
+check("⭐ with a reading gone the safety SET parks every battery at 0 W, at "
+      "N=1 and N=2, with nothing refused",
+      all(all(v == 0.0 for r in rows if 330.0 <= r[0] < 600.0
+              for v in cmd(r).values())
+          and any(r[6].startswith("safety -> 0 W") for r in rows)
+          and refused(rows) == 0 for rows, _c in STOP.values()))
+check("...the hold gave before the stop and gives again after it",
+      all(cmd(rows[50])["a"] > 0.0 and cmd(rows[-1])["a"] > 0.0
+          for rows, _c in STOP.values()))
+
+# ============================================================================ #
+print("\n6. the level dips below the hold's target")
+
+
+def draining(u, t):
+    """A's level falls 1 % a minute from 99 % (set, not integrated); B 60 %."""
+    return max(20.0, 99.0 - t / 60.0) if u == "a" else 60.0
+
+
+DIP = {n: run(6500.0, n, hold_min=30, dur=2400.0, soc=draining)
+       for n in (1, 2)}
+for n, (rows, c) in DIP.items():
+    print(f"    N={n}: phases {sorted({tuple(x[-1][3]) for x in rows})}, "
+          f"last event {[e for e in c.events if 'maintenance' in e][:1]}")
+check("⭐ the level falls from 99 % to 70 % while the hold gives, and "
+      "the hold still ends on its 30 min clock: HOLD -> IDLE, cycle "
+      "COMPLETE, never back to CHARGE, at N=1 and N=2",
+      all(c.maints[0].phase == IDLE
+          and any("cycle COMPLETE" in e for e in c.events)
+          and {x[-1][3][0] for x in rows} == {HOLD, IDLE}
+          and all(x[-1][3][0] == HOLD for x in rows if x[0] < 1795.0)
+          for rows, c in DIP.values()))
+check("...the hold gave on the way (the level really dipped)",
+      all(any(cmd(r)["a"] > 0.0 for r in rows if r[0] < 1795.0)
+          for rows, _c in DIP.values()))
+
+# ============================================================================ #
+print("\n7. the rule at its boundaries")
+_c = run(1500.0, 2, dur=10.0)[1]
+_c1 = run(1500.0, 1, dur=10.0)[1]
+HOLDING = [MaintResult(HOLD, True, 0.0, False, []),
+           MaintResult(IDLE, False, None, False, [])]
+
+
+def pin_after(floor, *extra, live=(A, B), results=HOLDING, c=_c):
+    fake = types.SimpleNamespace(inverters=c.inverters,
+                                 actuator_of=c.actuator_of,
+                                 live_units=lambda: tuple(live))
+    cl = [Claim.limit("supervised", P_SAFETY, -2500.0, 2500.0, actuator=A),
+          Claim.limit("supervised", P_SAFETY, -2500.0, 2500.0, actuator=B),
+          Claim.set("maintenance", P_MAINTENANCE, 0.0, HOLD, actuator=A),
+          Claim.limit("peak", P_PEAK, lo=floor, reason="quarter peak"),
+          *extra]
+    ceiling = Controller._peak_over_hold(fake, cl, results)
+    pin_after.ceiling = ceiling
+    return next((x.value for x in cl
+                 if x.source == "maintenance" and x.kind == "set"), None), cl
+
+
+check("⭐ at N=2 a floor under or at B's 2500 W leaves A at 0 W (never "
+      "charging); 0.5 W above it raises A's pin to 0.5 W",
+      pin_after(1000.0)[0] == 0.0 and pin_after(2500.0)[0] == 0.0
+      and pin_after(2500.5)[0] == 0.5)
+check("⭐ B not live, contradicting itself, or refusing to give: A gives the "
+      "whole floor",
+      pin_after(2000.0, live=(A,))[0] == 2000.0
+      and pin_after(2000.0, Claim.limit("pack", P_SAFETY, 10.0, -10.0,
+                                        actuator=B))[0] == 2000.0
+      and pin_after(2000.0, Claim.limit("refused", P_SAFETY, hi=0.0,
+                                        actuator=B))[0] == 2000.0)
+_probe = _ctl.PROBE_W
+check("⭐ B probing: giving counts its probe (A gives the rest), charging "
+      "leaves A at 0 W so the site is never split both ways",
+      pin_after(2000.0, Claim.set("refused", P_SAFETY, _probe, "probe",
+                                  actuator=B))[0] == 2000.0 - _probe
+      and pin_after(2000.0, Claim.set("refused", P_SAFETY, -_probe, "probe",
+                                      actuator=B))[0] == 0.0)
+_p, _cl = pin_after(4000.0, Claim.set("safety", P_SAFETY, 0.0,
+                                      "inputs missing"))
+check("⭐ a site safety SET: A stays at 0 W and the site resolves to 0 W; "
+      "a unit safety SET on B does not stop A giving",
+      _p == 0.0 and resolve(_cl, units=(A, B), live={A, B}).target_w == 0.0
+      and pin_after(3000.0, Claim.set("failsafe", P_SAFETY, 0.0,
+                                      actuator=B))[0] == 3000.0)
+_p, _cl = pin_after(2600.0)
+_hi = pin_after.ceiling
+_cl += [Claim.limit("hold", P_MAINTENANCE, hi=_hi, actuator="site"),
+        Claim.set("loop", 10, 5000.0, "the loop asks for the whole house")]
+pin_after(2500.0)
+check("⭐ at N=2 the `hold` site ceiling is the floor itself, only when A "
+      "gives, and a 5000 W loop ask resolves to that floor",
+      _hi == 2600.0 and pin_after.ceiling is None
+      and resolve(_cl, units=(A, B), live={A, B}).target_w == 2600.0)
+check("only a HOLD: a DRAIN pin or an idle pack is not raised",
+      pin_after(4000.0, results=[MaintResult(DRAIN, True, 0.0, False, []),
+                                 HOLDING[1]])[0] == 0.0
+      and pin_after(4000.0, results=[HOLDING[1], HOLDING[1]])[0] == 0.0)
+
+
+def refile(lo, results=(HOLDING[0],)):
+    fake = types.SimpleNamespace(inverters=_c1.inverters,
+                                 actuator_of=_c1.actuator_of)
+    cl = [Claim.set("maintenance", P_MAINTENANCE, 0.0, HOLD),
+          Claim.limit("peak", P_PEAK, lo=lo, reason="quarter peak")]
+    Controller._peak_over_hold(fake, cl, list(results))
+    return cl
+
+
+check("⭐ at N=1 a floor of 0 W stays at P_PEAK, 0.5 W is refiled at "
+      "P_MAINTENANCE, and a blind claim (a ceiling only) is not refiled",
+      refile(0.0)[1].priority == P_PEAK
+      and refile(0.5)[1].priority == P_MAINTENANCE
+      and refile(float("-inf"))[1].priority == P_PEAK)
+_stop = refile(3000.0) + [Claim.set("safety", P_SAFETY, 0.0,
+                                    "inputs missing")]
+check("⭐ at N=1 a refiled floor stays under a safety SET: 0 W",
+      resolve(_stop).target_w == 0.0)
+check("at N=1 outside a HOLD (idle, or a CHARGE) the claim is not touched "
+      "by this rule",
+      refile(3000.0, (MaintResult(IDLE, False, None, False, []),))[1]
+      .priority == P_PEAK
+      and refile(3000.0, (MaintResult(CHARGE, False, None, True, []),))[1]
+      .priority == P_PEAK)
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED: " + ", ".join(fails))
+    sys.exit(1)
+print("all SAFETY-86 checks passed")