SAFETY-90: the peak guard goes on after a P1 meter gap

A meter gap past meter_max_age_s left seconds the quarter average never
counted, so the guard was blind for the rest of the quarter (rig case 7b:
3964.8 W against a 3450 W aim). Once readings are back, the missed seconds
now count at the higher of the limit and the import seen just before and
just after the gap with the batteries at 0 W (peakguard.gap_w/fill_gaps);
the gap's first meter_max_age_s is topped up to that too. Seconds before
the add-on's first reading count at the aim, and the page says part of the
period was not measured (EN/NL). The bill record is unchanged; no option.

DECISIONS/SAFETY-90-peak-guard-after-a-gap.md, test_safety90.py.

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

goodwe_controller/DECISIONS/SAFETY-90-peak-guard-after-a-gap.md

@@ -0,0 +1,73 @@
+# SAFETY-90 - the peak guard goes on after a P1 meter gap
+
+**Status: DEV 2026-10-05, for review.** Tests: `test_safety90.py`.
+
+**The defect.** A meter gap longer than `meter_max_age_s` leaves seconds the
+quarter average does not count (`p1.QuarterAverager`, on purpose: a guess must
+not reach the bill record). `peakguard.blindness()` then called the projection
+blind for the rest of the quarter, so the floor was 0 W. During a HOLD the
+battery gave nothing and the quarter went over: SAFETY-86 rig case 7b, N=2,
+3964.8 W against a 3450 W aim, the month peak raised; N=1, 3701 W (review).
+The page still read "Keeping the grid near zero". At start-up mid-quarter the
+same rule left the start-up quarter blind, and a held battery was drawn as last
+resort (SAFETY-91 test, O2).
+
+**The fix** (`peakguard.fill_gaps()`, `peakguard.gap_w()`,
+`Controller._fill_peak_gaps()`).
+- Once readings are back, every second of a silence longer than
+  `meter_max_age_s` counts at `gap_w = max(the limit, the import just before
+  the gap, the import just after it)`, where "import" is the grid plus the
+  batteries' discharge, i.e. with the batteries at 0 W. The gap's first
+  `meter_max_age_s`, which the averager credits at the last reading, is topped
+  up to `gap_w` too. The guard then grades the quarter as usual, so it is not
+  blind and gives what keeps the quarter at the aim.
+- **The bound.** The count is at or above the true import as long as, during
+  the gap, the house load stayed at or under the higher of the loads seen just
+  before and just after it, or under the limit, and the batteries did not
+  start charging (during a gap they hold their last command, then go to 0 W).
+  It never counts the gap below the limit. A load that rises above all three
+  only inside the gap is the one case it can miss.
+- **Cost of the bound.** It is conservative: with the house at 6500 W, a 30 s
+  to 5 min gap lands the quarter 100-180 W under the aim (table below), paid in
+  battery energy.
+- **While the gap lasts** nothing changes: the reading is too old to steer by,
+  the guard is blind (floor 0 W) and the "inputs missing" stop sets 0 W, as
+  before.
+- **Before the first reading** (start-up or restart mid-quarter) nothing is
+  known, not even the house load, so a conservative count has no basis: the
+  connection rating would make every restart drain at full power and raise
+  "Load too high". Those seconds count **at the aim**, so the guard keeps the
+  rest of the quarter on the aim, and the page says so: the status row "Peak
+  guard: part of this 15-minute period was not measured, counted as on
+  target" and the home line "Part of this 15-minute period was not measured." /
+  "Een deel van dit kwartier is niet gemeten." until the quarter ends.
+- The bill record (`quarter_blocks`) is unchanged: it still banks only what
+  was measured. The estimate steers the guard and is never banked.
+- The action log gets one `PEAK_GUARD_GAP` row per gap, with the seconds and
+  the watts it was counted at.
+- No option. The threshold is `meter_max_age_s`, the limit and the aim are
+  `peak_cap_w` and `peak_margin_w`.
+
+**Before -> after** (`test_safety90.py`, quarter as the meter really averages
+it, gap starting 300 s in; 10 s battery lag; aim 3450 W):
+
+| case | before | after |
+|---|---|---|
+| N=1 HOLD 6500 W, gap 30 s / 60 s | 3456.6 / 5329.2 | 3346.2 / 3272.2 |
+| N=1 HOLD 5500 W, gap 5 min | 4717.9 | 3330.6 |
+| N=2 HOLD, B able, 6500 W, gap 30 s / 60 s | 3456.6 / 3865.6 | 3346.3 / 3272.7 |
+| N=2 HOLD, B able, 5500 W, gap 5 min | 3745.4 | 3288.9 |
+| N=2 HOLD, B held, 5500 W, gap 60 s | 3254.9, B drawn 2500 W | 3335.4, B 0 W |
+| gap across the edge, next quarter, N=1 / N=2 | 4100.0 / 3621.2 | 3446.3 / 3382.2 |
+| house 3000 -> 6500 W inside a 60 s gap, N=1 | 5255.6 | 3368.3 |
+| house 6500 -> 2000 W late in a 5 min gap, N=1 | 3629.2 | 3145.1 |
+| start-up 450 s into the quarter, N=1 (first half at the aim) | 4975.0 | 3450.0 |
+| meter never drops, N=1 and N=2 | | rows identical |
+
+**Not changed.** The late-reading edge (SAFETY-91 test, O1): a reading late by
+less than `meter_max_age_s` is not a gap, and the seconds since it are not
+counted in the projection (on any quarter, not only after a gap). Near a
+quarter's end the 900/remaining gain turns that into a large negative excess
+and the floor drops for a cycle or more. Counting those seconds at the held
+reading would fix it, but it changes every quarter's last seconds on a
+healthy meter; it is handed back as its own card.

