SAFETY-83: peak guard outranks the maintenance drain at N=1

At N=1 the drain is the site SET at P_MAINTENANCE (60), so the peak floor
(P_PEAK, 50) was not applied and the house imported what the drain left.
Where the floor asks for more than the drain rate, main.py now refiles the
peak claim at P_MAINTENANCE, so it wins; safety SETs still park everything.
Nothing else changes: safety78_identity --diff against 35cdffe changes only
cycles where the guard wanted more than the drain; N=2 is byte-identical.

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

goodwe_controller/test_safety83.py

@@ -0,0 +1,234 @@
+"""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, and without the fix the same runs go over.
+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
+
+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.arbiter import P_MAINTENANCE, P_PEAK                  # noqa: E402
+from app.main import Controller                                # noqa: E402
+from app.maintenance import CHARGE, DRAIN                      # 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, **extra):
+    """N=1 on a P1 meter; returns (rows, controller, banked quarters)."""
+    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")
+    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"]
+
+
+# ============================================================================ #
+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)
+
+# ============================================================================ #
+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))
+
+# ============================================================================ #
+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))
+_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 _bad 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 _bad)
+
+
+
+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")