SAFETY-126: test reads through named_absence; points at MUTATION.md

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

.gitea/ci/run-tests.sh

@@ -42,12 +42,12 @@
 SERIAL="test_loop.py test_freeze.py test_safety87.py"
 #
 # LONGEST FIRST (SAFETY-125's CI run, job 2580; then a local parallel run, seconds:
-# modes_p4 758, safety125 508 on hotfix-0.19.2 only, modes_p3 364, safety119 328,
+# modes_p4 758, safety126 634 (SAFETY-126, local, one core), safety125 508 on hotfix-0.19.2 only, modes_p3 364, safety119 328,
 # price_run_fixes 184, safety90 160, safety103 134, safety94 114, peakguard 113,
 # opt20 102, safety97 101, safety81 100). With JOBS workers pulling from one
 # queue, starting the longest files first keeps the tail short: the batch takes
 # about as long as the longest file. A sort order, not a selection.
-HEAVY="test_modes_p4.py test_safety125.py test_modes_p3.py test_safety119.py test_price_run_fixes.py test_safety90.py test_safety103.py test_safety94.py test_peakguard.py test_opt20.py test_safety97.py test_safety81.py"
+HEAVY="test_modes_p4.py test_safety126.py test_safety125.py test_modes_p3.py test_safety119.py test_price_run_fixes.py test_safety90.py test_safety103.py test_safety94.py test_peakguard.py test_opt20.py test_safety97.py test_safety81.py"
 #
 JOBS=${CI_JOBS:-6}
 cd "${TESTS_DIR:-/w/goodwe-addon/goodwe_controller}"

goodwe_controller/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## 0.19.2
 
+### Not in a release yet
+
+- The night grid charge no longer raises "Battery stopped limiting the peak" when the 15-minute period stays on the aim ([[SAFETY-126]]).
+
 **A patch bump: one fix for charging from the grid under the peak limit, and no option changes.**
 No option was added, removed or renamed, and no default changed (86 options).
 **Existing installs need no option change:** options that start 0.19.1 also

goodwe_controller/DECISIONS/SAFETY-126-landing-error-is-not-an-overrun.md

@@ -0,0 +1,42 @@
+# SAFETY-126 - a landing error on the aim is not an overrun
+
+**Status: FIXED on release/1.0 (after 0.19.2).** Tests: `test_safety126.py`.
+
+**Cause.** A grid charge under the peak limit is sized to land the 15-minute
+period on the aim (SAFETY-125), so the period ends 0-7 W over it. In the last
+second `peakguard.excess_w()` multiplies that by 900 / remaining (3 000 at
+0.3 s left, 90 000 at 0.01 s), so `decide()` wants tens of kW
+(`wanted > 0`, peakguard.py `decide()`). With every battery under the peak
+reserve that is floor-limited at 0 W, and `check_peak_floor_limited()`
+alarmed on it at once: SAFETY-87 graded only the case where another battery
+could give. With one battery able, the guard asked it for its rating for that
+cycle and the Batteries "Why" said "The peak limit needs power". The
+add-on's own charge is not the cause: `wanted` already counts it as
+stoppable.
+
+**Rule.** Each cycle the guard wants power, `Controller._peak_landing()`
+asks `peakguard.misses_aim()`: with the batteries at 0 W from now on, does
+the period end more than `ALARM_MISS_W` (10 W) over the aim? This is
+`load_too_high()` with nothing to give, without `decide()`'s gain. If not, and
+no real miss was seen since the guard took hold, the decision is
+`within_aim`. Such a decision raises no reserve alarm (ERROR,
+notification, red row), no red "Load too high" row, no "Why" peak and
+no "Battery stopped limiting the peak" event hint. The claim, the floor and the
+commands are unchanged.
+
+**Still alarms.** House load the batteries cannot cover under the reserve:
+the period would end over the aim with the batteries at 0 W, so the alarm
+fires on the same cycle as before (both under the reserve; a 6 500 W house
+during the night charge). SAFETY-87's case (others able, graded on the limit
+for `ALARM_HOLD_S`) and SAFETY-81's 'Load too high' notification are unchanged.
+A period heading at most 10 W over the aim no longer alarms with nothing able
+to give. That is SAFETY-81's landing tolerance. `misses_aim()` takes the house
+as meter plus batteries, so an export nets off. `decide()`'s projection counts
+offtake only. So a re-alarm while the loop's draw on held batteries covers the
+house with some export, and the period lands under the aim, is gone too
+(test_safety126.py part 2, modes off: 2 notifications, was 4).
+
+**Not changed.** The one-cycle floor itself: at a period's last instant the
+guard still asks an able battery for its rating, or stops a charge, for a
+landing error. That is control, outside the alarm path. Raise a card if it
+should go.