goodwe_controller/DOCS.md

@@ -1752,10 +1752,16 @@ ceiling at the cap, so a broken instrument makes the guard stricter, never loose
 **When it cannot see, it forbids importing.** Three states are not evidence and
 are checked before the projection is graded: no open quarter (a fresh averager
 projects a confident 0 W, which is under every ceiling there is), a meter gap
-longer than `meter_max_age_s`, and SAFETY-18's frozen-meter conjunction. In all
+longer than `meter_max_age_s` while it lasts, and SAFETY-18's frozen-meter
-three the claim becomes a floor of **0 W**. That is the backstop for the frozen
+conjunction. In all three the claim becomes a floor of **0 W**. That is the
-meter: the alarm arrives at `meter_freeze_s` (300 s) of a 900 s billing quarter
+backstop for the frozen meter: the alarm arrives at `meter_freeze_s` (300 s) of
-and commands nothing, and this stops the import.
+a 900 s billing quarter and commands nothing, and this stops the import.
+
+**After a gap it goes on (SAFETY-90).** The seconds the meter missed count at
+the higher of the peak limit and the import seen just before and just after
+the gap, with the batteries at 0 W. Seconds before the add-on's first reading
+count as on target, and the page says part of the period was not measured.
+DECISIONS/SAFETY-90-peak-guard-after-a-gap.md has the bound.
 
 **It trips, where the freeze detector only notifies,** and the captures are why.
 On the healthy reference day (2026-08-23) it engages on **zero** of 95 quarters.

goodwe_controller/app/homeview.py

@@ -904,7 +904,9 @@ def month_peak(c, now):
     if pg is not None:
         out["period_w"] = pg.projected_w
         ceiling = pg.ceiling_w or limit_w
-    out["line"] = line_period(out["period_w"], ceiling, c.shadow)
+    out["line"] = line_period(out["period_w"], ceiling, c.shadow,
+                              unmeasured=getattr(c, "peak_unmeasured_s", 0.0)
+                              > 0.0)
     return out
 
 
@@ -1294,8 +1296,11 @@ def line_peak_not_covered() -> str:
     return _t("peak_not_covered")
 
 
-def line_period(period_w, limit_w, shadow) -> str:
+def line_period(period_w, limit_w, shadow, unmeasured=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
     return _t("period", kw=_kw(period_w)) + " " + step
@@ -1369,6 +1374,7 @@ def _samples() -> list:
                 ("plan", line_peak_limit(3500, sh)),
                 ("plan", line_period(1240, 3500, sh)),
                 ("plan", line_period(None, 3500, sh)),
+                ("plan", line_period(1240, 3500, sh, unmeasured=True)),
                 ("plan", control_text(True, sh)),
                 ("plan", control_text(False, sh))]
     out += [("plan", run_grid_why(0.1412)), ("plan", run_grid_amount(1.23)),

goodwe_controller/app/main.py

@@ -947,6 +947,16 @@ class Controller:
         # SAFETY-81: what the alarm is graded on, beside the guard's decision.
         self._peak_proj = None
         self._peak_batt_prev = None
+        # SAFETY-90: what a meter gap is counted at (`_fill_peak_gaps()`).
+        self._gap_seen_w = None
+        self._gap_w = None
+        self._gap_block = None
+        self._gap_filled_s = 0.0
+        self._gap_held_ws = 0.0
+        self._gap_pre = None
+        self._unseen = None
+        self.peak_gap_s = 0.0
+        self.peak_unmeasured_s = 0.0
         self._peak_alarm_block = None
         self._peak_high_since = None
         self._peak_floor_limited = False
@@ -2693,6 +2703,58 @@ class Controller:
                                             lo=p, reason=c.reason,
                                             actuator=u)
 
