SAFETY-84 review: check the charging pack gives under the guard, and no-P1 at N=2

A 6500 W house needs more than B alone: A must give (the N>=2 suspension),
with a non-vacuous run without it (banks over the cap, 'Load too high').
Without P1 the charge is not sized, as at N=1: A charges at 2000 W.

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

goodwe_controller/DECISIONS/SAFETY-78-drain-at-two.md

@@ -35,10 +35,9 @@ direction. Opposite splits stay forbidden.
   (`tools/safety78_identity.py`, which hashes the claims too).
 
 **Not changed (measured in the PR #154 review, follow-ups).**
-- At N=1 a drain (60) outranks the peak floor (50), so the guard does not add
-  to it: a 5000 W house imports 3000 W against a 2500 W limit, a 6000 W house
-  4000 W. Pre-existing, kept by the N=1 identity.
+- At N=1 a drain (60) outranked the peak floor (50): a 5000 W house imported
+  3000 W against a 2500 W limit. Fixed by [[SAFETY-83]].
 - At N=2 a maintenance CHARGE in a cheap window has the same stall: the
   charging pack's `hi = -floor` against the loop asking the other to give is
   refused on 141 of 261 cycles. The mirror fix is a site `hi`, which can meet
-  the peak floor's `lo` (0 W), so it is not this card.
+  the peak floor's `lo` (0 W), so it is not this card. Fixed by [[SAFETY-84]].

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

@@ -0,0 +1,35 @@
+# SAFETY-83 - the peak guard outranks the maintenance drain at N=1
+
+**Status: DECIDED 2026-09-29 (owner go, P1).** Follows SAFETY-78's "Not
+changed" list. Tests: `test_safety83.py`.
+
+**The defect.** At N=1 the drain is the site SET at `P_MAINTENANCE` (60). The
+peak floor is a limit at `P_PEAK` (50), and clause 2 applies only limits at or
+above the winner, so the floor was ignored and the house imported what the
+drain left. A 5000 W house under a 2500 W limit and a 2000 W drain banked
+3000 W quarters; on the owner's settings (limit 3500, drain 2500, margin 50) a
+breach starts at about 5950 W.
+
+**The fix.** While a drain owns the setpoint at N=1 and the peak floor asks
+for more than the drain rate, `main.py` (`_peak_over_drain()`) refiles that
+peak claim at `P_MAINTENANCE`. The arbiter then applies it: the target is the
+floor, still inside the safety limits, and a safety SET (100) still parks
+everything. Where the floor is at or below the drain rate nothing changes.
+
+- The order in effect is now the same at N=1 and N=2: safety limits and the
+  SAFETY-73 refusal > peak guard (with the SAFETY-79 margin) > maintenance >
+  modes > trade > loop (`arbiter.py`, beside the priorities).
+- The drain keeps counting: the battery discharges at least the drain rate on
+  every cycle, so the cycle is not reset or stalled; it only ends sooner.
+- Below the drain rate the rest is still exported (owner, 2026-09-29).
+- Identity: `tools/safety78_identity.py --diff` against 35cdffe changes only
+  cycles where the guard asked for more than the drain; after such an episode
+  the commands are identical again and only the log and the idle loop's
+  integrator differ. N=2 is unchanged.
+
+**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.

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

@@ -0,0 +1,46 @@
+# SAFETY-84 - a maintenance charge at two inverters
+
+**Status: DECIDED 2026-09-30.** Follows SAFETY-78's "Not changed" list.
+Tests: `test_safety84.py`; identity: `tools/safety84_identity.py`.
+
+**The defect.** In a cheap window the charging pack files its own
+`hi = -charge` (P_MAINT_SHAPE). At N>=2 the loop's site target sat above it
+(a house to cover, or its 400 W slew step from 0 W), so the other pack had to
+give while this one charged. The allocator refuses opposite directions
+(SAFETY-31), so the site sat at 0 W whatever the house
+(`tools/safety84_identity.py` on 9d5b7d1: 436 of 480 cycles refused). N=1
+was fine.
+
+**The fix.**
+- While a pack's own limits make it charge at N>=2, `main.py` files a site
+  limit `charge`, `hi` = what that pack's own `hi` places
+  (`allocator.hi_w()`, whole W as the allocator rounds). The site target is
+  never above it, so the other pack only adds power in the same direction.
+  Opposite splits stay forbidden. Mirror of SAFETY-78's `drain` floor.
+- Peak guard first (owner order, SAFETY-83). The charge is sized to the
+  quarter's headroom under the aim (`charge_headroom_w()`, cap or banked peak
+  minus `peak_margin_w`), so it charges less instead of stopping: the meter
+  sits on the aim and the guard does not need to step in. Where the house
+  alone is over the aim the guard engages, the shaping is suspended as before
+  (`shaping_suspended()`) and no `charge` is filed, so the guard's `lo` and
+  this `hi` never meet in one cycle (no `lo > hi`, no park at 0 W).
+  Without the headroom sizing the aim still held (the suspension did it), but
+  a battery gave power back on 285 of 540 cycles at a 3000 W house
+  (test_safety84.py part 2).
+- A site stop (a safety SET above P_MAINT_SHAPE) sizes the charge to 0 W:
+  the unit `hi` is a hard limit, so a charge under a 0 W stop needs the other
+  pack to give. It was refused, with an ARBITER line, on most cycles of the
+  stop (test_safety84.py part 4). The stop still parked the site at 0 W.
+- A pack that cannot take the charge (a refusal, a full pack, a stale link)
+  files no ceiling; the other pack follows the loop.
+- Not filed at N=1: byte-identical (`tools/safety84_identity.py`, which hashes
+  every claim, in and out of a cheap window and under a banked peak).
+
+**Not changed.**
+- At N=1 the charge is not sized to the headroom: the guard still suspends it
+  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).

goodwe_controller/DOCS.md

@@ -1639,7 +1639,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). 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). 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 |
@@ -1664,6 +1664,11 @@ maintenance schedule.
 With two or more batteries, the one being emptied gives its drain rate and the
 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).
+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).
 
 #### Capacity-tariff exposure of the grid-following loop
 

goodwe_controller/app/allocator.py

@@ -354,6 +354,11 @@ def pin_w(b):
     return float(pv) if lo > hi else float(_clamp_int(round(float(pv)), lo, hi))
 
 
+def hi_w(b):
+    """A unit's upper hard limit as `plan()` narrows it to whole W (SAFETY-84)."""
+    return float(_int_envelope(b)[1])
+
+
 def loop_bound(bounds, live=None):
     """OPT-11. `(lo, hi)`: what the units can place when the LOOP decides.
 

goodwe_controller/app/arbiter.py

@@ -76,6 +76,12 @@ from . import allocator
 P_SAFETY = 100        # stopped, inputs missing, shutting down, device limits
 P_MAINTENANCE = 60    # maintenance owns the actuator outright (drain, hold)
 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).
+# 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)
 # OPT-04. §10.1 priority 3 - price arbitrage. ⚠️ ITS POSITION IS THE SPEC, NOT A
 # preference: P_PEAK (50) is ABOVE it so clause 2 lets the capacity-tariff floor
@@ -190,6 +196,12 @@ CLAIM_SCOPE = {
     # charge against the drain (circulation, refused, 0 W). A site `lo`,
     # never filed at N=1. DECISIONS/SAFETY-78-drain-at-two.md.
     "drain": SCOPE_SITE,
+    # main.py: `Claim.limit("charge", P_MAINT_SHAPE, hi=...)`, SAFETY-84. The
+    # mirror of `drain` for a cheap-window maintenance charge at N>=2: the site
+    # target may not go above what the charging pack's own `hi` asks, or a
+    # sibling would have to give against it. Never filed at N=1.
+    # DECISIONS/SAFETY-84-charge-at-two.md.
+    "charge": 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

@@ -25,6 +25,7 @@ battery to zero on its own. Never "hold the last value to be safe".
 
 import asyncio
 import contextlib
+import dataclasses
 import functools
 import json
 import logging
@@ -2513,6 +2514,58 @@ class Controller:
         self._drain_waits = waits
         return floor if floor > 0.0 else None
 
+    def _peak_over_drain(self, claims, maint_results) -> None:
+        """SAFETY-83: at N=1 a peak floor above the drain rate outranks it.
+
+        The drain is the site SET at P_MAINTENANCE, above the peak floor
+        (P_PEAK), so the floor did not bind and the house imported what the
+        drain left. Where the floor asks for more than the drain rate, it is
+        refiled at P_MAINTENANCE and wins; nothing else changes. Safety claims
+        (P_SAFETY) still outrank it. At N>=2 the `drain` floor does this.
+        """
+        if len(self.inverters) != 1:
+            return
+        res = maint_results[0]
+        if res.phase != DRAIN or not res.owns_setpoint:
+            return
+        rate = float(res.setpoint_w or 0.0)
+        for i, c in enumerate(claims):
+            if c.source == peakguard.SOURCE and c.lo > rate:
+                claims[i] = dataclasses.replace(c, priority=P_MAINTENANCE)
+
+    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
+        yields just enough to the peak guard, and nothing under a site stop
+        (its 0 W would need the other pack to give). N=1 is unchanged."""
+        if len(self.inverters) < 2:
+            return floor
+        if any(c.kind == SET and c.actuator == ACTUATOR_SITE
+               and c.priority > P_MAINT_SHAPE for c in claims):
+            return 0.0
+        room = peakguard.charge_headroom_w(self.peak_guard)
+        return floor if room is None else min(floor, room)
+
+    def _charge_ceiling(self, claims):
+        """SAFETY-84: the site `hi` for a cheap-window charge at N>=2, or None.
+
+        The charging pack's own `hi` (P_MAINT_SHAPE) makes it charge; above
+        that the other pack would have to give (opposite directions, refused,
+        0 W). So the site target stays at or below it. Not filed at N=1, nor
+        when the pack's own limits cancel the charge (a refusal, a stale link).
+        DECISIONS/SAFETY-84-charge-at-two.md.
+        """
+        if len(self.inverters) < 2:
+            return None
+        live = set(self.live_units())
+        ceiling = 0.0
+        for inv in self.inverters:
+            u = self.actuator_of("maintenance", inv)
+            b = pre_resolve([c for c in claims if c.actuator == u])
+            if u in live and not b.contradiction and b.hi < 0.0:
+                ceiling += allocator.hi_w(b)
+        return ceiling if ceiling < 0.0 else None
+
     def _maint_stop_off(self, machine, now) -> None:
         """OPT-11: stop this battery's maintenance cycle because it is off."""
         if machine.phase == IDLE:
