SAFETY-125: a grid charge counts a late meter reading's seconds; the guard row says "over" when it is

A meter reading late near a quarter's end left seconds the projection did
not count (SAFETY-99), and charge_headroom_w() read them as room: the cheap
run (and price trading, and the N=2 maintenance charge) went back to full
power in the quarter's last seconds and the quarter could land over a banked
month peak or the cap. peakguard.charge_room_w() counts those seconds at the
held reading for the charge only; the guard's floor is unchanged.

The guard row said "idle, heading for 5 814 W, under this month's 5 413 W
peak". Over the limit at the current rate the row and the month-peak line
now say so, and that the grid charge is cut back.

Version 0.19.2, CHANGELOG, DECISIONS. test_safety125.py new; the lag pins
in test_price_run_fixes.py and test_opt20.py and the AST pin in
test_arbitrage.py follow the change (see the pull request).

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

goodwe_controller/CHANGELOG.md

@@ -1,5 +1,32 @@
 # Changelog
 
+## 0.19.2
+
+**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
+start 0.19.2. The change below applies with one inverter or two.
+
+**Charging from the grid now stays under this month's peak.** When a meter reading came late near the end of a 15-minute period, a grid charge could go back to full power and push that period over the peak; now it is cut back. The peak guard line no longer says "under" when the period is over the limit at the current rate.
+
+Contents of this release, in merge order since 0.19.1 (`2932f3a`):
+
+| | |
+|---|---|
+| `SAFETY-125` | A grid charge (a battery mode's cheap run, price trading or the maintenance charge at two inverters) counts the seconds since a late meter reading at that reading, so it is cut back near the end of a 15-minute period instead of going back to full power. Over the limit at the current rate, the peak guard row and the month peak line say so, and that the grid charge is cut back ([[SAFETY-125]]). |
+
+### Known issues
+
+The list under `## 0.19.0` below still applies. The SAFETY-99 line there now
+covers the guard's discharge only, not a grid charge.
+
+### If you are crossing from a version older than 0.10.0
+
+`## 0.10.0` below can refuse an old config: `slew_w` above its limit, and the
+withdrawn `price_provider` ids `engie_dynamic`, `totalenergies_dynamic` and
+`eneco_eddy`. Read it before you update. Run `tools/predeploy_check.py` with the
+bare options (`jq '.data.options'`) first.
+
 ## 0.19.1
 
 **A patch bump: two fixes for battery modes, and no option changes.**

goodwe_controller/DECISIONS/SAFETY-125-grid-charge-under-the-month-peak.md

@@ -0,0 +1,9 @@
+# SAFETY-125 - a grid charge under the month peak
+
+**Status: FIXED in 0.19.2 (hotfix from 0.19.1).** Tests: `test_safety125.py`.
+
+A grid charge is sized by `peakguard.charge_room_w()`, the guard's headroom
+with the seconds since a late meter reading counted at that reading (the
+projection leaves them out, SAFETY-99, the bill does not), and the guard's own
+floor is unchanged. Evidence (repro on 0.19.1, sweep, digests): the SAFETY-125
+pull request.

goodwe_controller/app/homeview.py

@@ -963,7 +963,8 @@ def month_peak(c, now):
         ceiling = pg.ceiling_w or limit_w
     out["line"] = line_period(out["period_w"], ceiling, c.shadow,
                               unmeasured=getattr(c, "peak_unmeasured_s", 0.0)
-                              > 0.0)
+                              > 0.0,
+                              cut=getattr(c, "room_charge", False))
     return out
 
 
@@ -1367,13 +1368,19 @@ def line_peak_not_covered() -> str:
     return _t("peak_not_covered")
 
 
-def line_period(period_w, limit_w, shadow, unmeasured=False) -> str:
+def line_period(period_w, limit_w, shadow, unmeasured=False,
+                cut=False) -> str:
     step = _w(shadow, "step_in", kw=_kw(limit_w))
     if unmeasured:
         # SAFETY-90: the add-on started during this period.
         return _t("period_unmeasured") + " " + step
     if period_w is None:
         return step
+    if period_w > limit_w:
+        # SAFETY-125: the figure is the current rate, and a grid charge is
+        # what gets cut back.
+        return _t("period_over", kw=_kw(period_w)) + " " + (
+            _w(shadow, "charge_cut", kw=_kw(limit_w)) if cut else step)
     return _t("period", kw=_kw(period_w)) + " " + step
 
 

goodwe_controller/app/main.py