+    def _fill_peak_gaps(self, proj, now, reading):
+        """SAFETY-90: `proj` with a meter gap's uncounted seconds counted at
+        `peakguard.gap_w()` of the imports seen just before and just after it,
+        and those before this run's first reading at the aim (shown as not
+        measured). DECISIONS/SAFETY-90-peak-guard-after-a-gap.md."""
+        self.peak_gap_s = self.peak_unmeasured_s = 0.0
+        avg = self.p1.averager
+        since = avg.held_since
+        if proj is None or proj.block_start is None or since is None:
+            return proj
+        if self._unseen is None and self.p1.last is not None:
+            self._unseen = (proj.block_start, max(0.0, (
+                since - proj.block_start).total_seconds() - avg.elapsed_s))
+        seen = (None if self.grid is None or self.site_batt_w is None
+                else float(self.grid) + max(0.0, float(self.site_batt_w)))
+        if proj.block_start != self._gap_block:
+            self._gap_block, self._gap_w = proj.block_start, None
+            self._gap_filled_s = self._gap_held_ws = 0.0
+        cap = float(self.o.get("peak_cap_w", 3500))
+        ceiling = peakguard.quarter_ceiling_w(cap, reading)[0]
+        kw = dict(tail_s=(now - since).total_seconds(),
+                  max_hold_s=avg.max_hold_s,
+                  unseen_s=(self._unseen[1] if self._unseen is not None
+                            and self._unseen[0] == proj.block_start else 0.0),
+                  unseen_w=peakguard.quarter_target_w(
+                      cap, self.peak_margin_w, reading))
+        _p, gap_s, _u = peakguard.fill_gaps(proj, gap_w=0.0, **kw)
+        if gap_s > self._gap_filled_s + peakguard.GAP_TOL_S:
+            self._gap_w = peakguard.gap_w(ceiling, self._gap_w,
+                                          self._gap_seen_w, seen)
+            if self._gap_pre is not None:
+                # The averager credited the gap's first max_hold_s at the
+                # reading before it; top that part up to gap_w as well.
+                s0, g0 = self._gap_pre
+                h = (min(s0 + timedelta(seconds=avg.max_hold_s), since)
+                     - max(s0, proj.block_start)).total_seconds()
+                self._gap_held_ws += max(0.0, h) * max(
+                    0.0, self._gap_w - max(0.0, g0))
+            self.log_event(f"PEAK_GUARD_GAP: the meter missed {gap_s:.0f} s "
+                           f"of this 15-minute period, counted at "
+                           f"{self._gap_w:.0f} W",
+                           hint=messages.hint("peak_gap"))
+        self._gap_filled_s = max(self._gap_filled_s, gap_s)
+        if seen is not None:
+            self._gap_seen_w = seen
+        if self.grid is not None:
+            self._gap_pre = (since, float(self.grid))
+        out, self.peak_gap_s, self.peak_unmeasured_s = peakguard.fill_gaps(
+            proj, gap_w=self._gap_w if self._gap_w is not None else ceiling,
+            held_ws=self._gap_held_ws, **kw)
+        return out
+
     def _peak_over_drain(self, claims, maint_results) -> None:
         """SAFETY-83: at N=1 a peak floor above the drain rate outranks it.
 
@@ -4218,6 +4280,10 @@ class Controller:
         # gives for SAFETY-18 on a transport that cannot see `unchanged_s`.
         self._peak_proj = (None if not self.p1_enabled
                            else self.peak.project(self.p1.averager, now))
+        _reading = self.peak.peak(now) if self.p1_enabled else None
+        # SAFETY-90: seconds a meter gap left uncounted are counted, so the
+        # guard goes on after the readings return.
+        self._peak_proj = self._fill_peak_gaps(self._peak_proj, now, _reading)
         # SAFETY-91: at N>=2 the quarter's edge is graded on the next quarter,
         # so a HOLD's ceiling does not lapse for a cycle onto a held battery.
         if self._peak_proj is not None and len(self.inverters) > 1:
@@ -4232,7 +4298,7 @@ class Controller:
             batt_w=self.site_batt_w,
             cap_w=float(self.o.get("peak_cap_w", 3500)),
             margin_w=self.peak_margin_w,
-            reading=self.peak.peak(now),
+            reading=_reading,
             # ⚠️ SUMMED FROM THE PER-UNIT CONJUNCTION, WHICH IS AC 4. Each
             # unit contributes `min(its max_w, |its own device bounds|)`, so
             # inverter B's rating bounds inverter B. Conjoining the SUMS instead
@@ -6390,6 +6456,11 @@ class Controller:
                 out.append({"ok": False, "warn": True, "id": "peak_guard",
                             "text": messages.row_peak_guard_blind(
                                 peakguard.blind_kind(_pg.blind))})
+            elif self.peak_unmeasured_s > 0.0:
+                # SAFETY-90: the start of this quarter came before the first
+                # reading, and the guard counts it at the aim.
+                out.append({"ok": False, "warn": True, "id": "peak_guard",
+                            "text": messages.row_peak_guard_unmeasured()})
             elif _pg.engaged:
                 out.append({"ok": False, "warn": True, "id": "peak_guard",
                             "text": messages.row_peak_guard_on(

goodwe_controller/app/messages.py

@@ -793,6 +793,12 @@ def row_peak_guard_blind(why: str) -> str:
             f"meanwhile")
 
 
+def row_peak_guard_unmeasured() -> str:
+    """SAFETY-90. Before the first reading; that part counts as on target."""
+    return ("Peak guard: part of this 15-minute period was not measured, "
+            "counted as on target")
+
+
 def row_peak_guard_on(floor_w, limit_w, *, shadow: bool = False) -> str:
     if shadow:
         return (f"Peak guard: would ask for {watts(floor_w)} to stay under "
@@ -912,6 +918,7 @@ HINTS = {
     "peak_on": "Peak guard started",
     "peak_off": "Peak guard idle",
     "peak_blind": "Peak guard cannot read the meter well",
+    "peak_gap": "Meter is back, the peak guard counts the missed time as high use",
     "peak_saturated": T_PEAK_SATURATED,
     "peak_reserve": T_PEAK_RESERVE,
     "meter_frozen": T_METER_FROZEN,
@@ -1097,6 +1104,7 @@ def status_samples() -> list:
         ("row_peak_guard_idle", GREEN, row_peak_guard_idle(1300, 2500, True)),
         ("row_peak_guard_idle", GREEN, row_peak_guard_idle(1300, 2500, False)),
         ("row_peak_guard_quiet", GREEN, row_peak_guard_quiet()),
+        ("row_peak_guard_unmeasured", AMBER, row_peak_guard_unmeasured()),
         *(("row_peak_guard_blind", AMBER, row_peak_guard_blind(k))
           for k in BLIND_WORDS),
         ("row_peak_guard_on", AMBER, row_peak_guard_on(1200, 2500)),

goodwe_controller/app/p1.py

@@ -312,6 +312,13 @@ class QuarterAverager:
         """
         return self._last_net
 