goodwe_controller/DOCS.md

@@ -3458,7 +3458,7 @@ Charging from the grid stays blocked meanwhile. With several inverters the
 notification names the battery at its limit (SAFETY-32), each graded against
 its own cut-out level. That battery is held (weight 0 while the site gives)
 and the guard's need goes to the others, up to their rating; the alarm fires
-only if they miss the limit at full power for `ALARM_HOLD_S` (SAFETY-87). ⚠️ **The reserve binds the peak guard only
+only if they miss the limit at full power for `ALARM_HOLD_S` (SAFETY-87). With no battery able to give it fires at once, but only if the period would end more than 10 W over the aim with the batteries at 0 W. A grid charge the add-on runs counts as one it can stop, and the last second's landing error on the aim is not an alarm (SAFETY-126). ⚠️ **The reserve binds the peak guard only
 (SAFETY-25).** Self-consumption has no such floor. See *What `peak_soc_floor`
 does not protect*.
 

goodwe_controller/app/homeview.py

@@ -818,9 +818,11 @@ def refusing(c, inv) -> int:
 def peak_engaged(c) -> bool:
     """The peak guard asks the batteries for power this cycle. Not when it is
     blind or floor-limited: those are engaged with a 0 W floor (no charging
-    only), so no battery gives power for it."""
+    only), so no battery gives power for it. Nor when the quarter ends on
+    the aim anyway (SAFETY-126, `within_aim`)."""
     pg = getattr(c, "peak_guard", None)
-    return bool(pg is not None and pg.engaged and (pg.floor_w or 0.0) > 0.0)
+    return bool(pg is not None and pg.engaged and (pg.floor_w or 0.0) > 0.0
+                and not pg.within_aim)
 
 
 def battery_rows(c, shadow, running, held=False):

goodwe_controller/app/main.py

@@ -950,6 +950,9 @@ class Controller:
         # SAFETY-87: the reserve alarm's hold, as SAFETY-81's above.
         self._reserve_batt_prev = None
         self._reserve_high_since = None
+        # SAFETY-126: `_peak_landing()`'s last battery reading and latch.
+        self._land_batt_prev = None
+        self._peak_needed = False
         self._maint_shape_suspended = False
         # OPT-11. Battery modes, off by default so an install that has not
         # turned them on behaves exactly as before (SAFETY-10 section 1). Each
@@ -2989,6 +2992,27 @@ class Controller:
             soc=min(float(p.soc) for p in able),
             soc_floor=peakguard.binding_soc_floor(able, floor))
 
+    def _peak_landing(self, decision):
+        """SAFETY-126: `decision` marked `within_aim` while the guard wants
+        power but the batteries at 0 W would end the quarter within
+        `ALARM_MISS_W` of the aim (`peakguard.misses_aim()`). Once a real miss
+        is seen, the need stands until the guard lets go. The claim and the
+        commands are unchanged; only the alarms and the "Why" read it.
+        DECISIONS/SAFETY-126-landing-error-is-not-an-overrun.md."""
+        batt, prev = self.site_batt_w, self._land_batt_prev
+        self._land_batt_prev = batt
+        if not (decision.engaged and decision.blind is None
+                and (decision.wanted_w or 0.0) > 0.0):
+            self._peak_needed = False
+            return decision
+        if batt is not None and prev is not None:
+            batt = min(float(batt), float(prev))
+        self._peak_needed = self._peak_needed or (
+            self._peak_proj is not None and peakguard.misses_aim(
+                self._peak_proj, batt_w=batt, target_w=decision.target_w))
+        return (decision if self._peak_needed
+                else dataclasses.replace(decision, within_aim=True))
+
     def _peak_reserve(self) -> frozenset:
         """SAFETY-87: the batteries that give nothing for the peak this
         cycle: under their reserve while the others carry a peak floor."""
@@ -4845,6 +4869,7 @@ class Controller:
         if self.peak_guard is not None:
             self.peak_guard = self._peak_on_able(self.peak_guard,
                                                  maint_results)
+            self.peak_guard = self._peak_landing(self.peak_guard)
         # SITE, and peakguard.claim() says so on the object it builds. The
         # quantity is the 15-minute average OFFTAKE AT THE GRID CONNECTION,
         # which is what the capacity tariff bills; there is one quarter-hour
@@ -4869,6 +4894,7 @@ class Controller:
                 self.log_event(_pg.explain(), hint=messages.hint(
                     "peak_off" if not _pg.engaged else
                     "peak_blind" if _pg.blind else
+                    "peak_on" if _pg.within_aim else
                     "peak_reserve" if _pg.floor_limited else
                     "peak_saturated" if _pg.saturated else "peak_on"))
 
@@ -6166,6 +6192,9 @@ class Controller:
             limited = self._reserve_short(g)
         else:
             self._reserve_batt_prev = self._reserve_high_since = None
+            # SAFETY-126: nothing can give, so it alarms at once, but only
+            # when the quarter really misses the aim (`_peak_landing()`).
+            limited = limited and not g.within_aim
         was, self._peak_floor_limited = self._peak_floor_limited, limited
         if not limited or was:
             return
@@ -7043,12 +7072,13 @@ class Controller:
             if _pg is None:
                 out.append({"ok": True, "warn": True, "id": "peak_guard",
                             "text": messages.row_peak_guard_waiting()})
-            elif _pg.saturated:
+            elif _pg.saturated and not _pg.within_aim:
                 out.append({"ok": False, "warn": False, "id": "peak_guard",
                             "runs": True,
                             "text": messages.row_peak_saturated(_pg.ceiling_w)})
-            elif _pg.floor_limited and (not (_pg.floor_w or 0.0) > 0.0
-                                        or self._peak_floor_limited):
+            elif _pg.floor_limited and (
+                    not ((_pg.floor_w or 0.0) > 0.0 or _pg.within_aim)
+                    or self._peak_floor_limited):
                 # SAFETY-24. NOT ok, like saturation: in both states the money
                 # is going to be spent and a human is the only actuator left. A
                 # separate row from saturation because the two ask for opposite

goodwe_controller/app/peakguard.py

@@ -504,6 +504,11 @@ class PeakDecision:
     # when they are not asked). See `place_on_able()`.
     held: tuple = ()
     able_w: float | None = None
+    # SAFETY-126: the guard wants power, but with the batteries at 0 W the
+    # quarter still ends within ALARM_MISS_W of the aim (`misses_aim()`), as
+    # for a landing error in its last second. Set by main.py. No alarm and no
+    # "Why" peak while it holds; the floor and the claim are unchanged.
+    within_aim: bool = False
 
     def explain(self) -> str:
         """Idle: the status row's plain words (UX-08). Engaged: `reason`, the