@@ -3849,12 +3902,10 @@ class Controller:
         # claim is not sticky. Filed once and skipped next cycle, the floor
         # lapses and the actuator goes silently back to whatever was pushing the
         # peak. Which of the higher-priority claims outranks it is arbiter.py's
-        # business, not this block's - and the answer is already right: a
-        # maintenance DRAIN or HOLD owns the setpoint at P_MAINTENANCE and
-        # commands discharge or zero, neither of which can create offtake, while
-        # a maintenance CHARGE is only ever a P_MAINT_SHAPE limit (40) and is
-        # therefore bound by this floor. That is §9.1's "money outranks the
-        # maintenance schedule", now as a consequence of the priorities.
+        # 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).
         #
         # ⚠️ AND IT IS NOT RUN AT ALL WITHOUT P1. `meter_source: off` is the
         # legacy single-entity path: there is no telegram stream, so
@@ -4170,6 +4221,7 @@ class Controller:
                         floor = maintenance_charge_floor(
                             self.maint.cfg.charge_w,
                             self.peak_fc, self.peak_aim_w())
+                        floor = self._charge_headroom(floor, claims)
                         if floor > 0:
                             hi, reason = -floor, "cheap-window charge"
                     # PER-INVERTER, like the SET above and for the same reason -
@@ -4254,6 +4306,13 @@ class Controller:
             claims.append(Claim.limit("drain", P_MAINTENANCE, lo=drain,
                                       reason="maintenance drain",
                                       actuator=self.actuator_of("drain")))