@@ -1217,6 +1217,8 @@ class Controller:
         # because `self.peak_guard` is a different cycle's decision by the time
         # the status page renders.
         self.arb_headroom_w = None
+        # SAFETY-125: a grid charge sized to the quarter's room this cycle.
+        self.room_charge = False
         # UX-13 P1: today's prices as the Home chart has seen them. Display
         # only; nothing reads it back into a decision.
         self.home_prices = homeview.DayPrices()
@@ -3239,8 +3241,10 @@ class Controller:
         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)
+        room = peakguard.charge_room_w(self.peak_guard, self._peak_proj)
+        floor = floor if room is None else min(floor, room)
+        self.room_charge = self.room_charge or floor > 0.0
+        return floor
 
     def _charge_ceiling(self, claims):
         """SAFETY-84: the site `hi` for a cheap-window charge at N>=2, or None.
@@ -4885,6 +4889,7 @@ class Controller:
         # decision presented as current.
         self.arb, self.arb_at = None, now
         self.arb_headroom_w = None
+        self.room_charge = False
         # OPT-11 P3: with battery modes on, each battery's mode decides who
         # trades (`_mode_roles()`, `_mode_trade()`, below, which read the
         # same headroom), and this site-wide trade is not run.
@@ -4895,7 +4900,9 @@ class Controller:
             # `arbitrage.sized_from_meter()` in `_shadow_settle()`, so the
             # caveat on the recorded watts can never be answered from a
             # different cycle's guard than the watts were.
-            self.arb_headroom_w = peakguard.charge_headroom_w(self.peak_guard)
+            # SAFETY-125: a late meter reading's seconds count, as the bill.
+            self.arb_headroom_w = peakguard.charge_room_w(self.peak_guard,
+                                                          self._peak_proj)
         if site_trade:
             floor_pct, ceiling_pct = arbitrage.soc_bounds(self.o)
             charge_w, discharge_w = self.trade_power_w(floor_pct, ceiling_pct)
@@ -4950,6 +4957,7 @@ class Controller:
             arb_claim = arbitrage.claim(self.arb)
             if arb_claim is not None:
                 claims.append(arb_claim)
+            self.room_charge = self.arb.action == arbitrage.CHARGE
             # Edge-logged, for the peak guard's reason: the action log holds 40
             # lines and a trade lasts a whole hour, so a line per cycle would
             # flush the context that explains it. The transition is what an
@@ -5169,6 +5177,8 @@ class Controller:
         mode_trade, trade_was = False, self._mode_trade_was
         if self.modes_on:
             mode_trade = self._mode_trade(claims, now)
+            self.room_charge = self.room_charge or (
+                mode_trade and self.arb.action == arbitrage.CHARGE)
         self._mode_trade_was = mode_trade
         self._charge_track(claims, maint_results)
         _live96 = set(self.live_units())
@@ -7062,6 +7072,15 @@ class Controller:
                             "text": messages.row_peak_guard_on(
                                 _pg.floor_w or 0.0, _pg.ceiling_w,
                                 shadow=self.shadow)})
+            elif _pg.projected_w is not None:
+                # SAFETY-125: "under" only when it is; over the limit at this
+                # rate, a grid charge is cut back to the quarter's room.
+                out.append({"ok": True, "warn": False, "id": "peak_guard",
+                            "text": messages.row_peak_guard_idle(
+                                _pg.projected_w, _pg.ceiling_w,
+                                month_peak=(_pg.ceiling_from
+                                            == peakguard.CEILING_MONTH_PEAK),
+                                cut=self.room_charge and not self.shadow)})
             else:
                 out.append({"ok": True, "warn": False, "id": "peak_guard",
                             "text": _pg.explain()})

goodwe_controller/app/messages.py

@@ -1170,12 +1170,22 @@ def row_periods_measured(n) -> str:
               f"Gemeten kwartieren deze maand: {getoond}")
 
 
-def row_peak_guard_idle(projected_w, limit_w, month_peak: bool) -> str:
-    """The projection first and the limit second (SAFETY-38's ordered pair)."""
+def row_peak_guard_idle(projected_w, limit_w, month_peak: bool,
+                        cut: bool = False) -> str:
+    """The projection first and the limit second (SAFETY-38's ordered pair).
+    SAFETY-125: over the limit at this rate it says so, and `cut` that a
+    grid charge is cut back to the room left."""
     limit = (f"this month's {watts(limit_w)} peak" if month_peak
              else f"the {watts(limit_w)} peak limit")
     grens = (f"de maandpiek van {watts(limit_w)}" if month_peak
              else f"de piekgrens van {watts(limit_w)}")
+    if float(projected_w) > float(limit_w):
+        doing, doet = (("cutting the grid charge", "schroeft het laden van "
+                        "het net terug") if cut else ("idle", "in rust"))
+        return Bi(f"Peak guard: {doing}, at this rate {watts(projected_w)}, "
+                  f"over {limit}",
+                  f"Piekbewaking: {doet}, aan dit tempo "
+                  f"{watts(projected_w)}, boven {grens}")
     return Bi(f"Peak guard: idle, heading for {watts(projected_w)}, under "
               f"{limit}",
               f"Piekbewaking: in rust, richting {watts(projected_w)}, onder "

goodwe_controller/app/peakguard.py

@@ -1136,6 +1136,21 @@ def charge_headroom_w(decision) -> float | None:
     return max(0.0, -float(decision.wanted_w))
 
 
+def charge_room_w(decision, projection) -> float | None:
+    """SAFETY-125: `charge_headroom_w()` with the seconds since the last
+    meter reading counted at that reading, so a late reading near a
+    quarter's end is not read as room for a grid charge (SAFETY-99).
+    DECISIONS/SAFETY-125-grid-charge-under-the-month-peak.md."""
+    room = charge_headroom_w(decision)
+    if not room or projection is None or projection.block_start is None:
+        return room
+    left = QUARTER_S - float(projection.elapsed_s)
+    if left <= 0.0:
+        return room
+    late_ws = max(0.0, float(projection.held_w)) * uncovered_s(projection)
+    return max(0.0, room - late_ws / left)
+
+
 def shaping_suspended(decision, peak_forecast_w, cap_w) -> bool:
     """Must the maintenance charge-only shaping stand down this cycle?
 

goodwe_controller/app/ui_text/en.json

@@ -416,6 +416,9 @@
     "step_in_would": "The batteries would step in above {kw}.",
     "period": "This 15-minute period is heading for {kw}.",
     "period_unmeasured": "Part of this 15-minute period was not measured.",
+    "period_over": "At this rate this 15-minute period reaches {kw}.",
+    "charge_cut": "The grid charge is cut back to stay under {kw}.",
+    "charge_cut_would": "The grid charge would be cut back to stay under {kw}.",
     "prices": {
       "now": "Price now {eur} per kWh, all in.",
       "today": "Today: lowest {lo} at {lo_at}, highest {hi} at {hi_at}.",

goodwe_controller/app/ui_text/nl.json

@@ -351,6 +351,9 @@
     "step_in_would": "De batterijen zouden bijspringen boven {kw}.",
     "period": "Dit kwartier gaat richting {kw}.",
     "period_unmeasured": "Een deel van dit kwartier is niet gemeten.",
+    "period_over": "Aan dit tempo komt dit kwartier op {kw}.",
+    "charge_cut": "Het laden van het net wordt teruggeschroefd om onder {kw} te blijven.",
+    "charge_cut_would": "Het laden van het net zou teruggeschroefd worden om onder {kw} te blijven.",
     "prices": {"now": "Prijs nu {eur} per kWh, alles inbegrepen.", "today": "Vandaag: laagste {lo} om {lo_at}, hoogste {hi} om {hi_at}.", "none": "Geen prijzen voor vandaag."},
     "tomorrow": {"known": "De prijzen van morgen zijn binnen.", "unknown": "De prijzen van morgen komen meestal na 13:00."},
     "region": {"preset": "Heffingen en btw: {region}, cijfers van {year}.", "own": "Heffingen en btw: uw eigen bedrag."},

goodwe_controller/config.yaml

@@ -1,5 +1,5 @@
 name: GoodWe RS485 Controller
-version: "0.19.1"
+version: "0.19.2"
 slug: goodwe_controller
 description: >-
   Drives a GoodWe ES/BP battery inverter over its RS485 EMS port by sending it a

goodwe_controller/test_arbitrage.py

@@ -936,6 +936,8 @@ def _s45_assigned(name, clear=("None",)):
 
 
 _s45_head_src = _s45_assigned("arb_headroom_w")
+# SAFETY-125: the guard's headroom with a late reading's seconds counted.
+S125_ROOM = "peakguard.charge_room_w(self.peak_guard, self._peak_proj)"
 print(f"    main.py has {len(_s45_calls)} `arbitrage.decide(` call site(s); "
       f"charge_headroom_w= is {_s45_arg}, and self.arb_headroom_w is assigned "
       f"from {_s45_head_src}")
@@ -944,7 +946,7 @@ check("main.py passes the guard's OWN measured headroom at its one "
       "transcription of it, so a mutation back to a literal or to max_w is red "
       "rather than silent",
       _s45_arg == ["self.arb_headroom_w"]
-      and _s45_head_src == ["peakguard.charge_headroom_w(self.peak_guard)"])
+      and _s45_head_src == [S125_ROOM])
 # ⚠️ AND THE HOIST DID NOT MOVE IT OUT OF THE CYCLE. An `arb_headroom_w` set
 # once in `__init__` and never refreshed would satisfy the pin above and would
 # size every trade from the guard decision of whatever cycle happened to run
@@ -960,8 +962,7 @@ check("...and the assignment is INSIDE the same `cycle()` that makes the call, "
       "and not one carried from an arbitrary earlier one",
       any(isinstance(n, ast.Assign)
           and [ast.unparse(t) for t in n.targets] == ["self.arb_headroom_w"]
-          and ast.unparse(n.value)
-          == "peakguard.charge_headroom_w(self.peak_guard)"
+          and ast.unparse(n.value) == S125_ROOM
           for n in ast.walk(_s45_cycle)))
 
 

goodwe_controller/test_opt20.py

@@ -243,12 +243,16 @@ check("no opposite-direction split on any of these cycles",
 # ============================================================================ #
 print("\n4. battery modes on, no probe: byte-identical to release/1.0 bf9019e")
 # Printed by tools/opt20_scen.py on release/1.0 bf9019e (2026-10-15),
-# goodwe-addon-checks:1.
+# goodwe-addon-checks:1. SAFETY-125 (0.19.2) moved the two runs that charge
+# against the peak aim: the rig's P1 sample is 0.35 s old, and a grid charge
+# now counts those seconds, so ch_ch's quarters land on 3450.0 W where they
+# landed 1.3-1.4 W over it (bf9019e: 15fc25dd...), and ch_ch_maint's first,
+# partial quarter charges a little less at its end (bf9019e: e46e3c66...).
 AT_BASE = {
     "sf_ch": "422d6578a7decf1ae3749edd1898ff3cd0541d904c95f636cdd9b1de8c1a07c9",
-    "ch_ch": "15fc25dd34186e56d34b752d37c9e8613295ba737938c92454acca08c83080fb",
+    "ch_ch": "2b5879ec95810c39a5f0f82c04b9f8e358ea03ddce4ba187cc2b7e0885dfcbfa",
     "ch_ch_maint":
-        "e46e3c666dd63ab33b56e83c0a560c07cec7774a9a70560f55c62a6ae6f55920",
+        "e56eddabdb785f81793e7afa525f8e6da237b0b2165860aa8e0f8a6499b275a9",
 }
 _got = sc.no_probe_digests()
 for _k in AT_BASE:
@@ -256,7 +260,8 @@ for _k in AT_BASE:
           f"{_got[_k][:16]}")
 check("with no refusal every share, grid, trade sentence, role and plant "
       "power of a scripted cheap run equals bf9019e's (A Solar first, A "
-      "charging, A in maintenance charge)", _got == AT_BASE)
+      "charging, A in maintenance charge; the last two as SAFETY-125 moved "
+      "them)", _got == AT_BASE)
 
 print()
 if fails:

goodwe_controller/test_price_run_fixes.py

@@ -609,6 +609,10 @@ print("\n5. modes off: byte-identical to release/1.0 e3e4ba6")
 #   AT_BASE      the same rows through `ident.trade_digest()`, equal to what
 #                e3e4ba6 gives through it;
 #   AT_NOW       this tree's own digests with nothing hidden.
+# SAFETY-125 (0.19.2) moved the two lag pins: their P1 sample is 0.35 s old on
+# every cycle, and a grid charge now counts those seconds at the held reading,
+# so 16 of 1080 cycles charge about 1.7 W less (before: n1_lag f6c67b35...
+# masked, 78aac892... full; n2_lag as n2). n1 and n2 still equal e3e4ba6.
 AT_BASE_OLD = {
     "n1": "71245b227ca40bada4f71f57ea6154475f4b0bd60239c9ad4f80bffb0094514c",
     "n2": "56cf28010b06765c29e5479abb815d7503d570b39240d82dfc818efa90c93722",
@@ -618,14 +622,14 @@ AT_BASE_OLD = {
 AT_BASE = {
     "n1": "33886ef03f364852883f0370230e6e04823fbb46a4704f88ca75bf9082d98c01",
     "n2": "666a31f41491c22750720524853d95944d5fba395d1d14040c460d71414b275f",
-    "n1_lag": "f6c67b351bd732fb69e62cafbfdb2bcff6032f5d8845836659d096b8c677d3c0",
-    "n2_lag": "666a31f41491c22750720524853d95944d5fba395d1d14040c460d71414b275f",
+    "n1_lag": "28e3182ae73c52dae0e65b2495bc04b2b71003889bf9de1a394b047d2c0e0edf",
+    "n2_lag": "45bfb095b5707a91316f56afc408dba616afc729ce4c83a10b337f36948fa3f2",
 }
 AT_NOW = {
     "n1": "939b3d47bbd64fa50e3d70538f092fb98e5c4bd5ead85a7fc696a1dbb514039e",
     "n2": "c51008213dd739e1e5142c236d9a950aa585992393e4af120086f1ba52ff3e0d",
-    "n1_lag": "78aac892975be81a7ceac0c000ed441a731bd4f42ff9f729996cbee53d6b7f02",
-    "n2_lag": "c51008213dd739e1e5142c236d9a950aa585992393e4af120086f1ba52ff3e0d",
+    "n1_lag": "1827918d570bfa0cc371c35dbbb64378b4937557061e7e1a396aba7663e68117",
+    "n2_lag": "6acc1d1b5a11f3481c7a30d384612f37b59f8e5e9cf89204edd07d09a5b90d8d",
 }
 _cmd_digest = ident.trade_digest
 _rows5 = {k: ident.run(**ident.SCENARIOS[k])[0] for k in AT_BASE}
@@ -636,7 +640,8 @@ for _k in AT_BASE:
           f"{_got[_k][:16]}")
 check("with battery modes off every command, integrator, allocation and "
       "trade action equals e3e4ba6's (the sentence masked as above), also when "
-      "every quarter boundary's cycle still sees the old P1 block",
+      "every quarter boundary's cycle still sees the old P1 block (there, as "
+      "SAFETY-125 moved it)",
       _got == AT_BASE)
 check("...the mask is needed and nothing more: unmasked, the digests differ "
       "from e3e4ba6's printed ones, and equal this tree's own pins, so a "

goodwe_controller/test_safety125.py

@@ -0,0 +1,409 @@
+"""Runnable check for SAFETY-125.  `python3 test_safety125.py`
+
+A meter reading that is late near a quarter's end left seconds the
+projection did not count (SAFETY-99), and `charge_headroom_w()` read them as
+room: a grid charge went back to full power in the quarter's last seconds and
+the billed quarter went over the aim, above a banked month peak or the cap.
+And the peak guard's row said "under" while the period was over the limit at
+the current rate. DECISIONS/SAFETY-125-grid-charge-under-the-month-peak.md.
+
+Drives the REAL Controller (read_inputs + cycle, the tariff task on a
+day/night profile, a P1 meter on a simulated clock) in the live shape: modes
+on, inverter-a Cheap hours at 5000 W charging harder than asked (as it did
+live), inverter-b Solar first, a house of 1 kW, cap 3500 W, margin 50 W.
+
+1. `charge_room_w()` on its own.
+2. A reading 30 s late at each quarter's end: every quarter at or under the
+   aim, with a month peak banked above the cap and with none, with the P1
+   projection and with a `peak_forecast_entity`; the charge is cut late in
+   the quarter and back in the next one.
+3. The same for a price trade with modes off, and a maintenance charge at N=2.
+4. A healthy meter: the quarters as before, and the row and the month-peak
+   line say what is true.
+5. The words on their own.
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import asyncio
+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"))
+logging.disable(logging.WARNING)
+
+import opt11_identity as rig                                   # noqa: E402
+import opt11p3_replay as rp                                    # noqa: E402
+import safety84_identity as s84                                # noqa: E402
+import app.main as m                                           # noqa: E402
+import app.p1 as p1mod                                         # noqa: E402
+from named_absence import first, present                       # 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
+
+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_PEAK = PEAK - 50.0
+AIM_CAP = 3500.0 - 50.0
+TOL = 10.0              # W over the aim a quarter may land (SAFETY-81)
+FC = "sensor.piek_prognose_kwartier"
+L = timezone(timedelta(hours=1))
+START = datetime(2026, 10, 29, 0, 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}]}')
+
+
+def opts(*, fc=False, 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)
+    if fc:
+        o["peak_forecast_entity"] = FC
+    return o
+
+
+class Harder(rp.Plant):
+    """Charges harder than asked, as inverter-a did live (asked -4271 W,
+    measured about -4750 W; asked -5000 W, about -5210 W)."""
+
+    def step(self, dt, floor):
+        c = self.cmd
+        if c < 0:
+            self.cmd = max(c * 1.11, -5216.0)
+        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 drive(o, *, peak=None, late=0.0, minutes=31, every=0):
+    """The real Controller from 00:00 local for `minutes` at a 1 s cycle
+    (OPT-15's lag). `late`: no P1 telegram in each quarter's last `late` s.
+    -> (controller, rows, {quarter start: true grid average}, samples), the
+    samples taken every `every` cycles from the second quarter (the first
+    starts before the first reading): (projection, ceiling, engaged,
+    peak_guard row, month-peak line)."""
+    store = rp._Store()
+    for u, mode in (("a", "cheap_hours"), ("b", "solar_first")):
+        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
+    rows, samples, sums = [], [], {}
+
+    async def go():
+        lag = rp.Lag()
+        hass = rp.Hass(("a", "b"), lambda t: 1000.0, {"a": 11.0, "b": 10.0},
+                       limits={"a": (-5000.0, 5000.0),
+                               "b": (-4600.0, 4600.0)}, lag=lag)
+        hass.plants["a"].__class__ = Harder
+        fc_kw = [0.0]
+
+        async def state(e):
+            # The HA quarter projection, in kW like the live template.
+            return None if e != FC else {
+                "state": f"{fc_kw[0]:.3f}",
+                "attributes": {"unit_of_measurement": "kW"}}
+        hass.state = state
+        tmp = tempfile.mkdtemp(prefix="s125-")
+        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(START), 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 = [START], [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())
+        try:
+            hass.advance(0.0, 10.0)
+            t, n = 0.0, 0
+            while t < minutes * 60:
+                now = START + timedelta(seconds=t)
+                hass.t, simt[0], clock[0] = t, t, now
+                if t == 0.0:
+                    for _ in range(500):
+                        if c.tariff_plan is not None:
+                            break
+                        await asyncio.sleep(0)
+                q = int(now.timestamp() // QUARTER_S)
+                acc = sums.setdefault(q, [0.0, 0])
+                g_now = hass.grid()
+                fc_kw[0] = (acc[0] + g_now * (QUARTER_S - acc[1])) \
+                    / QUARTER_S / 1000.0
+                if now.timestamp() % QUARTER_S < QUARTER_S - late \
+                        and 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
+                rows.append((now, c.target, hass.plants["a"].p,
+                             getattr(c.arb, "action", None)))
+                pg = c.peak_guard
+                if every and n % every == 0 and t >= QUARTER_S \
+                        and pg is not None and pg.projected_w is not None:
+                    row = first((r["text"] for r in c.checks(now)
+                                 if r["id"] == "peak_guard"), "guard row")
+                    samples.append((pg.projected_w, pg.ceiling_w, pg.engaged,
+                                    row, homeview.month_peak(c, now)["line"]))
+                t += lag.tick
+                n += 1
+        finally:
+            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
+    c = asyncio.run(go())
+    full = {datetime.fromtimestamp(q * QUARTER_S, L).strftime("%H:%M"):
+            s / k for q, (s, k) in sorted(sums.items()) if k >= QUARTER_S}
+    return c, rows, full, samples
+
+
+def into(now):
+    return now.timestamp() % QUARTER_S
+
+
+# ============================================================================ #
+print("1. charge_room_w(): a late reading's seconds count at that reading")
+_dec = peakguard.decide(
+    Projection(5800.0, 5800.0, 450.0, 1.0, True, START, 5800.0),
+    batt_w=-4800.0, cap_w=3500, margin_w=50,
+    reading=types.SimpleNamespace(is_observed=True, peak_w=PEAK),
+    achievable_w=9600.0, soc=50.0, soc_floor=20.0)
+_room = peakguard.charge_headroom_w(_dec)
+_late = Projection(5800.0, 5800.0, 450.0, 430.0 / 450.0, False, START,
+                   5800.0)
+print(f"    headroom {_room:.0f} W; 20 s late, 450 s left: "
+      f"{peakguard.charge_room_w(_dec, _late):.0f} W")
+check("with every second counted it is charge_headroom_w() exactly",
+      peakguard.charge_room_w(
+          _dec, Projection(5800.0, 5800.0, 450.0, 1.0, True, START, 5800.0))
+      == _room and _room > 0.0)
+check("20 s late at a 5800 W reading with 450 s left: 5800 x 20 / 450 W "
+      "less room",
+      abs(peakguard.charge_room_w(_dec, _late)
+          - (_room - 5800.0 * 20.0 / 450.0)) < 1e-6)
+check("never below 0 W, and an export held counts nothing",
+      peakguard.charge_room_w(_dec, Projection(
+          5800.0, 5800.0, 880.0, 850.0 / 880.0, False, START, 5800.0)) == 0.0
+      and peakguard.charge_room_w(_dec, Projection(
+          5800.0, 5800.0, 450.0, 430.0 / 450.0, False, START, -900.0))
+      == _room)
+check("None, blind (0 W) and no projection pass through",
+      peakguard.charge_room_w(None, _late) is None
+      and peakguard.charge_room_w(_dec, None) == _room
+      and peakguard.charge_room_w(
+          peakguard.PeakDecision(engaged=True, floor_w=0.0, ceiling_w=3500.0,
+                                 ceiling_from="cap", blind="frozen"),
+          _late) == 0.0)
+
+
+# ============================================================================ #
+def graded(label, full, aim):
+    print(f"    {label}: quarters "
+          f"{ {k: round(v, 1) for k, v in full.items()} } W, aim {aim:.0f}")
+    return len(full) >= 2 and all(v <= aim + TOL for v in full.values())
+
+
+def cut_and_back(rows):
+    """The charge is off in each quarter's last 5 s, and back (A charging
+    over 2 kW) within the next quarter's first minute."""
+    ends = [r for r in rows if into(r[0]) >= QUARTER_S - 5]
+    starts = {}
+    for r in rows:
+        if into(r[0]) < 60:
+            q = int(r[0].timestamp() // QUARTER_S)
+            starts[q] = starts.get(q, False) or r[2] < -2000.0
+    return (present(first(iter(ends), "a quarter end"))
+            and all(r[1] > -1.0 for r in ends)
+            and len(starts) >= 2 and all(starts.values()))
+
+
+print("\n2. a reading 30 s late at each quarter's end, the live shape")
+GAP = {}
+for name, kw, aim in (
+        ("month peak 5413 W banked, P1 projection",
+         dict(peak=PEAK), AIM_PEAK),
+        ("month peak banked, peak_forecast_entity (kW)",
+         dict(peak=PEAK, fc=True), AIM_PEAK),
+        ("no peak banked (aim = cap - margin), P1 projection",
+         dict(), AIM_CAP),
+        ("no peak banked, peak_forecast_entity", dict(fc=True), AIM_CAP)):
+    fc = kw.pop("fc", False)
+    _c, _rows, _full, _s = drive(opts(fc=fc), late=30.0, **kw)
+    GAP[name] = (_c, _rows, _full)
+    check(f"{name}: every quarter at or under the aim "
+          f"({aim:.0f} W, +{TOL:.0f} W landing)", graded(name, _full, aim))
+    check(f"{name}: the charge is off in the quarter's last seconds and "
+          f"back in the next quarter", cut_and_back(_rows))
+_c_fc = GAP["month peak banked, peak_forecast_entity (kW)"][0]
+check("the forecast entity is read, in kW, as the live template (SAFETY-118), "
+      "and the cheap run's charge does not depend on it",
+      _c_fc.peak_fc is not None and _c_fc.peak_fc > 1000.0
+      and [r[1] for r in GAP["month peak banked, peak_forecast_entity (kW)"]
+           [1]] == [r[1] for r in GAP[
+               "month peak 5413 W banked, P1 projection"][1]])
+
+
+# ============================================================================ #
+print("\n3. the same for a price trade (modes off) and a maintenance charge")
+_ca, _ra, _fa = drive(opts(modes=False, arb=True), peak=PEAK, late=30.0)[:3]
+check("modes off, price trading on, a reading 30 s late: every quarter at "
+      "or under the aim", graded("trade", _fa, AIM_PEAK)
+      and any(r[3] == "ARBITRAGE_CHARGE" for r in _ra))
+
+
+def late_feed(gap):
+    """safety84's cheap-window feed, no telegram in each quarter's last
+    `gap` s."""
+    feed = s84.feed(True)
+
+    def before(c, t):
+        if (rig.BASE.timestamp() + t) % QUARTER_S < QUARTER_S - gap \
+                or not getattr(c.hass, "_s84", False):
+            feed(c, t)
+    return before
+
+
+for peak, house, aim in ((5413.0, 4000.0, 5363.0), (None, 2500.0, AIM_CAP)):
+    _mr, _mc = s84.run(2, load=lambda t, h=house: h, dur=2700.0,
+                       month_peak=peak, before=late_feed(30.0))
+    _qs = {}
+    for r in _mr:
+        q = int((rig.BASE.timestamp() + r[0]) // QUARTER_S)
+        a = _qs.setdefault(q, [0.0, 0])
+        a[0] += max(0.0, house - sum(v for _u, v in r[2]))
+        a[1] += 1
+    _mf = {datetime.fromtimestamp(q * QUARTER_S, L).strftime("%H:%M"): s / k
+           for q, (s, k) in sorted(_qs.items())
+           if k * rig.TICK >= QUARTER_S}
+    check(f"N=2 maintenance charge, house {house:.0f} W, "
+          f"{'peak banked' if peak else 'no peak banked'}, a reading 30 s "
+          f"late: every quarter at or under the aim",
+          graded("maintenance", _mf, aim)
+          and any(mm.phase == "charge" for mm in _mc.maints))
+
+
+# ============================================================================ #
+print("\n4. a healthy meter: the quarters, and the words")
+for name, kw, aim, kw_text in (("month peak banked", dict(peak=PEAK),
+                                AIM_PEAK, "5.4 kW"),
+                               ("no peak banked", dict(), AIM_CAP, "3.5 kW")):
+    _c, _rows, _full, _s = drive(opts(), every=10, **kw)
+    check(f"{name}: every quarter at or under the aim",
+          graded(name, _full, aim))
+    over = [s for s in _s if not s[2] and s[0] > s[1]]
+    under = [s for s in _s if not s[2] and s[0] <= s[1]]
+    _o, _u = first(iter(over), "an over sample"), \
+        first(iter(under), "an under sample")
+    print(f"    over the limit at this rate ({len(over)} samples): "
+          f"{_o[3] if present(_o) else _o!s} / "
+          f"{_o[4] if present(_o) else _o!s}")
+    print(f"    under it ({len(under)}): "
+          f"{_u[3] if present(_u) else _u!s} / "
+          f"{_u[4] if present(_u) else _u!s}")
+    check(f"{name}: over the limit at this rate the guard row never says "
+          f"'under'; it says the grid charge is cut",
+          present(_o) and all(
+              "under" not in s[3] and "Peak guard: cutting the grid charge, "
+              "at this rate" in s[3] and " over " in s[3] for s in over))
+    check(f"{name}: ...and the month-peak line says the rate, and that the "
+          f"grid charge is cut back to stay under {kw_text}",
+          present(_o) and all(
+              s[4].startswith("At this rate this 15-minute period reaches")
+              and s[4].endswith("The grid charge is cut back to stay "
+                                f"under {kw_text}.") for s in over))
+    check(f"{name}: under the limit both read as before",
+          present(_u) and all(
+              s[3].startswith("Peak guard: idle, heading for")
+              and " under " in s[3]
+              and s[4].startswith("This 15-minute period is heading for")
+              and s[4].endswith(f"The batteries step in above {kw_text}.")
+              for s in under))
+
+
+# ============================================================================ #
+print("\n5. the words on their own")
+_en = messages.row_peak_guard_idle(5814.0, 5413.0, month_peak=True)
+_cut = messages.row_peak_guard_idle(5814.0, 5413.0, month_peak=True,
+                                    cut=True)
+_ok = messages.row_peak_guard_idle(5200.0, 5413.0, month_peak=True)
+print(f"    {_en}\n    {_cut}\n    {_ok}")
+check("over, nothing cut: 'idle, at this rate ..., over ...'",
+      _en == "Peak guard: idle, at this rate 5 814 W, over this month's "
+             "5 413 W peak")
+check("over, a grid charge cut",
+      _cut == "Peak guard: cutting the grid charge, at this rate 5 814 W, "
+              "over this month's 5 413 W peak")
+check("under: the 0.19.1 sentence, unchanged",
+      _ok == "Peak guard: idle, heading for 5 200 W, under this month's "
+             "5 413 W peak")
+check("the Dutch forms say the same",
+      _cut.nl == "Piekbewaking: schroeft het laden van het net terug, aan "
+                 "dit tempo 5 814 W, boven de maandpiek van 5 413 W"
+      and _en.nl.startswith("Piekbewaking: in rust, aan dit tempo"))
+check("the month-peak line in shadow mode says 'would'",
+      homeview.line_period(5814.0, 5413.0, True, cut=True).endswith(
+          "The grid charge would be cut back to stay under 5.4 kW.")
+      and homeview.line_period(5814.0, 5413.0, False).endswith(
+          "The batteries step in above 5.4 kW."))
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED:")
+    for f in fails:
+        print(f"  - {f}")
+    sys.exit(1)
+print("all checks passed")