Merge pull request 'SAFETY-83: peak guard outranks the maintenance drain at N=1' (#158) from safety-83-peak-over-drain into release/1.0

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

@@ -35,9 +35,8 @@ 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

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/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,8 @@ 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).
 
 #### Capacity-tariff exposure of the grid-following loop
 

goodwe_controller/app/arbiter.py

@@ -76,6 +76,10 @@ 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).
 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

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,25 @@ 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 _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 +3869,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
@@ -4254,6 +4272,7 @@ 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)
         # --- 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/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/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]