@@ -997,6 +1002,18 @@ def load_too_high(projection, *, batt_w, ceiling_w, achievable_w) -> bool:
     return ends_w - float(ceiling_w) > ALARM_MISS_W
 
 
+def misses_aim(projection, *, batt_w, target_w) -> bool:
+    """SAFETY-126. With the batteries at 0 W from now on, does the quarter
+    end more than `ALARM_MISS_W` over the aim (`target_w`)? `decide()`'s
+    `wanted > 0` asks the same question with no margin and with its gain, so a
+    1 W landing error in a quarter's last second reads as tens of kW. A grid
+    charge the add-on runs counts as stoppable: it is in `batt_w`. The caller
+    passes the lower of two cycles' battery readings, as for
+    `load_too_high()`. DECISIONS/SAFETY-126-landing-error-is-not-an-overrun.md."""
+    return load_too_high(projection, batt_w=batt_w, ceiling_w=target_w,
+                         achievable_w=0.0)
+
+
 def charge_headroom_w(decision) -> float | None:
     """SAFETY-45. The hardest this SITE may charge without putting the billed
     quarter over the ceiling, in W, positive - or None when nothing measured it.

goodwe_controller/test_safety126.py

@@ -0,0 +1,430 @@
+"""Runnable check for SAFETY-126.  `python3 test_safety126.py`
+
+During the night grid charge, with both batteries under the peak reserve,
+the add-on logged ERROR PEAK_GUARD_FLOOR_LIMITED and raised "Battery stopped
+limiting the peak" at quarter ends, and the Batteries "Why" blinked "The peak
+limit needs power from the batteries.", while every quarter landed on the aim.
+The cause: the charge lands the quarter on the aim, a 1 W landing error in
+the quarter's last second becomes tens of kW through `excess_w()`'s
+900 / remaining, and the alarm fired on any `wanted > 0` when no battery can
+give. DECISIONS/SAFETY-126-landing-error-is-not-an-overrun.md.
+
+Drives the REAL Controller (read_inputs + cycle, the tariff task on a
+day/night profile, a P1 meter on a simulated clock, OPT-15's lag), as
+test_safety125.py does, with the cycle a fraction of a second off the
+quarter's edge as on the rig.
+
+1. The night grid charge under the reserve: no ERROR, no notification, no
+   red peak row, no "Why" peak, every quarter on the aim; the guard did want
+   power at the quarter's end (not vacuous); with the fix reverted the same
+   run alarms. The commands are the same with and without the fix.
+2. A real overrun under the reserve still alarms, at the same cycle as
+   before, and names both batteries.
+3. `misses_aim()` on its own.
+4. The page and the "Why" on a decision within the aim.
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import asyncio
+import dataclasses
+import io
+import logging
+import os
+import sys
+import tempfile
+import types
+from datetime import datetime, timedelta, timezone
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+sys.path.insert(0, os.path.join(HERE, "tools"))
+
+import opt11p3_replay as rp                                    # noqa: E402
+from named_absence import first, nth, present                  # noqa: E402
+import app.main as m                                           # noqa: E402
+import app.p1 as p1mod                                         # noqa: E402
+from app import homeview, messages, peakguard, tariff          # noqa: E402
+from app import price as _price, static_tariff as _st          # noqa: E402
+from app.main import Controller                                # noqa: E402
+from app.p1 import make_sample, split_signed                   # noqa: E402
+from app.store import unit_key                                 # noqa: E402
+from app.tariff import QUARTER_S, Projection                   # noqa: E402
+
+logging.disable(logging.NOTSET)
+logging.getLogger().setLevel(logging.CRITICAL)
+_ERRORS = io.StringIO()
+_h = logging.StreamHandler(_ERRORS)
+_h.setLevel(logging.ERROR)
+logging.getLogger("goodwe").addHandler(_h)
+logging.getLogger("goodwe").setLevel(logging.ERROR)
+logging.getLogger("goodwe").propagate = False
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+PEAK = 5413.17          # October's banked peak on the live site
+AIM_CAP = 3500.0 - 50.0
+AIM_PEAK = PEAK - 50.0
+TOL = 10.0              # W over the aim a quarter may land (SAFETY-81)
+L = timezone(timedelta(hours=1))
+NIGHT = datetime(2026, 10, 30, 22, 0, tzinfo=L).astimezone(timezone.utc)
+PROFILE = ('{"periods": ['
+           '{"name": "night", "days": "all", "from": "22:00", "to": "07:00",'
+           ' "eur_per_kwh": 0.2333},'
+           '{"name": "day", "days": "all", "from": "07:00", "to": "22:00",'
+           ' "eur_per_kwh": 0.2801}]}')
+OWNER = {"a": "leftover_solar", "b": "cheap_hours"}
+
+
+def opts(*, modes=True, arb=False):
+    o = dict(rp.options(modes, arb, 2))
+    o.update(cycle_cost_threshold_eur=0.03, static_profile=PROFILE,
+             peak_cap_w=3500, peak_margin_w=50)
+    o["inverters"][0].update(max_w=5000, capacity_kwh=15.36)
+    o["inverters"][1].update(max_w=4600, capacity_kwh=20.48)
+    return o
+
+
+class Harder(rp.Plant):
+    """Charges harder than asked, as inverter-a did live (test_safety125.py:
+    asked -4271 W, measured about -4750 W; asked -5000 W, about -5210 W), up
+    to `cap` W. So a charge sized to the aim lands a few W over it, as on the
+    rig (worst +7 W)."""
+    cap = -5216.0
+
+    def step(self, dt, floor):
+        c = self.cmd
+        if c < 0:
+            self.cmd = max(c * 1.11, self.cap)
+        try:
+            super().step(dt, floor)
+        finally:
+            self.cmd = c
+
+
+async def _no_fetch(*a, **k):
+    raise AssertionError("the tariff task reached for a price feed")
+
+
+def without_fix(fn):
+    """release/1.0 cc9bbeb: no decision is ever within the aim."""
+    real = getattr(Controller, "_peak_landing", None)
+    if real is None:
+        return fn()
+    Controller._peak_landing = lambda self, decision: decision
+    try:
+        return fn()
+    finally:
+        Controller._peak_landing = real
+
+
+def drive(o, *, modes=OWNER, socs=(10.0, 10.0), house=600.0, minutes=46,
+          start=NIGHT, off=0.7, peak=None):
+    """The real Controller from `start` + `off` s for `minutes` at a 1 s
+    cycle with OPT-15's lag, so a cycle lands `1 - off` s before each
+    quarter's end. `house` is W or a function of t. Returns a dict: the
+    quarters' true grid averages, the commands, the reserve notifications,
+    the ERROR lines, and per cycle where the guard engaged: (t, floor
+    limited, within the aim, battery "Why" peak, red reserve row, red
+    saturated row, alarm latched)."""
+    store = rp._Store()
+    for u, mode in (modes or {}).items():
+        store.data[unit_key(f"inverter-{u}", "pack_use")] = True
+        store.data[unit_key(f"inverter-{u}", "pack_mode")] = mode
+        store.data[unit_key(f"inverter-{u}", "pack_sell")] = False
+    load = house if callable(house) else (lambda t: float(house))
+    t0 = start + timedelta(seconds=off)
+    out = {"cmds": [], "notes": [], "seen": [], "sums": {}}
+    _ERRORS.seek(0)
+    _ERRORS.truncate()
+
+    async def go():
+        lag = rp.Lag()
+        hass = rp.Hass(("a", "b"), load, {"a": socs[0], "b": socs[1]},
+                       limits={"a": (-5000.0, 5000.0),
+                               "b": (-4600.0, 4600.0)}, lag=lag)
+        for u, cap in (("a", -5216.0), ("b", -4800.0)):
+            hass.plants[u].__class__ = Harder
+            hass.plants[u].cap = cap
+        real_set = hass.set_number
+
+        async def set_number(e, v):
+            out["cmds"].append((round(hass.t, 3), e, float(v)))
+            return await real_set(e, v)
+
+        async def call(*a):
+            if a[:2] == ("persistent_notification", "create") \
+                    and a[2]["notification_id"] == messages.ID_PEAK_RESERVE:
+                out["notes"].append((hass.t, a[2]["message"]))
+            return True
+        hass.set_number, hass.call = set_number, call
+        tmp = tempfile.mkdtemp(prefix="s126-")
+        db = os.path.join(tmp, "tariff.db")
+        if peak is not None:
+            tariff.PeakStore(db)._db.execute(
+                "INSERT INTO monthly_peak VALUES (?, ?, ?, NULL)",
+                (tariff.month_key(t0), peak,
+                 "2026-10-21T20:15:00+00:00")).connection.commit()
+        c = Controller(dict(o), hass, store, rp._Pub(), stop=asyncio.Event(),
+                       peak_db_path=db,
+                       price_history_path=os.path.join(tmp, "ph.json"))
+        c.inverters[0].dev_min, c.inverters[0].dev_max = -5000.0, 5000.0
+        c.inverters[1].dev_min, c.inverters[1].dev_max = -4600.0, 4600.0
+        c.auto = True
+        real = m.datetime, homeview.datetime, p1mod.time
+        clock, simt = [t0], [0.0]
+
+        class _DT(real[0]):
+            @classmethod
+            def now(cls, tz=None):
+                return clock[0] if tz is None else clock[0].astimezone(tz)
+        m.datetime = homeview.datetime = _DT
+        p1mod.time = types.SimpleNamespace(monotonic=lambda: 1e6 + simt[0])
+        real_load = _price.load_prices, _st.load_prices
+        _price.load_prices = _st.load_prices = _no_fetch
+        task = (asyncio.create_task(c.run_tariff_refresh())
+                if o.get("battery_modes_enabled") or o.get("arbitrage_enabled")
+                else None)
+        try:
+            hass.advance(0.0, 10.0)
+            t = 0.0
+            while t < minutes * 60:
+                now = t0 + timedelta(seconds=t)
+                hass.t, simt[0], clock[0] = t, t, now
+                if t == 0.0 and task is not None:
+                    for _ in range(500):
+                        if c.tariff_plan is not None:
+                            break
+                        await asyncio.sleep(0)
+                q = int(now.timestamp() // QUARTER_S)
+                acc = out["sums"].setdefault(q, [0.0, 0])
+                if t >= lag.p1_s:
+                    imp, exp = split_signed(hass._past(lag.meter_s
+                                                       + lag.p1_s)[0])
+                    c.p1.submit(make_sample(
+                        "ha_signed", imp, exp, phases=1,
+                        ingest_ts=now - timedelta(seconds=lag.p1_s),
+                        ingest_mono=1e6 + t))
+                await c.read_inputs(now_mono=t)
+                await c.cycle()
+                hass.advance(lag.tick, 10.0)
+                acc[0] += max(0.0, hass.grid())
+                acc[1] += 1
+                g = c.peak_guard
+                if g is not None and g.engaged and g.blind is None:
+                    rows = {r["id"]: r for r in c.checks(now)}
+                    row = rows.get("peak_guard", {})
+                    red = row.get("ok") is False and not row.get("warn")
+                    out["seen"].append((
+                        t, g.floor_limited, getattr(g, "within_aim", False),
+                        homeview.peak_engaged(c),
+                        red and "banner" in row,
+                        red and "banner" not in row,
+                        c._peak_floor_limited))
+                t += lag.tick
+        finally:
+            if task is not None:
+                task.cancel()
+                try:
+                    await task
+                except asyncio.CancelledError:
+                    pass
+            m.datetime, homeview.datetime, p1mod.time = real
+            _price.load_prices, _st.load_prices = real_load
+        return c
+    out["c"] = asyncio.run(go())
+    out["quarters"] = {
+        datetime.fromtimestamp(q * QUARTER_S, L).strftime("%H:%M"): s / k
+        for q, (s, k) in sorted(out["sums"].items()) if k >= QUARTER_S - 1}
+    out["errors"] = [x for x in _ERRORS.getvalue().splitlines()
+                     if "PEAK_GUARD_FLOOR_LIMITED" in x]
+    return out
+
+
+def show(tag, r):
+    print(f"    {tag}: quarters "
+          f"{ {k: round(v, 1) for k, v in r['quarters'].items()} }, "
+          f"guard cycles {len(r['seen'])}, floor-limited "
+          f"{sum(1 for s in r['seen'] if s[1])}, within-aim "
+          f"{sum(1 for s in r['seen'] if s[2])}, Why peak "
+          f"{sum(1 for s in r['seen'] if s[3])}, notes {len(r['notes'])}, "
+          f"errors {len(r['errors'])}")
+
+
+def quiet(r):
+    """No reserve alarm in any form: ERROR, notification, red row, latch,
+    "Why" peak, and no red saturated row."""
+    return (not r["errors"] and not r["notes"]
+            and not any(s[3] or s[4] or s[5] or s[6] for s in r["seen"]))
+
+
+def on_aim(r, aim):
+    q = r["quarters"]
+    return len(q) >= 3 and all(v <= aim + TOL for v in q.values())
+
+
+# ============================================================================ #
+print("1. the night grid charge under the peak reserve")
+NIGHTS = {}
+# The owner's shape at the cap's aim. At the banked peak's aim it is
+# test_safety125.py's live shape (A Cheap hours, B Solar first), because
+# B's 4 600 W alone stays under that aim.
+LIVE = {"a": "cheap_hours", "b": "solar_first"}
+for tag, kw, aim in (
+        ("aim 3450, cycle 0.3 s before the edge", dict(off=0.7), AIM_CAP),
+        ("aim 3450, cycle 0.01 s before the edge", dict(off=0.99), AIM_CAP),
+        ("month peak 5413, cycle 0.01 s before the edge",
+         dict(off=0.99, peak=PEAK, modes=LIVE, socs=(11.0, 10.0)),
+         AIM_PEAK)):
+    head = drive(opts(), **kw)
+    base = without_fix(lambda: drive(opts(), **kw))
+    NIGHTS[tag] = (head, base)
+    show("head " + tag, head)
+    show("fix reverted", base)
+    check(f"⭐ {tag}: no ERROR, no notification, no red peak row and no "
+          f"'Why' peak", quiet(head))
+    check(f"{tag}: every quarter at or under the aim ({aim:.0f} W + "
+          f"{TOL:.0f} W)", on_aim(head, aim))
+    check(f"{tag}: not vacuous, the guard wanted power under the reserve "
+          f"at a quarter's end (floor-limited, within the aim)",
+          any(s[1] and s[2] for s in head["seen"]))
+    check(f"{tag}: with the fix reverted the same run alarms",
+          not quiet(base))
+    check(f"{tag}: the commands are the same with and without the fix",
+          head["cmds"] == base["cmds"] and len(head["cmds"]) > 100)
+_h, _b = NIGHTS["aim 3450, cycle 0.01 s before the edge"]
+check("fix reverted, at 0.01 s: the ERROR, the notification and the "
+      "'Why' peak all come back (the rig's three symptoms)",
+      _b["errors"] and _b["notes"] and any(s[3] for s in _b["seen"]))
+
+
+# ============================================================================ #
+print("\n2. a real overrun under the reserve still alarms")
+
+
+def evening(t):
+    """600 W, then an oven and a car charger from 20 minutes on."""
+    return 600.0 if t < 1200 else 6500.0
+
+
+for tag, onset, kw in (
+        ("night grid charge, then 6500 W", 1200.0, dict(house=evening)),
+        ("modes off, 5500 W all along", 0.0,
+         dict(house=5500.0, modes=None, socs=(15.0, 12.0)))):
+    o = opts(modes=kw.get("modes", OWNER) is not None)
+    head = drive(o, minutes=35, **kw)
+    base = without_fix(lambda: drive(o, minutes=35, **kw))
+    show("head " + tag, head)
+    show("fix reverted", base)
+    # From the load's onset: before it, the fix reverted shows the night's
+    # false alarm (part 1).
+    first_h = next((s[0] for s in head["seen"] if s[6] and s[0] >= onset),
+                   None)
+    first_b = next((s[0] for s in base["seen"] if s[6] and s[0] >= onset),
+                   None)
+    check(f"⭐ {tag}: the ERROR and the notification, naming both "
+          f"batteries",
+          head["errors"] and head["notes"]
+          and all("inverter-a and inverter-b" in n for _t, n in head["notes"]))
+    _th = [t for t, _n in head["notes"]]
+    _tb = [t for t, _n in base["notes"] if t >= onset]
+    print(f"    notifications at {_th} s; with the fix reverted {_tb} s")
+    check(f"{tag}: the first at the same cycle as before the fix "
+          f"({first_h} s, {first_b} s), and none that was not raised before",
+          first_h is not None and first_h == first_b
+          and present(nth(_th, 0)) and nth(_th, 0) == nth(_tb, 0)
+          and set(_th) <= set(_tb))
+    check(f"{tag}: the red peak row and its banner while it stands",
+          any(s[4] for s in head["seen"]))
+
+
+# ============================================================================ #
+print("\n3. misses_aim() on its own")
+START = datetime(2026, 10, 30, 21, 15, tzinfo=timezone.utc)
+
+
+def proj(projected, elapsed, held):
+    return Projection(projected, projected, elapsed, 1.0, True, START, held)
+
+
+check("the rig's case: 3452 W against 3450 W, 0.3 s left, charging "
+      "1250 W from the grid: not a miss",
+      not peakguard.misses_aim(proj(3452.0, 899.7, 3452.0), batt_w=-1250.0,
+                               target_w=AIM_CAP))
+check("...although decide() wants power there (its gain, 900 / 0.3 s)",
+      (peakguard.decide(proj(3452.0, 899.7, 3452.0), batt_w=-1250.0,
+                        cap_w=3500, margin_w=50, reading=None,
+                        achievable_w=0.0, soc=10.0, soc_floor=20.0)
+       .wanted_w or 0.0) > 1000.0)
+check("a quarter heading 11 W over the aim with the batteries at 0 W is a "
+      "miss", peakguard.misses_aim(proj(3461.0, 899.7, 3461.0), batt_w=0.0,
+                                   target_w=AIM_CAP))
+check("...and 10 W is not (ALARM_MISS_W, as for 'Load too high')",
+      not peakguard.misses_aim(proj(3460.0, 899.7, 3460.0), batt_w=0.0,
+                               target_w=AIM_CAP))
+check("the add-on's own grid charge counts as stoppable: 4000 W on the "
+      "meter mid-quarter, 3000 W of it charging, is not a miss",
+      not peakguard.misses_aim(proj(2500.0, 450.0, 4000.0), batt_w=-3000.0,
+                               target_w=AIM_CAP))
+check("...and a house of 6000 W the batteries are covering is one",
+      peakguard.misses_aim(proj(1000.0, 450.0, 0.0), batt_w=6000.0,
+                           target_w=AIM_CAP))
+check("no battery reading, or no seconds left: not a miss (decide() files "
+      "nothing there)",
+      not peakguard.misses_aim(proj(9000.0, 450.0, 9000.0), batt_w=None,
+                               target_w=AIM_CAP)
+      and not peakguard.misses_aim(proj(9000.0, 900.0, 9000.0), batt_w=0.0,
+                                   target_w=AIM_CAP))
+
+
+# ============================================================================ #
+print("\n4. the page and the 'Why' on a decision within the aim")
+_c = NIGHTS["aim 3450, cycle 0.01 s before the edge"][0]["c"]
+_dec = peakguard.PeakDecision(
+    engaged=True, floor_w=5000.0, ceiling_w=3500.0,
+    ceiling_from=peakguard.CEILING_CAP, projected_w=3451.0, excess_w=9e4,
+    wanted_w=9e4, target_w=AIM_CAP, saturated=True, soc=30.0, soc_floor=20.0)
+
+
+def page(d):
+    """(the peak row is red, the "Why" says peak) for decision `d`."""
+    _c.peak_guard, _c._peak_floor_limited = d, False
+    row = first((r for r in _c.checks(datetime.now(timezone.utc))
+                 if r["id"] == "peak_guard"), "peak row")
+    if not present(row):
+        return {"red": None}, None
+    return dict(row, red=row["ok"] is False and not row.get("warn")), \
+        homeview.peak_engaged(_c)
+
+
+_r1, _w1 = page(_dec)
+_r2, _w2 = page(dataclasses.replace(_dec, within_aim=True))
+check("a saturated decision: red 'Load too high' row and the 'Why' peak, "
+      "as before", _r1["red"] and _w1)
+check("...within the aim: neither", _r2["red"] is False and _w2 is False)
+_lim = dataclasses.replace(_dec, floor_w=0.0, saturated=False,
+                         floor_limited=True, soc=10.0,
+                         soc_floor_reason=peakguard.FLOOR_AT_SOC_FLOOR)
+_r3, _w3 = page(_lim)
+_r4, _w4 = page(dataclasses.replace(_lim, within_aim=True))
+check("floor-limited at 0 W: the red reserve row with its banner, as "
+      "before", _r3["red"] and "banner" in _r3 and not _w3)
+check("...within the aim: no red row",
+      _r4["red"] is False and _w4 is False)
+
+
+# ============================================================================ #
+print()
+if fails:
+    print(f"{len(fails)} SAFETY-126 check(s) FAILED:")
+    for f in fails:
+        print(f"  - {f}")
+    sys.exit(1)
+print("all SAFETY-126 checks passed")