+    @property
+    def held_since(self) -> datetime | None:
+        """SAFETY-90: when the held sample arrived (its `ingest_ts`), or None."""
+        if self._last_t is None:
+            return None
+        return datetime.fromtimestamp(self._last_t, timezone.utc)
+
     @property
     def offtake_avg_w(self) -> float:
         """Average offtake over the part of the open block seen so far."""

goodwe_controller/app/peakguard.py

@@ -183,6 +183,8 @@ guard passes its own tests while being switched off:
     record). Those seconds make the projection read LOW, which is the
     under-guarding direction. The threshold is `max_hold_s` itself - the point
     p1.py already stops trusting a held sample - not a new number.
+    SAFETY-90: once readings are back, main.py counts those seconds
+    (`fill_gaps()`), so this holds only while the gap lasts.
 
   * FROZEN. SAFETY-18's conjunction, evaluated through `freeze.looks_frozen()`
     rather than re-written here. A frozen meter is the worst case of all: the
@@ -599,6 +601,57 @@ def blindness(projection, max_hold_s, frozen: bool) -> str | None:
     return None
 
 
+# SAFETY-90. Seconds of uncounted time below which nothing is filled: float
+# noise between the averager's sums and the wall clock, never a real gap.
+GAP_TOL_S = 1e-3
+
+
+def gap_w(ceiling_w, *seen_w) -> float:
+    """SAFETY-90: the watts a meter gap is counted at. The highest of the
+    limit and `seen_w`, the import with the batteries at 0 W (grid + their
+    discharge) just before and just after the gap; None entries are skipped.
+    DECISIONS/SAFETY-90-peak-guard-after-a-gap.md."""
+    return max([float(ceiling_w)] + [float(w) for w in seen_w if w is not None])
+
+
+def fill_gaps(projection, *, tail_s, max_hold_s, unseen_s, gap_w, unseen_w,
+              held_ws=0.0):
+    """SAFETY-90: `projection` with the seconds the averager did not count
+    (a meter gap past `max_hold_s`) counted at `gap_w`, and those from before
+    this run's first reading (`unseen_s`) at `unseen_w`. `held_ws` is added
+    too: what the gap's first `max_hold_s`, credited at the reading before
+    it, lacks to reach `gap_w`. Returns
+    `(projection, gap seconds filled, unseen seconds filled)`.
+
+    `tail_s` is the age of the held reading: up to `max_hold_s` of it is left
+    as before (the averager credits it once the next reading lands). Past
+    that the gap is still open, nothing is filled and `blindness()` answers
+    as before. DECISIONS/SAFETY-90-peak-guard-after-a-gap.md."""
+    if projection.block_start is None or max_hold_s is None:
+        return projection, 0.0, 0.0
+    wall = float(projection.elapsed_s)
+    tail = min(max(0.0, float(tail_s)), wall)
+    if tail > float(max_hold_s):
+        return projection, 0.0, 0.0
+    covered = wall - uncovered_s(projection)
+    missing = wall - covered - tail
+    if missing <= GAP_TOL_S:
+        missing = 0.0
+        if held_ws <= 0.0:
+            return projection, 0.0, 0.0
+    unseen = min(missing, max(0.0, float(unseen_s)))
+    gap = missing - unseen
+    add_ws = gap * float(gap_w) + unseen * float(unseen_w) + float(held_ws)
+    done = covered + missing
+    return (replace(
+        projection,
+        projected_w=float(projection.projected_w) + add_ws / QUARTER_S,
+        elapsed_w=((float(projection.elapsed_w) * covered + add_ws) / done
+                   if done > 0.0 else float(projection.elapsed_w)),
+        covered_frac=(done / wall if wall > 0.0
+                      else float(projection.covered_frac))), gap, unseen)
+
+
 def discharge_permitted(soc, soc_floor) -> str | None:
     """SAFETY-24. May the guard size a forced discharge, or why may it not?
 

goodwe_controller/app/ui_text/en.json

@@ -410,6 +410,7 @@
     "step_in": "The batteries step in above {kw}.",
     "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.",
     "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

@@ -347,6 +347,7 @@
     "step_in": "De batterijen springen bij boven {kw}.",
     "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.",
     "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/test_safety90.py

@@ -0,0 +1,423 @@
+"""Runnable check for SAFETY-90: the peak guard goes on after a P1 meter gap.
+`python3 test_safety90.py`
+
+A gap past `meter_max_age_s` left seconds the quarter average never counted,
+so the guard was blind for the rest of the quarter: a HOLD gave 0 W and the
+quarter raised the month peak (SAFETY-86 rig case 7b: 3964.8 W against a
+3450 W aim). Now the missed seconds count at `peakguard.gap_w()`, and seconds
+before the add-on's first reading count at the aim, with a plain line on the
+page. Drives the REAL Controller on a lagging battery (10 s), with the P1
+telegrams stopped for each gap. DECISIONS/SAFETY-90-peak-guard-after-a-gap.md.
+
+1. N=1 HOLD: gaps of 30 s, 60 s and 5 min mid-quarter land at or under the aim.
+2. N=2 HOLD, B able: the same.
+3. N=2 HOLD, B held at its reserve: the same, and B gives nothing (O2).
+4. A gap across a quarter edge; the house rising during a gap.
+5. Start-up mid-quarter: the guard runs, the page says what it assumed.
+6. While the gap lasts the guard is blind as before; no gap, no change.
+7. fill_gaps() and gap_w() at their boundaries.
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import logging
+import math
+import os
+import sys
+import types
+from datetime import timedelta
+
+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 safety84_identity as s84                                # noqa: E402
+import app.p1 as p1mod                                         # noqa: E402
+from app import homeview, messages, peakguard                  # noqa: E402
+from app.maintenance import HOLD                               # 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 = []
+A, B = "inverter-a", "inverter-b"
+AIM = 3450.0            # peak_cap_w 3500 - peak_margin_w 50
+TOL = 5.0               # W over the aim a quarter may land (SAFETY-79)
+CUT = 10.0
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+def store(phase_of, n):
+    s = rig._Store()
+    for u in (A, B)[:n]:
+        key = (lambda k: k) if n == 1 else (lambda k, u=u: unit_key(u, k))
+        if u in phase_of:
+            s.data[key("phase")] = phase_of[u]
+            s.data[key("phase_started")] = rig.BASE.isoformat()
+        else:
+            s.data[key("last_completed")] = rig.BASE.isoformat()
+    return s
+
+
+def run(house, socs, dur, *, n, gaps=(), start=0.0, lag=10.0, on_cycle=None):
+    """Rows and controller. A holds full; `socs` are the starting levels
+    (B's follows its power). `gaps`: (t0, t1), no telegram for t0 <= t < t1.
+    `start`: the add-on starts this many s into the run (mid-quarter). The
+    P1 clock is simulated, so a gap really goes stale. `c.truth` is the grid
+    each second, which the quarter really averages."""
+    clock = [0.0]
+    soc, last, truth = dict(socs), [start], []
+
+    def before(c, t0):
+        if on_cycle is not None:
+            on_cycle(c, t0)         # the state the last cycle left
+        t = t0 + start
+        clock[0] = t
+        g = c.hass.grid()
+        truth.append((t, g))
+        for u in soc:
+            soc[u] = max(CUT, soc[u] - c.hass.power(u) * (t - last[0])
+                         / 3600.0 / 153.6)
+        last[0] = t
+        if any(a <= t < b for a, b in gaps):
+            return
+        imp, exp = split_signed(g)
+        c.p1.submit(make_sample("ha_signed", imp, exp, phases=1,
+                                ingest_ts=rig.BASE + timedelta(seconds=t0),
+                                ingest_mono=1e6 + t))
+
+    def power(self, u):
+        st = self.__dict__.setdefault("_plant", {})
+        a, tgt, tl = st.get(u, (0.0, 0.0, self.t))
+        for _ in range(int(round((self.t - tl) / 0.1))):
+            d = self.cmd[u] - tgt
+            tgt += math.copysign(min(abs(d), 80.0), d)
+            a += (tgt - a) * (1 - math.exp(-0.1 / lag))
+        st[u] = (a, tgt, self.t)
+        return 0.0 if self.soc(u, self.t) <= CUT and a > 0 else float(round(a))
+
+    real = (rig._Hass.power, rig.TICK, rig.BASE, p1mod.time)
+    rig._Hass.power, rig.TICK = power, 1.0
+    rig.BASE = real[2] + timedelta(seconds=start)
+    p1mod.time = types.SimpleNamespace(monotonic=lambda: 1e6 + clock[0])
+    try:
+        rows, c = s84.run(
+            n, load=house if callable(house) else (lambda t: house),
+            cheap=False, store=store({A: HOLD}, n), before=before, dur=dur,
+            soc=lambda u, t: soc[u],
+            extra={"maintenance_hold_min": 600})
+    finally:
+        rig._Hass.power, rig.TICK, rig.BASE, p1mod.time = real
+    c.truth = truth
+    return rows, c
+
+
+def without_fix(fn):
+    """release/1.0 3ed3302: nothing is filled, so a gap blinds the guard."""
+    real = peakguard.fill_gaps
+    peakguard.fill_gaps = lambda p, **kw: (p, 0.0, 0.0)
+    try:
+        return fn()
+    finally:
+        peakguard.fill_gaps = real
+
+
+def quarter(c, q):
+    """What quarter `q` (0 = the run's first) really averages, in W."""
+    return sum(max(g, 0.0) for t, g in c.truth if int(t // QUARTER_S) == q) \
+        / QUARTER_S
+
+
+def cmd(r):
+    return dict(r[2])
+
+
+def mean(rows, u, lo, hi):
+    rs = [cmd(r)[u] for r in rows if lo <= r[0] < hi]
+    return round(sum(rs) / len(rs))
+
+
+TABLE = []
+
+
+def case(tag, n, house, socs, gaps, q=0, dur=1800.0):
+    """One gap case, after and before the fix; prints and records it."""
+    new = run(house, socs, dur, n=n, gaps=gaps)
+    old = without_fix(lambda: run(house, socs, dur, n=n, gaps=gaps))
+    a, b = quarter(new[1], q), quarter(old[1], q)
+    print(f"    {tag}: quarter {a:.1f} W (before the fix {b:.1f} W), aim "
+          f"{AIM:.0f}")
+    TABLE.append((tag, n, round(a, 1), round(b, 1)))
+    return new, old, a, b
+
+
+# ============================================================================ #
+print("1. N=1 HOLD, house 6500 W (5500 W for 5 min, so it can be held)")
+for tag, gap, house in (("30 s", (300, 330), 6500.0),
+                        ("60 s", (300, 360), 6500.0),
+                        ("5 min", (300, 600), 5500.0)):
+    (rows, c), _old, a, b = case(f"N=1 gap {tag}", 1, house, {"a": 99.0},
+                                 [gap])
+    check(f"⭐ N=1, {tag} gap mid-quarter: the quarter lands at or under the "
+          f"aim ({a:.1f} W)", a <= AIM + TOL)
+    if tag != "30 s":
+        check(f"...non-vacuous: before the fix it went over ({b:.1f} W)",
+              b > AIM + 100)
+        check("...and once readings are back the battery gives at least the "
+              "house over the aim",
+              mean(rows, "a", gap[1] + 15, 900) >= house - AIM)
+    check("...the action log says what the gap was counted at",
+          any("PEAK_GUARD_GAP" in e for e in c.events))
+
+# ============================================================================ #
+print("\n2. N=2 HOLD (A holding), B able at 60 %, house 6500 W")
+for tag, gap, house in (("30 s", (300, 330), 6500.0),
+                        ("60 s", (300, 360), 6500.0),
+                        ("5 min", (300, 600), 5500.0)):
+    (rows, c), _old, a, b = case(f"N=2 gap {tag}", 2, house,
+                                 {"a": 99.0, "b": 60.0}, [gap])
+    check(f"⭐ N=2, {tag} gap: the quarter lands at or under the aim "
+          f"({a:.1f} W)", a <= AIM + TOL)
+    if tag != "30 s":
+        check(f"...non-vacuous: before the fix it went over ({b:.1f} W)",
+              b > AIM + 100)
+    check("...B gives first: at its full 2500 W once readings are back",
+          mean(rows, "b", gap[1] + 30, 900) >= 2450)
+
+# ============================================================================ #
+print("\n3. N=2 HOLD, B held at 15 % (under its 20 % reserve), house 5500 W "
+      "(4500 W for 5 min: A alone gives 2500 W)")
+for tag, gap, house in (("30 s", (300, 330), 5500.0),
+                        ("60 s", (300, 360), 5500.0),
+                        ("5 min", (300, 600), 4500.0)):
+    (rows, c), (orows, _oc), a, b = case(
+        f"N=2 B held, gap {tag}", 2, house, {"a": 99.0, "b": 15.0}, [gap])
+    check(f"⭐ N=2 B held, {tag} gap: the quarter lands at or under the aim "
+          f"({a:.1f} W)", a <= AIM + TOL)
+    check("⭐ O2: the held B gives nothing once readings are back; the "
+          "holding A gives the need",
+          all(cmd(r)["b"] == 0.0 for r in rows if gap[1] <= r[0] < 900)
+          and mean(rows, "a", gap[1] + 15, 900) >= house - AIM)
+    if tag != "30 s":
+        check("...non-vacuous: before the fix the blind guard left A at 0 W "
+              "and the held B was drawn at 2500 W as last resort",
+              mean(orows, "a", gap[1] + 30, 900) == 0
+              and mean(orows, "b", gap[1] + 30, 900) >= 2450)
+
+rows, c = run(5500.0, {"a": 99.0, "b": 15.0}, 1000.0, n=2,
+              gaps=[(300, 360), (899, 910)])
+edge = [(r[0], cmd(r)["a"], cmd(r)["b"]) for r in rows if 900 <= r[0] < 915]
+print(f"    a 60 s gap, then the edge's telegram 11 s late: {edge[:4]}; the "
+      f"second before the edge (O1, not this card): "
+      f"{[(r[0], cmd(r)['a'], cmd(r)['b']) for r in rows if r[0] == 899]}")
+check("⭐ ...a quarter that had a gap still hands over at its edge (SAFETY-91): "
+      "with the edge's telegram late the held B gives nothing and A keeps "
+      "giving", all(b == 0.0 and a > 1500.0 for _t, a, b in edge))
+
+# ============================================================================ #
+print("\n4. a gap across a quarter edge, and a house that rises in a gap")
+for n, socs in ((1, {"a": 99.0}), (2, {"a": 99.0, "b": 60.0})):
+    (rows, c), _o, a, b = case(f"N={n} gap across the edge", n, 6500.0, socs,
+                               [(870, 930)], q=1)
+    check(f"⭐ N={n}: the quarter the readings come back in lands at or "
+          f"under the aim ({a:.1f} W; before the fix {b:.1f} W)",
+          a <= AIM + TOL and b > AIM + 100)
+    check("...the quarter that ended in the gap could not be steered after "
+          "it, and stays under the limit itself",
+          quarter(c, 0) <= 3500.0)
+
+
+def rising(t):
+    return 3000.0 if t < 320 else 6500.0
+
+
+for n, socs in ((1, {"a": 99.0}), (2, {"a": 99.0, "b": 15.0})):
+    house = rising if n == 1 else (lambda t: 3000.0 if t < 320 else 5500.0)
+    (rows, c), _o, a, b = case(f"N={n} house rises in a 60 s gap", n, house,
+                               socs, [(300, 360)])
+    check(f"⭐ N={n}: the house goes from 3000 W to over the limit during "
+          f"the gap; the import seen after it is counted and the quarter "
+          f"lands at or under the aim ({a:.1f} W)", a <= AIM + TOL)
+
+for n, socs in ((1, {"a": 99.0}), (2, {"a": 99.0, "b": 15.0})):
+    big = 6500.0 if n == 1 else 5500.0
+    (rows, c), _o, a, b = case(
+        f"N={n} house falls in a 5 min gap", n,
+        lambda t, big=big: big if t < 560 else 2000.0, socs, [(300, 600)])
+    check(f"⭐ N={n}: the house falls to 2000 W just before readings return; "
+          f"the import before the gap (batteries at 0 W) is counted and the "
+          f"quarter lands at or under the aim ({a:.1f} W; before {b:.1f} W, "
+          f"at N=2 by drawing the held B)",
+          a <= AIM + TOL and (n == 2 or b > AIM + 100))
+
+(rows, c), _o, a, b = case(
+    "N=1 5 min gap at 3000 W, then 60 s gap at 6500 W next quarter", 1,
+    lambda t: 3000.0 if t < 900 else 6500.0, {"a": 99.0},
+    [(300, 600), (1200, 1260)], q=1)
+check(f"⭐ ...each quarter's gap is counted on its own: a long gap at a low "
+      f"load does not hide a shorter one in the next quarter ({a:.1f} W)",
+      a <= AIM + TOL and sum("PEAK_GUARD_GAP" in e for e in c.events) == 2)
+
+# ============================================================================ #
+print("\n5. start-up mid-quarter (the add-on starts 450 s into the quarter)")
+UI = {}
+
+
+def grab(c, t):
+    if t == 60.0:
+        UI["unmeasured"] = c.peak_unmeasured_s
+        UI["rows"] = [r for r in c.checks() if r.get("id") == "peak_guard"]
+        UI["en"] = homeview.month_peak(c, c.cycle_at)["line"]
+        homeview._LANG[0] = "nl"
+        try:
+            UI["nl"] = homeview.month_peak(c, c.cycle_at)["line"]
+        finally:
+            homeview._LANG[0] = "en"
+    if t == 500.0:
+        UI["later"] = (c.peak_unmeasured_s,
+                       [r for r in c.checks() if r.get("id") == "peak_guard"])
+
+
+for n, socs, house in ((1, {"a": 99.0}, 6500.0),
+                       (2, {"a": 99.0, "b": 15.0}, 5500.0)):
+    rows, c = run(house, socs, 950.0, n=n, start=450.0,
+                  on_cycle=grab if n == 1 else None)
+    orows, oc = without_fix(lambda: run(house, socs, 950.0, n=n, start=450.0))
+    seen = sum(max(g, 0.0) for t, g in c.truth if t < 900) / QUARTER_S
+    old_seen = sum(max(g, 0.0) for t, g in oc.truth if t < 900) / QUARTER_S
+    land = seen + 450.0 * AIM / QUARTER_S
+    TABLE.append((f"N={n} start-up mid-quarter", n, round(land, 1),
+                  round(old_seen + 450.0 * AIM / QUARTER_S, 1)))
+    print(f"    N={n}: measured part {seen:.1f} W + 450 s at the aim = "
+          f"{land:.1f} W (before the fix {old_seen:.1f} W + the same)")
+    check(f"⭐ N={n}: with the first 450 s at the aim the quarter lands at or "
+          f"under it; the holding A gives the need from the first minute",
+          land <= AIM + TOL and mean(rows, "a", 30, 450) >= house - AIM)
+    if n == 2:
+        check("...and the held B gives nothing (O2 at start-up)",
+              all(cmd(r)["b"] == 0.0 for r in rows if 1 <= r[0] < 450))
+    check("...non-vacuous: before the fix A gave 0 W for the quarter",
+          mean(orows, "a", 30, 450) == 0)
+
+rows, c = run(6500.0, {"a": 99.0}, 1350.0, n=1, start=450.0,
+              gaps=[(1200, 1260)])
+a = quarter(c, 1)
+TABLE.append(("N=1 start-up, then 60 s gap next quarter", 1, round(a, 1),
+              round(quarter(without_fix(lambda: run(
+                  6500.0, {"a": 99.0}, 1350.0, n=1, start=450.0,
+                  gaps=[(1200, 1260)]))[1], 1), 1)))
+check(f"...the start-up assumption stays in its own quarter: a 60 s gap in "
+      f"the next one is counted as a gap and lands at or under the aim "
+      f"({a:.1f} W)", a <= AIM + TOL)
+
+print(f"    page: {UI.get('rows')}\n          EN {UI.get('en')!r}\n"
+      f"          NL {UI.get('nl')!r}")
+check("⭐ the page says plainly that part of the period was not measured, "
+      "in English and Dutch",
+      UI.get("unmeasured", 0) > 400
+      and UI["rows"][0]["text"] == messages.row_peak_guard_unmeasured()
+      and UI["en"].startswith("Part of this 15-minute period was not "
+                              "measured.")
+      and UI["nl"].startswith("Een deel van dit kwartier is niet gemeten."))
+check("...and stops saying it in the next quarter",
+      UI["later"][0] == 0.0 and UI["later"][1][0]["text"]
+      != messages.row_peak_guard_unmeasured())
+
+# ============================================================================ #
+print("\n6. while the gap lasts, and with no gap at all")
+LIVE = {}
+
+
+def live(c, t):
+    if t == 350.0:
+        LIVE["g"] = c.peak_guard
+
+
+run(6500.0, {"a": 99.0}, 400.0, n=1, gaps=[(300, 360)], on_cycle=live)
+check("⭐ 50 s into a gap the guard is blind as before (floor 0 W, no "
+      "charging): nothing is filled while the gap is open",
+      LIVE["g"].blind is not None and LIVE["g"].floor_w == 0.0)
+for n, socs, house in ((1, {"a": 99.0}, 6500.0),
+                       (2, {"a": 99.0, "b": 60.0}, 6500.0),
+                       (2, {"a": 99.0, "b": 15.0}, 5500.0)):
+    rows, c = run(house, socs, 1000.0, n=n)
+    orows, _ = without_fix(lambda: run(house, socs, 1000.0, n=n))
+    check(f"⭐ N={n}, meter never drops: every row is identical with and "
+          f"without the fix, and nothing is filled",
+          rows == orows and not any("PEAK_GUARD_GAP" in e for e in c.events))
+
+# ============================================================================ #
+print("\n7. fill_gaps() and gap_w() at their boundaries")
+T0 = rig.BASE
+
+
+def proj(wall, covered, held=1000.0, acc_w=1000.0):
+    return Projection(projected_w=(acc_w * covered + held * (QUARTER_S - wall))
+                      / QUARTER_S, elapsed_w=acc_w, elapsed_s=wall,
+                      covered_frac=covered / wall, complete=False,
+                      block_start=T0, held_w=held)
+
+
+def fill(p, tail, unseen=0.0, gw=4000.0, uw=3450.0):
+    return peakguard.fill_gaps(p, tail_s=tail, max_hold_s=30.0,
+                               unseen_s=unseen, gap_w=gw, unseen_w=uw)
+
+
+p = proj(400.0, 300.0)                        # 100 s uncounted
+f, g, u = fill(p, 10.0)
+check("90 s missed (100 s uncounted less a 10 s tail): counted at gap_w, "
+      "the tail left as before, no longer blind",
+      g == 90.0 and u == 0.0
+      and abs(f.projected_w - (p.projected_w + 90.0 * 4000.0 / QUARTER_S))
+      < 1e-9 and abs(peakguard.uncovered_s(f) - 10.0) < 1e-9
+      and peakguard.blindness(f, 30.0, False) is None
+      and peakguard.blindness(p, 30.0, False) is not None)
+check("tail exactly max_hold_s: filled; one past it: the gap is open, "
+      "nothing filled",
+      fill(p, 30.0)[1] == 70.0 and fill(p, 30.001) == (p, 0.0, 0.0))
+f, g, u = fill(p, 10.0, unseen=60.0)
+check("unseen seconds come first and count at the aim, the rest at gap_w",
+      u == 60.0 and g == 30.0
+      and abs(f.projected_w - (p.projected_w + (60 * 3450.0 + 30 * 4000.0)
+                               / QUARTER_S)) < 1e-9)
+check("nothing missed (or under 1 ms): the projection object is returned "
+      "unchanged",
+      fill(proj(400.0, 400.0), 0.0)[0] is not None
+      and fill(proj(400.0, 400.0), 0.0) == (proj(400.0, 400.0), 0.0, 0.0)
+      and fill(proj(400.0, 399.9995), 0.0)[1] == 0.0)
+f, g, u = fill(proj(400.0, 400.0), 0.0)
+f2 = peakguard.fill_gaps(proj(400.0, 400.0), tail_s=0.0, max_hold_s=30.0,
+                         unseen_s=0.0, gap_w=4000.0, unseen_w=3450.0,
+                         held_ws=9000.0)
+check("held_ws (the gap's first 30 s topped up) is added even when nothing "
+      "is missing any more",
+      g == 0.0 and abs(f2[0].projected_w - f.projected_w - 10.0) < 1e-9)
+z = peakguard.fill_gaps(Projection(0, 0, 0, 0, False, T0, 0), tail_s=0.0,
+                        max_hold_s=30.0, unseen_s=0.0, gap_w=4000.0,
+                        unseen_w=3450.0, held_ws=9000.0)
+check("the first instant of a block (0 s elapsed) with held_ws: no division "
+      "by zero", abs(z[0].projected_w - 10.0) < 1e-9)
+check("no open block: unchanged (blindness says so)",
+      fill(Projection(0, 0, 0, 0, False, None, 0), 0.0)[1:] == (0.0, 0.0))
+check("gap_w(): the highest of the limit and the imports seen, None skipped",
+      peakguard.gap_w(3500, None, 3000.0) == 3500.0
+      and peakguard.gap_w(3500, 6500.0, None) == 6500.0
+      and peakguard.gap_w(3500, 3000.0, 4500.0) == 4500.0
+      and peakguard.gap_w(3500) == 3500.0)
+
+# ============================================================================ #
+print("\n   case                              N   quarter W   before W   aim")
+for tag, n, a, b in TABLE:
+    print(f"   {tag:<34}{n}   {a:>9.1f}  {b:>9.1f}   {AIM:.0f}")
+print()
+if fails:
+    print(f"{len(fails)} FAILED")
+    sys.exit(1)
+print("all SAFETY-90 checks passed")