+        self._peak_over_drain(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:
+            claims.append(Claim.limit("charge", P_MAINT_SHAPE, hi=charge,
+                                      reason="maintenance charge",
+                                      actuator=self.actuator_of("charge")))
         # --- OPT-11 P3: the site trade, after every unit claim --------------
         # Sized from what the batteries may place with their modes and hard
         # limits (the loop bound), so none is asked for watts it may not give.

goodwe_controller/app/messages.py

@@ -835,6 +835,7 @@ WHY_LIMIT = {
     "pack": "the battery level",
     "reserve": "the battery reserve",
     "maintenance": "battery maintenance",
+    "charge": "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: fourteen sources, each classified where it is BUILT")
+print("ARB-03 2. the split: fifteen 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
@@ -175,6 +175,10 @@ SITE_SOURCES = {
     "drain":     "the site target may not go below a pinned maintenance "
                  "drain, or a sibling would charge against it. Filed at N>1 "
                  "only",
+    # SAFETY-84.
+    "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",
 }
 UNIT_SOURCES = {
     "device":     "one inverter's own reported min/max rating",
@@ -206,9 +210,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 FOURTEEN claim sources are classified, and only those fourteen",
+check("all FIFTEEN claim sources are classified, and only those fifteen",
       set(CLAIM_SCOPE) == set(SITE_SOURCES) | set(UNIT_SOURCES))
-check("the seven that are about the GRID CONNECTION are site claims",
+check("the eight 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))
@@ -230,7 +234,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 "
-      "seven site sources stay on the grid connection",
+      "eight 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}})

goodwe_controller/test_peak_margin.py

@@ -64,9 +64,19 @@ _default = rig.digests()
 for k in ED739D4:
     print(f"    {k:7} margin 0 {'same' if _zero[k] == ED739D4[k] else 'DIFFERS'}"
           f" {_zero[k][:16]}, default {_default[k][:16]}")
+# SAFETY-84 changes n2_maint ON PURPOSE: on 9d5b7d1 its N=2 cheap-window
+# charge was refused on 214 of 540 cycles (site at 0 W); now none is.
+# Printed on the SAFETY-84 tree the same way (margin 0).
+SAFETY84_N2_MAINT = \
+    "95722b262f1bcf504493d7b5babc3895799ac56cc15f5d2ccdbe7c9f04adfcbb"
 check("⭐ at N=1 and N=2, with and without battery modes and with a "
       "maintenance charge, margin 0 is "
-      "byte-identical to ed739d4", _zero == ED739D4)
+      "byte-identical to ed739d4 (n2_maint apart, below)",
+      {k: v for k, v in _zero.items() if k != "n2_maint"}
+      == {k: v for k, v in ED739D4.items() if k != "n2_maint"})
+check("...and n2_maint at margin 0 is the SAFETY-84 digest: the N=2 charge "
+      "that 9d5b7d1 refused on 214 of 540 cycles now runs (test_safety84.py)",
+      _zero["n2_maint"] == SAFETY84_N2_MAINT != ED739D4["n2_maint"])
 check("...and the default (key absent) changes every scenario, so the "
       "scenarios do reach the margin", all(_default[k] != ED739D4[k]
                                           for k in ED739D4))

goodwe_controller/test_peakguard.py

@@ -2687,9 +2687,13 @@ check("...and so does the persistent notification, which outlives the page and "
 # state of charge, it never closes discharge (a `lo`, not a `hi`), and it is
 # never filed at N=1, so the sweeps below still see {device, supervised,
 # loop}. DECISIONS/SAFETY-78-drain-at-two.md.
+# ⚠️ SAFETY-84 ADDED `charge`, ITS MIRROR, AND IT IS NOT THE RESERVE EITHER. A
+# 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.
 CLAIM_SOURCES = {"device", "supervised", "failsafe", "safety", "peak",
                  "arbitrage", "maintenance", "loop", "circulation",
-                 "sign_site", "pack", "refused", "mode", "drain"}
+                 "sign_site", "pack", "refused", "mode", "drain", "charge"}
 # 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.
@@ -2731,7 +2735,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 FOURTEEN. THIS "
+check("...and the sources that cycle can file are EXACTLY these FIFTEEN. 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 "
@@ -2871,7 +2875,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 and SAFETY-78's additions: the seven about the GRID CONNECTION "
+      "SAFETY-30's, OPT-09's, SAFETY-78's and SAFETY-84's additions: the eight 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; "
@@ -2881,7 +2885,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"}
+          "drain", "charge"}
       and {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_UNIT}
       == {"device", "supervised", "failsafe", "maintenance", "pack",
           "refused", "mode"})
@@ -2986,7 +2990,7 @@ 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})
+                  "drain": ACTUATOR_SITE, "charge": 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)
@@ -3065,13 +3069,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 FIFTEEN of `Controller.cycle()`'s DIRECT appends route through "
+check("all SIXTEEN 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) == 15
+      len(_direct) == 16
       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 "
@@ -3088,7 +3092,7 @@ check("⚠️ AND EVERY PER-INVERTER APPEND NAMES THE INVERTER IT IS ABOUT. "
       "site-scoped sources and is the whole defect for a per-inverter one: a "
       "`device` limit built inside the roster loop and routed without `inv` "
       "files inverter B's rating on inverter A, where it binds the wrong "
-      "share. Nine of the fifteen appends are SCOPE_UNIT - `device`, "
+      "share. Nine of the sixteen appends are SCOPE_UNIT - `device`, "
       "`supervised`, `failsafe`, maintenance's two, OPT-09's `pack`, "
       "SAFETY-73's two `refused` and OPT-11's `mode` - and all nine pass one. "
       "SAFETY-30's `sign_site` is deliberately NOT one of them: it is filed on "
@@ -3105,8 +3109,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 seventeen appends account for exactly the "
-      "fourteen "
+check("...and between them the eighteen appends account for exactly the "
+      "fifteen "
       "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())
@@ -3167,7 +3171,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"})
+                          "refused", "mode", "drain", "charge"})
 # ⚠️ 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_safety78.py

@@ -15,7 +15,7 @@ DECISIONS/SAFETY-78-drain-at-two.md has the record.
 3. Battery modes on, a Cheap hours battery in a cheap run during the drain:
    no grid charge, and its card says why.
 4. The peak guard binds during a drain, and the arbiter's order holds.
-5. N=1 is byte-identical to release/1.0 5a4cdc2, and the harness bites.
+5. The N=1 and N=2 digests hold, and the harness bites.
 6. A seeded sweep of the floor against the loop, trades, the peak floor, a
    safety stop, refusals and an Off battery.
 
@@ -326,26 +326,32 @@ check("⭐ the order holds: a safety stop parks everything; the peak floor "
       and _got["a grid charge (trade)"][1] == {A: 2000, B: 0})
 
 # ============================================================================ #
-print("\n5. N=1: byte-identical to release/1.0 5a4cdc2")
-# Printed by `python3 tools/safety78_identity.py` on release/1.0 5a4cdc2 (the
-# same file copied in, nothing else changed), goodwe-addon-checks:1,
+print("\n5. the N=1 and N=2 digests")
+# Printed by `python3 tools/safety78_identity.py`, goodwe-addon-checks:1,
 # 2026-09-29. 480 cycles each, draining throughout, every claim and the
-# newest action-log line hashed.
-AT_5A4CDC2 = {
+# newest action-log line hashed. N=2: the same on release/1.0 35cdffe. N=1:
+# re-pinned by SAFETY-83, whose `--diff` against 35cdffe changes only cycles
+# where the peak guard asked for more than the drain (test_safety83.py).
+PINNED = {
     "n1_drain":
-        "55fcfc4a0b9251746f315cfdea607a33640d3a81faf6604c4718a04bbc7dd776",
+        "b0f9ef44149163bc0acece537fe047871c00f0d9ea213e7a7fc8efc4ad3aa500",
     "n1_drain_modes":
-        "93867eb4ee4132b47c030668bd7f07587a353bced2461524e09421125c3c4edf",
+        "95ab82426e23bce308db3edebe18b1ab37d956952e9f521adce37150ce87ee8a",
+    "n1_drain_owner":
+        "10f7f42a6a952e899d739eb10c674874215c440f41ead7a9f2408bc9b4cb9ea4",
+    "n2_drain":
+        "055d83de0552d3a014dc73d87fdd697dde8b5933b8032f9b3a464ff0d08795fb",
+    "n2_drain_owner":
+        "05750347f1829c249f99154613f068f442a59ef59f409093907f6d0add78dd4c",
 }
 _now = s78.digests()
-for k in AT_5A4CDC2:
-    print(f"    {k} {'same' if _now[k] == AT_5A4CDC2[k] else 'DIFFERS'} "
+for k in PINNED:
+    print(f"    {k} {'same' if _now[k] == PINNED[k] else 'DIFFERS'} "
           f"{_now[k][:16]}")
-check("⭐ at one inverter a drain under a house below, at and above its "
-      "rate, a surplus and a load over the peak limit, with battery modes "
-      "off and on, is byte-identical to 5a4cdc2: commands, reasons, roles, "
-      "trades and every claim",
-      _now == AT_5A4CDC2)
+check("⭐ a drain under a house below, at and above its rate, a surplus and a "
+      "load over the peak limit, at N=1 (modes off and on) and N=2, matches "
+      "the pinned digests: commands, reasons, roles, trades and every claim",
+      _now == PINNED)
 Controller._drain_floor = lambda self, claims, results: (
     2000.0 if any(r.phase == DRAIN for r in results) else None)
 try:

goodwe_controller/test_safety83.py

@@ -0,0 +1,308 @@
+"""Runnable check for SAFETY-83: the peak guard outranks the drain at N=1.
+`python3 test_safety83.py`
+
+At one inverter the monthly drain is the site SET at P_MAINTENANCE, above the
+peak floor, so the house imported what the drain left and quarters went over
+the peak limit. Now a peak floor above the drain rate is refiled at
+P_MAINTENANCE and wins. Drives the REAL Controller (tools/opt11_identity.py's
+harness on a P1 meter). DECISIONS/SAFETY-83-peak-over-drain.md has the record.
+
+1. A house of 5000, 6000 and 7000 W during a drain, limits 2500 and 3500,
+   margins 0 and 50: every quarter banks at or under the aim, the drain keeps
+   running, no 'Load too high' alarm, and without the fix the same runs go over.
+   Under a month peak banked above the limit the aim is that peak - margin.
+2. The drain never makes the banked quarter worse than no drain, below and
+   above the drain rate; below it the rest is still exported.
+3. The drain still ends, and the order holds: a lost reading parks the
+   battery, the rating caps the floor without a clash.
+4. Only cycles where the guard wanted more than the drain change, and N=2 is
+   untouched.
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import functools
+import logging
+import os
+import sys
+import tempfile
+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 safety78_identity as s78                                # noqa: E402
+import safety79_identity as s79                                # noqa: E402
+from app import tariff                                         # 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, MaintResult         # noqa: E402
+
+fails = []
+TOL = 1.0       # watts over the aim a banked quarter may show
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+def run(house, cap, margin, drain_w, *, maint=True, dur=2700.0, soc=None,
+        before=None, store=None, month_peak=None, **extra):
+    """N=1 on a P1 meter; returns (rows, controller, banked quarters).
+    `month_peak` banks that quarter earlier in the month."""
+    opts = dict(s79.P1, peak_cap_w=cap, peak_margin_w=margin, **extra)
+    if maint:
+        opts.update(maintenance_enabled=True, maintenance_start_hour=11,
+                    maintenance_discharge_w=drain_w)
+    db = os.path.join(tempfile.mkdtemp(prefix="s83-"), "tariff.db")
+    if month_peak is not None:
+        tariff.PeakStore(db)._db.execute(
+            "INSERT INTO monthly_peak VALUES (?, ?, ?, NULL)",
+            (tariff.month_key(rig.BASE), month_peak,
+             "2026-09-02T10:00:00+00:00")).connection.commit()
+    real = rig.Controller
+    rig.Controller = functools.partial(real, peak_db_path=db)
+    try:
+        rows, c = rig.run(1, extra=opts, dur=dur,
+                          load=house if callable(house) else lambda t: house,
+                          soc=soc or (lambda u, t: 60.0),
+                          cut=lambda u, t: False, before=before or s79._feed,
+                          store=store)
+    finally:
+        rig.Controller = real
+    return rows, c, {b.start.strftime("%H:%M"): b.offtake_avg_w
+                     for b in c.p1.blocks}
+
+
+def without_fix(fn):
+    real = Controller._peak_over_drain
+    Controller._peak_over_drain = lambda self, claims, results: None
+    try:
+        return fn()
+    finally:
+        Controller._peak_over_drain = real
+
+
+def cmd(r):
+    return dict(r[2])["a"]
+
+
+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)
+
+
+# ============================================================================ #
+print("1. banked quarters during a drain, N=1")
+CASES = [(cap, drain, m, h) for cap, drain in ((2500, 2000), (3500, 2500))
+         for m in (0, 50) for h in (5000.0, 6000.0, 7000.0)]
+RES, OLD = {}, {}
+for k in CASES:
+    cap, drain, m, h = k
+    RES[k] = run(h, cap, m, drain)
+    OLD[k] = without_fix(lambda: run(h, cap, m, drain))
+    print(f"    limit {cap} drain {drain} margin {m:>2} house {h:.0f}: "
+          f"banked {[round(v) for v in RES[k][2].values()]} W "
+          f"(before the fix {[round(v) for v in OLD[k][2].values()]} W)")
+check("⭐ every quarter banks at or under the aim (limit - margin) plus 1 W, "
+      "for a 5000, 6000 and 7000 W house, limits 2500 and 3500, margins 0 "
+      "and 50",
+      all(len(b) >= 2 and max(b.values()) <= k[0] - k[2] + TOL
+          for k, (_r, _c, b) in RES.items()))
+check("⭐ the drain keeps running: the phase is drain on every run and the "
+      "battery gives at least the drain rate on every cycle",
+      all(c.maints[0].phase == DRAIN and all(cmd(r) >= k[1] for r in rows)
+          for k, (rows, c, _b) in RES.items()))
+check("...and where the house is over drain + aim, the battery gives more "
+      "than the drain rate, bound by the peak floor",
+      all(any(cmd(r) > k[1] and "limited by peak" in r[6] for r in rows)
+          for k, (rows, _c, _b) in RES.items() if k[3] - k[1] > k[0] - k[2]))
+check("non-vacuous: without the fix the same runs bank over the limit "
+      "wherever the house is above drain + aim (the reported defect: 3000 W "
+      "at a 5000 W house against 2500 W)",
+      all(max(b.values()) > k[0] + 100.0 for k, (_r, _c, b) in OLD.items()
+          if k[3] - k[1] > k[0])
+      and round(OLD[(2500, 2000, 50, 5000.0)][2]["11:15"]) == 3000)
+# A hob cycling: one 5 s spike to 11000 W every 150 s on a 6000 W house. Each
+# spike saturates the guard for one cycle, shorter than ALARM_HOLD_S.
+_sat = []
+
+
+def _spiky_feed(c, t):
+    s79._feed(c, t)
+    _sat.append(bool(c.peak_guard is not None and c.peak_guard.saturated))
+
+
+_k_rows, _k, _kb = run(lambda t: 11000.0 if t % 150 < 5 else 6000.0,
+                       3500, 50, 2500, before=_spiky_feed)
+check("⭐ a drain the guard raises gives no 'Load too high' alarm (SAFETY-81), "
+      "steady or with 5 s spikes that saturate the guard",
+      not any(alarmed(c) for _r, c, _b in RES.values())
+      and any(_sat) and not alarmed(_k))
+
+# Owner (SAFETY-79): under a month peak banked above the limit, the aim is that
+# peak - margin, during a drain too. Owner's settings, peak banked at 4500 W.
+BANK = {h: (run(h, 3500, 50, 2500, month_peak=4500.0),
+            without_fix(lambda: run(h, 3500, 50, 2500, month_peak=4500.0)))
+        for h in (6500.0, 7500.0)}
+for h, (new, old) in BANK.items():
+    print(f"    month peak 4500, house {h:.0f}: banked "
+          f"{[round(v) for v in new[2].values()]} W (before the fix "
+          f"{[round(v) for v in old[2].values()]} W)")
+check("⭐ under a month peak of 4500 W: a 7500 W house banks at 4450 W "
+      "(peak - margin, not the 3450 W cap aim), was 5000 W; a 6500 W house "
+      "stays under the drain alone at 4000 W",
+      all(4430.0 <= v <= 4450.0 + TOL for v in BANK[7500.0][0][2].values())
+      and max(BANK[7500.0][1][2].values()) > 4900.0
+      and all(round(v) in (4000, 4014)
+              for v in BANK[6500.0][0][2].values()))
+
+# ============================================================================ #
+print("\n2. the drain never makes the banked quarter worse, N=1")
+CAP, DRAIN_W, MARGIN = 3500, 2500, 50          # the owner's settings
+AIM = CAP - MARGIN
+WORSE = {}
+for h in (1200.0, 3000.0, 5000.0, 6500.0, 7000.0):
+    _rows, _c, b = run(h, CAP, MARGIN, DRAIN_W)
+    _r0, _c0, b0 = run(h, CAP, MARGIN, DRAIN_W, maint=False)
+    WORSE[h] = (b, b0, _rows)
+    print(f"    house {h:.0f}: with the drain {[round(v) for v in b.values()]}"
+          f" W, without {[round(v) for v in b0.values()]} W")
+check("⭐ below and above the drain rate, every quarter with the drain banks "
+      "no more than the larger of the aim and the same quarter without it",
+      all(b[q] <= max(AIM, b0[q]) + TOL for b, b0, _r in WORSE.values()
+          for q in b))
+_r12 = WORSE[1200.0][2]
+check("below the drain rate the battery gives the drain rate and exports the "
+      "rest (owner, 2026-09-29): 2500 W against a 1200 W house",
+      all(cmd(r) == DRAIN_W for r in _r12[1:]))
+
+# ============================================================================ #
+print("\n3. the drain ends, and the order holds")
+_e_rows, _e, _eb = run(6500.0, CAP, MARGIN, DRAIN_W, dur=1200.0,
+                       soc=lambda u, t: max(5.0, 60.0 - t / 20.0))
+check("⭐ with the guard binding, the drain still reaches its floor and moves "
+      "to the charge phase, with no abort",
+      _e.maints[0].phase == CHARGE
+      and not any("abort" in e.lower() for e in _e.events))
+
+
+def blind_at(t0):
+    """The battery power reading goes missing at `t0`: a safety SET."""
+    def before(c, t):
+        s79._feed(c, t)
+        if t >= t0 and not getattr(c.hass, "_blind", False):
+            real = c.hass.number
+
+            async def number(e, invert=False):
+                return None if e == "sensor.batt_a" else await real(e, invert)
+            c.hass.number, c.hass._blind = number, True
+    return before
+
+
+_s_rows, _s, _sb = run(6500.0, CAP, MARGIN, DRAIN_W, dur=600.0,
+                       before=blind_at(300.0))
+print(f"    reading lost at 300 s: last {cmd(_s_rows[-1]):+.0f} W, "
+      f"{_s_rows[-1][6]}")
+check("⭐ a safety SET outranks the refiled peak floor: before the reading "
+      "goes the battery gives above the drain rate, after it it is at 0 W",
+      any(cmd(r) > DRAIN_W for r in _s_rows if r[0] < 300.0)
+      and all(cmd(r) == 0.0 for r in _s_rows if r[0] >= 400.0)
+      and _s_rows[-1][6].startswith("safety -> 0 W"))
+_m_rows, _m, _mb = run(7000.0, 2500, 50, 2000, dur=900.0, max_w=3000)
+print(f"    max_w 3000, house 7000: last {cmd(_m_rows[-1]):+.0f} W, "
+      f"{_m_rows[-1][6]}")
+check("⭐ the rating caps the floor without a clash: at max_w 3000 the "
+      "battery gives 3000 W, never 0 W, and nothing contradicts",
+      all(cmd(r) >= 2000.0 for r in _m_rows)
+      and cmd(_m_rows[-1]) == 3000.0
+      and not any("CONTRADICTION" in (r[6] or "") for r in _m_rows))
+check("...and there, with the battery at its rating and the house still over, "
+      "the 'Load too high' alarm does fire", alarmed(_m))
+
+# ============================================================================ #
+print("\n4. only the cycles where the guard wanted more change")
+
+
+def promoted(row):
+    return [c for c in row[-1][0] if c[0] == "peak"]
+
+
+_id = {k: s78.run(**v)[0] for k, v in s78.SCENARIOS.items()}
+_bad = []
+for k, rows in _id.items():
+    for r in rows:
+        claims, guard = r[-1][0], r[-1][1]
+        rate = [c[3] for c in claims if c[0] == "maintenance" and c[2] == "set"]
+        pk = promoted(r)
+        if k.startswith("n2"):
+            ok = all(c[1] == P_PEAK for c in pk)
+        elif s78.wanted_more(r):
+            ok = ([c[1] for c in pk] == [P_MAINTENANCE]
+                  and r[1] == guard[1] and "limited by peak" in r[6])
+        else:
+            ok = (all(c[1] == P_PEAK for c in pk)
+                  and (not rate or r[1] == rate[0]))
+        if not ok:
+            _bad.append((k, r[0], r[1], r[6], pk))
+_bad1 = [b for b in _bad if b[0].startswith("n1")]
+_bad2 = [b for b in _bad if b[0].startswith("n2")]
+_more = {k: sum(map(s78.wanted_more, rows)) for k, rows in _id.items()}
+print(f"    cycles with the guard above the drain: {_more}; bad {_bad[:3]}")
+check("⭐ at N=1, on every cycle where the guard asks for more than the drain "
+      "the peak claim is refiled at P_MAINTENANCE and the target is its floor; "
+      "on every other cycle the peak claim stays at P_PEAK and the target is "
+      "the drain rate",
+      not _bad1 and all(_more[k] > 50 for k in _id if k.startswith("n1")))
+check("⭐ at N=2 the peak claim is never refiled (the SAFETY-78 floor does "
+      "that job), and test_safety78.py part 5 holds N=2 to the digests of "
+      "release/1.0 35cdffe", not _bad2)
+
+
+# The rule itself, at the boundary: a floor AT the drain rate changes nothing.
+_one = types.SimpleNamespace(inverters=[None])
+_edge = {}
+for lo in (1999.5, 2000.0, 2000.5):
+    cl = [Claim.set("maintenance", P_MAINTENANCE, 2000.0, DRAIN),
+          Claim.limit("peak", P_PEAK, lo=lo, reason="quarter peak")]
+    Controller._peak_over_drain(_one, cl, [MaintResult(DRAIN, True, 2000.0,
+                                                       False, [])])
+    _edge[lo] = cl[1].priority
+check("at the boundary: a floor at or below the 2000 W drain stays at P_PEAK, "
+      "a floor 0.5 W above it is refiled at P_MAINTENANCE",
+      _edge == {1999.5: P_PEAK, 2000.0: P_PEAK, 2000.5: P_MAINTENANCE})
+
+_stop = [Claim.set("maintenance", P_MAINTENANCE, 2000.0, DRAIN),
+         Claim.limit("peak", P_PEAK, lo=4000.0, reason="quarter peak"),
+         Claim.set("safety", P_SAFETY, 0.0, "inputs missing")]
+Controller._peak_over_drain(_one, _stop, [MaintResult(DRAIN, True, 2000.0,
+                                                      False, [])])
+check("⭐ and a refiled floor stays under a safety SET: the site resolves "
+      "to 0 W", _stop[1].priority == P_MAINTENANCE
+      and resolve(_stop).target_w == 0.0)
+
+
+def store_hold():
+    st = rig._Store()
+    st.data["phase"], st.data["phase_started"] = "hold", rig.BASE.isoformat()
+    return st
+
+
+_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",
+      _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))
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED: {', '.join(fails)}")
+    sys.exit(1)
+print("all checks passed")

goodwe_controller/test_safety84.py

@@ -0,0 +1,503 @@
+"""Runnable check for SAFETY-84: a maintenance charge at N=2 no longer stalls.
+`python3 test_safety84.py`
+
+In a cheap window the charging battery's own `hi` makes it charge. At two
+inverters the loop's site target sat above it, the other battery had to give,
+the allocator refused the opposite directions and the site sat at 0 W. Now a
+site `hi` (`charge`, P_MAINT_SHAPE) keeps the site target at or below the
+charge, and the charge is sized to the quarter's headroom so it yields to the
+peak guard. Drives the REAL Controller (tools/safety84_identity.py's harness).
+DECISIONS/SAFETY-84-charge-at-two.md has the record.
+
+1. N=2 in a cheap window, house below, at and above the charge rate and a
+   solar surplus: the charge runs, no opposite split, no refusal, and it ends.
+2. The peak guard: the charge yields just enough to hold the aim (cap -
+   margin, and a banked month peak - margin); over the aim the guard wins,
+   the charging battery gives too, and nothing parks; no false 'Load too
+   high'. Without P1 the charge is not sized, as at N=1.
+3. Outside a cheap window nothing changes.
+4. A safety stop parks both batteries without a refusal, and the charge
+   resumes after it.
+5. One battery refusing, or dropped, and B Off with battery modes on.
+6. A trade outranked by the charge (P_ARBITRAGE < P_MAINT_SHAPE).
+7. A seeded sweep of the ceiling against the loop, trades, refusals, packs
+   and modes.
+8. The N=1 digests equal release/1.0 9d5b7d1, and the N=2 ones are pinned.
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import logging
+import os
+import random
+import sys
+
+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 opt11p3_identity as p3                                  # noqa: E402
+import safety84_identity as s84                                # noqa: E402
+from app import arbitrage, packmode                            # noqa: E402
+from app.arbiter import (Claim, P_ARBITRAGE, P_LOOP, P_MAINT_SHAPE,  # noqa
+                         P_MODE, P_SAFETY, resolve)
+from app.main import Controller                                # noqa: E402
+from app.maintenance import CHARGE, HOLD                       # noqa: E402
+from app.store import unit_key                                 # noqa: E402
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+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 cmd(r):
+    return dict(r[2])
+
+
+def refused(rows):
+    return sum(1 for r in rows if r[7] and "REFUSED" in r[7])
+
+
+def opposite(rows):
+    return sum(1 for r in rows if cmd(r)["a"] * cmd(r)["b"] < 0)
+
+
+def charge_claims(r):
+    """The site `charge` claims main.py handed resolve() on row `r`."""
+    return [c for c in r[-1][0] if c[0] == "charge"]
+
+
+def banked(c):
+    return [b.offtake_avg_w for b in c.p1.blocks]
+
+
+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 run(house, n=2, **kw):
+    return s84.run(n, load=house if callable(house) else lambda t: house,
+                   **kw)
+
+
+def rise(u, t):
+    """A fills from 60 % (1 % a cycle); B stays at 50 %."""
+    return min(100.0, 60.0 + t / 5.0) if u == "a" else 50.0
+
+
+# ============================================================================ #
+print("1. N=2 in a cheap window: the charge runs whatever the house uses")
+RUNS = {h: run(h, dur=900.0) for h in (800.0, 2000.0, 3000.0, -1500.0)}
+for h, (rows, c) in RUNS.items():
+    print(f"    house {h:+.0f} W: last A {cmd(rows[-1])['a']:+.0f} W, B "
+          f"{cmd(rows[-1])['b']:+.0f} W; {rows[-1][6]}; refusals "
+          f"{refused(rows)}")
+check("⭐ below, at and above the charge rate and under a solar surplus, "
+      "battery A charges on EVERY cycle (it used to sit at 0 W from the "
+      "first cycle)",
+      all(cmd(r)["a"] < 0.0 for rows, _c in RUNS.values() for r in rows)
+      and all(c.maints[0].phase == CHARGE for _r, c in RUNS.values()))
+check("⭐ no cycle splits the site both ways and the allocator refuses "
+      "nothing: battery B never gives while A charges",
+      all(opposite(rows) == 0 and refused(rows) == 0
+          and all(cmd(r)["b"] <= 0.0 for r in rows)
+          for rows, _c in RUNS.values()))
+check("...and nothing is logged per cycle: no ARBITER line in the action log",
+      not any("ARBITER" in e for _r, c in RUNS.values() for e in c.events))
+check("with the quarter's headroom above the charge rate, A charges at the "
+      "full 2000 W and the Why names the charge",
+      all(cmd(r)["a"] == -2000.0 for h in (800.0, -1500.0)
+          for r in RUNS[h][0])
+      and "limited by charge" in RUNS[800.0][0][-1][6])
+print(f"    Why: {RUNS[800.0][1].why_words}")
+check("...and the page's Why reads as at one battery: 'held back by battery "
+      "maintenance'",
+      RUNS[800.0][1].why_words.endswith("held back by battery maintenance"))
+
+_end_rows, _end = run(800.0, dur=600.0, soc=rise)
+check("⭐ and the charge ENDS: A reaches its 99 % target and the cycle moves "
+      "on to its hold, with no refusal on the way",
+      _end.maints[0].phase == HOLD and refused(_end_rows) == 0
+      and any("charge complete" in e for e in _end.events))
+
+_real = Controller._charge_ceiling
+Controller._charge_ceiling = lambda self, claims: None
+try:
+    _bite, _bc = run(800.0, dur=120.0)
+finally:
+    Controller._charge_ceiling = _real
+print(f"    without the ceiling: refusals {refused(_bite)}/{len(_bite)}, "
+      f"last A {cmd(_bite[-1])['a']:+.0f} W")
+check("non-vacuous: with the ceiling taken out, the same fixture refuses "
+      "every cycle and parks A at 0 W (the reported defect)",
+      refused(_bite) == len(_bite) and cmd(_bite[-1])["a"] == 0.0)
+
+# ============================================================================ #
+print("\n2. the peak guard outranks the charge")
+_q_rows, _q = run(3000.0, dur=2700.0)
+print(f"    house 3000: banked {[round(v) for v in banked(_q)]} W, last A "
+      f"{cmd(_q_rows[-1])['a']:+.0f} W")
+check("⭐ house 3000 W, aim 3450 W: the charge yields just enough - every "
+      "banked quarter lands on the aim, not on house + 2000 W",
+      len(banked(_q)) >= 2
+      and all(AIM - TOL <= v <= AIM + TOL for v in banked(_q)))
+check("...and it yields by charging less, not by stopping: A charges and B "
+      "is 0 W on every cycle, nothing is refused, no 'Load too high'",
+      all(cmd(r)["a"] < 0.0 and cmd(r)["b"] == 0.0 for r in _q_rows)
+      and refused(_q_rows) == 0 and not alarmed(_q))
+check("...and the peak guard never has to step in (the charge stays under "
+      "what the quarter can afford)",
+      not any(r[-1][1] and r[-1][1][0] for r in _q_rows))
+
+_o_rows, _o = run(5000.0, dur=2700.0)
+_both = [r for r in _o_rows
+         if charge_claims(r) and any(c[0] == "peak" for c in r[-1][0])]
+print(f"    house 5000: banked {[round(v) for v in banked(_o)]} W, last "
+      f"{cmd(_o_rows[-1])}, {_o_rows[-1][6]}")
+check("⭐ house 5000 W, over the aim alone: the peak guard wins - every "
+      "banked quarter is at or under the aim, both batteries give in the "
+      "same direction, nothing is refused and nothing parks at 0 W",
+      all(v <= AIM + TOL for v in banked(_o)) and opposite(_o_rows) == 0
+      and refused(_o_rows) == 0
+      and sum(cmd(r)["a"] + cmd(r)["b"] > 0.0 for r in _o_rows) > 100)
+check("...and the guard's `lo` never meets the charge's `hi` in one cycle "
+      "(the charge is suspended while the guard holds)", not _both)
+check("...with no false 'Load too high' alarm", not alarmed(_o))
+
+_k_rows, _k = run(3000.0, dur=2700.0, month_peak=4000.0)
+print(f"    month peak 4000 banked, house 3000: banked "
+      f"{[round(v) for v in banked(_k)]} W")
+check("⭐ under a month peak banked at 4000 W the aim is 3950 W: the charge "
+      "takes the extra headroom and every banked quarter lands on 3950 W",
+      len(banked(_k)) >= 2
+      and all(3950.0 - TOL <= v <= 3950.0 + TOL for v in banked(_k))
+      and refused(_k_rows) == 0 and not alarmed(_k))
+
+_real = Controller._charge_headroom
+Controller._charge_headroom = lambda self, floor, claims: floor
+try:
+    _nh_rows, _nh = run(3000.0, dur=2700.0)
+finally:
+    Controller._charge_headroom = _real
+_gave = sum(1 for r in _nh_rows if cmd(r)["a"] > 0.0 or cmd(r)["b"] > 0.0)
+print(f"    without the headroom: cycles a battery gives {_gave}, banked "
+      f"{[round(v) for v in banked(_nh)]} W")
+check("non-vacuous: charging at the full rate instead, the guard has to "
+      "step in and the batteries give back on many cycles",
+      _gave > 50)
+
+
+def peak_run(house):
+    rows, c = run(house, dur=1800.0)
+    return rows, c, [r for r in rows if r[-1][1] and r[-1][1][0]]
+
+
+# A 6500 W house needs 3050 W off the meter: more than B's 2500 W alone.
+_p_rows, _p, _p_eng = peak_run(6500.0)
+print(f"    house 6500: banked {[round(v) for v in banked(_p)]} W, A gives "
+      f"on {sum(cmd(r)['a'] > 0.0 for r in _p_eng)} of {len(_p_eng)} "
+      f"engaged cycles")
+check("⭐ house 6500 W, more than B alone can cover: the charge is suspended "
+      "and A, the battery being filled, gives too; the quarter lands on the "
+      "aim, nothing is refused, no 'Load too high'",
+      len(_p_eng) > 100 and all(cmd(r)["a"] > 0.0 for r in _p_eng)
+      and all(v <= AIM + TOL for v in banked(_p))
+      and refused(_p_rows) == 0 and not alarmed(_p))
+_real = s84.m.peakguard.shaping_suspended
+s84.m.peakguard.shaping_suspended = lambda *a: False
+try:
+    _ns_rows, _ns, _ns_eng = peak_run(6500.0)
+finally:
+    s84.m.peakguard.shaping_suspended = _real
+check("non-vacuous: with the suspension taken out, A is held at 0 W, the "
+      "quarter goes over the cap and 'Load too high' fires",
+      not any(cmd(r)["a"] > 0.0 for r in _ns_eng)
+      and max(banked(_ns)) > 3500.0 and alarmed(_ns))
+
+
+def prices_only(c, t):
+    """A cheap price and no P1 telegram (`meter_source: off`)."""
+    if not getattr(c.hass, "_px", False):
+        read, px = c.hass.number, {"sensor.pnow": 0.05, "sensor.pavg": 0.20}
+
+        async def number(e, invert=False):
+            return px[e] if e in px else await read(e, invert)
+        c.hass.number, c.hass._px = number, True
+
+
+_p1 = s84.s79.P1
+s84.s79.P1 = {"meter_source": "off", "peak_cap_w": 3500}
+try:
+    _off_rows, _off = run(800.0, dur=600.0, before=prices_only)
+finally:
+    s84.s79.P1 = _p1
+check("⭐ with no P1 meter nothing measures the quarter and the charge is not "
+      "sized, as at N=1: A charges at 2000 W on every cycle, nothing refused",
+      _off.peak_guard is None and all(cmd(r)["a"] == -2000.0 for r in _off_rows)
+      and refused(_off_rows) == 0)
+
+# ============================================================================ #
+print("\n3. outside a cheap window nothing changes")
+_d = {h: run(h, cheap=False, dur=600.0) for h in (800.0, -1500.0)}
+check("⭐ no `charge` claim is filed outside a cheap window: B covers the "
+      "house, the charge takes solar only, nothing is refused",
+      not any(charge_claims(r) for rows, _c in _d.values() for r in rows)
+      and cmd(_d[800.0][0][-1]) == {"a": 0.0, "b": 800.0}
+      and cmd(_d[-1500.0][0][-1])["a"] < 0.0
+      and all(refused(rows) == 0 for rows, _c in _d.values()))
+
+
+# ============================================================================ #
+print("\n4. a safety stop outranks the charge")
+
+
+def blind_at(t0, t1):
+    """A's power reading is missing from `t0` to `t1`: a safety SET."""
+    def before(c, t):
+        s84.feed(True)(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
+
+
+_s_rows, _s = run(800.0, dur=600.0, before=blind_at(150.0, 400.0))
+_stop = [r for r in _s_rows if 175.0 <= r[0] < 400.0]
+_last = _s_rows[79]                     # t = 395 s, still without the reading
+print(f"    A's reading lost 150-400 s: {_last[2]}, {_last[6]}; after: "
+      f"{_s_rows[-1][2]}")
+check("⭐ with a reading gone the safety SET parks both batteries at 0 W "
+      "with no refusal and no ARBITER line (it used to refuse every cycle)",
+      all(cmd(r) == {"a": 0.0, "b": 0.0} for r in _stop)
+      and _last[0] == 395.0 and _last[6].startswith("safety -> 0 W")
+      and refused(_s_rows) == 0 and not any("ARBITER" in e for e in _s.events))
+check("...and the charge ran before and runs again after",
+      cmd(_s_rows[20])["a"] == -2000.0 and cmd(_s_rows[-1])["a"] == -2000.0)
+
+
+_full = _s._charge_headroom(2000.0, [])
+check("the stop test reads only SITE SETs: a unit's own safety SET (on B) "
+      "or a site limit does not cut A's charge, a site safety SET does",
+      _s._charge_headroom(2000.0, [Claim.set("failsafe", P_SAFETY, 0.0,
+                                             actuator=B)]) == _full > 0.0
+      and _s._charge_headroom(2000.0, [Claim.limit("peak", P_SAFETY,
+                                                   lo=0.0)]) == _full
+      and _s._charge_headroom(2000.0, [Claim.set("safety", P_SAFETY,
+                                                 0.0)]) == 0.0)
+
+
+def stop_at(t0):
+    def before(c, t):
+        s84.feed(True)(c, t)
+        if t >= t0:
+            c.auto = False
+    return before
+
+
+_st_rows, _st = run(800.0, dur=400.0, before=stop_at(200.0))
+check("control switched off mid-charge: both batteries at 0 W, no refusal",
+      cmd(_st_rows[-1]) == {"a": 0.0, "b": 0.0} and refused(_st_rows) == 0)
+
+# ============================================================================ #
+print("\n5. one battery refusing, dropped, or Off")
+
+
+def a_takes_nothing(c, t):
+    """A reads 0 W whenever it is asked to charge (a battery that refuses)."""
+    s84.feed(True)(c, t)
+    if not getattr(c.hass, "_rt", False):
+        real = c.hass.power
+        c.hass.power = lambda u: (0.0 if u == "a" and c.hass.cmd["a"] < 0.0
+                                  else real(u))
+        c.hass._rt = True
+
+
+_r_rows, _r = run(800.0, dur=900.0, before=a_takes_nothing)
+print(f"    A refuses to charge: {_r_rows[-1][2]}, {_r_rows[-1][6]}")
+check("⭐ A refuses to charge (SAFETY-73): no ceiling is filed for it, B "
+      "covers the 800 W house, and no cycle is split both ways or refused",
+      _r.inverters[0].refused == -1 and cmd(_r_rows[-1]) == {"a": 0.0,
+                                                           "b": 800.0}
+      and not charge_claims(_r_rows[-1])
+      and opposite(_r_rows) == 0 and refused(_r_rows) == 0)
+
+_g_rows, _g = run(3000.0, dur=600.0, cut=lambda u, t: u == "b")
+check("B refuses to give: A still charges and nothing is refused",
+      all(cmd(r)["a"] < 0.0 for r in _g_rows) and refused(_g_rows) == 0)
+
+
+def reject_a(c, t):
+    """Every write to A is rejected, so A's link goes stale (not live)."""
+    s84.feed(True)(c, t)
+    if not getattr(c.hass, "_rej", False):
+        real = c.hass.set_number
+
+        async def set_number(e, v):
+            return False if e == "number.sp_a" else await real(e, v)
+        c.hass.set_number, c.hass._rej = set_number, True
+
+
+_l_rows, _l = run(1500.0, dur=400.0, before=reject_a)
+print(f"    A's writes rejected: {_l_rows[-1][2]}, {_l_rows[-1][7][-40:]}")
+check("⭐ a charging battery that is dropped (stale link) adds no ceiling: "
+      "B covers the 1500 W house instead of charging for A",
+      "dropped inverter-a" in _l_rows[-1][7]
+      and 1490.0 <= cmd(_l_rows[-1])["b"] <= 1530.0
+      and not charge_claims(_l_rows[-1]))
+
+
+def modes_store(b_mode=None, b_off=False):
+    s = rig._Store()
+    for k in ("phase", "phase_started"):
+        s.data[unit_key(A, k)] = ("charge" if k == "phase"
+                                  else rig.BASE.isoformat())
+    if b_mode:
+        s.data[unit_key(B, packmode.MODE_KEY)] = b_mode
+    if b_off:
+        s.data[unit_key(B, packmode.USE_KEY)] = False
+    return s
+
+
+_m = {h: run(h, dur=600.0, extra={"battery_modes_enabled": True},
+             store=modes_store(b_off=True)) for h in (800.0, -3000.0)}
+print(f"    B Off: {[(h, cmd(rows[-1])) for h, (rows, _c) in _m.items()]}")
+check("⭐ battery modes on, B Off: A charges at 2000 W over an 800 W house "
+      "and takes 2500 W of a 3000 W surplus, B stays at 0 W, nothing is "
+      "refused",
+      cmd(_m[800.0][0][-1]) == {"a": -2000.0, "b": 0.0}
+      and cmd(_m[-3000.0][0][-1]) == {"a": -2500.0, "b": 0.0}
+      and all(refused(rows) == 0 and opposite(rows) == 0
+              for rows, _c in _m.values()))
+
+# ============================================================================ #
+print("\n6. a trade does not outrank the charge")
+
+
+def sell_feed(c, t):
+    p3._feed(c, t)
+    s84.feed(True)(c, t)
+
+
+_trades = []
+_t_rows, _t = run(800.0, dur=5400.0, before=sell_feed,
+                  extra=dict(p3.ARB),
+                  soc=lambda u, t: 60.0)
+print(f"    modes off, a sell at 75-90 min: last {_t_rows[-1][2]}")
+_sells = [r for r in _t_rows if any(c[0] == "arbitrage" and c[3] > 0.0
+                                    for c in r[-1][0])]
+check("⭐ battery modes off: the site trade asks to SELL while A charges; "
+      "the charge (P_MAINT_SHAPE) outranks it, so A keeps charging and "
+      "nothing is refused",
+      len(_sells) > 50
+      and all(cmd(r)["a"] < 0.0 and cmd(r)["b"] <= 0.0 for r in _sells)
+      and refused(_t_rows) == 0)
+
+# ============================================================================ #
+print("\n7. a seeded sweep of the ceiling")
+_c = run(800.0, dur=30.0)[1]
+_rnd = random.Random(84)
+_bad = []
+for _i in range(20000):
+    ra, rb = _rnd.choice((2000.0, 2500.0)), _rnd.choice((1000.0, 2500.0))
+    f = _rnd.uniform(1.0, ra)
+    cl = [Claim.limit("supervised", P_SAFETY, -ra, ra, actuator=A),
+          Claim.limit("supervised", P_SAFETY, -rb, rb, actuator=B),
+          Claim.limit("maintenance", P_MAINT_SHAPE, hi=-f, actuator=A)]
+    x = _rnd.random()
+    if x < 0.1:
+        cl.append(Claim.limit("refused", P_SAFETY, 0.0, actuator=A))
+    elif x < 0.2:
+        cl.append(Claim.limit("pack", P_SAFETY, 0.0, actuator=A))
+    y = _rnd.random()
+    if y < 0.15:
+        cl.append(Claim.limit("refused", P_SAFETY, *_rnd.choice(
+            ((0.0, float("inf")), (float("-inf"), 0.0))), actuator=B))
+    elif y < 0.3:
+        cl.append(Claim.limit("mode", P_MODE, *_rnd.choice(
+            ((0.0, 0.0), (0.0, rb), (-rb, 0.0))), actuator=B))
+    cl.append(Claim.set("loop", P_LOOP, _rnd.uniform(-6000, 6000)))
+    if _rnd.random() < 0.4:
+        cl.append(Claim.set("arbitrage", P_ARBITRAGE,
+                            _rnd.uniform(-5000, 5000)))
+    ceil = _c._charge_ceiling(cl)
+    if ceil is not None:
+        cl.append(Claim.limit("charge", P_MAINT_SHAPE, hi=ceil))
+    res = resolve(cl, units=(A, B), live={A, B})
+    sh = res.shares or {}
+    why = []
+    if not res.allocated or res.contradiction:
+        why.append("refused or contradicted")
+    if sh.get(A, 0.0) * sh.get(B, 0.0) < 0.0:
+        why.append("opposite")
+    a_free = x >= 0.2
+    if a_free and (ceil is None or sh.get(A, 0.0) > -f + 1.0):
+        why.append("A does not charge")
+    if not a_free and ceil is not None:
+        why.append("ceiling filed for a pack that cannot charge")
+    if why:
+        _bad.append((_i, why))
+print(f"    20000 claim lists, bad {len(_bad)}: {_bad[:3]}")
+check("⭐ 20000 random N=2 claim lists (a charge on A, refusals, a full "
+      "pack, B's mode, the loop and trades in both directions): never "
+      "refused, never both ways, A always charges at least its rate, and "
+      "no ceiling for a pack that cannot charge", not _bad)
+
+# ============================================================================ #
+print("\n8. the N=1 and N=2 digests")
+# Printed by `python3 tools/safety84_identity.py` in goodwe-addon-checks:1,
+# 2026-09-30, 480 cycles each, every claim hashed. N=1 and n2_charge_dear:
+# the same on release/1.0 9d5b7d1 (the tool copied in). N=2 in a cheap
+# window: new, the charge that 9d5b7d1 refused on every cycle.
+PINNED = {
+    "n1_charge":
+        "0ed25c856456b97acb07a39299a8cb7b1a762c4d0005b88a0823c03057a80be5",
+    "n1_charge_dear":
+        "c25340b56060d4581f7204e200b07e7c8618711be2baca3b1667b7ebeb600dc3",
+    "n1_charge_banked":
+        "2619acdd06b04e56c000b168d64e1d044346c6ba3f929c94258d6ad8189ec609",
+    "n2_charge":
+        "748abe785a79084bc738c82d0d6f90b951e1cb78e9629195c0c01d3bb67b55e2",
+    "n2_charge_dear":
+        "c34f08b3b5b6256e68cf05030fb0c29522916dc8f57c351008f48384657966c0",
+    "n2_charge_banked":
+        "fb600ee1dd3202910715c97c5809584d50c6b8b57b55aaa4dc83bcaaa97627ef",
+}
+_now = s84.digests()
+for k in PINNED:
+    print(f"    {k} {'same' if _now[k] == PINNED[k] else 'DIFFERS'} "
+          f"{_now[k][:16]}")
+check("⭐ at N=1 a charge in and out of a cheap window and under a banked "
+      "month peak is byte-identical to 9d5b7d1 (commands, reasons and every "
+      "claim), and so is N=2 outside a cheap window",
+      all(_now[k] == PINNED[k] for k in ("n1_charge", "n1_charge_dear",
+                                         "n1_charge_banked",
+                                         "n2_charge_dear")))
+check("...and the N=2 cheap-window digests match the pins",
+      _now["n2_charge"] == PINNED["n2_charge"]
+      and _now["n2_charge_banked"] == PINNED["n2_charge_banked"])
+_n1 = s84.run(1, dur=600.0)[0]
+check("...and no N=1 cycle files a `charge` claim",
+      not any(charge_claims(r) for r in _n1))
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED: " + ", ".join(fails))
+    sys.exit(1)
+print("all SAFETY-84 checks passed")

goodwe_controller/tools/safety78_identity.py

@@ -13,11 +13,18 @@ no watt still moves the digest. The `_modes` scenarios run battery modes on,
 Cheap hours only, over opt11p3_identity's prices. Uses only the API release/1.0 5a4cdc2 has, so the same file runs on
 both trees; test_safety78.py part 5 holds this tree to the digests printed
 there.
+
+SAFETY-83 adds the owner's settings (limit 3500, drain 2500, margin 50) with a
+house that goes over drain + aim and back, and N=2 runs of both. At N=1 the
+peak floor now outranks the drain where it asks for more, so the N=1 digests
+moved; `--dump FILE` on the base tree and `--diff FILE` here list every
+changed cycle and whether the peak guard wanted more than the drain on it.
 """
 
 import functools
 import hashlib
 import os
+import pickle
 import sys
 import tempfile
 
@@ -47,14 +54,25 @@ def store_cheap(n):
     return s
 
 
+def load_owner(t):
+    """Below the 2500 W drain, above it, over drain + aim (5950 W), back."""
+    return (1200.0, 3000.0, 6500.0, 3000.0, -1500.0)[min(int(t // 480), 4)]
+
+
+OWNER = {"peak_cap_w": 3500, "peak_margin_w": 50,
+         "maintenance_discharge_w": 2500}
+
 SCENARIOS = {
     "n1_drain": dict(n=1),
     "n1_drain_modes": dict(n=1, modes=True),
+    "n1_drain_owner": dict(n=1, load=load_owner, extra=OWNER),
+    "n2_drain": dict(n=2),
+    "n2_drain_owner": dict(n=2, load=load_owner, extra=OWNER),
 }
 
 
-def run(n, *, modes=False, dur=DUR):
-    extra = dict(s79.P1, **DRAIN)
+def run(n, *, modes=False, dur=DUR, load=load, extra=None):
+    extra = dict(s79.P1, **DRAIN, **(extra or {}))
     if modes:
         extra.update(p3.ARB, battery_modes_enabled=True)
     seen, extra_rows = [], []
@@ -99,7 +117,65 @@ def digests() -> dict:
     return {k: digest(run(**v)[0]) for k, v in SCENARIOS.items()}
 
 
+FIELDS = ("t", "target", "commands", "loop reason", "integrator",
+          "saturation", "why", "allocation")
+XFIELDS = ("claims", "guard", "roles", "trade", "phase", "log length",
+           "newest log line")
+
+
+def wanted_more(row) -> bool:
+    """Did the peak guard ask for more than the drain rate on this cycle?"""
+    claims, guard = row[-1][0] or (), row[-1][1]
+    rate = [c[3] for c in claims if c[0] == "maintenance" and c[2] == "set"]
+    return (bool(rate) and guard is not None and bool(guard[0])
+            and guard[1] is not None and guard[1] > max(rate))
+
+
+def diff(base: dict, now: dict) -> dict:
+    """Per scenario: (cycles, changed, changed where the guard wanted more on
+    either tree, changed cycles whose commands differ without that)."""
+    out = {}
+    for k, rows in now.items():
+        old = base.get(k)
+        if old is None:
+            continue
+        ch = [i for i, (a, b) in enumerate(zip(old, rows)) if a != b]
+        ch += list(range(min(len(old), len(rows)), max(len(old), len(rows))))
+        more = [i for i in ch if i < min(len(old), len(rows))
+                and (wanted_more(old[i]) or wanted_more(rows[i]))]
+        cmd = [i for i in ch if i not in more
+               and (i >= min(len(old), len(rows))
+                    or old[i][1:3] != rows[i][1:3])]
+        fields = set()
+        for i in set(ch) - set(more) - set(cmd):
+            fields |= {FIELDS[j] for j in range(8) if old[i][j] != rows[i][j]}
+            fields |= {XFIELDS[j] for j in range(7)
+                       if old[i][8][j] != rows[i][8][j]}
+        out[k] = (len(rows), ch, more, cmd, sorted(fields))
+    return out
+
+
+def all_rows() -> dict:
+    return {k: run(**v)[0] for k, v in SCENARIOS.items()}
+
+
 if __name__ == "__main__":
+    if "--dump" in sys.argv:
+        with open(sys.argv[sys.argv.index("--dump") + 1], "wb") as f:
+            pickle.dump(all_rows(), f)
+        sys.exit(0)
+    if "--diff" in sys.argv:
+        with open(sys.argv[sys.argv.index("--diff") + 1], "rb") as f:
+            base = pickle.load(f)
+        for k, (n, ch, more, cmd, fields) in diff(base, all_rows()).items():
+            rest = sorted(set(ch) - set(more))
+            print(f"{k}: {n} cycles, {len(ch)} changed, {len(more)} with the "
+                  f"guard above the drain, {len(rest)} others "
+                  f"({len(cmd)} with a different target or command"
+                  + (f"; the rest differ only in {', '.join(fields)}"
+                     if fields else "") + ")"
+                  + (f", first changed t={ch[0] * rig.TICK:.0f} s" if ch else ""))
+        sys.exit(0)
     if "--summary" in sys.argv:
         for k, v in SCENARIOS.items():
             rows = run(**v)[0]

goodwe_controller/tools/safety84_identity.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""SAFETY-84: at one inverter, is a maintenance charge byte-identical to before?
+
+    python3 tools/safety84_identity.py          # prints one digest per scenario
+
+safety78_identity's harness (the REAL Controller on a P1 meter, every claim
+hashed) with the battery in its maintenance CHARGE phase from the first cycle,
+on the owner's settings (limit 3500, margin 50, charge 2000 W). The house steps
+below the charge rate, at it, under the aim, over it and into a solar surplus,
+in a cheap window, outside one, and under a month peak banked at 4000 W. Uses
+only the API release/1.0 9d5b7d1 has, so the same file runs on both trees;
+test_safety84.py holds this tree to the digests printed there (N=1) and to the
+new N=2 ones.
+"""
+
+import functools
+import hashlib
+import os
+import sys
+import tempfile
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import opt11_identity as rig                               # noqa: E402
+import safety79_identity as s79                            # noqa: E402
+import app.main as m                                       # noqa: E402
+from app import tariff                                     # noqa: E402
+
+DUR = 2400.0
+OPT = {"peak_cap_w": 3500, "peak_margin_w": 50, "maintenance_enabled": True,
+       "maintenance_start_hour": 11, "maintenance_charge_w": 2000,
+       "price_now_entity": "sensor.pnow", "price_avg_entity": "sensor.pavg"}
+
+
+def load(t):
+    """Below the 2000 W charge, at it, under the aim, over it, a surplus."""
+    return (800.0, 2000.0, 3000.0, 5000.0, -1500.0)[min(int(t // 480), 4)]
+
+
+def feed(cheap):
+    """s79's P1 feed plus a price now under (cheap) or over the average."""
+    now = 0.05 if cheap else 0.30
+
+    def before(c, t):
+        if not getattr(c.hass, "_s84", False):
+            read = c.hass.number
+
+            async def number(e, invert=False):
+                if e == "sensor.pnow":
+                    return now
+                if e == "sensor.pavg":
+                    return 0.20
+                return await read(e, invert)
+            c.hass.number, c.hass._s84 = number, True
+        s79._feed(c, t)
+    return before
+
+
+SCENARIOS = {
+    "n1_charge": dict(n=1),
+    "n1_charge_dear": dict(n=1, cheap=False),
+    "n1_charge_banked": dict(n=1, month_peak=4000.0),
+    "n2_charge": dict(n=2),
+    "n2_charge_dear": dict(n=2, cheap=False),
+    "n2_charge_banked": dict(n=2, month_peak=4000.0),
+}
+
+
+def run(n, *, cheap=True, month_peak=None, dur=DUR, load=load, soc=None,
+        before=None, cut=None, extra=None, store=None):
+    """Rows (opt11_identity's fields plus every claim, the guard and the
+    phases) and the controller. `month_peak` banks that peak this month."""
+    seen, extra_rows = [], []
+    real_resolve = m.resolve
+
+    def spy(claims, **kw):
+        seen.append(tuple((c.source, c.priority, c.kind, c.value, c.lo, c.hi,
+                           c.reason, c.actuator) for c in claims))
+        return real_resolve(claims, **kw)
+
+    def on_cycle(c, t):
+        g = c.peak_guard
+        extra_rows.append((
+            seen[-1] if seen else None,
+            None if g is None else (g.engaged, g.floor_w, g.reason),
+            c._maint_shape_suspended, [mm.phase for mm in c.maints],
+            len(c.events), c.events[0] if c.events else None))
+    db = os.path.join(tempfile.mkdtemp(prefix="s84-"), "tariff.db")
+    if month_peak is not None:
+        tariff.PeakStore(db)._db.execute(
+            "INSERT INTO monthly_peak VALUES (?, ?, ?, NULL)",
+            (tariff.month_key(rig.BASE), month_peak,
+             "2026-09-02T10:00:00+00:00")).connection.commit()
+    real = rig.Controller
+    rig.Controller = functools.partial(real, peak_db_path=db)
+    m.resolve = spy
+    try:
+        rows, c = rig.run(n, extra=dict(s79.P1, **OPT, **(extra or {})),
+                          dur=dur, load=load, soc=soc or s79.soc,
+                          cut=cut or (lambda u, t: False),
+                          before=before or feed(cheap), on_cycle=on_cycle,
+                          store=store or s79.store_charging(n))
+    finally:
+        rig.Controller = real
+        m.resolve = real_resolve
+    return [r + (x,) for r, x in zip(rows, extra_rows)], c
+
+
+def digest(rows) -> str:
+    return hashlib.sha256(repr(rows).encode()).hexdigest()
+
+
+def digests() -> dict:
+    return {k: digest(run(**v)[0]) for k, v in SCENARIOS.items()}
+
+
+if __name__ == "__main__":
+    for k, v in digests().items():
+        print(f"{k} {v}")